Loading src/open_exposure_gateway/dependencies.py +30 −8 Changes for src/open_exposure_gateway/dependencies.py: 30 added lines, 8 removed lines. Original line number Diff line number Diff line Loading @@ -70,6 +70,15 @@ class CallerContext: app_provider_id: str roles: list[str] scopes: list[str] client_id: Optional[str] = None def caller_log_fields(caller: CallerContext) -> dict[str, Optional[str]]: return { "tenant_id": caller.tenant_id, "app_provider_id": caller.app_provider_id, "client_id": caller.client_id, } def get_app_state(request: Request) -> AppState: Loading @@ -93,14 +102,23 @@ _bearer_scheme = HTTPBearer(auto_error=False) logger = structlog.get_logger(__name__) def _unverified_caller_context(x_correlator: Optional[str]) -> CallerContext: def _remember_caller(request: Request, caller: CallerContext) -> CallerContext: request.state.caller = caller structlog.contextvars.bind_contextvars(**caller_log_fields(caller)) return caller def _unverified_caller_context(request: Request) -> CallerContext: """Stand-in caller used only while `enforcement_enabled` is False.""" return CallerContext( x_correlator=x_correlator, return _remember_caller( request, CallerContext( x_correlator=x_correlator_header(request), tenant_id="unverified", app_provider_id="unverified", roles=[], scopes=[], ), ) Loading @@ -117,7 +135,7 @@ async def get_caller_context( if enforcement_enabled: raise UnauthorizedException(message="Missing bearer token") logger.warning("auth_would_reject", reason="missing_bearer_token", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) try: decoded = await token_verifier.verify(credentials.credentials) Loading @@ -125,12 +143,12 @@ async def get_caller_context( if enforcement_enabled: raise logger.warning("auth_would_reject", reason="invalid_token", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) except DownstreamServiceException: if enforcement_enabled: raise logger.warning("auth_would_reject", reason="jwks_unreachable", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) claims = decoded.claims subject = claims.get("sub") Loading @@ -141,17 +159,21 @@ async def get_caller_context( raise UnauthorizedException(message="Access token has no subject") raise ForbiddenException(message="Access token has no roles") logger.warning("auth_would_reject", reason="missing_claims", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) # A user without an `organization` is a tenant of their own. tenant_id: str = claims.get("organization") or subject return CallerContext( return _remember_caller( request, CallerContext( x_correlator=x_correlator_header(request), tenant_id=tenant_id, app_provider_id=subject, roles=roles, scopes=claims.get("scope", "").split(), client_id=claims.get("azp"), ), ) Loading src/open_exposure_gateway/main.py +24 −0 Changes for src/open_exposure_gateway/main.py: 24 added lines, 0 removed lines. Original line number Diff line number Diff line import asyncio import time from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager, suppress from datetime import timedelta Loading Loading @@ -61,6 +62,7 @@ from open_exposure_gateway.application.services.quality_on_demand_service import ) from open_exposure_gateway.core.config import get_settings from open_exposure_gateway.core.logging import configure_logging from open_exposure_gateway.dependencies import caller_log_fields from open_exposure_gateway.domain.edge_application_management import ( SRMOperationCompleted, Subject, Loading Loading @@ -419,6 +421,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: await db_engine.dispose() logger.info("Database engine disposed") _PROBE_PATHS = frozenset({"/platform/healthz", "/platform/readyz"}) def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: """Build the application. Loading Loading @@ -448,6 +451,27 @@ def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: response.headers["x-correlator"] = x_correlator return response @app.middleware("http") async def log_request( request: Request, call_next: Callable[[Request], Awaitable[Response]] ) -> Response: started = time.perf_counter() response = await call_next(request) logger = structlog.get_logger() log = logger.debug if request.url.path in _PROBE_PATHS else logger.info caller = getattr(request.state, "caller", None) route = request.scope.get("route") log( "http_request", method=request.method, path=getattr(route, "path", request.url.path), status=response.status_code, duration_ms=round((time.perf_counter() - started) * 1000, 1), x_correlator=x_correlator_header(request), **(caller_log_fields(caller) if caller else {}), ) return response app.include_router(health_router, prefix="/platform") app.include_router(edge_application_management_router) app.include_router(quality_on_demand_router) Loading tests/unit/test_caller_context.py +36 −0 Changes for tests/unit/test_caller_context.py: 36 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -3,9 +3,11 @@ caller's tenant_id/app_provider_id, and which missing claims reject the request. JWT validation itself is covered by test_core_auth.py; here the verifier is a stub returning fixed claims.""" from collections.abc import Iterator from typing import Any import pytest import structlog from fastapi import Request from fastapi.security import HTTPAuthorizationCredentials Loading Loading @@ -88,6 +90,40 @@ class TestTenantId: assert caller.app_provider_id == "user-123" class TestClientId: async def test_client_id_is_the_azp_claim(self) -> None: caller = await _caller(_claims(azp="acme-backend")) assert caller.client_id == "acme-backend" async def test_client_id_is_none_without_azp(self) -> None: caller = await _caller(_claims()) assert caller.client_id is None class TestLogContext: @pytest.fixture(autouse=True) def _clean_contextvars(self) -> Iterator[None]: structlog.contextvars.clear_contextvars() yield structlog.contextvars.clear_contextvars() async def test_caller_is_bound_for_later_log_lines(self) -> None: await _caller(_claims(azp="acme-backend")) assert structlog.contextvars.get_contextvars() == { "tenant_id": "acme-labs", "app_provider_id": "user-123", "client_id": "acme-backend", } async def test_unverified_caller_is_bound_when_enforcement_disabled(self) -> None: await _caller(_claims(), enforcement_enabled=False, with_token=False) assert structlog.contextvars.get_contextvars() == { "tenant_id": "unverified", "app_provider_id": "unverified", "client_id": None, } class TestMissingClaims: async def test_missing_sub_is_rejected(self) -> None: with pytest.raises(UnauthorizedException): Loading tests/unit/test_request_logging.py 0 → 100644 +86 −0 Changes for tests/unit/test_request_logging.py: 86 added lines, 0 removed lines. Original line number Diff line number Diff line """Flow tests for the request log line: one `http_request` event per request, carrying the caller resolved by get_caller_context.""" from collections.abc import Iterator, MutableMapping from typing import Any import pytest import structlog from fastapi.testclient import TestClient from open_exposure_gateway.core.auth.tokens import DecodedClaims from open_exposure_gateway.dependencies import get_caller_context, get_token_verifier from open_exposure_gateway.main import app from tests.unit.fakes import FakeTokenVerifier _Logs = list[MutableMapping[str, Any]] _CLAIMS = { "sub": "user-123", "realm_access": {"roles": ["app-provider"]}, "organization": "acme-labs", "azp": "acme-backend", } class _StubVerifier(FakeTokenVerifier): async def verify(self, token: str) -> DecodedClaims: return DecodedClaims(claims=_CLAIMS) @pytest.fixture() def client(api_client: TestClient) -> TestClient: app.dependency_overrides.pop(get_caller_context) app.dependency_overrides[get_token_verifier] = lambda: _StubVerifier() return api_client @pytest.fixture() def logs() -> Iterator[_Logs]: with structlog.testing.capture_logs() as captured: yield captured def _request_lines(logs: _Logs) -> _Logs: return [line for line in logs if line["event"] == "http_request"] class TestRequestLog: def test_request_is_logged_with_its_caller(self, client: TestClient, logs: _Logs) -> None: response = client.get( "/qod/v0/qos-profiles/QOS_E", headers={"Authorization": "Bearer token", "x-correlator": "corr-1"}, ) [line] = _request_lines(logs) assert ( line.items() >= { "method": "GET", "path": "/qod/v0/qos-profiles/{name}", "status": response.status_code, "x_correlator": "corr-1", "tenant_id": "acme-labs", "app_provider_id": "user-123", "client_id": "acme-backend", }.items() ) assert "duration_ms" in line def test_caller_does_not_leak_into_the_next_request( self, client: TestClient, logs: _Logs ) -> None: client.get("/qod/v0/qos-profiles", headers={"Authorization": "Bearer token"}) response = client.get("/qod/v0/qos-profiles") assert response.status_code == 401 unauthenticated = _request_lines(logs)[1] assert unauthenticated["status"] == 401 assert "tenant_id" not in unauthenticated def test_health_probes_are_logged_at_debug(self, client: TestClient, logs: _Logs) -> None: client.get("/platform/healthz") client.get("/platform/readyz") client.get("/qod/v0/qos-profiles", headers={"Authorization": "Bearer token"}) assert [line["log_level"] for line in _request_lines(logs)] == ["debug", "debug", "info"] Loading
src/open_exposure_gateway/dependencies.py +30 −8 Changes for src/open_exposure_gateway/dependencies.py: 30 added lines, 8 removed lines. Original line number Diff line number Diff line Loading @@ -70,6 +70,15 @@ class CallerContext: app_provider_id: str roles: list[str] scopes: list[str] client_id: Optional[str] = None def caller_log_fields(caller: CallerContext) -> dict[str, Optional[str]]: return { "tenant_id": caller.tenant_id, "app_provider_id": caller.app_provider_id, "client_id": caller.client_id, } def get_app_state(request: Request) -> AppState: Loading @@ -93,14 +102,23 @@ _bearer_scheme = HTTPBearer(auto_error=False) logger = structlog.get_logger(__name__) def _unverified_caller_context(x_correlator: Optional[str]) -> CallerContext: def _remember_caller(request: Request, caller: CallerContext) -> CallerContext: request.state.caller = caller structlog.contextvars.bind_contextvars(**caller_log_fields(caller)) return caller def _unverified_caller_context(request: Request) -> CallerContext: """Stand-in caller used only while `enforcement_enabled` is False.""" return CallerContext( x_correlator=x_correlator, return _remember_caller( request, CallerContext( x_correlator=x_correlator_header(request), tenant_id="unverified", app_provider_id="unverified", roles=[], scopes=[], ), ) Loading @@ -117,7 +135,7 @@ async def get_caller_context( if enforcement_enabled: raise UnauthorizedException(message="Missing bearer token") logger.warning("auth_would_reject", reason="missing_bearer_token", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) try: decoded = await token_verifier.verify(credentials.credentials) Loading @@ -125,12 +143,12 @@ async def get_caller_context( if enforcement_enabled: raise logger.warning("auth_would_reject", reason="invalid_token", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) except DownstreamServiceException: if enforcement_enabled: raise logger.warning("auth_would_reject", reason="jwks_unreachable", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) claims = decoded.claims subject = claims.get("sub") Loading @@ -141,17 +159,21 @@ async def get_caller_context( raise UnauthorizedException(message="Access token has no subject") raise ForbiddenException(message="Access token has no roles") logger.warning("auth_would_reject", reason="missing_claims", path=request.url.path) return _unverified_caller_context(x_correlator_header(request)) return _unverified_caller_context(request) # A user without an `organization` is a tenant of their own. tenant_id: str = claims.get("organization") or subject return CallerContext( return _remember_caller( request, CallerContext( x_correlator=x_correlator_header(request), tenant_id=tenant_id, app_provider_id=subject, roles=roles, scopes=claims.get("scope", "").split(), client_id=claims.get("azp"), ), ) Loading
src/open_exposure_gateway/main.py +24 −0 Changes for src/open_exposure_gateway/main.py: 24 added lines, 0 removed lines. Original line number Diff line number Diff line import asyncio import time from collections.abc import AsyncGenerator, Awaitable, Callable from contextlib import asynccontextmanager, suppress from datetime import timedelta Loading Loading @@ -61,6 +62,7 @@ from open_exposure_gateway.application.services.quality_on_demand_service import ) from open_exposure_gateway.core.config import get_settings from open_exposure_gateway.core.logging import configure_logging from open_exposure_gateway.dependencies import caller_log_fields from open_exposure_gateway.domain.edge_application_management import ( SRMOperationCompleted, Subject, Loading Loading @@ -419,6 +421,7 @@ async def default_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: await db_engine.dispose() logger.info("Database engine disposed") _PROBE_PATHS = frozenset({"/platform/healthz", "/platform/readyz"}) def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: """Build the application. Loading Loading @@ -448,6 +451,27 @@ def create_app(lifespan: Optional[Lifespan[FastAPI]] = None) -> FastAPI: response.headers["x-correlator"] = x_correlator return response @app.middleware("http") async def log_request( request: Request, call_next: Callable[[Request], Awaitable[Response]] ) -> Response: started = time.perf_counter() response = await call_next(request) logger = structlog.get_logger() log = logger.debug if request.url.path in _PROBE_PATHS else logger.info caller = getattr(request.state, "caller", None) route = request.scope.get("route") log( "http_request", method=request.method, path=getattr(route, "path", request.url.path), status=response.status_code, duration_ms=round((time.perf_counter() - started) * 1000, 1), x_correlator=x_correlator_header(request), **(caller_log_fields(caller) if caller else {}), ) return response app.include_router(health_router, prefix="/platform") app.include_router(edge_application_management_router) app.include_router(quality_on_demand_router) Loading
tests/unit/test_caller_context.py +36 −0 Changes for tests/unit/test_caller_context.py: 36 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -3,9 +3,11 @@ caller's tenant_id/app_provider_id, and which missing claims reject the request. JWT validation itself is covered by test_core_auth.py; here the verifier is a stub returning fixed claims.""" from collections.abc import Iterator from typing import Any import pytest import structlog from fastapi import Request from fastapi.security import HTTPAuthorizationCredentials Loading Loading @@ -88,6 +90,40 @@ class TestTenantId: assert caller.app_provider_id == "user-123" class TestClientId: async def test_client_id_is_the_azp_claim(self) -> None: caller = await _caller(_claims(azp="acme-backend")) assert caller.client_id == "acme-backend" async def test_client_id_is_none_without_azp(self) -> None: caller = await _caller(_claims()) assert caller.client_id is None class TestLogContext: @pytest.fixture(autouse=True) def _clean_contextvars(self) -> Iterator[None]: structlog.contextvars.clear_contextvars() yield structlog.contextvars.clear_contextvars() async def test_caller_is_bound_for_later_log_lines(self) -> None: await _caller(_claims(azp="acme-backend")) assert structlog.contextvars.get_contextvars() == { "tenant_id": "acme-labs", "app_provider_id": "user-123", "client_id": "acme-backend", } async def test_unverified_caller_is_bound_when_enforcement_disabled(self) -> None: await _caller(_claims(), enforcement_enabled=False, with_token=False) assert structlog.contextvars.get_contextvars() == { "tenant_id": "unverified", "app_provider_id": "unverified", "client_id": None, } class TestMissingClaims: async def test_missing_sub_is_rejected(self) -> None: with pytest.raises(UnauthorizedException): Loading
tests/unit/test_request_logging.py 0 → 100644 +86 −0 Changes for tests/unit/test_request_logging.py: 86 added lines, 0 removed lines. Original line number Diff line number Diff line """Flow tests for the request log line: one `http_request` event per request, carrying the caller resolved by get_caller_context.""" from collections.abc import Iterator, MutableMapping from typing import Any import pytest import structlog from fastapi.testclient import TestClient from open_exposure_gateway.core.auth.tokens import DecodedClaims from open_exposure_gateway.dependencies import get_caller_context, get_token_verifier from open_exposure_gateway.main import app from tests.unit.fakes import FakeTokenVerifier _Logs = list[MutableMapping[str, Any]] _CLAIMS = { "sub": "user-123", "realm_access": {"roles": ["app-provider"]}, "organization": "acme-labs", "azp": "acme-backend", } class _StubVerifier(FakeTokenVerifier): async def verify(self, token: str) -> DecodedClaims: return DecodedClaims(claims=_CLAIMS) @pytest.fixture() def client(api_client: TestClient) -> TestClient: app.dependency_overrides.pop(get_caller_context) app.dependency_overrides[get_token_verifier] = lambda: _StubVerifier() return api_client @pytest.fixture() def logs() -> Iterator[_Logs]: with structlog.testing.capture_logs() as captured: yield captured def _request_lines(logs: _Logs) -> _Logs: return [line for line in logs if line["event"] == "http_request"] class TestRequestLog: def test_request_is_logged_with_its_caller(self, client: TestClient, logs: _Logs) -> None: response = client.get( "/qod/v0/qos-profiles/QOS_E", headers={"Authorization": "Bearer token", "x-correlator": "corr-1"}, ) [line] = _request_lines(logs) assert ( line.items() >= { "method": "GET", "path": "/qod/v0/qos-profiles/{name}", "status": response.status_code, "x_correlator": "corr-1", "tenant_id": "acme-labs", "app_provider_id": "user-123", "client_id": "acme-backend", }.items() ) assert "duration_ms" in line def test_caller_does_not_leak_into_the_next_request( self, client: TestClient, logs: _Logs ) -> None: client.get("/qod/v0/qos-profiles", headers={"Authorization": "Bearer token"}) response = client.get("/qod/v0/qos-profiles") assert response.status_code == 401 unauthenticated = _request_lines(logs)[1] assert unauthenticated["status"] == 401 assert "tenant_id" not in unauthenticated def test_health_probes_are_logged_at_debug(self, client: TestClient, logs: _Logs) -> None: client.get("/platform/healthz") client.get("/platform/readyz") client.get("/qod/v0/qos-profiles", headers={"Authorization": "Bearer token"}) assert [line["log_level"] for line in _request_lines(logs)] == ["debug", "debug", "info"]