Loading .env.example +7 −0 Changes for .env.example: 7 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -23,6 +23,13 @@ # OBSERVABILITY_SETTINGS__LOG_LEVEL="INFO" # CALLBACK_SETTINGS__TIMEOUT=10.0 # Encrypts the callback tokens clients send (EAM sinkCredential, QoD # notificationAuthToken) before they are stored. Must be base64url of exactly # 32 bytes; generate one with: # python -c "import os,base64;print(base64.urlsafe_b64encode(os.urandom(32)).decode())" # Unset (the default) means OEG rejects any request carrying such a token with 501. # Changing it makes every token already stored unreadable, so keep it stable. # CALLBACK_SETTINGS__SECRET_KEY="" # QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c" # LOCATION_RETRIEVAL_SETTINGS__SERVICE_SPECIFICATION_ID="0129e7ce-8e02-5dfd-bae1-0303bcc7676e" Loading src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py +1 −0 Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -132,6 +132,7 @@ async def delete_qod_session( responses=_responses(400, 401, 403, 404, 500, 501, 503), ) async def extend_qod_session_duration(sessionId: str, request: ExtendSessionDuration) -> Any: # TODO: when implemented, also push back the notificationAuthToken's expires_at. raise NotImplementedException(message="Session duration extension is not implemented") Loading src/open_exposure_gateway/application/services/quality_on_demand_service.py +10 −5 Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 10 added lines, 5 removed lines. Original line number Diff line number Diff line from datetime import datetime, timezone from datetime import datetime, timedelta, timezone from typing import Literal, Optional from uuid import UUID, uuid4 Loading Loading @@ -64,6 +64,9 @@ _QOS_STATUS_INFO_MAP: dict[str, Literal["DURATION_EXPIRED", "NETWORK_TERMINATED" "network_terminated": "NETWORK_TERMINATED", } # Lets the final DURATION_EXPIRED / NETWORK_TERMINATED notification still carry the token. NOTIFICATION_AUTH_TOKEN_GRACE = timedelta(hours=1) _TERMINAL_STATES = frozenset( {QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR} ) Loading Loading @@ -138,6 +141,8 @@ class QualityOnDemandService: operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator) session_id = uuid4() started_at = int(datetime.now(timezone.utc).timestamp()) expires_at = started_at + request.duration # The platform-seeded QoD specification, identical on every session (ADR-0035). # Stored per row so the offering a session was provisioned under stays auditable # once per-offering specifications exist. Loading Loading @@ -208,14 +213,15 @@ class QualityOnDemandService: raise RuntimeError( "CallbackCredentialRepository/callback secret key is not available" ) # CAMARA QoD tokens carry no expiry. # CAMARA QoD tokens carry no expiry, so the token expires with the session. await self._callback_credential_repo.save( CallbackCredential( callback_registration_id=registration.id, encrypted_token=encrypt_secret( notification_auth_token, self._callback_secret_key ), expires_at=None, expires_at=datetime.fromtimestamp(expires_at, timezone.utc) + NOTIFICATION_AUTH_TOKEN_GRACE, ) ) Loading @@ -225,7 +231,6 @@ class QualityOnDemandService: "Failed to publish QoD session activation", ) started_at = int(datetime.now(timezone.utc).timestamp()) return SessionInfo( sessionId=session_id, device=request.device, Loading @@ -236,7 +241,7 @@ class QualityOnDemandService: webhook=request.webhook, duration=request.duration, startedAt=started_at, expiresAt=started_at + request.duration, expiresAt=expires_at, qosStatus=QosStatus.REQUESTED, ) Loading tests/unit/test_qod_service.py +58 −4 Changes for tests/unit/test_qod_service.py: 58 added lines, 4 removed lines. Original line number Diff line number Diff line from datetime import datetime, timedelta, timezone from typing import Any from unittest.mock import AsyncMock from uuid import UUID, uuid4 Loading @@ -13,8 +14,13 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import ( PortsSpec, Webhook, ) from open_exposure_gateway.application.callback_secrets import decrypt_secret, encrypt_secret from open_exposure_gateway.application.callback_secrets import ( decrypt_secret, encrypt_secret, purge_expired_callback_credentials, ) from open_exposure_gateway.application.services.quality_on_demand_service import ( NOTIFICATION_AUTH_TOKEN_GRACE, QualityOnDemandService, ) from open_exposure_gateway.core.config import DEFAULT_QOD_SERVICE_SPECIFICATION_ID Loading Loading @@ -300,7 +306,7 @@ class TestCreateSession: callback_credential_repo=callback_credential_repo, callback_secret_key=CALLBACK_KEY, ) await service.create_session( info = await service.create_session( request=_make_request( webhook=Webhook( notificationUrl="https://client.example.com/cb", Loading @@ -314,7 +320,10 @@ class TestCreateSession: registration = next(iter(callback_registration_repo.rows.values())) (credential,) = list(callback_credential_repo.rows.values()) assert credential.callback_registration_id == registration.id assert credential.expires_at is None assert info.expiresAt == info.startedAt + 3600 assert credential.expires_at == ( datetime.fromtimestamp(info.expiresAt, timezone.utc) + NOTIFICATION_AUTH_TOKEN_GRACE ) assert decrypt_secret(credential.encrypted_token, CALLBACK_KEY) == RAW_TOKEN assert RAW_TOKEN.encode() not in credential.encrypted_token saved_rows = [ Loading @@ -326,6 +335,41 @@ class TestCreateSession: assert RAW_TOKEN not in row.model_dump_json() publisher.publish.assert_awaited_once() async def test_notification_auth_token_is_purged_only_after_session_expiry_plus_grace( self, operation_repo: FakeOperationRepository, qod_session_repo: FakeQodSessionRepository, ) -> None: callback_credential_repo = FakeCallbackCredentialRepository() service = QualityOnDemandService( srm_client=AsyncMock(), publisher=AsyncMock(), operation_repo=operation_repo, qod_session_repo=qod_session_repo, callback_registration_repo=FakeCallbackRegistrationRepository(), callback_credential_repo=callback_credential_repo, callback_secret_key=CALLBACK_KEY, ) info = await service.create_session( request=_make_request( webhook=Webhook( notificationUrl="https://client.example.com/cb", notificationAuthToken=RAW_TOKEN, ) ), tenant_id="tenant-1", app_provider_id="provider-1", ) session_expiry = datetime.fromtimestamp(info.expiresAt, timezone.utc) purge_at = session_expiry + NOTIFICATION_AUTH_TOKEN_GRACE for now in (session_expiry, purge_at - timedelta(seconds=1)): assert await purge_expired_callback_credentials(callback_credential_repo, now) == 0 assert len(callback_credential_repo.rows) == 1 assert await purge_expired_callback_credentials(callback_credential_repo, purge_at) == 1 assert callback_credential_repo.rows == {} async def test_notification_auth_token_rejected_with_501_when_no_key( self, operation_repo: FakeOperationRepository, Loading Loading @@ -662,6 +706,9 @@ class TestHandleCompleted: [ pytest.param(False, CALLBACK_KEY, [None], id="no-credential-sends-without-auth"), pytest.param(True, CALLBACK_KEY, [RAW_TOKEN], id="credential-sends-token"), pytest.param( "live", CALLBACK_KEY, [RAW_TOKEN], id="credential-within-session-sends-token" ), pytest.param(True, OTHER_CALLBACK_KEY, [], id="undecryptable-skips"), pytest.param(True, None, [], id="missing-key-skips"), ], Loading @@ -670,7 +717,7 @@ class TestHandleCompleted: self, operation_repo: FakeOperationRepository, qod_session_repo: FakeQodSessionRepository, stored: bool, stored: bool | str, key: bytes | None, expected_tokens: list[str | None], ) -> None: Loading Loading @@ -704,6 +751,13 @@ class TestHandleCompleted: CallbackCredential( callback_registration_id=registration.id, encrypted_token=encrypt_secret(RAW_TOKEN, CALLBACK_KEY), expires_at=( datetime.now(timezone.utc) + timedelta(seconds=3600) + NOTIFICATION_AUTH_TOKEN_GRACE if stored == "live" else None ), ) ) event = SRMOperationCompleted.model_validate( Loading Loading
.env.example +7 −0 Changes for .env.example: 7 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -23,6 +23,13 @@ # OBSERVABILITY_SETTINGS__LOG_LEVEL="INFO" # CALLBACK_SETTINGS__TIMEOUT=10.0 # Encrypts the callback tokens clients send (EAM sinkCredential, QoD # notificationAuthToken) before they are stored. Must be base64url of exactly # 32 bytes; generate one with: # python -c "import os,base64;print(base64.urlsafe_b64encode(os.urandom(32)).decode())" # Unset (the default) means OEG rejects any request carrying such a token with 501. # Changing it makes every token already stored unreadable, so keep it stable. # CALLBACK_SETTINGS__SECRET_KEY="" # QOD_SETTINGS__SERVICE_SPECIFICATION_ID="7608e902-b927-559f-b448-e7e9061dfa5c" # LOCATION_RETRIEVAL_SETTINGS__SERVICE_SPECIFICATION_ID="0129e7ce-8e02-5dfd-bae1-0303bcc7676e" Loading
src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py +1 −0 Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/router.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -132,6 +132,7 @@ async def delete_qod_session( responses=_responses(400, 401, 403, 404, 500, 501, 503), ) async def extend_qod_session_duration(sessionId: str, request: ExtendSessionDuration) -> Any: # TODO: when implemented, also push back the notificationAuthToken's expires_at. raise NotImplementedException(message="Session duration extension is not implemented") Loading
src/open_exposure_gateway/application/services/quality_on_demand_service.py +10 −5 Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 10 added lines, 5 removed lines. Original line number Diff line number Diff line from datetime import datetime, timezone from datetime import datetime, timedelta, timezone from typing import Literal, Optional from uuid import UUID, uuid4 Loading Loading @@ -64,6 +64,9 @@ _QOS_STATUS_INFO_MAP: dict[str, Literal["DURATION_EXPIRED", "NETWORK_TERMINATED" "network_terminated": "NETWORK_TERMINATED", } # Lets the final DURATION_EXPIRED / NETWORK_TERMINATED notification still carry the token. NOTIFICATION_AUTH_TOKEN_GRACE = timedelta(hours=1) _TERMINAL_STATES = frozenset( {QodSessionState.DELETION_REQUESTED, QodSessionState.DELETED, QodSessionState.ERROR} ) Loading Loading @@ -138,6 +141,8 @@ class QualityOnDemandService: operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator) session_id = uuid4() started_at = int(datetime.now(timezone.utc).timestamp()) expires_at = started_at + request.duration # The platform-seeded QoD specification, identical on every session (ADR-0035). # Stored per row so the offering a session was provisioned under stays auditable # once per-offering specifications exist. Loading Loading @@ -208,14 +213,15 @@ class QualityOnDemandService: raise RuntimeError( "CallbackCredentialRepository/callback secret key is not available" ) # CAMARA QoD tokens carry no expiry. # CAMARA QoD tokens carry no expiry, so the token expires with the session. await self._callback_credential_repo.save( CallbackCredential( callback_registration_id=registration.id, encrypted_token=encrypt_secret( notification_auth_token, self._callback_secret_key ), expires_at=None, expires_at=datetime.fromtimestamp(expires_at, timezone.utc) + NOTIFICATION_AUTH_TOKEN_GRACE, ) ) Loading @@ -225,7 +231,6 @@ class QualityOnDemandService: "Failed to publish QoD session activation", ) started_at = int(datetime.now(timezone.utc).timestamp()) return SessionInfo( sessionId=session_id, device=request.device, Loading @@ -236,7 +241,7 @@ class QualityOnDemandService: webhook=request.webhook, duration=request.duration, startedAt=started_at, expiresAt=started_at + request.duration, expiresAt=expires_at, qosStatus=QosStatus.REQUESTED, ) Loading
tests/unit/test_qod_service.py +58 −4 Changes for tests/unit/test_qod_service.py: 58 added lines, 4 removed lines. Original line number Diff line number Diff line from datetime import datetime, timedelta, timezone from typing import Any from unittest.mock import AsyncMock from uuid import UUID, uuid4 Loading @@ -13,8 +14,13 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import ( PortsSpec, Webhook, ) from open_exposure_gateway.application.callback_secrets import decrypt_secret, encrypt_secret from open_exposure_gateway.application.callback_secrets import ( decrypt_secret, encrypt_secret, purge_expired_callback_credentials, ) from open_exposure_gateway.application.services.quality_on_demand_service import ( NOTIFICATION_AUTH_TOKEN_GRACE, QualityOnDemandService, ) from open_exposure_gateway.core.config import DEFAULT_QOD_SERVICE_SPECIFICATION_ID Loading Loading @@ -300,7 +306,7 @@ class TestCreateSession: callback_credential_repo=callback_credential_repo, callback_secret_key=CALLBACK_KEY, ) await service.create_session( info = await service.create_session( request=_make_request( webhook=Webhook( notificationUrl="https://client.example.com/cb", Loading @@ -314,7 +320,10 @@ class TestCreateSession: registration = next(iter(callback_registration_repo.rows.values())) (credential,) = list(callback_credential_repo.rows.values()) assert credential.callback_registration_id == registration.id assert credential.expires_at is None assert info.expiresAt == info.startedAt + 3600 assert credential.expires_at == ( datetime.fromtimestamp(info.expiresAt, timezone.utc) + NOTIFICATION_AUTH_TOKEN_GRACE ) assert decrypt_secret(credential.encrypted_token, CALLBACK_KEY) == RAW_TOKEN assert RAW_TOKEN.encode() not in credential.encrypted_token saved_rows = [ Loading @@ -326,6 +335,41 @@ class TestCreateSession: assert RAW_TOKEN not in row.model_dump_json() publisher.publish.assert_awaited_once() async def test_notification_auth_token_is_purged_only_after_session_expiry_plus_grace( self, operation_repo: FakeOperationRepository, qod_session_repo: FakeQodSessionRepository, ) -> None: callback_credential_repo = FakeCallbackCredentialRepository() service = QualityOnDemandService( srm_client=AsyncMock(), publisher=AsyncMock(), operation_repo=operation_repo, qod_session_repo=qod_session_repo, callback_registration_repo=FakeCallbackRegistrationRepository(), callback_credential_repo=callback_credential_repo, callback_secret_key=CALLBACK_KEY, ) info = await service.create_session( request=_make_request( webhook=Webhook( notificationUrl="https://client.example.com/cb", notificationAuthToken=RAW_TOKEN, ) ), tenant_id="tenant-1", app_provider_id="provider-1", ) session_expiry = datetime.fromtimestamp(info.expiresAt, timezone.utc) purge_at = session_expiry + NOTIFICATION_AUTH_TOKEN_GRACE for now in (session_expiry, purge_at - timedelta(seconds=1)): assert await purge_expired_callback_credentials(callback_credential_repo, now) == 0 assert len(callback_credential_repo.rows) == 1 assert await purge_expired_callback_credentials(callback_credential_repo, purge_at) == 1 assert callback_credential_repo.rows == {} async def test_notification_auth_token_rejected_with_501_when_no_key( self, operation_repo: FakeOperationRepository, Loading Loading @@ -662,6 +706,9 @@ class TestHandleCompleted: [ pytest.param(False, CALLBACK_KEY, [None], id="no-credential-sends-without-auth"), pytest.param(True, CALLBACK_KEY, [RAW_TOKEN], id="credential-sends-token"), pytest.param( "live", CALLBACK_KEY, [RAW_TOKEN], id="credential-within-session-sends-token" ), pytest.param(True, OTHER_CALLBACK_KEY, [], id="undecryptable-skips"), pytest.param(True, None, [], id="missing-key-skips"), ], Loading @@ -670,7 +717,7 @@ class TestHandleCompleted: self, operation_repo: FakeOperationRepository, qod_session_repo: FakeQodSessionRepository, stored: bool, stored: bool | str, key: bytes | None, expected_tokens: list[str | None], ) -> None: Loading Loading @@ -704,6 +751,13 @@ class TestHandleCompleted: CallbackCredential( callback_registration_id=registration.id, encrypted_token=encrypt_secret(RAW_TOKEN, CALLBACK_KEY), expires_at=( datetime.now(timezone.utc) + timedelta(seconds=3600) + NOTIFICATION_AUTH_TOKEN_GRACE if stored == "live" else None ), ) ) event = SRMOperationCompleted.model_validate( Loading