auth-fixes (#3)
## Summary
Resolve `{ENV_VAR}` placeholders in the OpenAPI spec before serving it. Replace the monolithic `x-introspect-url` extension with composable `x-server-url` + individual `x-*-path` fields so auth endpoints are configurable per-environment without hardcoding.
## Changes
- **`openapi_first/app.py`** — add `_resolve_spec_env_vars()` that replaces `{ENV_VAR}` patterns (e.g. `{AUTH_SERVER}`) with the corresponding OS environment variable before returning the spec JSON. Called in `__init__` after spec load.
- **`openapi_first/security.py`** — build the introspection URL dynamically from `x-server-url` + `x-introspect-path` extensions on the `bearerAuth` security scheme, instead of reading a single `x-introspect-url`.
## Migration
Existing specs using `x-introspect-url: "https://auth.example.com/introspect"` must switch to the new extension format:
```yaml
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
x-server-url: "{AUTH_SERVER}"
x-login-path: "/login"
x-register-path: "/register"
x-logout-path: "/logout"
x-me-path: "/me"
x-introspect-path: "/introspect"
Reviewed-on: #3
Co-authored-by: Vishesh 'ironeagle' Bangotra <aetoskia@gmail.com>
Co-committed-by: Vishesh 'ironeagle' Bangotra <aetoskia@gmail.com>
This commit is contained in:
145
openapi_first/security.py
Normal file
145
openapi_first/security.py
Normal file
@@ -0,0 +1,145 @@
|
||||
"""
|
||||
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":
|
||||
server_url = scheme.get("x-server-url", "")
|
||||
introspect_path = scheme.get("x-introspect-path", "/introspect")
|
||||
introspect_url = server_url + introspect_path if server_url else None
|
||||
return _make_bearer_dependency(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
|
||||
Reference in New Issue
Block a user