From 600030199db329a30b6293eef2124529131bf585 Mon Sep 17 00:00:00 2001 From: TAMDeveloper13 Date: Mon, 6 Apr 2026 13:03:05 +0000 Subject: [PATCH] Update: Sheet 7 Adjustment --- src/auth/service.py | 16 +- src/config.py | 1 + src/exceptions.py | 488 ++++++++++++++++++++++++++++---------------- src/logging.py | 165 ++++++++------- src/main.py | 158 +++++++++----- src/middleware.py | 95 ++++----- 6 files changed, 564 insertions(+), 359 deletions(-) diff --git a/src/auth/service.py b/src/auth/service.py index f69f078..dc68909 100644 --- a/src/auth/service.py +++ b/src/auth/service.py @@ -15,7 +15,11 @@ from .util import extract_template class JWTBearer(HTTPBearer): def __init__(self, auto_error: bool = True): - super(JWTBearer, self).__init__(auto_error=auto_error) + # Always pass auto_error=False so that HTTPBearer returns None (instead of + # raising HTTP 403) when the Authorization header is absent. We raise 401 + # ourselves in the else-branch below, which is the correct status for a + # missing / unauthenticated credential. + super(JWTBearer, self).__init__(auto_error=False) async def __call__(self, request: Request): credentials: HTTPAuthorizationCredentials = await super( @@ -24,7 +28,7 @@ class JWTBearer(HTTPBearer): if credentials: if not credentials.scheme == "Bearer": raise HTTPException( - status_code=403, detail="Invalid authentication scheme." + status_code=401, detail="Invalid authentication scheme." ) method = request.method @@ -40,7 +44,7 @@ class JWTBearer(HTTPBearer): if not user_info: message = message.get("message", "Invalid token or expired token.") raise HTTPException( - status_code=403, detail=message + status_code=401, detail=message ) request.state.user = message @@ -57,7 +61,7 @@ class JWTBearer(HTTPBearer): return message else: - raise HTTPException(status_code=403, detail="Invalid authorization code.") + raise HTTPException(status_code=401, detail="Invalid authorization code.") def verify_jwt(self, jwtoken: str, method: str, endpoint: str): try: @@ -100,7 +104,7 @@ async def internal_key(request: Request): if api_key != config.API_KEY: raise HTTPException( - status_code=403, detail="Invalid Key." + status_code=401, detail="Invalid Key." ) try: @@ -136,7 +140,7 @@ async def internal_key(request: Request): if not response.ok: raise HTTPException( - status_code=403, detail="Invalid token." + status_code=401, detail="Invalid token." ) return token.split(" ")[1] diff --git a/src/config.py b/src/config.py index e75959e..933332c 100644 --- a/src/config.py +++ b/src/config.py @@ -46,6 +46,7 @@ config = get_config() LOG_LEVEL = config("LOG_LEVEL", default="INFO") +SERVICE_NAME = config("SERVICE_NAME", default="Be-OptimumOH") ENV = config("ENV") PORT = config("PORT", cast=int) HOST = config("HOST") diff --git a/src/exceptions.py b/src/exceptions.py index c99b8c8..8ac9e5a 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -1,25 +1,94 @@ -# Define base error model +# Define base error model and exception hierarchy import logging +import re +import uuid from typing import Any, Dict, List, Optional from asyncpg.exceptions import DataError as AsyncPGDataError -from asyncpg.exceptions import PostgresError -from fastapi import FastAPI, HTTPException, Request +from fastapi import Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from pydantic import BaseModel -from slowapi import _rate_limit_exceeded_handler from slowapi.errors import RateLimitExceeded -from sqlalchemy.exc import (DataError, DBAPIError, IntegrityError, - SQLAlchemyError) +from sqlalchemy.exc import DataError, DBAPIError, IntegrityError, SQLAlchemyError, NoResultFound +from starlette.exceptions import HTTPException as StarletteHTTPException from src.enums import ResponseStatus -from src.auth.service import notify_admin_on_rate_limit_sync - -from starlette.exceptions import HTTPException as StarletteHTTPException log = logging.getLogger(__name__) + +# ── Custom exception classes ────────────────────────────────────────────────── + +class OptimumOHBaseException(Exception): + def __init__(self, message: str, data: Any = None): + self.message = message + self.data = data + super().__init__(message) + + +class BadRequestError(OptimumOHBaseException): pass +class UnauthorizedError(OptimumOHBaseException): pass +class ForbiddenError(OptimumOHBaseException): pass +class NotFoundError(OptimumOHBaseException): pass +class RequestTimeoutError(OptimumOHBaseException): pass +class UnprocessableEntityError(OptimumOHBaseException): pass +class TooManyRequestsError(OptimumOHBaseException): pass +class InternalServerError(OptimumOHBaseException): pass + + +# ── Fixed client response messages (generic, not for logging) ───────────────── + +_FIXED_MESSAGES = { + 400: "Malformed JSON body input", + 401: "Invalid authorization token", + 403: "Forbidden Access", + 404: "Resource not found", + 408: "Request timeout", + 422: "Invalid request parameter", + 429: "Please try again later", +} + +# Human-readable error types for logging (no underscores) +_ERROR_TYPE_MAP = { + BadRequestError: "Bad Request", + UnauthorizedError: "Unauthorized", + ForbiddenError: "Forbidden", + NotFoundError: "Not Found", + RequestTimeoutError: "Request Timeout", + UnprocessableEntityError: "Unprocessable Entity", + TooManyRequestsError: "Too Many Requests", + InternalServerError: "Internal Server Error", +} + +_EXCEPTION_STATUS_MAP = { + BadRequestError: 400, + UnauthorizedError: 401, + ForbiddenError: 403, + NotFoundError: 404, + RequestTimeoutError: 408, + UnprocessableEntityError: 422, + TooManyRequestsError: 429, + InternalServerError: 500, +} + +_HTTP_STATUS_NAMES = { + 400: "Bad Request", + 401: "Unauthorized", + 403: "Forbidden", + 404: "Not Found", + 405: "Method Not Allowed", + 408: "Request Timeout", + 409: "Conflict", + 413: "Payload Too Large", + 414: "URI Too Long", + 415: "Unsupported Media Type", + 422: "Unprocessable Entity", + 429: "Too Many Requests", + 500: "Internal Server Error", +} + + class ErrorDetail(BaseModel): field: Optional[str] = None message: str @@ -34,30 +103,17 @@ class ErrorResponse(BaseModel): errors: Optional[List[ErrorDetail]] = None -# Custom exception handler setup - +# ── Helpers ─────────────────────────────────────────────────────────────────── -def get_request_context(request: Request): - """ - Get detailed request context for logging. - """ +def get_request_context(request: Request) -> dict: + """Get detailed request context for logging.""" - def get_client_ip(): - """ - Get the real client IP address from Kong Gateway headers. - Kong sets X-Real-IP and X-Forwarded-For headers by default. - """ - # Kong specific headers + def get_client_ip() -> str: if "X-Real-IP" in request.headers: return request.headers["X-Real-IP"] - - # Fallback to X-Forwarded-For if "X-Forwarded-For" in request.headers: - # Get the first IP (original client) return request.headers["X-Forwarded-For"].split(",")[0].strip() - - # Last resort - return request.client.host + return request.client.host if request.client else "unknown" return { "endpoint": request.url.path, @@ -67,169 +123,255 @@ def get_request_context(request: Request): } -def handle_sqlalchemy_error(error: SQLAlchemyError): +def _json_response(status_code: int, message: str, data: Any = None) -> JSONResponse: + return JSONResponse( + status_code=status_code, + content={"data": data, "message": message, "status": ResponseStatus.ERROR}, + ) + + +def _store_log_state( + request: Request, + status_code: int, + error_id: str, + error_type: str, + error_message: str, +) -> None: + """Store logging metadata on request.state so db_session_middleware emits a single structured log.""" + request.state.log_status_code = status_code + request.state.log_error_id = error_id + request.state.log_error_type = error_type + request.state.log_error_message = error_message + + +def _short_db_error(orig_str: str) -> str: + """Extract a concise, human-readable reason from a raw DB/asyncpg error string.""" + s = orig_str.lower() + + if "uuid" in s: + if "length must be between" in s or "got " in s: + return "Invalid UUID length" + if "badly formed" in s or "invalid hexadecimal" in s or "invalid uuid" in s: + return "Invalid UUID format" + + if "invalid input syntax for type" in s: + m = re.search(r"invalid input syntax for type (\w+)", s) + type_name = m.group(1) if m else "value" + return f"Invalid format for type {type_name}" + + if "invalid input for query argument" in s: + m = re.search(r"\(([^)]+)\)\s*$", orig_str) + if m: + return _short_db_error(m.group(1)) + return "Invalid query argument value" + + if "unique constraint" in s: + return "Unique constraint violation" + if "foreign key constraint" in s: + return "Foreign key constraint violation" + if "null value in column" in s: + m = re.search(r'null value in column "([^"]+)"', orig_str) + col = f' "{m.group(1)}"' if m else "" + return f"Required field{col} is missing" + if "check constraint" in s: + return "Check constraint violation" + if "out of range" in s: + return "Value out of allowed range" + if "numeric field overflow" in s: + return "Numeric field overflow" + + first_line = orig_str.split("\n")[0].split(". ")[0].strip() + first_line = re.sub(r"^:\s*", "", first_line) + return first_line if first_line else "Database error" + + +# ── Core error response builder ─────────────────────────────────────────────── + +def _build_error_response(request: Request, exc: Exception, error_id: str) -> JSONResponse: """ - Handle SQLAlchemy errors and return user-friendly error messages. + Build JSON error response and store log metadata on request.state. + + Status codes handled: + 400 – BadRequestError + 401 – UnauthorizedError (missing/expired token) + 403 – ForbiddenError (valid token, access denied) + 404 – NotFoundError + 408 – RequestTimeoutError + 422 – UnprocessableEntityError / validation errors / security payloads + 429 – TooManyRequestsError / RateLimitExceeded + 500 – InternalServerError / SQLAlchemy errors / unexpected """ + + # 1. Custom OptimumOH exceptions + for exc_class, status_code in _EXCEPTION_STATUS_MAP.items(): + if isinstance(exc, exc_class): + error_type = _ERROR_TYPE_MAP[exc_class] + client_msg = _FIXED_MESSAGES.get(status_code, exc.message) + _store_log_state(request, status_code, error_id, error_type, exc.message) + return _json_response(status_code, f"{client_msg}. Error ID: {error_id}") + + # 2. SlowAPI RateLimitExceeded + if isinstance(exc, RateLimitExceeded): + try: + from src.auth.service import notify_admin_on_rate_limit_sync + request_info = get_request_context(request) + notify_admin_on_rate_limit_sync( + endpoint_name=request_info["endpoint"], + ip_address=request_info["remote_addr"], + method=request_info["method"], + request=request, + ) + except Exception: + pass + _store_log_state(request, 429, error_id, "Too Many Requests", "Please try again later") + return _json_response(429, f"Please try again later. Error ID: {error_id}") + + # 3. Pydantic / FastAPI RequestValidationError + if isinstance(exc, RequestValidationError): + errors = exc.errors() + detail_parts = [] + for err in errors: + loc = " -> ".join(str(part) for part in err.get("loc", [])) + msg = err.get("msg", "") + detail_parts.append(f"{loc}: {msg}" if loc else msg) + detail_str = "; ".join(detail_parts) if detail_parts else "Input validation failed" + _store_log_state(request, 422, error_id, "Unprocessable Entity", detail_str) + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + + # 4. Starlette/FastAPI HTTPException + if isinstance(exc, StarletteHTTPException): + status_code = exc.status_code + error_type = _HTTP_STATUS_NAMES.get(status_code, f"HTTP {status_code}") + + if status_code == 401: + log_msg = str(exc.detail) if exc.detail else "Invalid authorization token" + _store_log_state(request, 401, error_id, "Unauthorized", log_msg) + return _json_response(401, f"{_FIXED_MESSAGES[401]}. Error ID: {error_id}") + + if status_code == 422: + log_msg = str(exc.detail) if exc.detail else _FIXED_MESSAGES[422] + _store_log_state(request, 422, error_id, "Unprocessable Entity", log_msg) + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + + if isinstance(exc.detail, dict): + log_msg = exc.detail.get("error", error_type) + else: + log_msg = str(exc.detail) if exc.detail else error_type + + client_msg = _FIXED_MESSAGES.get(status_code, log_msg) + _store_log_state(request, status_code, error_id, error_type, log_msg) + return _json_response(status_code, f"{client_msg}. Error ID: {error_id}") + + # 5. SQLAlchemy errors + if isinstance(exc, SQLAlchemyError): + orig = getattr(exc, "orig", None) + orig_str = str(orig) if orig else str(exc) + short = _short_db_error(orig_str) + + # NoResultFound: .one() / .scalar_one() found no row → 404 + if isinstance(exc, NoResultFound): + _store_log_state(request, 404, error_id, "Not Found", "Resource not found") + return _json_response(404, f"Resource not found. Error ID: {error_id}") + + if isinstance(exc, IntegrityError): + orig_lower = orig_str.lower() + if "unique constraint" in orig_lower: + _store_log_state(request, 409, error_id, "Conflict", f"Unique constraint violation: {short}") + return _json_response(409, f"This record already exists. Error ID: {error_id}") + elif "foreign key constraint" in orig_lower: + _store_log_state(request, 422, error_id, "Unprocessable Entity", f"Foreign key constraint violation: {short}") + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + else: + _store_log_state(request, 422, error_id, "Unprocessable Entity", f"Database integrity error: {short}") + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + + if isinstance(exc, DataError): + _store_log_state(request, 422, error_id, "Unprocessable Entity", f"Invalid data format: {short}") + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + + if isinstance(exc, DBAPIError): + orig_lower = orig_str.lower() + if "invalid input syntax" in orig_lower or "invalid uuid" in orig_lower or "invalid input for query argument" in orig_lower: + _store_log_state(request, 422, error_id, "Unprocessable Entity", f"Invalid input format: {short}") + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + elif "unique constraint" in orig_lower: + _store_log_state(request, 409, error_id, "Conflict", f"Unique constraint violation: {short}") + return _json_response(409, f"This record already exists. Error ID: {error_id}") + elif "foreign key constraint" in orig_lower: + _store_log_state(request, 422, error_id, "Unprocessable Entity", f"Foreign key constraint violation: {short}") + return _json_response(422, f"{_FIXED_MESSAGES[422]}. Error ID: {error_id}") + else: + _store_log_state(request, 500, error_id, "Internal Server Error", f"Database query error: {short}") + return _json_response(500, f"An unexpected error occurred. Error ID: {error_id}") + + _store_log_state(request, 500, error_id, "Internal Server Error", f"Database error: {short}") + return _json_response(500, f"An unexpected error occurred. Error ID: {error_id}") + + # 5b. httpx upstream errors (raise_for_status() from external service calls) try: - original_error = error.orig - except AttributeError: - original_error = None + import httpx as _httpx + if isinstance(exc, _httpx.HTTPStatusError): + upstream_status = exc.response.status_code + error_type = _HTTP_STATUS_NAMES.get(upstream_status, f"HTTP {upstream_status}") + client_msg = _FIXED_MESSAGES.get(upstream_status, "Upstream service error") + log_msg = f"Upstream service returned {upstream_status}: {exc.request.url}" + _store_log_state(request, upstream_status, error_id, error_type, log_msg) + return _json_response(upstream_status, f"{client_msg}. Error ID: {error_id}") + if isinstance(exc, (_httpx.ConnectError, _httpx.TimeoutException, _httpx.RequestError)): + log_msg = f"Upstream service unreachable: {type(exc).__name__}: {exc}" + _store_log_state(request, 502, error_id, "Bad Gateway", log_msg) + return _json_response(502, f"Upstream service unavailable. Error ID: {error_id}") + except ImportError: + pass + + # 6. Catch-all + detail = type(exc).__name__ + _store_log_state(request, 500, error_id, "Internal Server Error", detail) + return _json_response(500, f"An unexpected error occurred. Error ID: {error_id}") + + +# ── Single exception handler registered with FastAPI ───────────────────────── + +async def handle_exception(request: Request, exc: Exception) -> JSONResponse: + """ + Global exception handler. + Sets request.state.log_* so that db_session_middleware emits a single structured log. + """ + error_id = str(uuid.uuid4()) + + if not hasattr(request.state, "request_id"): + request.state.request_id = error_id + + return _build_error_response(request, exc, error_id) + + +# ── Legacy helper kept for backward compatibility ───────────────────────────── + +def handle_sqlalchemy_error(error: SQLAlchemyError): + """Return (user_message, status_code) for a SQLAlchemy error.""" + original_error = error.orig if hasattr(error, "orig") else None if isinstance(error, IntegrityError): if "unique constraint" in str(error).lower(): - return "This record already exists.", 422 + return "This record already exists.", 409 elif "foreign key constraint" in str(error).lower(): - return "Related record not found.", 422 + return "Related record not found.", 400 else: - return "Data integrity error.", 422 + return "Data integrity error.", 400 elif isinstance(error, DataError) or isinstance(original_error, AsyncPGDataError): - return "Invalid data provided.", 422 + return "Invalid data provided.", 400 elif isinstance(error, DBAPIError): if "unique constraint" in str(error).lower(): - return "This record already exists.", 422 + return "This record already exists.", 409 elif "foreign key constraint" in str(error).lower(): - return "Related record not found.", 422 + return "Related record not found.", 400 elif "null value in column" in str(error).lower(): - return "Required data missing.", 422 + return "Required data missing.", 400 elif "invalid input for query argument" in str(error).lower(): - return "Invalid data provided.", 422 + return "Invalid data provided.", 400 else: return "Database error.", 500 else: - # Log the full error for debugging purposes log.error(f"Unexpected database error: {str(error)}") return "An unexpected database error occurred.", 500 - -def handle_exception(request: Request, exc: Exception): - """ - Global exception handler for Fastapi application. - """ - import uuid - error_id = str(uuid.uuid1()) - request_info = get_request_context(request) - - # Store error_id in request.state for middleware/logging - request.state.error_id = error_id - - if isinstance(exc, RateLimitExceeded): - # Kirim notifikasi ke admin - notify_admin_on_rate_limit_sync( - endpoint_name=request_info["endpoint"], - ip_address=request_info["remote_addr"], - method=request_info["method"], - request=request, - ) - - log.warning( - f"Rate limit exceeded: {str(exc.description) if hasattr(exc, 'description') else str(exc)} | Error ID: {error_id}", - extra={ - "error_id": error_id, - "error_category": "rate_limit", - "detail": str(exc.description) if hasattr(exc, "description") else str(exc), - "request": request_info, - }, - ) - return JSONResponse( - status_code=429, - content={ - "data": None, - "message": "Rate limit exceeded", - "status": ResponseStatus.ERROR, - "error_id": error_id - } - ) - - if isinstance(exc, RequestValidationError): - log.warning( - f"Validation error: {exc.errors()} | Error ID: {error_id}", - extra={ - "error_id": error_id, - "error_category": "validation", - "errors": exc.errors(), - "request": request_info, - }, - ) - return JSONResponse( - status_code=422, - content={ - "data": exc.errors(), - "message": "Validation Error", - "status": ResponseStatus.ERROR, - "error_id": error_id - }, - ) - - if isinstance(exc, (HTTPException, StarletteHTTPException)): - # Log as warning for 4xx, error for 5xx - status_code = exc.status_code if hasattr(exc, "status_code") else 500 - detail = exc.detail if hasattr(exc, "detail") else str(exc) - - log_level = logging.WARNING if 400 <= status_code < 500 else logging.ERROR - log.log( - log_level, - f"HTTP {status_code}: {detail} | Error ID: {error_id}", - extra={ - "error_id": error_id, - "error_category": "http", - "status_code": status_code, - "request": request_info, - }, - ) - - return JSONResponse( - status_code=status_code, - content={ - "data": None, - "message": str(detail), - "status": ResponseStatus.ERROR, - "error_id": error_id - }, - ) - - if isinstance(exc, SQLAlchemyError): - error_message, status_code = handle_sqlalchemy_error(exc) - # Log integrity errors as warning, others as error - log_level = logging.WARNING if 400 <= status_code < 500 else logging.ERROR - log.log( - log_level, - f"Database error: {error_message} | Error ID: {error_id}", - extra={ - "error_id": error_id, - "error_category": "database", - "violation": str(exc).split('\n')[0], - "request": request_info, - }, - ) - - return JSONResponse( - status_code=status_code, - content={ - "data": None, - "message": error_message, - "status": ResponseStatus.ERROR, - "error_id": error_id - }, - ) - - # Log unexpected errors - log.error( - f"Unexpected error: {str(exc)} | Error ID: {error_id}", - extra={ - "error_id": error_id, - "error_category": "unexpected", - "request": request_info, - }, - ) - - return JSONResponse( - status_code=500, - content={ - "data": None, - "message": "An unexpected error occurred", - "status": ResponseStatus.ERROR, - "error_id": error_id - }, - ) - diff --git a/src/logging.py b/src/logging.py index 787d24b..a3ea1da 100644 --- a/src/logging.py +++ b/src/logging.py @@ -1,27 +1,39 @@ import logging +import logging.handlers import json import datetime import os import sys -from typing import Optional -from src.config import LOG_LEVEL +try: + from zoneinfo import ZoneInfo +except ImportError: + from backports.zoneinfo import ZoneInfo # type: ignore[no-redef] + +from src.config import LOG_LEVEL, SERVICE_NAME from src.enums import OptimumOHEnum -LOG_FORMAT_DEBUG = "%(levelname)s:%(message)s:%(pathname)s:%(funcName)s:%(lineno)d" +_TZ_LOCAL = ZoneInfo("Asia/Jakarta") # ANSI Color Codes RESET = "\033[0m" COLORS = { - "DEBUG": "\033[36m", # Cyan - "INFO": "\033[32m", # Green - "WARNING": "\033[33m", # Yellow - "WARN": "\033[33m", # Yellow - "ERROR": "\033[31m", # Red + "DEBUG": "\033[36m", # Cyan + "INFO": "\033[32m", # Green + "WARNING": "\033[33m", # Yellow + "WARN": "\033[33m", # Yellow + "ERROR": "\033[31m", # Red "CRITICAL": "\033[1;31m", # Bold Red } +# Fields in the log record injected by the logging middleware or extra= kwargs +_REQUEST_FIELDS = frozenset({ + "method", "path", "status_code", "duration_ms", + "request_id", "user_id", "user_ip", + "error_id", "result", "error_type", "error_message", +}) + class LogLevels(OptimumOHEnum): info = "INFO" @@ -32,102 +44,89 @@ class LogLevels(OptimumOHEnum): class JSONFormatter(logging.Formatter): """ - Custom formatter to output logs in JSON format. + Outputs exactly 14 fixed fields per log line in JSON format. + + Fields (always present, null when not applicable): + timestamp, service_name, level, method, path, status_code, + duration_ms, request_id, user_id, user_ip, error_id, + result, error_type, error_message """ - def format(self, record): - from src.context import get_request_id, get_user_id, get_username, get_role - - request_id = None - user_id = None - username = None - role = None - - try: - request_id = get_request_id() - user_id = get_user_id() - username = get_username() - role = get_role() - except Exception: - pass - - # Standard fields from requirements + + def format(self, record: logging.LogRecord) -> str: + dt = datetime.datetime.fromtimestamp(record.created, tz=_TZ_LOCAL) + timestamp = dt.isoformat(timespec="microseconds") + + status_code = getattr(record, "status_code", None) + result = getattr(record, "result", None) + if result is None and status_code is not None: + result = "Request completed" if int(status_code) < 400 else "Request failed" + log_record = { - "timestamp": datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S"), - "level": record.levelname, - "message": record.getMessage(), + "timestamp": timestamp, + "service_name": SERVICE_NAME, + "level": record.levelname, + "method": getattr(record, "method", None), + "path": getattr(record, "path", None), + "status_code": status_code, + "duration_ms": getattr(record, "duration_ms", None), + "request_id": getattr(record, "request_id", None), + "user_id": getattr(record, "user_id", None), + "user_ip": getattr(record, "user_ip", None), + "error_id": getattr(record, "error_id", None), + "result": result, + "error_type": getattr(record, "error_type", None), + "error_message": getattr(record, "error_message", None), } - # Add Context information if available - if request_id: - log_record["request_id"] = request_id - - # Add Error context if available - if hasattr(record, "error_id"): - log_record["error_id"] = record.error_id - elif "error_id" in record.__dict__: - log_record["error_id"] = record.error_id - - if user_id: - log_record["user_id"] = user_id - - # Add any extra attributes passed to the log call - standard_attrs = { - "args", "asctime", "created", "exc_info", "exc_text", "filename", - "funcName", "levelname", "levelno", "lineno", "module", "msecs", - "message", "msg", "name", "pathname", "process", "processName", - "relativeCreated", "stack_info", "thread", "threadName", "error_id" - } - for key, value in record.__dict__.items(): - if key not in standard_attrs and not key.startswith("_"): - log_record[key] = value - - log_json = json.dumps(log_record) + log_json = json.dumps(log_record, default=str) - # Apply color if the output is a terminal if sys.stdout.isatty(): - level_color = COLORS.get(record.levelname, "") - return f"{level_color}{log_json}{RESET}" + color = COLORS.get(record.levelname, "") + return f"{color}{log_json}{RESET}" return log_json -def configure_logging(): - log_level = str(LOG_LEVEL).upper() # cast to string - log_levels = list(LogLevels) - - if log_level not in log_levels: +def configure_logging() -> None: + log_level = str(LOG_LEVEL).upper() + if log_level not in list(LogLevels): log_level = LogLevels.error - # Get the root logger root_logger = logging.getLogger() root_logger.setLevel(log_level) - # Clear existing handlers to avoid duplicate logs if root_logger.hasHandlers(): root_logger.handlers.clear() - # Create a stream handler that outputs to stdout - handler = logging.StreamHandler(sys.stdout) - - # Use JSONFormatter for all environments, or could be conditional - # For now, let's assume the user wants JSON everywhere as requested formatter = JSONFormatter() - - # If debug mode is specifically requested and we want the old format for debug: - # if log_level == LogLevels.debug: - # formatter = logging.Formatter(LOG_FORMAT_DEBUG) - - handler.setFormatter(formatter) - root_logger.addHandler(handler) - - # Reconfigure uvicorn loggers to use our JSON formatter + + # stdout handler + stream_handler = logging.StreamHandler(sys.stdout) + stream_handler.setFormatter(formatter) + root_logger.addHandler(stream_handler) + + # rotating file handler — logs/app.log 10 MB × 5 backups + os.makedirs("logs", exist_ok=True) + file_handler = logging.handlers.RotatingFileHandler( + "logs/app.log", + maxBytes=10 * 1024 * 1024, + backupCount=5, + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + root_logger.addHandler(file_handler) + + # Reconfigure uvicorn loggers to propagate through our formatter for logger_name in ["uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"]: - logger = logging.getLogger(logger_name) - logger.handlers = [] - logger.propagate = True + uvicorn_logger = logging.getLogger(logger_name) + uvicorn_logger.handlers = [] + uvicorn_logger.propagate = True - # Silence the chatty uvicorn access logs as we have custom middleware logging + # Silence chatty loggers — we emit one structured log per request ourselves logging.getLogger("uvicorn.access").setLevel(logging.WARNING) - # sometimes the slack client can be too verbose + slowapi_logger = logging.getLogger("slowapi") + slowapi_logger.setLevel(logging.ERROR) + slowapi_logger.propagate = False + logging.getLogger("slack_sdk.web.base_client").setLevel(logging.CRITICAL) \ No newline at end of file diff --git a/src/main.py b/src/main.py index f7e969c..44922d3 100644 --- a/src/main.py +++ b/src/main.py @@ -7,7 +7,8 @@ from os import path from typing import Final, Optional from uuid import uuid1 -from fastapi import FastAPI, HTTPException, status +from fastapi import Depends, FastAPI, HTTPException, status +from src.auth.service import JWTBearer from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import ValidationError @@ -25,6 +26,8 @@ from starlette.responses import FileResponse, Response, StreamingResponse from starlette.routing import compile_path from starlette.staticfiles import StaticFiles +import uuid as _uuid_module + from src.api import api_router from src.database.core import async_session, engine, async_collector_session from src.enums import ResponseStatus @@ -34,6 +37,7 @@ from src.middleware import RequestValidationMiddleware, RequestTimeoutMiddleware from src.rate_limiter import limiter from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException +from src.csrf_protect import csrf_protect log = logging.getLogger(__name__) @@ -50,12 +54,25 @@ app = FastAPI( version="0.1.0", ) +# Sync handler for RateLimitExceeded — SlowAPIMiddleware falls back to default +# when the registered handler is async, so this must be a plain sync function. +def _sync_rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONResponse: + error_id = str(_uuid_module.uuid4()) + request.state.log_status_code = 429 + request.state.log_error_id = error_id + request.state.log_error_type = "Too Many Requests" + request.state.log_error_message = "Please try again later" + return JSONResponse( + status_code=429, + content={"data": None, "message": f"Please try again later. Error ID: {error_id}", "status": ResponseStatus.ERROR}, + ) + # we define the exception handlers app.add_exception_handler(Exception, handle_exception) app.add_exception_handler(HTTPException, handle_exception) app.add_exception_handler(StarletteHTTPException, handle_exception) app.add_exception_handler(RequestValidationError, handle_exception) -app.add_exception_handler(RateLimitExceeded, handle_exception) +app.add_exception_handler(RateLimitExceeded, _sync_rate_limit_handler) app.state.limiter = limiter app.add_middleware(GZipMiddleware, minimum_size=1000) @@ -70,7 +87,12 @@ async def slow_endpoint(request: Request): await asyncio.sleep(REQUEST_TIMEOUT_SECONDS + 10) return JSONResponse(content={"message": "This should timeout"}, status_code=200) - +@app.post('/testcsrf', dependencies=[Depends(JWTBearer()), Depends(csrf_protect)]) +async def test_csrf(request: Request): + return JSONResponse( + content={"data": None, "message": "CSRF token is valid", "status": ResponseStatus.SUCCESS}, + status_code=200, + ) def security_headers_middleware(app: FastAPI): @@ -148,9 +170,10 @@ app.add_middleware(RequestValidationMiddleware) async def db_session_middleware(request: Request, call_next): request_id = str(uuid1()) - # we create a per-request id such that we can ensure that our session is scoped for a particular request. + # Per-request ID scopes the SQLAlchemy session. # see: https://github.com/tiangolo/fastapi/issues/726 ctx_token = set_request_id(request_id) + request.state.request_id = request_id try: session = async_scoped_session(async_session, scopefunc=get_request_id) @@ -160,60 +183,103 @@ async def db_session_middleware(request: Request, call_next): request.state.collector_db = collector_session() start_time = time.time() - response = await call_next(request) - process_time = (time.time() - start_time) * 1000 - - # Skip logging in middleware if it's an error (already logged in handle_exception) - if response.status_code >= 400: - return response - - from src.context import get_username, get_role, get_user_id, set_user_id, set_username, set_role - - # Pull from context or fallback to request.state.user - username = get_username() - role = get_role() + try: + response = await call_next(request) + except Exception as exc: + # Non-HTTP exceptions (SQLAlchemy, httpx, ValueError, etc.) are NOT caught by + # ExceptionMiddleware because app.add_exception_handler(Exception, ...) registers + # with ServerErrorMiddleware, not ExceptionMiddleware. If we let such exceptions + # escape here, db_session_middleware exits via finally WITHOUT logging, then + # ServerErrorMiddleware calls handle_exception (giving the 500 response) and + # re-raises, causing uvicorn to emit an all-null JSON log entry. + # Catching here ensures: proper error response, structured log, no re-raise. + response = await handle_exception(request, exc) + duration_ms = round((time.time() - start_time) * 1000, 2) + + # ── Determine status_code ───────────────────────────────────────────── + # RequestTimeoutMiddleware flags the scope when a 408 bypasses exception handlers + timed_out = request.scope.get("x_timed_out", False) + if timed_out: + status_code = 408 + error_id = request.scope.get("x_timeout_error_id", None) + error_type = "Request Timeout" + error_message = "Request timeout over 20 seconds" + else: + status_code = getattr(request.state, "log_status_code", None) or response.status_code + error_id = getattr(request.state, "log_error_id", None) + error_type = getattr(request.state, "log_error_type", None) + error_message = getattr(request.state, "log_error_message", None) + + # ── Fallback error labels for responses that skip exception handler ─── + _FALLBACK_ERROR_TYPES = { + 400: "Bad Request", 401: "Unauthorized", 403: "Forbidden", + 404: "Not Found", 408: "Request Timeout", 409: "Conflict", + 422: "Unprocessable Entity", 429: "Too Many Requests", + 500: "Internal Server Error", + } + _FALLBACK_ERROR_MESSAGES = { + 400: "Bad request", 401: "Invalid authorization token", + 403: "Forbidden Access", 404: "Resource not found", + 408: "Request timeout", 409: "This record already exists", + 422: "Input validation failed", 429: "Please try again later", + 500: "An unexpected error occurred", + } + if status_code >= 400 and not error_type: + error_type = _FALLBACK_ERROR_TYPES.get(status_code, f"HTTP {status_code}") + if status_code >= 400 and not error_message: + error_message = _FALLBACK_ERROR_MESSAGES.get(status_code, "Request failed") + + # ── User info ───────────────────────────────────────────────────────── + from src.context import get_user_id, set_user_id user_id = get_user_id() - user_obj = getattr(request.state, "user", None) - if user_obj: - # message is UserBase dict/obj in this project + if user_obj and not user_id: if isinstance(user_obj, dict): u_id = user_obj.get("user_id") - u_name = user_obj.get("name") or user_obj.get("username") - u_role = user_obj.get("role") else: u_id = getattr(user_obj, "user_id", None) - u_name = getattr(user_obj, "name", None) or getattr(user_obj, "username", None) - u_role = getattr(user_obj, "role", None) - - if not user_id and u_id: + if u_id: user_id = str(u_id) set_user_id(user_id) - if not username and u_name: - username = u_name - set_username(username) - if not role and u_role: - role = u_role - set_role(role) - - user_info_str = "" - if user_id: - user_info_str = f" | User ID: {user_id}" - - error_id = getattr(request.state, "error_id", None) - log_msg = f"HTTP {request.method} {request.url.path} completed in {round(process_time, 2)}ms{user_info_str}" - if error_id: - log_msg += f" | Error ID: {error_id}" - - log.info( - log_msg, + + # ── Client IP ───────────────────────────────────────────────────────── + headers = request.headers + if "X-Real-IP" in headers: + user_ip = headers["X-Real-IP"] + elif "X-Forwarded-For" in headers: + user_ip = headers["X-Forwarded-For"].split(",")[0].strip() + else: + user_ip = request.client.host if request.client else None + + # ── Full path with query string ─────────────────────────────────────── + path = request.url.path + if request.url.query: + path = f"{path}?{request.url.query}" + + # ── Emit single structured log ──────────────────────────────────────── + result = "Request completed" if status_code < 400 else "Request failed" + if status_code >= 500: + log_level = logging.ERROR + elif status_code >= 400: + log_level = logging.WARNING + else: + log_level = logging.INFO + + log.log( + log_level, + f"{request.method} {path} {status_code}", extra={ "method": request.method, - "path": request.url.path, - "status_code": response.status_code, - "duration_ms": round(process_time, 2), + "path": path, + "status_code": status_code, + "duration_ms": duration_ms, + "request_id": request_id, "user_id": user_id, + "user_ip": user_ip, "error_id": error_id, + "result": result, + "error_type": error_type if status_code >= 400 else None, + "error_message": error_message if status_code >= 400 else None, }, ) finally: diff --git a/src/middleware.py b/src/middleware.py index 0fc7ea8..020b189 100644 --- a/src/middleware.py +++ b/src/middleware.py @@ -191,41 +191,35 @@ def has_control_chars(value: str) -> bool: def inspect_value(value: str, source: str): if not isinstance(value, str) or value == "*/*": return - if XSS_PATTERN.search(value): - log.warning(f"Security violation: Potential XSS payload detected in {source}, value: {value}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"XSS payload detected in {source}", ) if SQLI_PATTERN.search(value): - log.warning(f"Security violation: Potential SQL injection payload detected in {source}, value: {value}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"SQL injection payload detected in {source}", ) if RCE_PATTERN.search(value): - log.warning(f"Security violation: Potential RCE payload detected in {source}, value: {value}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Remote code execution payload detected in {source}", ) if TRAVERSAL_PATTERN.search(value): - log.warning(f"Security violation: Potential Path Traversal payload detected in {source}, value: {value}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Path traversal payload detected in {source}", ) if has_control_chars(value): - log.warning(f"Security violation: Invalid control characters detected in {source}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Invalid control characters detected in {source}", ) @@ -233,17 +227,15 @@ def inspect_json(obj, path="body", check_whitelist=True): if isinstance(obj, dict): for key, value in obj.items(): if key in FORBIDDEN_JSON_KEYS: - log.warning(f"Security violation: Forbidden JSON key detected: {path}.{key}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Forbidden JSON key detected: {path}.{key}", ) - + if check_whitelist and key not in ALLOWED_DATA_PARAMS: - log.warning(f"Security violation: Unknown JSON key detected: {path}.{key}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Unknown JSON key detected: {path}.{key}", ) # Recurse. If the key is a dynamic container, we stop whitelist checking for children. @@ -262,6 +254,18 @@ def inspect_json(obj, path="body", check_whitelist=True): class RequestValidationMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): + # Wrap the entire dispatch body so that any HTTPException raised during + # validation is handled by our central handle_exception rather than + # escaping through call_next into the outer middleware chain. + # Without this wrapper, exceptions from dispatch() bypass db_session_middleware's + # logging code and cause an all-nulls ERROR log from uvicorn's internal logger. + try: + return await self._validate_and_forward(request, call_next) + except HTTPException as exc: + from src.exceptions import handle_exception + return await handle_exception(request, exc) + + async def _validate_and_forward(self, request: Request, call_next): # ------------------------- # 0. Header validation # ------------------------- @@ -274,21 +278,19 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): ALLOW_DUPLICATE_HEADERS = {'accept', 'accept-encoding', 'accept-language', 'accept-charset', 'cookie'} real_duplicates = [h for h in duplicate_headers if h not in ALLOW_DUPLICATE_HEADERS] if real_duplicates: - log.warning(f"Security violation: Duplicate headers detected: {real_duplicates}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Duplicate headers detected: {real_duplicates}", ) - + # Whitelist headers unknown_headers = [key for key in header_keys if key not in ALLOWED_HEADERS] if unknown_headers: - filtered_unknown = [h for h in unknown_headers if not h.startswith('sec-')] - if filtered_unknown: - log.warning(f"Security violation: Unknown headers detected: {filtered_unknown}") + filtered_unknown = [h for h in unknown_headers if not h.startswith('sec-')] + if filtered_unknown: raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Unknown headers detected: {filtered_unknown}", ) # Inspect header values @@ -300,28 +302,25 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): # 1. Query string limits # ------------------------- if len(request.url.query) > MAX_QUERY_LENGTH: - log.warning(f"Security violation: Query string too long") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail="Query string too long", ) params = request.query_params.multi_items() if len(params) > MAX_QUERY_PARAMS: - log.warning(f"Security violation: Too many query parameters") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail="Too many query parameters", ) # Check for unknown query parameters unknown_params = [key for key, _ in params if key not in ALLOWED_DATA_PARAMS] if unknown_params: - log.warning(f"Security violation: Unknown query parameters detected: {unknown_params}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Unknown query parameters detected: {unknown_params}", ) # ------------------------- @@ -334,10 +333,9 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): ] if duplicates: - log.warning(f"Security violation: Duplicate query parameters detected: {duplicates}") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Duplicate query parameters: {duplicates}", ) # ------------------------- @@ -352,22 +350,19 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): try: size_val = int(value) if size_val > 50: - log.warning(f"Security violation: Pagination size too large ({size_val})") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Pagination size too large: {size_val} (max 50)", ) if size_val % 5 != 0: - log.warning(f"Security violation: Pagination size not multiple of 5 ({size_val})") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Pagination size not a multiple of 5: {size_val}", ) except ValueError: - log.warning(f"Security violation: Pagination size invalid value") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail=f"Pagination size invalid value for '{key}'", ) # ------------------------- @@ -378,8 +373,7 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): content_type.startswith(t) for t in ("application/json", "multipart/form-data", "application/x-www-form-urlencoded") ): - log.warning(f"Security violation: Unsupported Content-Type: {content_type}") - raise HTTPException(status_code=422, detail="Invalid request parameters") + raise HTTPException(status_code=422, detail=f"Unsupported Content-Type: {content_type}") # ------------------------- # 5. Single source check (Query vs JSON Body) @@ -395,10 +389,9 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): has_body = True if has_query and has_body: - log.warning(f"Security violation: Mixed parameters (query + JSON body)") raise HTTPException( status_code=422, - detail="Invalid request parameters", + detail="Mixed parameters: both query string and JSON body are not allowed", ) # ------------------------- @@ -413,8 +406,9 @@ class RequestValidationMiddleware(BaseHTTPMiddleware): try: payload = json.loads(body) except json.JSONDecodeError: - log.warning(f"Security violation: Invalid JSON body") - raise HTTPException(status_code=422, detail="Invalid request parameters") + # Malformed JSON is a client-side structural error → 400 Bad Request, + # NOT 422 Unprocessable Entity (which is for semantically invalid input). + raise HTTPException(status_code=400, detail="Invalid JSON body: malformed JSON") inspect_json(payload) @@ -465,16 +459,15 @@ class RequestTimeoutMiddleware: ) except asyncio.TimeoutError: if not response_started: - log.warning( - "Request timeout (%ds): %s %s", - self.timeout, - scope.get("method", ""), - scope.get("path", ""), - ) + import uuid as _uuid + error_id = str(_uuid.uuid4()) + # Flag the scope so db_session_middleware can log the 408 with correct fields + scope["x_timed_out"] = True + scope["x_timeout_error_id"] = error_id body = json.dumps({ - "status": False, - "message": f"Request timeout. Processing exceeded {self.timeout} seconds.", "data": None, + "message": f"Request timeout. Error ID: {error_id}", + "status": False, }).encode("utf-8") await send({ "type": "http.response.start",