Middleware & Security

Middleware auto-discovery, the request processing chain, CORS configuration, rate limiting, exception handling, JWT authentication, and WebSocket security.

Singularity's middleware and security layers sit between the incoming request and your service handlers. Middleware is auto-discovered from the singularity/middleware/ directory, and authentication is handled through FastAPI dependencies. This page covers the full request processing chain, how to write custom middleware, and how JWT authentication works for both HTTP and WebSocket connections.

Middleware Chain

Every request passes through a stack of middleware before reaching the route handler. The following diagram shows the processing order:

Middleware executes in reverse registration order on the way in, and in registration order on the way out. The CORS middleware is registered first in app.py, so it is the outermost layer. Auto-discovered middlewares from singularity/middleware/ are registered after CORS.

Processing Order

Order	Layer	Source	Purpose
1	CORS	`app.py` (explicit)	Handle preflight requests, set `Access-Control-*` headers
2	Rate Limiter	`singularity/middleware/ratelimit.py` (auto-discovered)	Sliding-window per-IP throttling
3	Exception Handler	`singularity/middleware/exceptions.py` (auto-discovered)	Catch unhandled exceptions, return structured errors
4	JWTBearer	`singularity/security/security.py` (FastAPI dependency)	Validate Bearer tokens on protected routes

Middleware Auto-Discovery

The Manager discovers and registers middleware automatically by scanning the singularity/middleware/ directory. The process mirrors service discovery:

Scan directory -- The Manager lists all .py files in singularity/middleware/, skipping files starting with __.
Import module -- Each file is imported dynamically using importlib.import_module().
Find class -- The Manager looks for a class named exactly Middleware (not just any class -- the name must be Middleware).
Inspect signature -- If the Middleware class __init__ accepts an acquire parameter, the Acquire DI container is injected. Otherwise, only the app argument is passed.
Register -- The middleware is added to the FastAPI application via app.add_middleware().

Here is the relevant code from singularity/core/manager.py:

def register_middlewares(self) -> None:
    for mw_name in os.listdir(self.mws_dir):
        if mw_name.startswith("__"):
            continue
        mw_path = os.path.join(self.mws_dir, mw_name)
        if os.path.isfile(mw_path) and mw_name.endswith(".py"):
            mw_module_name = mw_name[:-3]
            mw_module_path = f"middlewares.{mw_module_name}"
            mw_module = importlib.import_module(mw_module_path)
            mw_class = getattr(mw_module, "Middleware", None)
            if mw_class:
                init_params = inspect.signature(mw_class.__init__).parameters
                if "acquire" in init_params:
                    self.app.add_middleware(mw_class, acquire=self.acquire)
                else:
                    self.app.add_middleware(mw_class)

To add a new middleware, create a Python file in singularity/middleware/ containing a class named Middleware that inherits from BaseHTTPMiddleware. It will be picked up automatically on the next server restart.

Built-in Middleware

Rate Limiter

The rate limiter in singularity/middleware/ratelimit.py implements a sliding-window algorithm that tracks request timestamps per client IP:

# singularity/middleware/ratelimit.py
from starlette.middleware.base import BaseHTTPMiddleware

class Middleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests: int = 100, window_size: int = 60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window_size = window_size
        self.ip_requests: dict[str, list[float]] = {}

    async def dispatch(self, request, call_next):
        client_ip = request.client.host if request.client else None
        if client_ip is None:
            raise HTTPException(status_code=400, detail="Invalid client IP")

        current_time = time.time()
        window_start = current_time - self.window_size

        # Filter out timestamps older than the window
        request_times = self.ip_requests.get(client_ip, [])
        request_times = [t for t in request_times if t > window_start]

        # Reject if over limit
        if len(request_times) >= self.max_requests:
            return JSONResponse(
                status_code=429,
                content={"detail": "Too many requests"},
            )

        # Record this request
        request_times.append(current_time)
        self.ip_requests[client_ip] = request_times

        return await call_next(request)

Configuration	Default	Description
`max_requests`	`100`	Maximum requests allowed per window
`window_size`	`60`	Sliding window size in seconds

The default rate limiter stores request timestamps in-memory. In a multi-worker deployment (e.g., gunicorn with multiple workers), each worker maintains its own counter. For distributed rate limiting, replace the in-memory store with Redis.

Exception Handler

