fix: enforce session ownership, echo sink, ignore (cdf7d20f) · Commits · OOP / code / open-exposure-gateway

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

+1 −0

Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/router.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -102,6 +102,7 @@ async def get_qod_session(
		) -> Any:
		return await service.get_session(
		session_id=sessionId,
		app_provider_id=caller.app_provider_id,
		x_correlator=caller.x_correlator,
		)

src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py

+4 −1

Changes for src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py: 4 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -104,7 +104,9 @@ def build_activate_command(
		)


		def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability) -> SessionInfo:
		def build_session_info(
		qod_session: QodSession, capability: SRMNetworkCapability, sink: str \| None = None
		) -> SessionInfo:
		"""Combines `qod_sessions` (identity, ownership, cached `qosStatus`) with SRM's live
		capability detail (target device/application server) into the CAMARA response -- the
		read-side counterpart of `build_activate_command`, and the boundary that keeps SRM's
		@@ -170,6 +172,7 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability
		devicePorts=device_ports,
		applicationServerPorts=application_server_ports,
		qosProfile=qod_session.qos_profile,
		sink=sink,
		duration=duration,
		startedAt=started_at,
		expiresAt=expires_at,

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+27 −13

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 27 added lines, 13 removed lines.
Original line number	Diff line number	Diff line
		@@ -138,10 +138,13 @@ class QualityOnDemandService:
		message="The device cannot be identified.",
		)
		if device.networkAccessIdentifier is not None:
		if not (device.phoneNumber or device.ipv4Address or device.ipv6Address):
		raise UnprocessableEntityException(
		error_code=ErrorCode.UNSUPPORTED_IDENTIFIER,
		message="The identifier provided is not supported.",
		)
		# Unsupported but accompanied by a supported identifier: ignore it.
		device = device.model_copy(update={"networkAccessIdentifier": None})

		# TODO: remove once callback credentials can be stored.
		if request.sinkCredential is not None:
		@@ -437,23 +440,38 @@ class QualityOnDemandService:
		occurred_at=event.emitted_at,
		)

		async def _get_owned_session(self, session_id: str, app_provider_id: str) -> QodSession:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(self._parse_session_id(session_id))
		if qod_session is None or qod_session.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Session {session_id} not found")
		# 404 rather than 403 so other callers can't probe which session IDs exist.
		activate_operation = await self._operation_repo.get_by_id(qod_session.operation_id)
		if activate_operation is None or activate_operation.app_provider_id != app_provider_id:
		raise NotFoundException(message=f"Session {session_id} not found")
		return qod_session

		async def get_session(
		self,
		session_id: str,
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> SessionInfo:
		if self._qod_session_repo is None:
		raise RuntimeError("QodSession repository is not available")

		qod_session = await self._qod_session_repo.get_by_id(self._parse_session_id(session_id))
		if qod_session is None or qod_session.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Session {session_id} not found")
		qod_session = await self._get_owned_session(session_id, app_provider_id)

		capability = await self.srm_client.get_network_capability(
		service_instance_id=session_id,
		x_correlator=x_correlator,
		)
		return build_session_info(qod_session, capability)
		sink: Optional[str] = None
		if self._callback_registration_repo is not None:
		registration = await self._callback_registration_repo.get_by_operation_id(
		qod_session.operation_id
		)
		sink = registration.sink if registration is not None else None
		return build_session_info(qod_session, capability, sink=sink)

		async def delete_session(
		self,
		@@ -462,12 +480,8 @@ class QualityOnDemandService:
		app_provider_id: str,
		x_correlator: Optional[str] = None,
		) -> None:
		if self._operation_repo is None or self._qod_session_repo is None:
		raise RuntimeError("Operation/QodSession repositories are not available")

		qod_session = await self._qod_session_repo.get_by_id(self._parse_session_id(session_id))
		if qod_session is None or qod_session.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Session {session_id} not found")
		qod_session = await self._get_owned_session(session_id, app_provider_id)
		assert self._operation_repo is not None and self._qod_session_repo is not None

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

tests/unit/test_qod_flows.py

+31 −0

Changes for tests/unit/test_qod_flows.py: 31 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -237,6 +237,18 @@ class TestQodSessionFlow:
		assert body["applicationServer"]["ipv4Address"] == "192.168.0.1"
		assert body["device"]["ipv4Address"]["publicPort"] == 59765

		def test_get_session_echoes_sink(
		self, api_client: TestClient, fake_bus: FakeDataBus, fake_srm: FakeSRMClient
		) -> None:
		wire_srm_worker(fake_bus, fake_srm)
		body = {**SESSION_BODY, "sink": "https://application-server.com/notifications"}

		session_id = api_client.post(f"{QOD_BASE}/sessions", json=body).json()["sessionId"]
		response = api_client.get(f"{QOD_BASE}/sessions/{session_id}")

		assert response.status_code == 200
		assert response.json()["sink"] == "https://application-server.com/notifications"

		def test_get_session_returns_404_when_unknown(self, api_client: TestClient) -> None:
		response = api_client.get(f"{QOD_BASE}/sessions/{uuid4()}")
		assert response.status_code == 404
		@@ -394,6 +406,25 @@ class TestQodSessionValidation:
		assert response.status_code == 422
		assert response.json()["code"] == "UNSUPPORTED_IDENTIFIER"

		def test_network_access_identifier_is_ignored_alongside_a_supported_identifier(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		body = {
		**SESSION_BODY,
		"device": {
		"networkAccessIdentifier": "123456789@example.com",
		"phoneNumber": "+123456789",
		},
		}

		response = api_client.post(f"{QOD_BASE}/sessions", json=body)

		assert response.status_code == 201
		assert response.json()["device"] == {"phoneNumber": "+123456789"}
		activates = [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE]
		command = SRMNetworkCapabilityActivateCommand.model_validate(activates[0])
		assert command.network_capability.target.device.network_access_id is None

		def test_sink_credential_is_501(self, api_client: TestClient) -> None:
		body = {
		**SESSION_BODY,

tests/unit/test_qod_service.py

+205 −55

File changed.

Preview size limit exceeded, changes collapsed.