add spec-driven auth dependency injection

- security.py: parse securitySchemes, resolve {ENV_VAR} from env,
  generate FastAPI Depends for Bearer JWT introspection
- app.py: extract schemes, build deps, pass to binder at init
- binder.py: inject Depends() per operation based on spec's security
- __init__.py: export security module
- pyproject.toml: add httpx dependency
This commit is contained in:
2026-07-17 21:00:23 +05:30
parent 7d075b3904
commit a2169b2bb1
5 changed files with 186 additions and 11 deletions

142
openapi_first/security.py Normal file
View File

@@ -0,0 +1,142 @@
"""
OpenAPI security scheme parsing and auto-generated auth dependencies.
Reads `securitySchemes` and per-operation `security` from an OpenAPI spec,
resolves `{ENV_VAR}` placeholders in `x-` extension fields, and generates
FastAPI dependencies for token validation (e.g., Bearer JWT introspection).
"""
import os
import re
from typing import Any, Callable
import httpx
from fastapi import Depends, Request, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
_env_pattern = re.compile(r"\{(\w+)\}")
def _resolve_env(value: str) -> str:
"""Replace {ENV_VAR} placeholders with values from os.environ."""
def _replace(m: re.Match) -> str:
return os.environ.get(m.group(1), "")
return _env_pattern.sub(_replace, value)
def _resolve_scheme(scheme: dict) -> dict:
"""Recursively resolve env vars in all string-valued fields of a scheme."""
resolved = {}
for key, value in scheme.items():
if isinstance(value, str):
resolved[key] = _resolve_env(value)
elif isinstance(value, dict):
resolved[key] = _resolve_scheme(value)
else:
resolved[key] = value
return resolved
def parse_security_schemes(spec: dict) -> dict[str, dict]:
"""Extract and resolve environment variables in security schemes."""
raw = spec.get("components", {}).get("securitySchemes", {})
return {name: _resolve_scheme(scheme) for name, scheme in raw.items()}
def _make_bearer_dependency(introspect_url: str | None) -> Callable:
"""
Create a FastAPI dependency that validates a Bearer JWT.
If *introspect_url* is provided the dependency calls that endpoint
with ``{"token": "<token>"}`` and expects ``{"active": true, "user": …}``
back. The resolved user dict is stored on ``request.state.user``.
Without an introspection URL the dependency only extracts the raw token
and stores it on ``request.state.token`` — no remote validation.
"""
bearer = HTTPBearer(auto_error=False)
async def _bearer_dep(
request: Request,
credentials: HTTPAuthorizationCredentials | None = Depends(bearer),
) -> None:
if credentials is None:
raise HTTPException(status_code=401, detail="Not authenticated")
token = credentials.credentials
if introspect_url:
try:
async with httpx.AsyncClient(timeout=5) as client:
resp = await client.post(
introspect_url,
json={"token": token},
)
except httpx.RequestError:
raise HTTPException(
status_code=503,
detail="Authentication service unavailable",
)
if resp.status_code != 200:
raise HTTPException(status_code=401, detail="Invalid or expired token")
body = resp.json()
if not body.get("active", False):
raise HTTPException(status_code=401, detail="Invalid or expired token")
request.state.user = body.get("user")
else:
request.state.token = token
return _bearer_dep
def _build_dependency(scheme_name: str, scheme: dict) -> Callable | None:
"""Return a FastAPI dependency callable for *scheme*, or ``None``."""
scheme_type = scheme.get("type")
if scheme_type == "http" and scheme.get("scheme") == "bearer":
return _make_bearer_dependency(scheme.get("x-introspect-url"))
return None
def make_security_dependencies(
spec: dict,
security_schemes: dict[str, dict],
) -> dict[str, list[Any]]:
"""
Build a mapping of ``METHOD:/path`` → list of ``Depends(...)``.
The effective security for each operation is resolved by:
1. Using the operation-level ``security`` field if present.
2. Falling back to the top-level ``security`` field.
3. An empty list means *no auth required* for that operation.
"""
global_security = spec.get("security", [])
paths = spec.get("paths", {})
result: dict[str, list[Any]] = {}
for path, methods in paths.items():
for http_method, operation in methods.items():
if http_method.startswith("x-"):
continue
op_security = operation.get("security", global_security)
depends_list: list[Any] = []
for sec_req in op_security:
for scheme_name in sec_req:
scheme = security_schemes.get(scheme_name)
if scheme:
dep = _build_dependency(scheme_name, scheme)
if dep:
depends_list.append(Depends(dep))
key = f"{http_method.upper()}:{path}"
result[key] = depends_list
return result