The exception handler middleware in singularity/middleware/exceptions.py catches unhandled exceptions and returns structured error responses. It uses the Acquire container to access the logger:

# singularity/middleware/exceptions.py
class Middleware(BaseHTTPMiddleware):
    def __init__(self, app, acquire: Acquire):
        super().__init__(app)
        self.ws_manager = acquire.ws_manager
        self.logger = acquire.logger

    async def dispatch(self, request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception:
            exc_type, exc_value, exc_traceback = sys.exc_info()

            stack_trace = []
            for frame in traceback.extract_tb(exc_traceback):
                stack_trace.append({
                    "filename": frame.filename,
                    "line": frame.line,
                    "lineno": frame.lineno,
                    "name": frame.name,
                })

            error_info = {
                "error_type": exc_type.__name__,
                "error_message": str(exc_value),
                "traceback": stack_trace,
            }

            if settings.environment == "dev":
                self.logger.exception("Exception occurred:", error_info=error_info)
                return JSONResponse(status_code=500, content=error_info)
            return JSONResponse(status_code=500, content="Internal server error!!!")

Key behavior differences by environment:

Environment	Response Body	Logging
`dev`	Full error info with traceback	`logger.exception()` with stack trace
`beta` / `prod`	Generic `"Internal server error!!!"`	Same logging, sanitized response

Writing Custom Middleware

To add your own middleware, create a file in singularity/middleware/ with a class named Middleware. Here is a complete example of a request timing middleware:

# singularity/middleware/timing.py
import time

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint


class Middleware(BaseHTTPMiddleware):
    """Add X-Process-Time header to all responses."""

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start
        response.headers["X-Process-Time"] = f"{duration:.4f}"
        return response

If your middleware needs access to shared resources (database, settings, logger), accept acquire in __init__:

# singularity/middleware/audit_log.py
from starlette.middleware.base import BaseHTTPMiddleware
from singularity.core.acquire import Acquire


class Middleware(BaseHTTPMiddleware):
    """Log every request to the audit trail."""

    def __init__(self, app, acquire: Acquire):
        super().__init__(app)
        self.logger = acquire.logger
        self.settings = acquire.settings

    async def dispatch(self, request, call_next):
        self.logger.info(
            f"{request.method} {request.url.path}",
            client=request.client.host if request.client else "unknown",
            env=self.settings.environment,
        )
        response = await call_next(request)
        self.logger.info(
            f"Response {response.status_code} for {request.url.path}"
        )
        return response

The class must be named Middleware (exactly). The Manager looks for getattr(module, "Middleware", None) -- any other class name will be ignored during auto-discovery.

CORS Configuration

Cross-Origin Resource Sharing (CORS) is configured directly in app.py as the first middleware layer, before auto-discovered middleware:

# app.py
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # Allow all origins
    allow_credentials=True,    # Allow cookies and auth headers
    allow_methods=["*"],       # Allow all HTTP methods
    allow_headers=["*"],       # Allow all request headers
)

Parameter	Value	Description
`allow_origins`	`["*"]`	Origins permitted to make cross-origin requests. Restrict to specific domains in production.
`allow_credentials`	`True`	Allow cookies and `Authorization` headers in cross-origin requests.
`allow_methods`	`["*"]`	HTTP methods allowed (GET, POST, PUT, DELETE, etc.).
`allow_headers`	`["*"]`	Request headers allowed in cross-origin requests.

The default configuration allows all origins (["*"]). For production deployments, replace "*" with specific domain names to prevent unauthorized cross-origin access:

allow_origins=["https://app.example.com", "https://admin.example.com"]

JWT Authentication

The JWTBearer class in singularity/security/security.py provides token-based authentication for both HTTP and WebSocket endpoints. It extends FastAPI's HTTPBearer scheme.

How JWTBearer Works

# singularity/security/security.py
import jwt
from fastapi import HTTPException, Request, WebSocket
from fastapi.security import HTTPBearer
from singularity.config import settings


