fix: improve error handling for missing claims and roles in caller context (d5187566) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/dependencies.py

+9 −3

Changes for src/open_exposure_gateway/dependencies.py: 9 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -43,7 +43,11 @@ from open_exposure_gateway.application.services.quality_on_demand_service import
		)
		from open_exposure_gateway.core.auth.tokens import TokenFlow
		from open_exposure_gateway.core.config import Settings, get_settings
		from open_exposure_gateway.core.exceptions import DownstreamServiceException, UnauthorizedException
		from open_exposure_gateway.core.exceptions import (
		DownstreamServiceException,
		ForbiddenException,
		UnauthorizedException,
		)
		from open_exposure_gateway.core.state import AppState
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		@@ -136,10 +140,12 @@ async def get_caller_context(

		claims = decoded.claims
		subject = claims.get("sub")
		roles = claims.get("realm_access", {}).get("roles")
		roles = (claims.get("realm_access") or {}).get("roles")
		if not subject or not roles:
		if enforcement_enabled:
		raise UnauthorizedException(message="Access token missing required claims")
		if not subject:
		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))

tests/unit/test_authorization.py

+13 −1

Changes for tests/unit/test_authorization.py: 13 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -8,6 +8,7 @@ from typing import Optional

		import pytest
		from fastapi import Request
		from fastapi.testclient import TestClient

		from open_exposure_gateway.core.auth.tokens import TokenFlow
		from open_exposure_gateway.core.authorization import (
		@@ -17,7 +18,8 @@ from open_exposure_gateway.core.authorization import (
		)
		from open_exposure_gateway.core.config import KeycloakSettings, Settings
		from open_exposure_gateway.core.exceptions import ForbiddenException
		from open_exposure_gateway.dependencies import CallerContext
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context
		from open_exposure_gateway.main import app

		_TWO_LEGGED_ONLY_ROUTE = (
		"POST",
		@@ -166,3 +168,13 @@ class TestRequireScope:
		caller = _caller(roles=["app-provider"], flow=TokenFlow.THREE_LEGGED, scopes=[])
		result = require_scope(request, caller, _settings(enforcement_enabled=False))
		assert result is caller


		class TestRouterWithoutToken:
		def test_request_without_token_is_401(self, api_client: TestClient) -> None:
		app.dependency_overrides.pop(get_caller_context)

		response = api_client.get("/edge-application-management/v0.1alpha1/edge-cloud-zones")

		assert response.status_code == 401
		assert response.json()["code"] == "UNAUTHENTICATED"

tests/unit/test_caller_context.py

+65 −7

Changes for tests/unit/test_caller_context.py: 65 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -11,15 +11,22 @@ from fastapi.security import HTTPAuthorizationCredentials

		from open_exposure_gateway.core.auth.tokens import DecodedClaims
		from open_exposure_gateway.core.config import KeycloakSettings, Settings
		from open_exposure_gateway.core.exceptions import UnauthorizedException
		from open_exposure_gateway.core.exceptions import (
		DownstreamServiceException,
		ForbiddenException,
		UnauthorizedException,
		)
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context


		class _StubVerifier:
		def __init__(self, claims: dict[str, Any]) -> None:
		def __init__(self, claims: dict[str, Any], error: Exception \| None = None) -> None:
		self._claims = claims
		self._error = error

		async def verify(self, token: str) -> DecodedClaims:
		if self._error is not None:
		raise self._error
		return DecodedClaims(claims=self._claims)


		@@ -48,12 +55,21 @@ def _claims(**overrides: Any) -> dict[str, Any]:
		return {k: v for k, v in claims.items() if v is not None}


		async def _caller(claims: dict[str, Any], enforcement_enabled: bool = True) -> CallerContext:
		async def _caller(
		claims: dict[str, Any],
		enforcement_enabled: bool = True,
		with_token: bool = True,
		verify_error: Exception \| None = None,
		) -> CallerContext:
		return await get_caller_context(
		request=_request(),
		x_correlator=None,
		credentials=HTTPAuthorizationCredentials(scheme="Bearer", credentials="token"),
		token_verifier=_StubVerifier(claims), # type: ignore[arg-type]
		credentials=(
		HTTPAuthorizationCredentials(scheme="Bearer", credentials="token")
		if with_token
		else None
		),
		token_verifier=_StubVerifier(claims, verify_error), # type: ignore[arg-type]
		settings=Settings(
		keycloak_settings=KeycloakSettings(enforcement_enabled=enforcement_enabled)
		),
		@@ -77,10 +93,52 @@ class TestMissingClaims:
		with pytest.raises(UnauthorizedException):
		await _caller(_claims(sub=None))

		async def test_missing_roles_is_rejected(self) -> None:
		with pytest.raises(UnauthorizedException):
		async def test_missing_roles_is_forbidden(self) -> None:
		with pytest.raises(ForbiddenException):
		await _caller(_claims(realm_access=None))

		async def test_null_realm_access_is_forbidden(self) -> None:
		with pytest.raises(ForbiddenException):
		await _caller({"sub": "user-123", "realm_access": None})

		async def test_missing_sub_is_unverified_when_enforcement_disabled(self) -> None:
		caller = await _caller(_claims(sub=None), enforcement_enabled=False)
		assert caller.tenant_id == "unverified"


		class TestTokenRejected:
		@pytest.mark.parametrize(
		("with_token", "verify_error", "expected"),
		[
		(False, None, UnauthorizedException),
		(True, UnauthorizedException(), UnauthorizedException),
		(True, DownstreamServiceException(), DownstreamServiceException),
		],
		ids=["missing_token", "invalid_token", "keycloak_unreachable"],
		)
		async def test_is_raised_when_enforcement_enabled(
		self, with_token: bool, verify_error: Exception \| None, expected: type[Exception]
		) -> None:
		with pytest.raises(expected):
		await _caller(_claims(), with_token=with_token, verify_error=verify_error)

		@pytest.mark.parametrize(
		("with_token", "verify_error"),
		[
		(False, None),
		(True, UnauthorizedException()),
		(True, DownstreamServiceException()),
		],
		ids=["missing_token", "invalid_token", "keycloak_unreachable"],
		)
		async def test_is_unverified_when_enforcement_disabled(
		self, with_token: bool, verify_error: Exception \| None
		) -> None:
		caller = await _caller(
		_claims(),
		enforcement_enabled=False,
		with_token=with_token,
		verify_error=verify_error,
		)
		assert caller.tenant_id == "unverified"
		assert caller.roles == []