FastAPI Middleware — Logging, CORS and Custom Middleware

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Configure the 'Onion' layers
# CORSMiddleware should generally be added early in the stack
app.add_middleware(
    CORSMiddleware,
    # List specific trusted origins for production
    allow_origins=[
        'https://api.thecodeforge.io',
        'https://dashboard.thecodeforge.io',
        'http://localhost:3000'
    ],
    # Required if your frontend sends cookies or Authorization headers
    allow_credentials=True,
    allow_methods=['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS'],
    allow_headers=['Authorization', 'X-Forge-Trace-ID', 'Content-Type'],
)

# Note: If allow_credentials is True, allow_origins cannot be ['*']

from fastapi import FastAPI, Request
import time
import uuid
import logging

app = FastAPI()
logger = logging.getLogger("thecodeforge.access")

@app.middleware('http')
async def add_process_time_header(request: Request, call_next):
    # 1. Logic BEFORE the route (Request Phase)
    start_time = time.perf_counter()
    request_id = str(uuid.uuid4())

    # Inject trace ID into request state for downstream access
    request.state.trace_id = request_id

    # 2. Hand off to the next middleware or route handler
    response = await call_next(request)

    # 3. Logic AFTER the route (Response Phase)
    process_time = time.perf_counter() - start_time

    # Log the performance metric
    logger.info(f"RID: {request_id} | Path: {request.url.path} | Time: {process_time:.4f}s")

    # Standardize our response headers
    response.headers['X-Forge-Process-Time'] = str(process_time)
    response.headers['X-Forge-Trace-ID'] = request_id

    return response

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Execution order (request direction):
# 1. Security headers middleware (added last)
# 2. CORSMiddleware
# 3. Logging middleware (added first)
# 4. Route handler
# Response order is reverse

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_headers=["*"],
)

@app.middleware("http")
async def security_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    return response

@app.middleware("http")
async def logging_middleware(request, call_next):
    print(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    print(f"Response status: {response.status_code}")
    return response

from fastapi import FastAPI, Request, Response
import logging

app = FastAPI()
logger = logging.getLogger("thecodeforge.security")

BLOCKED_IPS = {"10.0.0.99", "192.168.1.200"}

@app.middleware("http")
async def ip_blocker(request: Request, call_next):
    client_ip = request.client.host if request.client else "unknown"
    if client_ip in BLOCKED_IPS:
        logger.warning(f"Blocked IP: {client_ip}")
        return Response(
            content='{"detail": "Forbidden"}',
            status_code=403,
            media_type="application/json"
        )
    # If not blocked, continue
    response = await call_next(request)
    return response

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
import logging
import traceback

thecodeforge_logger = logging.getLogger("thecodeforge.error")

app = FastAPI()

@app.middleware("http")
async def catch_exceptions_middleware(request: Request, call_next):
    try:
        response = await call_next(request)
        return response
    except Exception as exc:
        # Log the full traceback internally
        thecodeforge_logger.error(
            f"Unhandled exception on {request.method} {request.url.path}: {traceback.format_exc()}"
        )
        # Return a generic error to the client
        return JSONResponse(
            status_code=500,
            content={"detail": "Internal server error. Please try again later.",
                      "trace_id": request.state.trace_id if hasattr(request, 'state') else None}
        )

import pytest
from fastapi.testclient import TestClient
from io.thecodeforge.main import app  # assuming your FastAPI app is here

client = TestClient(app)

class TestCORS:
    def test_cors_headers_present(self):
        response = client.get("/", headers={"Origin": "https://dashboard.thecodeforge.io"})
        assert response.headers.get("access-control-allow-origin") == "https://dashboard.thecodeforge.io"
        assert response.headers.get("access-control-allow-credentials") == "true"

    def test_cors_rejects_wrong_origin(self):
        response = client.get("/", headers={"Origin": "https://evil.com"})
        # No CORS headers means browser blocks it
        assert "access-control-allow-origin" not in response.headers

class TestShortCircuit:
    def test_blocked_ip_returns_403(self):
        # Simulate an IP - this might not work in TestClient because request.client is None
        # We need to mock request.client
        with patch("fastapi.Request.client", new_callable=PropertyMock) as mock_client:
            mock_client.return_value = Mock(host="10.0.0.99")
            response = client.get("/")
            assert response.status_code == 403
            assert response.json() == {"detail": "Forbidden"}