"""Error logging endpoint — accepts frontend errors and logs them. No auth required so errors can be logged even during logout. Rate-limited to 10 requests per minute per IP (simple in-memory implementation). Context data is sanitized to prevent leaking sensitive information. """ from __future__ import annotations import time import logging import re from collections import defaultdict, deque from typing import Any from fastapi import APIRouter, Request, Response, status from pydantic import BaseModel, Field logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/errors", tags=["errors"]) # -- Simple in-memory rate limiter (10 req/min per IP) -- RATE_LIMIT = 10 # max requests RATE_WINDOW = 60 # seconds _ip_requests: dict[str, deque[float]] = defaultdict(deque) # -- Sensitive key patterns to strip from context -- _SENSITIVE_PATTERNS = re.compile( r"(?i)(token|password|secret|authorization|cookie|session|api[_-]?key|" r"access[_-]?token|refresh[_-]?token|csrf|bearer|private[_-]?key|" r"client[_-]?secret|x[_-]?auth|x[_-]?api[_-]?key)", ) def _is_rate_limited(client_ip: str) -> bool: """Return True if the IP has exceeded the rate limit.""" now = time.monotonic() dq = _ip_requests[client_ip] # Remove timestamps outside the window while dq and now - dq[0] > RATE_WINDOW: dq.popleft() if len(dq) >= RATE_LIMIT: return True dq.append(now) return False def _sanitize_context(context: Any, max_depth: int = 3, _depth: int = 0) -> Any: """Recursively remove sensitive keys and limit depth/size of context data.""" if _depth > max_depth: return "[truncated]" if isinstance(context, dict): sanitized = {} for key, value in context.items(): if _SENSITIVE_PATTERNS.search(str(key)): sanitized[key] = "[redacted]" else: sanitized[key] = _sanitize_context(value, max_depth, _depth + 1) return sanitized if isinstance(context, list): return [_sanitize_context(item, max_depth, _depth + 1) for item in context[:20]] if isinstance(context, str) and len(context) > 500: return context[:500] + "[truncated]" return context # -- Request schema -- class ErrorReport(BaseModel): timestamp: str | None = None message: str = Field(..., max_length=2000) stack: str | None = Field(None, max_length=10000) context: dict[str, Any] | None = None url: str | None = Field(None, max_length=500) userAgent: str | None = Field(None, max_length=500) @router.post("", status_code=status.HTTP_204_NO_CONTENT) async def report_error(error: ErrorReport, request: Request) -> Response: """Log a frontend error. No auth required. Rate-limited per IP.""" from app.core.rate_limit import get_client_ip client_ip = get_client_ip(request) if _is_rate_limited(client_ip): return Response(status_code=status.HTTP_429_TOO_MANY_REQUESTS) # Sanitize context to prevent leaking sensitive data sanitized_context = _sanitize_context(error.context) if error.context else None # Log with structured info logger.error( "Frontend error reported: %s", error.message, extra={ "error_timestamp": error.timestamp, "error_message": error.message, "error_stack": error.stack, "error_context": sanitized_context, "error_url": error.url, "error_user_agent": error.userAgent, "client_ip": client_ip, }, ) # If forgejo_error_reporter plugin is active, forward sanitized error try: from app.plugins.builtins.forgejo_error_reporter.service import report_error_to_forgejo entry = { "message": error.message, "stack": error.stack, "url": error.url, "userAgent": error.userAgent, "timestamp": error.timestamp, "context": sanitized_context, } await report_error_to_forgejo(entry) except Exception: pass # Plugin not active or error in reporting return Response(status_code=status.HTTP_204_NO_CONTENT)