File size: 3,278 Bytes
0125724
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
"""
TradeSense AI — Production Error Sanitizer & Information Leakage Defense (Module 08)
Prevents internal stack traces, database schema details, file paths, and secrets
from leaking in production API error responses.
"""

import uuid
import logging
from datetime import datetime, timezone
from typing import Dict, Any, Optional
from fastapi import Request, HTTPException, status
from fastapi.responses import JSONResponse

logger = logging.getLogger("tradesense.security.errors")


class ProductionErrorSanitizer:
    """
    Sanitizes unhandled exceptions and security errors into stable, opaque schemas.
    """

    @staticmethod
    def format_sanitized_error(
        status_code: int,
        error_code: str,
        message: str,
        request_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        return {
            "error_code": error_code,
            "message": message,
            "request_id": request_id or str(uuid.uuid4()),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }

    @classmethod
    def sanitize_exception(cls, exc: Exception, request_id: Optional[str] = None) -> Dict[str, Any]:
        req_id = request_id or str(uuid.uuid4())
        return {
            "error": {
                "code": "INTERNAL_SERVER_ERROR",
                "message": "An unexpected error occurred while processing your request. Please contact support with the request ID.",
                "request_id": req_id,
            }
        }

    @classmethod
    async def unhandled_exception_handler(cls, request: Request, exc: Exception) -> JSONResponse:
        request_id = str(uuid.uuid4())
        # Log full internal diagnostics securely on server
        logger.error(
            f"Unhandled exception [req_id={request_id}] at {request.method} {request.url.path}: {exc}",
            exc_info=True,
        )

        sanitized_payload = cls.format_sanitized_error(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            error_code="INTERNAL_SERVER_ERROR",
            message="An unexpected error occurred while processing your request. Please contact support with the request ID.",
            request_id=request_id,
        )
        return JSONResponse(status_code=500, content=sanitized_payload)

    @classmethod
    async def http_exception_handler(cls, request: Request, exc: HTTPException) -> JSONResponse:
        request_id = str(uuid.uuid4())
        # Preserve client-facing message but sanitize any internal substrings
        raw_detail = str(exc.detail) if exc.detail else "Operation failed."
        # Strip potential SQL or path leakage
        if "sqlite" in raw_detail.lower() or "postgresql" in raw_detail.lower() or "traceback" in raw_detail.lower():
            sanitized_detail = "Database error occurred. Please contact administrator."
        else:
            sanitized_detail = raw_detail

        sanitized_payload = cls.format_sanitized_error(
            status_code=exc.status_code,
            error_code=f"HTTP_{exc.status_code}",
            message=sanitized_detail,
            request_id=request_id,
        )
        headers = dict(exc.headers) if exc.headers else {}
        return JSONResponse(status_code=exc.status_code, content=sanitized_payload, headers=headers)