fix: scope traffic influence GET and DELETE by caller app_provider_id (c4b28c1f) · Commits · OOP / code / open-exposure-gateway

.cursor/hook-logs/audit.jsonl

0 → 100644

+218 −0

File added.

Preview size limit exceeded, changes collapsed.

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

+6 −1

Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/router.py: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -73,10 +73,13 @@ def _responses(*codes: int) -> dict[int \| str, dict[str, Any]]:
		)
		async def get_all_traffic_influences(
		service: TrafficInfluenceServiceDep,
		caller: Caller,
		appId: Annotated[Optional[UUID], Query()] = None,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.list_traffic_influences(app_id=appId, x_correlator=x_correlator)
		return await service.list_traffic_influences(
		app_provider_id=caller.app_provider_id, app_id=appId, x_correlator=x_correlator
		)


		@router.post(
		@@ -124,10 +127,12 @@ async def post_traffic_influence(
		async def get_traffic_influence_by_id(
		trafficInfluenceID: str,
		service: TrafficInfluenceServiceDep,
		caller: Caller,
		x_correlator: XCorrelatorHeader = None,
		) -> Any:
		return await service.get_traffic_influence(
		traffic_influence_id=trafficInfluenceID,
		app_provider_id=caller.app_provider_id,
		x_correlator=x_correlator,
		)

src/open_exposure_gateway/application/services/traffic_influence_service.py

+31 −16

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 31 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -558,19 +558,32 @@ class TrafficInfluenceService:
		)
		return build_traffic_influence_response(traffic_influence, request_metadata)

		async def get_traffic_influence(
		self,
		traffic_influence_id: str,
		x_correlator: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		if self._traffic_influence_repo is None:
		raise RuntimeError("TrafficInfluence repository is not available")
		async def _get_owned_traffic_influence(
		self, traffic_influence_id: str, app_provider_id: str
		) -> TrafficInfluenceRecord:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(
		self._parse_traffic_influence_id(traffic_influence_id)
		)
		if traffic_influence is None:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")
		# 404 rather than 403 so other callers can't probe which IDs exist.
		create_operation = await self._operation_repo.get_by_id(traffic_influence.operation_id)
		if create_operation is None or create_operation.app_provider_id != app_provider_id:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")
		return traffic_influence

		async def get_traffic_influence(
		self,
		traffic_influence_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> TrafficInfluenceResponse:
		traffic_influence = await self._get_owned_traffic_influence(
		traffic_influence_id, app_provider_id
		)

		# SRM only holds a capability_instance while the policy is realized, so a 404 is
		# expected unless OEG believes it is ACTIVE: before activation completes there is
		@@ -601,11 +614,12 @@ class TrafficInfluenceService:

		async def list_traffic_influences(
		self,
		app_provider_id: str,
		app_id: Optional[UUID] = None,
		x_correlator: Optional[str] = None,
		) -> list[TrafficInfluenceResponse]:
		if self._traffic_influence_repo is None:
		raise RuntimeError("TrafficInfluence repository is not available")
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		# An unknown appId filters to nothing rather than erroring, matching
		# EdgeApplicationManagementService.get_app_instances.
		@@ -637,7 +651,10 @@ class TrafficInfluenceService:

		responses = []
		for record in records:
		request_metadata = await self._fetch_request_metadata(record.operation_id)
		create_operation = await self._operation_repo.get_by_id(record.operation_id)
		if create_operation is None or create_operation.app_provider_id != app_provider_id:
		continue
		request_metadata = create_operation.metadata.get("request", {})
		capability = capabilities_by_id.get(str(record.traffic_influence_id))
		responses.append(self._build_response_best_effort(record, request_metadata, capability))
		return responses
		@@ -649,14 +666,12 @@ class TrafficInfluenceService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(
		self._parse_traffic_influence_id(traffic_influence_id)
		traffic_influence = await self._get_owned_traffic_influence(
		traffic_influence_id, app_provider_id
		)
		if traffic_influence is None or traffic_influence.state in _TEARDOWN_STATES:
		if traffic_influence.state in _TEARDOWN_STATES:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")
		assert self._operation_repo is not None and self._traffic_influence_repo is not None

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		command = build_deactivate_command(

tests/unit/test_traffic_influence_flows.py

+36 −0

Changes for tests/unit/test_traffic_influence_flows.py: 36 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -15,6 +15,7 @@ from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.router import (
		BASE_PATH as TI_BASE,
		)
		from open_exposure_gateway.core.exceptions import DownstreamServiceException
		from open_exposure_gateway.dependencies import CallerContext, get_caller_context
		from open_exposure_gateway.domain.edge_application_management import (
		SRMServiceInstance,
		SRMZone,
		@@ -40,6 +41,7 @@ from open_exposure_gateway.domain.traffic_influence import (
		SRMTrafficInfluenceCapability,
		Subject,
		)
		from open_exposure_gateway.main import app
		from tests.unit.fakes import (
		FakeAppInstanceRepository,
		FakeAppRegistrationRepository,
		@@ -847,6 +849,40 @@ class TestTrafficInfluenceGetFlow:
		assert response.json() == []


		class TestTrafficInfluenceTenantScope:
		@pytest.fixture()
		def created_id(self, api_client: TestClient) -> str:
		traffic_influence_id: str = api_client.post(
		f"{TI_BASE}/traffic-influences", json=TI_BODY
		).json()["trafficInfluenceID"]
		app.dependency_overrides[get_caller_context] = lambda: CallerContext(
		x_correlator=None, tenant_id="other-tenant", app_provider_id="other-provider"
		)
		return traffic_influence_id

		def test_get_by_id_hides_other_providers_resource(
		self, api_client: TestClient, created_id: str
		) -> None:
		response = api_client.get(f"{TI_BASE}/traffic-influences/{created_id}")
		assert response.status_code == 404

		def test_list_excludes_other_providers_resources(
		self, api_client: TestClient, created_id: str
		) -> None:
		assert api_client.get(f"{TI_BASE}/traffic-influences").json() == []
		response = api_client.get(
		f"{TI_BASE}/traffic-influences", params={"appId": TI_BODY["appId"]}
		)
		assert response.json() == []

		def test_delete_rejects_other_providers_resource(
		self, api_client: TestClient, created_id: str, fake_bus: FakeDataBus
		) -> None:
		response = api_client.delete(f"{TI_BASE}/traffic-influences/{created_id}")
		assert response.status_code == 404
		assert not [p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]


		class TestTrafficInfluenceDeleteFlow:
		def test_delete_succeeds_before_activation_confirmed(
		self, api_client: TestClient, fake_bus: FakeDataBus