feat: enforce role and three-legged checks per CAMARA route (c46f9dee) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/edge_application_management/v0_1_0_alpha_1/router.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -19,6 +19,7 @@ from open_exposure_gateway.api.camara.edge_application_management.v0_1_0_alpha_1
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		)
		from open_exposure_gateway.core.authorization import require_role, require_three_legged
		from open_exposure_gateway.core.exceptions import (
		AbortedException,
		AlreadyExistsException,
		@@ -42,7 +43,12 @@ from open_exposure_gateway.schemas.common import ErrorInfo
		# by this package, {apiRoot} by the deployment.
		BASE_PATH = "/edge-application-management/v0.1alpha1"

		router = APIRouter(prefix=BASE_PATH)
		# require_role/require_three_legged apply to every route in this router, looking each
		# request's requirements up in the one explicit table (core/authorization.py) -- so
		# authorization is enforced identically here without touching individual handlers.
		router = APIRouter(
		prefix=BASE_PATH, dependencies=[Depends(require_role), Depends(require_three_legged)]
		)

		EdgeAppService = Annotated[EdgeApplicationManagementService, Depends(get_edge_app_service)]
		Caller = Annotated[CallerContext, Depends(get_caller_context)]

src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/router.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ from open_exposure_gateway.api.camara.location_retrieval.v0_5_0.schemas import (
		from open_exposure_gateway.application.services.location_retrieval_service import (
		LocationRetrievalService,
		)
		from open_exposure_gateway.core.authorization import require_role, require_three_legged
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		DownstreamServiceException,
		@@ -30,7 +31,12 @@ from open_exposure_gateway.schemas.common import ErrorInfo
		# v0.5.0), mounted bare per ADR-0020.
		BASE_PATH = "/location-retrieval/v0.5"

		router = APIRouter(prefix=BASE_PATH)
		# require_role/require_three_legged apply to every route in this router, looking each
		# request's requirements up in the one explicit table (core/authorization.py) -- so
		# authorization is enforced identically here without touching individual handlers.
		router = APIRouter(
		prefix=BASE_PATH, dependencies=[Depends(require_role), Depends(require_three_legged)]
		)

		LocationService = Annotated[LocationRetrievalService, Depends(get_location_retrieval_service)]
		Caller = Annotated[CallerContext, Depends(get_caller_context)]

src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py

+7 −1

Original line number	Diff line number	Diff line
		@@ -11,6 +11,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import (
		from open_exposure_gateway.application.services.quality_on_demand_service import (
		QualityOnDemandService,
		)
		from open_exposure_gateway.core.authorization import require_role, require_three_legged
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		@@ -26,7 +27,12 @@ from open_exposure_gateway.schemas.common import ErrorInfo
		# CAMARA base path: the spec serves at {apiRoot}/qod/v0 (wire version of v0.10.1).
		BASE_PATH = "/qod/v0"

		router = APIRouter(prefix=BASE_PATH)
		# require_role/require_three_legged apply to every route in this router, looking each
		# request's requirements up in the one explicit table (core/authorization.py) -- so
		# authorization is enforced identically here without touching individual handlers.
		router = APIRouter(
		prefix=BASE_PATH, dependencies=[Depends(require_role), Depends(require_three_legged)]
		)

		QoDService = Annotated[QualityOnDemandService, Depends(get_qod_service)]
		Caller = Annotated[CallerContext, Depends(get_caller_context)]

src/open_exposure_gateway/core/authorization.py

0 → 100644

+120 −0

Original line number	Diff line number	Diff line
		from dataclasses import dataclass
		from typing import Annotated

		from fastapi import Depends, Request

		from open_exposure_gateway.core.auth.tokens import TokenFlow
		from open_exposure_gateway.core.exceptions import ForbiddenException
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context


		@dataclass(frozen=True)
		class OperationRequirements:
		role: str
		requires_three_legged: bool


		# One explicit table per CAMARA operation, not per-handler if statements: this is what keeps
		# authorization behavior auditable and identical regardless of which client type the caller
		# came through. `require_role`/`require_three_legged` below look an incoming request up here
		# by "{method} {path}" (the route's registered path template, e.g. ".../{sessionId}") rather
		# than taking arguments, so a router only has to declare the two dependencies once -- see
		# their use as router-level `dependencies=[...]` in each CAMARA router module.
		OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = {
		# Quality on Demand -- app-provider only; the spec allows a three-legged token on
		# get/delete/extend but never requires one, and createSession is two-legged only.
		"POST /qod/v0/sessions": OperationRequirements("app-provider", False),
		"GET /qod/v0/sessions/{sessionId}": OperationRequirements("app-provider", False),
		"DELETE /qod/v0/sessions/{sessionId}": OperationRequirements("app-provider", False),
		"POST /qod/v0/sessions/{sessionId}/extend": OperationRequirements("app-provider", False),
		"GET /qod/v0/qos-profiles": OperationRequirements("app-provider", False),
		"GET /qod/v0/qos-profiles/{name}": OperationRequirements("app-provider", False),
		# Edge Application Management -- app-provider manages its own apps/instances/deployments;
		# no end-user consent is ever involved.
		"GET /edge-application-management/v0.1alpha1/edge-cloud-zones": OperationRequirements(
		"app-provider", False
		),
		"GET /edge-application-management/v0.1alpha1/apps": OperationRequirements(
		"app-provider", False
		),
		"GET /edge-application-management/v0.1alpha1/apps/{appId}": OperationRequirements(
		"app-provider", False
		),
		"POST /edge-application-management/v0.1alpha1/apps": OperationRequirements(
		"app-provider", False
		),
		"DELETE /edge-application-management/v0.1alpha1/apps/{appId}": OperationRequirements(
		"app-provider", False
		),
		"POST /edge-application-management/v0.1alpha1/appinstances": OperationRequirements(
		"app-provider", False
		),
		"GET /edge-application-management/v0.1alpha1/appinstances": OperationRequirements(
		"app-provider", False
		),
		"DELETE /edge-application-management/v0.1alpha1/appinstances/{appInstanceId}": (
		OperationRequirements("app-provider", False)
		),
		"POST /edge-application-management/v0.1alpha1/deployments": OperationRequirements(
		"app-provider", False
		),
		"GET /edge-application-management/v0.1alpha1/deployments": OperationRequirements(
		"app-provider", False
		),
		"DELETE /edge-application-management/v0.1alpha1/deployments/{appDeploymentId}": (
		OperationRequirements("app-provider", False)
		),
		"PATCH /edge-application-management/v0.1alpha1/deployments/{appDeploymentId}": (
		OperationRequirements("app-provider", False)
		),
		"GET /edge-application-management/v0.1alpha1/clusters": OperationRequirements(
		"app-provider", False
		),
		# Location Retrieval -- locates a real end user's device, so a role-correct
		# app-provider service-account token is not enough: this always requires a
		# three-legged (user-consented) token, regardless of role/audience validity.
		"POST /location-retrieval/v0.5/retrieve": OperationRequirements("app-provider", True),
		}


		def _operation_key(request: Request) -> str:
		route = request.scope.get("route")
		path = getattr(route, "path", request.url.path)
		return f"{request.method} {path}"


		def _requirements_for(request: Request) -> OperationRequirements:
		key = _operation_key(request)
		requirements = OPERATION_REQUIREMENTS.get(key)
		if requirements is None:
		# A route with no registered entry is a config gap, not a client error --
		# fail closed rather than silently letting an unclassified operation through.
		raise RuntimeError(f"No CAMARA operation requirements registered for {key!r}")
		return requirements


		def require_role(
		request: Request,
		caller: Annotated[CallerContext, Depends(get_caller_context)],
		) -> CallerContext:
		requirements = _requirements_for(request)
		if requirements.role not in caller.roles:
		raise ForbiddenException(
		message=f"Caller is missing the required role: {requirements.role}"
		)
		return caller


		def require_three_legged(
		request: Request,
		caller: Annotated[CallerContext, Depends(get_caller_context)],
		) -> CallerContext:
		requirements = _requirements_for(request)
		# Independent of require_role: a legitimate app-provider service-account token
		# (TWO_LEGGED) must still be rejected here, since role correctness says nothing
		# about whether a real end-user actually consented.
		if requirements.requires_three_legged and caller.flow != TokenFlow.THREE_LEGGED:
		raise ForbiddenException(
		message="This operation requires a three-legged (user-consented) access token"
		)
		return caller