CIzz22
/
be-optimumoh
1
0
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update: Sheet 7 Adjustment

Browse Source
oh_security
TAMDeveloper13 3 months ago
parent d4175f75d0
commit 600030199d
6 changed files with 564 additions and 359 deletions
Show Stats Download Patch File Download Diff File

16
src/auth/service.py
Unescape Escape View File

@ -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]

1
src/config.py
Unescape Escape View File

@ -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")

484
src/exceptions.py
Unescape Escape View File

@ -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):
"""
Handle SQLAlchemy errors and return user-friendly error messages.
"""
try:
original_error = error.orig
except AttributeError:
original_error = None
if isinstance(error, IntegrityError):
if "unique constraint" in str(error).lower():
return "This record already exists.", 422
elif "foreign key constraint" in str(error).lower():
return "Related record not found.", 422
else:
return "Data integrity error.", 422
elif isinstance(error, DataError) or isinstance(original_error, AsyncPGDataError):
return "Invalid data provided.", 422
elif isinstance(error, DBAPIError):
if "unique constraint" in str(error).lower():
return "This record already exists.", 422
elif "foreign key constraint" in str(error).lower():
return "Related record not found.", 422
elif "null value in column" in str(error).lower():
return "Required data missing.", 422
elif "invalid input for query argument" in str(error).lower():
return "Invalid data provided.", 422
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 _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 handle_exception(request: Request, exc: Exception):
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"^<class '[^']+'>:\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:
"""
Global exception handler for Fastapi application.
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
"""
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
# 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):
# Kirim notifikasi ke admin
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}")
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
}
)
# 3. Pydantic / FastAPI RequestValidationError
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,
},
)
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
return JSONResponse(
status_code=status_code,
content={
"data": None,
"message": str(detail),
"status": ResponseStatus.ERROR,
"error_id": error_id
},
)
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):
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,
},
)
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}")
return JSONResponse(
status_code=status_code,
content={
"data": None,
"message": error_message,
"status": ResponseStatus.ERROR,
"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}")
# 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,
},
)
# 5b. httpx upstream errors (raise_for_status() from external service calls)
try:
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())
return JSONResponse(
status_code=500,
content={
"data": None,
"message": "An unexpected error occurred",
"status": ResponseStatus.ERROR,
"error_id": error_id
},
)
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.", 409
elif "foreign key constraint" in str(error).lower():
return "Related record not found.", 400
else:
return "Data integrity error.", 400
elif isinstance(error, DataError) or isinstance(original_error, AsyncPGDataError):
return "Invalid data provided.", 400
elif isinstance(error, DBAPIError):
if "unique constraint" in str(error).lower():
return "This record already exists.", 409
elif "foreign key constraint" in str(error).lower():
return "Related record not found.", 400
elif "null value in column" in str(error).lower():
return "Required data missing.", 400
elif "invalid input for query argument" in str(error).lower():
return "Invalid data provided.", 400
else:
return "Database error.", 500
else:
log.error(f"Unexpected database error: {str(error)}")
return "An unexpected database error occurred.", 500

147
src/logging.py
Unescape Escape View File

@ -1,15 +1,20 @@
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"
@ -22,6 +27,13 @@ COLORS = {
"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
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"
try:
request_id = get_request_id()
user_id = get_user_id()
username = get_username()
role = get_role()
except Exception:
pass
# Standard fields from requirements
log_record = {
"timestamp": datetime.datetime.fromtimestamp(record.created).strftime("%Y-%m-%d %H:%M:%S"),
"timestamp": timestamp,
"service_name": SERVICE_NAME,
"level": record.levelname,
"message": record.getMessage(),
"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, default=str)
log_json = json.dumps(log_record)
# 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)

154
src/main.py
Unescape Escape View File

@ -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()
try:
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
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")
# Pull from context or fallback to request.state.user
username = get_username()
role = get_role()
# ── 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:

87
src/middleware.py
Unescape Escape View File

@ -192,40 +192,34 @@ 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,10 +278,9 @@ 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
@ -285,10 +288,9 @@ class RequestValidationMiddleware(BaseHTTPMiddleware):
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}")
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",

Write Preview
Loading…
Cancel
Save