Merge branch 'feat/oeg-auth-changes' of... (f5188f34) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/core/authorization.py

+4 −10

Changes for src/open_exposure_gateway/core/authorization.py: 4 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		@@ -15,24 +15,20 @@ logger = structlog.get_logger(__name__)
		class OperationRequirements:
		role: str
		scope: Optional[str] = None
		# False where the spec lets client-credentials callers skip the scope (QoD).
		scope_required_for_two_legged: bool = True


		# Keyed by "{METHOD} {route path template}"; scopes are the ones each CAMARA spec declares.
		OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = {
		# Quality on Demand
		"POST /qod/v0/sessions": OperationRequirements(
		"app-provider", scope="qod-sessions-write", scope_required_for_two_legged=False
		),
		"POST /qod/v0/sessions": OperationRequirements("app-provider", scope="qod-sessions-write"),
		"GET /qod/v0/sessions/{sessionId}": OperationRequirements(
		"app-provider", scope="qod-sessions-read", scope_required_for_two_legged=False
		"app-provider", scope="qod-sessions-read"
		),
		"DELETE /qod/v0/sessions/{sessionId}": OperationRequirements(
		"app-provider", scope="qod-sessions-delete", scope_required_for_two_legged=False
		"app-provider", scope="qod-sessions-delete"
		),
		"POST /qod/v0/sessions/{sessionId}/extend": OperationRequirements(
		"app-provider", scope="qod-sessions-write", scope_required_for_two_legged=False
		"app-provider", scope="qod-sessions-write"
		),
		"GET /qod/v0/qos-profiles": OperationRequirements("app-provider"),
		"GET /qod/v0/qos-profiles/{name}": OperationRequirements("app-provider"),
		@@ -156,8 +152,6 @@ def require_scope(
		requirements = _requirements_for(request)
		if requirements.scope is None:
		return caller
		if not requirements.scope_required_for_two_legged:
		return caller
		if requirements.scope not in caller.scopes:
		if not settings.keycloak_settings.enforcement_enabled:
		logger.warning(

tests/unit/test_authorization.py

+34 −17

Changes for tests/unit/test_authorization.py: 34 added lines, 17 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,6 +7,7 @@ produces."""
		from typing import Optional

		import pytest
		import structlog
		from fastapi import Request
		from fastapi.routing import APIRoute
		from fastapi.testclient import TestClient
		@@ -21,16 +22,17 @@ from open_exposure_gateway.core.exceptions import ForbiddenException
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context
		from open_exposure_gateway.main import app

		_QOD_ROUTE = (
		"POST",
		"/qod/v0/sessions",
		) # scope: qod-sessions-write, two-legged exempt
		_QOD_ROUTE = ("POST", "/qod/v0/sessions")
		_QOD_SESSION_ROUTES = [
		("POST", "/qod/v0/sessions", "qod-sessions-write"),
		("GET", "/qod/v0/sessions/{sessionId}", "qod-sessions-read"),
		("DELETE", "/qod/v0/sessions/{sessionId}", "qod-sessions-delete"),
		("POST", "/qod/v0/sessions/{sessionId}/extend", "qod-sessions-write"),
		]
		_LOCATION_RETRIEVAL_ROUTE = (
		"POST",
		"/location-retrieval/v0.5/retrieve",
		) # scope: location-retrieval:read
		# EAM's openId scheme requires its scope -- unlike QoD's routes above.
		_SCOPE_ALWAYS_REQUIRED_ROUTE = ("GET", "/edge-application-management/v0.1alpha1/apps")
		_FEDERATION_ROUTE = ("POST", "/platform/v1/federation/partners/{partnerOpId}/federations")


		@@ -131,20 +133,35 @@ class TestRequireScope:
		caller = _caller(roles=["app-provider"], scopes=["location-retrieval:read"])
		assert require_scope(request, caller, _settings()) is caller

		def test_caller_is_exempt_on_a_qod_route(self) -> None:
		# QoD's spec offers oAuth2ClientCredentials (no scope) as an alternative to
		# threeLegged+scope -- a caller needs no scope at all here.
		request = _request(*_QOD_ROUTE)
		caller = _caller(roles=["app-provider"], scopes=[])
		@pytest.mark.parametrize(("method", "path", "scope"), _QOD_SESSION_ROUTES)
		def test_qod_session_route_without_its_scope_is_rejected(
		self, method: str, path: str, scope: str
		) -> None:
		request = _request(method, path)
		other_scopes = [s for _, _, s in _QOD_SESSION_ROUTES if s != scope]
		caller = _caller(roles=["app-provider"], scopes=other_scopes)
		with pytest.raises(ForbiddenException):
		require_scope(request, caller, _settings())

		@pytest.mark.parametrize(("method", "path", "scope"), _QOD_SESSION_ROUTES)
		def test_qod_session_route_with_its_scope_is_allowed(
		self, method: str, path: str, scope: str
		) -> None:
		request = _request(method, path)
		caller = _caller(roles=["app-provider"], scopes=[scope])
		assert require_scope(request, caller, _settings()) is caller

		def test_caller_still_needs_scope_on_an_eam_route(self) -> None:
		# EAM's openIdConnect scheme has no client-credentials alternative: the scope
		# is required.
		request = _request(*_SCOPE_ALWAYS_REQUIRED_ROUTE)
		def test_missing_qod_scope_is_logged_not_rejected_when_enforcement_disabled(self) -> None:
		request = _request(*_QOD_ROUTE)
		caller = _caller(roles=["app-provider"], scopes=[])
		with pytest.raises(ForbiddenException):
		require_scope(request, caller, _settings())
		with structlog.testing.capture_logs() as logs:
		result = require_scope(request, caller, _settings(enforcement_enabled=False))
		assert result is caller
		assert {
		"event": "authz_would_reject",
		"reason": "missing_scope",
		"required_scope": "qod-sessions-write",
		}.items() <= logs[0].items()

		def test_operation_with_no_declared_scope_is_allowed(self) -> None:
		# QoD's qos-profiles reads declare no scope in the spec at all.