fix: enforce aut, app ownership and CAMARA callback rules (!36) · Merge requests · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/traffic_influence_callback_client.py

+5 −1

Changes for src/open_exposure_gateway/adapters/http/traffic_influence_callback_client.py: 5 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		import httpx

		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.domain.errors import CallbackSinkGoneError
		from open_exposure_gateway.domain.traffic_influence import TrafficInfluenceChangedCloudEvent


		@@ -13,11 +14,14 @@ class HttpTrafficInfluenceCallbackClient:
		self,
		sink: str,
		event: TrafficInfluenceChangedCloudEvent,
		x_correlator: str,
		bearer_token: str \| None = None,
		) -> None:
		headers = {"Content-Type": "application/cloudevents+json"}
		headers = {"Content-Type": "application/cloudevents+json", "x-correlator": x_correlator}
		if bearer_token is not None:
		headers["Authorization"] = f"Bearer {bearer_token}"
		async with httpx.AsyncClient(timeout=self.timeout) as client:
		response = await client.post(sink, content=event.model_dump_json(), headers=headers)
		if response.status_code == httpx.codes.GONE:
		raise CallbackSinkGoneError(f"sink answered {response.status_code}")
		response.raise_for_status()

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py

+8 −1

Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py: 8 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -13,6 +13,10 @@ from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		from open_exposure_gateway.application.services.traffic_influence_service import (
		TrafficInfluenceService,
		)
		from open_exposure_gateway.core.authorization import (
		require_role,
		require_scope,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		@@ -32,7 +36,10 @@ from open_exposure_gateway.schemas.common import ErrorInfo
		# CAMARA base path: the spec serves at {apiRoot}/traffic-influence/v0.10.
		BASE_PATH = "/traffic-influence/v0.10"

		router = APIRouter(prefix=BASE_PATH)
		router = APIRouter(
		prefix=BASE_PATH,
		dependencies=[Depends(require_role), Depends(require_scope)],
		)

		TrafficInfluenceServiceDep = Annotated[
		TrafficInfluenceService, Depends(get_traffic_influence_service)

src/open_exposure_gateway/application/services/traffic_influence_service.py

+25 −5

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 25 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -36,6 +36,7 @@ from open_exposure_gateway.domain.errors import (
		CallbackCredentialExpiredError,
		CallbackCredentialRevokedError,
		CallbackSecretError,
		CallbackSinkGoneError,
		)
		from open_exposure_gateway.domain.models import (
		CallbackCredential,
		@@ -176,11 +177,11 @@ class TrafficInfluenceService:
		return instances[0].zone_id
		return None

		async def _resolve_app_registration_id(self, app_id: UUID) -> UUID:
		async def _resolve_app_registration_id(self, app_id: UUID, tenant_id: str) -> UUID:
		if self._app_registration_repo is None:
		raise RuntimeError("AppRegistrationRepository is not available")
		app_registration = await self._app_registration_repo.get_by_app_id(app_id)
		if app_registration is None:
		if app_registration is None or app_registration.tenant_id != tenant_id:
		raise BadRequestException(message=f"App {app_id} is not registered")
		return app_registration.app_registration_id

		@@ -241,7 +242,7 @@ class TrafficInfluenceService:
		if existing is not None:
		return await self._replay_create(existing)

		app_registration_id = await self._resolve_app_registration_id(request.appId)
		app_registration_id = await self._resolve_app_registration_id(request.appId, tenant_id)

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		traffic_influence_id = uuid4()
		@@ -416,7 +417,13 @@ class TrafficInfluenceService:
		f"{get_settings().public_base_url}/traffic-influence/v0.10/traffic-influences/"
		f"{traffic_influence.traffic_influence_id}"
		)
		request_metadata = await self._fetch_request_metadata(registration_operation_id)
		operation = (
		await self._operation_repo.get_by_id(registration_operation_id)
		if self._operation_repo is not None
		else None
		)
		request_metadata = operation.metadata.get("request", {}) if operation is not None else {}
		correlation_id = operation.correlation_id if operation is not None else str(uuid4())
		selected_app_instance_ids: list[UUID] = []
		if (
		traffic_influence.state == RecordTrafficInfluenceState.ACTIVE
		@@ -445,9 +452,22 @@ class TrafficInfluenceService:
		last_error: Optional[str] = None
		try:
		await self._callback_delivery_port.deliver(
		registration.sink, cloud_event, bearer_token=bearer_token
		registration.sink, cloud_event, correlation_id, bearer_token=bearer_token
		)
		state = "delivered"
		except CallbackSinkGoneError as exc:
		state = "gone"
		last_error = str(exc)
		logger.info(
		"traffic_influence_callback_sink_gone",
		callback_registration_id=str(registration.id),
		sink=registration.sink,
		)
		await self._callback_registration_repo.save(
		registration.model_copy(update={"is_active": False})
		)
		if self._callback_credential_repo is not None:
		await self._callback_credential_repo.revoke_by_registration_id(registration.id)
		except Exception as exc:
		state = "failed"
		last_error = str(exc)

src/open_exposure_gateway/core/authorization.py

+19 −0

Changes for src/open_exposure_gateway/core/authorization.py: 19 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -83,6 +83,25 @@ OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = {
		"POST /location-retrieval/v0.5/retrieve": OperationRequirements(
		"app-provider", scope="location-retrieval:read"
		),
		# Traffic Influence
		"GET /traffic-influence/v0.10/traffic-influences": OperationRequirements(
		"app-provider", scope="traffic-influence:traffic-influences:read"
		),
		"POST /traffic-influence/v0.10/traffic-influences": OperationRequirements(
		"app-provider", scope="traffic-influence:traffic-influences:write"
		),
		"GET /traffic-influence/v0.10/traffic-influences/{trafficInfluenceID}": (
		OperationRequirements("app-provider", scope="traffic-influence:traffic-influences:read")
		),
		"PATCH /traffic-influence/v0.10/traffic-influences/{trafficInfluenceID}": (
		OperationRequirements("app-provider", scope="traffic-influence:traffic-influences:update")
		),
		"DELETE /traffic-influence/v0.10/traffic-influences/{trafficInfluenceID}": (
		OperationRequirements("app-provider", scope="traffic-influence:traffic-influences:delete")
		),
		"POST /traffic-influence/v0.10/traffic-influence-devices": OperationRequirements(
		"app-provider", scope="traffic-influence:traffic-influence-devices:write"
		),
		# Federation Manager (platform namespace, ADR-0051): operator-only, no CAMARA scope.
		"POST /platform/v1/federation/partners/{partnerOpId}/federations": (
		OperationRequirements("operator")

src/open_exposure_gateway/ports/traffic_influence_callback_port.py

+5 −1

Changes for src/open_exposure_gateway/ports/traffic_influence_callback_port.py: 5 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -5,5 +5,9 @@ from open_exposure_gateway.domain.traffic_influence import TrafficInfluenceChang

		class TrafficInfluenceCallbackDeliveryPort(Protocol):
		async def deliver(
		self, sink: str, event: TrafficInfluenceChangedCloudEvent, bearer_token: str \| None = None
		self,
		sink: str,
		event: TrafficInfluenceChangedCloudEvent,
		x_correlator: str,
		bearer_token: str \| None = None,
		) -> None: ...

fix: enforce aut, app ownership and CAMARA callback rules

Merge request reports