simple-nanoshare/routes/handeling/errorsAndBots.py

from my_modules.app.constens import (
  BLOCKED_IPS_ACCESSING_TIMES,
  BLOCKED_IPS_STORED_TIMEFRAME,
)
from my_modules.app.setup import app, LIMITER
from my_modules.app.logger import logger

from quart import request, render_template, jsonify, current_app, make_response, g
from werkzeug.exceptions import HTTPException
from my_modules.functions import (
  get_ip,
  enforce_custom_limit,
  get_request_context,
)
from quart_common.web.env import is_development_environment
from quart_common.web.wide_event import add_httpx_error_wide_event_context, add_wide_event_context, httpx_error_wide_event_context, log_when_wide_event_disabled
import asyncio, os
import httpx

CONVEX_ERROR_TRACKING_TIMEOUT_SECS = float(os.getenv("CONVEX_ERROR_TRACKING_TIMEOUT_SECS", "120"))
DATABASE_RETRY_AFTER_SECS = int(os.getenv("DATABASE_RETRY_AFTER_SECS", "30"))

CONVEX_DOWN_ERROR_MESSAGES = (
  "convex runtime not running",
  "convex job timed out",
  "convex worker",
  "connection refused",
  "connect call failed",
  "all connection attempts failed",
  "name or service not known",
  "temporary failure in name resolution",
)

def _is_convex_unavailable_error(error: BaseException) -> bool:
  if isinstance(error, (asyncio.TimeoutError, httpx.RequestError)):
    return True

  error_text = str(error).casefold()
  return any(message in error_text for message in CONVEX_DOWN_ERROR_MESSAGES)

def _is_convex_timeout_error(error: BaseException) -> bool:
  if isinstance(error, (asyncio.TimeoutError, httpx.TimeoutException)):
    return True

  error_text = str(error).casefold()
  return "timeout" in error_text or "timed out" in error_text

def _convex_unavailable_status_code(error: BaseException) -> int:
  return 504 if _is_convex_timeout_error(error) else 503

async def _safe_convex_error_tracking(label: str, awaitable):
  try:
    return await asyncio.wait_for(awaitable, timeout=CONVEX_ERROR_TRACKING_TIMEOUT_SECS)
  except (asyncio.TimeoutError, asyncio.CancelledError) as error:
    add_wide_event_context(error_tracking={f"{label}_failed": True}, error={"type": type(error).__name__, "message": str(error)})
    await log_when_wide_event_disabled(logger, "warning", f"[ERROR_TRACKING] Convex {label} failed ({type(error).__name__}); rendering error response anyway")
    return None
  except Exception as error:
    add_wide_event_context(error_tracking={f"{label}_failed": True}, error={"type": type(error).__name__, "message": str(error)})
    await log_when_wide_event_disabled(logger, "error", f"[ERROR_TRACKING] Convex {label} failed: {error}")
    return None

IGNORED_404_PATHS = [
  "/.well-known/",
]

IGNORE_CONTAIN_404_PATHS = [
  "/.htaccess",
]

FEATURE_FLAG_DISABLED_PREFIX = "feature_flag_disabled:"

AUTH_STATUS_TITLES = {
  400: '400 - OAuth Time Paradox',
  401: '401 - Not Authenticated',
  403: '403 - Access Forbidden',
  500: '500 - Auth Reactor Meltdown',
  504: '504 - Auth Gateway Timeout',
}

AUTH_STATUS_MESSAGES = {
  400: (
    'The fox courier dropped your login form in a puddle before it reached the gatekeeper. '
    'Please send it again with all required fields so the checkpoint can read it.'
  ),
  401: (
    'The fox guards checked your badge and it did not pass the sniff test this round. '
    'Please sign in again so they can issue a fresh one.'
  ),
  403: (
    'The fox guards found your badge valid, but this den is still off-limits for your current role. '
    'If you should have access, ask an admin to update your permissions.'
  ),
  500: (
    'The auth engine coughed up a spark and the fox mechanics are tightening bolts right now. '
    'Please try again in a moment while they get the reactor stable.'
  ),
  504: (
    'The fox guards are still waiting for the auth mothership to answer the walkie-talkie. '
    'Please try again in a moment before they start howling at the server rack.'
  ),
}

async def auth_error(message:str, status_code:int=400):
  context = get_request_context()
  if status_code in AUTH_STATUS_MESSAGES:
    funny_message = AUTH_STATUS_MESSAGES[status_code]
  else:
    funny_message = (
      'The fox guards tripped over a cable while checking your badge. '
      'Authentication failed. Please try again or contact an administrator.'
    )

  add_wide_event_context(auth={"operation_status": "error"}, error={"type": "AuthenticationError", "message": message})
  await log_when_wide_event_disabled(logger, "error", f"[AUTH:{status_code}] {message}")

  if context and context.path.startswith("/api"):
    return jsonify({"error": "Authentication Error", "message": funny_message}), status_code

  return await render_template('views/basics/error.htm',
    title='Authentication Error',
    header={'title': AUTH_STATUS_TITLES.get(status_code, f'{status_code} - Authentication Error'), 'message': funny_message},
    file={'name': 'auth_error.webp', 'alt': 'A monitor flashes unauthorized access in blinking red warning text'},
  ), status_code

