leocrm/app/routes/errors.py

"""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.

    Combines regex-based pattern matching with the central
    :mod:`app.core.sensitive_data` module to ensure entity-specific
    sensitive fields are also redacted.
    """
    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


def _sanitize_entity_context(context: dict[str, Any]) -> dict[str, Any]:
    """Sanitize context dict using both pattern matching and entity-aware redaction.

    If the context contains an ``entity_type`` key, uses :func:`sanitize_dict`
    from :mod:`app.core.sensitive_data` to redact entity-specific sensitive
    fields.  Falls back to pattern-based sanitization otherwise.
    """
    from app.core.sensitive_data import sanitize_dict

    entity_type = context.get("entity_type") or context.get("type")
    if entity_type:
        sanitized = sanitize_dict(context, str(entity_type))
    else:
        sanitized = dict(context)
    return _sanitize_context(sanitized)


# -- 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)
    user_agent: 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
    if error.context:
        sanitized_context = _sanitize_entity_context(error.context)
    else:
        sanitized_context = 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.contracts import get_contract
        reporter_contract = get_contract("forgejo_error_reporter")
        if reporter_contract is not None:
            entry = {
            "message": error.message,
            "stack": error.stack,
            "url": error.url,
            "userAgent": error.userAgent,
            "timestamp": error.timestamp,
            "context": sanitized_context,
        }
        await reporter_contract.report_error_to_forgejo(entry)
    except Exception:
        pass  # Plugin not active or error in reporting

    return Response(status_code=status.HTTP_204_NO_CONTENT)