"""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 via central check_rate_limit(). Context data is sanitized to prevent leaking sensitive information. """ from __future__ import annotations import logging import re from typing import Any from fastapi import APIRouter, HTTPException, Request, Response, status from pydantic import BaseModel, Field logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/errors", tags=["errors"]) # -- Rate limit constants (10 req/min per IP) -- _RATE_LIMIT_MAX = 10 _RATE_LIMIT_WINDOW = 60 # seconds # -- 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 _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 check_rate_limit, get_client_ip client_ip = get_client_ip(request) try: await check_rate_limit( f"rate:errors:{client_ip}", _RATE_LIMIT_MAX, _RATE_LIMIT_WINDOW, ) except HTTPException: 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)