@app.errorhandler(401)
async def handle_unauthorized(e):
  add_wide_event_context(auth={"operation_status": "unauthorized"}, error={"type": type(e).__name__, "message": str(e)})
  context = get_request_context()
  if context.path.startswith("/api"):
    return jsonify({"error": "Unauthorized Access", "message": "Gandalf has spoken: You shall not pass… until you log in."}), 401

  await log_when_wide_event_disabled(logger, "error", e)
  return await render_template('views/basics/error.htm',
    title='Unauthorized Access',
    header={'title': '401 - Unauthorized', 'message': "Gandalf has spoken: You shall not pass… until you log in."},
    file={'name': '401.gif', 'alt': "Gandalf blocking the bridge – You shall not pass!"},
  ), 401

@app.errorhandler(404)
async def not_found(e):
  try:
    enforce_custom_limit(LIMITER, "404")
  except LookupError as e:
    return await to_many_requests(e)

  context = get_request_context()
  error_description = str(getattr(e, "description", ""))
  is_feature_flag_disabled_404 = error_description.startswith(FEATURE_FLAG_DISABLED_PREFIX)

  if (
    not is_development_environment()
    and not is_feature_flag_disabled_404
    and context.path not in IGNORED_404_PATHS
    and not any(p in context.path for p in IGNORE_CONTAIN_404_PATHS)
  ):
    await _safe_convex_error_tracking(
      "404_increment",
      current_app.convex.increment_page_not_found_error(path=context.path, status=404),
    )

  add_wide_event_context(error={"type": "NotFound", "message": str(e)})
  await log_when_wide_event_disabled(logger, "error", f"[404] Page Not Found: {context.path}")

  if context.path.startswith("/api"):
    return jsonify({"error": "Page Not Found", "message": "Oops! The page you are looking for does not exist."}), 404

  return await render_template('views/basics/error.htm',
    title='Page Not Found',
    header={'title': '404 - Page Not Found', 'message': "Oops! The page you are looking for does not exist."},
    file={'name': '404.webp', 'alt': "Matrix - Neo stoping the Bullets by holding his hand up"},
  ), 404

@app.errorhandler(418)
async def maybe_a_hacker(e=None):
  add_wide_event_context(security={"blocked": True, "block_reason": "honeypot"})
  try:
    enforce_custom_limit(LIMITER, "BotScan", BLOCKED_IPS_ACCESSING_TIMES, BLOCKED_IPS_STORED_TIMEFRAME)
  except LookupError as e:
    client_ip=get_ip()
    await _safe_convex_error_tracking(
      "honeypot_ip_increment",
      current_app.convex.increment_blocked_ip_address_access(
        ip_address=client_ip,
        method=request.method,
        path=request.path,
      ),
    )
    await log_when_wide_event_disabled(logger, "warning", f"[HONEYPOT] Blocked {client_ip} after accessing {request.path}")
    return await to_many_requests(e)

  rendered = await render_template('views/basics/error.htm',
    title='Oops! Something Went AWOL!',
    header={'title': "418 - I'm a Teapot", 'message': f"You don't say the Magic Word. By the way, we might have your IP now, but don’t worry, it's in safe hands (probably). Feel free to keep poking around, just maybe give us a sec to catch our breath."},
    file={'name': 'hacker_crap.webp', 'alt': "Someone got Hacked and he says I hate this Hacker crap - Jurassic Park Movie"},
  )

  response = await make_response((rendered, 418))
  response.headers['X-Honeypot-Triggered'] = 'true'
  response.headers['X-Reason'] = 'Unauthorized access attempt'

  return response

@app.errorhandler(429)
async def to_many_requests(e):
  add_wide_event_context(rate_limit={"limited": True}, error={"type": type(e).__name__, "message": str(e)})
  message = "We love your enthusiasm, but our server thought it was being DDoSed… by you. The keyboard needs a new set of keys and we need a nap. Try again soon!"

  context = get_request_context()
  if context.path.startswith("/api") or context.path.endswith('/auth/userinfo') or context.path.endswith('/auth/refresh'):
    return jsonify({"error": "Too Many Requests - YOU SHALL NOT PASS (for now)", "message": message}), 429

  return await render_template('views/basics/error.htm',
    title='Too Many Requests',
    header={'title': '429 - YOU SHALL NOT PASS (for now)', 'message': message},
    file={'name': '429_JimCarrey.gif', 'alt': "Jim Carrey Tips very fast on a computer keyboard"},
  ), 429