class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super().__init__(auto_error=auto_error)

    async def __call__(
        self,
        request: Request = None,
        websocket: WebSocket = None,
    ) -> Optional[dict[str, Any]]:
        if request:
            # HTTP: Extract from Authorization header
            credentials = await super().__call__(request)
            if not credentials or credentials.scheme != "Bearer":
                raise HTTPException(status_code=403, detail="Invalid authorization")
            try:
                payload = jwt.decode(
                    credentials.credentials,
                    settings.jwt_secret,
                    algorithms=["HS256"],
                )
                return payload
            except jwt.InvalidTokenError:
                raise HTTPException(status_code=403, detail="Invalid or expired token")

        elif websocket:
            # WebSocket: Extract from query parameter
            token = websocket.query_params.get("token")
            if not token:
                raise HTTPException(status_code=403, detail="Invalid authorization")
            try:
                payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
                return payload
            except jwt.InvalidTokenError:
                raise HTTPException(status_code=403, detail="Invalid or expired token")

        else:
            raise HTTPException(status_code=403, detail="Invalid authorization")

Protecting HTTP Endpoints

Use JWTBearer as a FastAPI dependency on routes that require authentication:

from fastapi import Depends
from singularity.security.jwt import JWTBearer

# Option 1: Require auth, access token payload
@router.get("/profile")
async def get_profile(token_data: dict = Depends(JWTBearer())):
    user_id = token_data["sub"]
    return {"user_id": user_id, "email": token_data.get("email")}

# Option 2: Require auth on all routes in a router
protected_router = APIRouter(
    dependencies=[Depends(JWTBearer())]
)

@protected_router.get("/settings")
async def get_settings():
    return {"theme": "dark"}

Protecting Services with Blacklists

Services can declare a blacklist attribute to control which RPC callers are blocked. Combined with JWTBearer, you can protect both HTTP and RPC access:

from singularity.core import Service
from singularity.security.jwt import JWTBearer
from fastapi import Depends

class AdminService(Service):
    rpc_exposed = True
    blacklist = ["public"]  # Block RPC calls from "public" service

    async def get(self, token_data: dict = Depends(JWTBearer())):
        # HTTP endpoint requires JWT
        return {"admin": True, "user": token_data["sub"]}

WebSocket Authentication

WebSocket connections cannot use the Authorization header (the WebSocket protocol does not support custom headers during the handshake). Instead, JWTBearer extracts the token from a query parameter:

ws://localhost:8000/api/v1/events?token=eyJhbGciOiJIUzI1NiIs...

The JWTBearer class detects whether it is being called with a Request or WebSocket object and handles each case accordingly.

WebSocket Connection Flow

Client connects with the JWT token as a query parameter: ws://host/api/v1/events?token=<jwt>.
JWTBearer validates the token using jwt.decode() with the HS256 algorithm and the application's jwt_secret.
On success, the decoded payload is returned to the endpoint handler, which can extract user_id, org_id, and permissions.
On failure, an HTTPException with status 403 is raised, and the WebSocket connection is rejected.
WebSocketManager takes over, accepting the connection, storing it indexed by channel, and routing messages based on organization and permissions.

Example: Authenticated WebSocket Endpoint

from singularity.core import Service
from singularity.security.jwt import JWTBearer
from fastapi import Depends, WebSocket

class EventsService(Service):
    http_exposed = ["ws=stream"]

    def __init__(self, acquire):
        super().__init__(acquire)
        self.ws_manager = acquire.ws_manager

    async def ws_stream(
        self,
        websocket: WebSocket,
        token_data: dict = Depends(JWTBearer()),
    ):
        user_id = token_data["sub"]
        org_id = token_data["org_id"]
        permissions = token_data.get("permissions", [])

        await self.ws_manager.connect(
            websocket=websocket,
            org_id=org_id,
            user_id=user_id,
            permissions=permissions,
        )

The WebSocketManager handles the persistent connection lifecycle after authentication: accepting the socket, routing messages based on permissions, and cleaning up on disconnect.

Security Checklist

CORS Origins

Replace allow_origins=["*"] with specific domains in production to prevent unauthorized cross-origin requests.

JWT Secret

Set a strong, unique JWT_SECRET in your .env file. The default "dummy_value" must never be used in production.

Rate Limiting

The default 100 requests per 60 seconds is a starting point. Adjust max_requests and window_size based on your traffic patterns. Consider Redis-backed limiting for multi-worker deployments.

Error Responses

The exception handler exposes full stack traces in dev mode. Verify that ENVIRONMENT is set to prod or beta in production to return sanitized error responses.

Middleware & Security

CORS Origins

JWT Secret

Rate Limiting

Error Responses

On this page