@app.errorhandler(405)
async def method_not_allowed(e):
  allowed_methods = getattr(e, "valid_methods", None) or getattr(e, "description", None)
  if not isinstance(allowed_methods, (list, tuple, set)):
    allowed_methods = getattr(e, "have_match_for", None) or []
  allowed_methods = sorted(str(method) for method in allowed_methods)
  allowed_methods_text = ", ".join(allowed_methods) if allowed_methods else "the supported method"
  message = f"Nice try, but {request.method} tried to enter this endpoint wearing fake glasses and a moustache. The bouncer only accepts {allowed_methods_text}."

  add_wide_event_context(error={"type": type(e).__name__, "message": str(e)}, http={"allowed_methods": allowed_methods})
  await log_when_wide_event_disabled(logger, "warning", f"[405] Method Not Allowed: {request.method} {request.path}; allowed={allowed_methods_text}")

  context = get_request_context()
  if context.path.startswith("/api"):
    response = await make_response(jsonify({"error": "Method Not Allowed", "message": message, "allowed_methods": allowed_methods}), 405)
  else:
    rendered = await render_template('views/basics/error.htm',
      title='Method Not Allowed',
      header={'title': '405 - Method Not Allowed', 'message': message},
      file={'name': '405_hal_9000_hal.gif', 'alt': "HAL 9000 calmly refuses with I'm Afraid i can't do that Dave"},
    )
    response = await make_response(rendered, 405)

  if allowed_methods:
    response.headers['Allow'] = ", ".join(allowed_methods)

  return response

@app.errorhandler(Exception)
async def handle_unexpected_exception(e):
  if isinstance(e, HTTPException):
    return e

  add_httpx_error_wide_event_context(e)

  if _is_convex_unavailable_error(e):
    status_code = _convex_unavailable_status_code(e)
    g.wide_event_handled_error_status = status_code
    add_wide_event_context(database={"provider": "convex", "available": False}, error={"type": type(e).__name__, "message": str(e)})
    await log_when_wide_event_disabled(logger, "error", f"[CONVEX_DOWN] Rendering database error response status={status_code} after Convex failure: {e}")
    return await database_server_error(e, status_code=status_code)

  raise e

@app.errorhandler(500)
async def internal_server_error(e):
  add_wide_event_context(**httpx_error_wide_event_context(e), error={"type": type(e).__name__, "message": str(e)})
  try:
    enforce_custom_limit(LIMITER, "500")
  except LookupError as e:
    return await to_many_requests(e)

  context = get_request_context()
  if context.path.startswith("/api"):
    return jsonify({"error": "Internal Server Error", "message": "It looks like you broke something... but don't worry, we're fixing it! In the meantime, we may or may not have logged your IP address (just kidding... or are we?). Either way, thanks for helping us find new ways to crash our system. Stay curious, hacker-friend!"}), 500

  await log_when_wide_event_disabled(logger, "error", e)
  return await render_template('views/basics/error.htm',
    title='Internal Server Error',
    header={'title': '500 - Internal Server Error', 'message': "It looks like you broke something... but don't worry, we're fixing it! In the meantime, we may or may not have logged your IP address (just kidding... or are we?). Either way, thanks for helping us find new ways to crash our system. Stay curious, hacker-friend!"},
    file={'name': '500.webp', 'alt': "Astronaut jumping and clicking on random Buttons as a red alert gone off - They is a Text on the Image saying: Why don't shit Work!?!"},
  ), 500

@app.errorhandler(503)
@app.errorhandler(504)
async def database_server_error(e, status_code:int|None=None):
  status_code = status_code or getattr(e, "code", None) or 504
  add_wide_event_context(**httpx_error_wide_event_context(e), error={"type": type(e).__name__, "message": str(e)})
  try:
    enforce_custom_limit(LIMITER, str(status_code))
  except LookupError as e:
    return await to_many_requests(e)

  retry_after = str(DATABASE_RETRY_AFTER_SECS)
  title = f'{status_code} - Database Error'
  message = "It looks like something is broke on our end... but don't worry, we're fixing it! Either way, thanks for helping us find new ways to crash our system. Stay curious, hacker-friend!"

  await log_when_wide_event_disabled(logger, "error", e)
  context = get_request_context()
  if context.path.startswith("/api"):
    response = await make_response(jsonify({"error": "Database Error", "message": message}), status_code)
  else:
    rendered = await render_template('views/basics/error.htm',
      title='Database Error',
      header={'title': title, 'message': message},
      file={'name': '504.gif', 'alt': "Hex Code running over a screen and ends with Error"},
    )
    response = await make_response(rendered, status_code)

  response.headers['Retry-After'] = retry_after
  return response