feat(callback): clean up callback credentials (79566312) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/callback_secrets.py

+7 −0

Changes for src/open_exposure_gateway/application/callback_secrets.py: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -57,3 +57,10 @@ async def load_bearer_token(
		if key is None:
		raise CallbackSecretError("no callback secret key is configured")
		return decrypt_secret(credential.encrypted_token, key)


		async def purge_expired_callback_credentials(
		repo: CallbackCredentialRepository, now: datetime
		) -> int:
		"""Deletes credentials past their expiry; a skipped delivery never needs them again."""
		return await repo.delete_expired(now)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+29 −7

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 29 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -1283,7 +1283,20 @@ class EdgeApplicationManagementService:
		)
		if registration is None or not registration.is_active:
		continue
		await self._deliver_to_registration(registration, operation_id, occurred_at, group)
		await self._forget_credential_after_final_notification(
		registration.id, creating_operation_id
		)

		async def _deliver_to_registration(
		self,
		registration: CallbackRegistration,
		operation_id: UUID,
		occurred_at: str,
		group: list[AppInstance],
		) -> None:
		assert self._callback_delivery_port is not None
		assert self._callback_delivery_repo is not None
		try:
		bearer_token = await load_bearer_token(
		self._callback_credential_repo,
		@@ -1296,22 +1309,22 @@ class EdgeApplicationManagementService:
		"callback_skipped_credential_expired",
		callback_registration_id=str(registration.id),
		)
		continue
		return
		except CallbackSecretError as exc:
		logger.error(
		"callback_skipped_credential_unusable",
		callback_registration_id=str(registration.id),
		error=str(exc),
		)
		continue
		return

		app_id = await self._resolve_app_id(group[0].app_registration_id)
		if app_id is None:
		logger.warning(
		"callback_skipped_unresolvable_app_id",
		operation_id=str(creating_operation_id),
		operation_id=str(registration.operation_id),
		)
		continue
		return

		deliveries: list[CallbackEvent] = []
		app_deployment_id = group[0].app_deployment_id
		@@ -1337,9 +1350,7 @@ class EdgeApplicationManagementService:
		except Exception as exc:
		state = "failed"
		last_error = str(exc)
		logger.warning(
		"callback_delivery_failed", sink=registration.sink, error=str(exc)
		)
		logger.warning("callback_delivery_failed", sink=registration.sink, error=str(exc))

		await self._callback_delivery_repo.save(
		CallbackDelivery(
		@@ -1352,6 +1363,17 @@ class EdgeApplicationManagementService:
		)
		)

		async def _forget_credential_after_final_notification(
		self, callback_registration_id: UUID, creating_operation_id: UUID
		) -> None:
		"""Once every instance the subscription covers is terminated, nothing more will
		be sent to the sink, so its token is no longer needed."""
		if self._callback_credential_repo is None or self._app_instance_repo is None:
		return
		instances = await self._app_instance_repo.list_by_operation_id(creating_operation_id)
		if instances and all(i.state == AppInstanceState.TERMINATED for i in instances):
		await self._callback_credential_repo.delete_by_registration_id(callback_registration_id)

		def _require_callback_secret_key_for(self, subscription: SubscriptionRequest \| None) -> None:
		"""Runs before anything is saved: without a key, OEG would accept a token it
		cannot store and the sink would never get an authenticated notification."""

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+10 −0

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -519,3 +519,13 @@ class QualityOnDemandService:
		status_info="DELETE_REQUESTED",
		occurred_at=requested_at,
		)

		# DELETE_REQUESTED is the session's last notification: drop the token.
		await self._forget_callback_credential(qod_session.operation_id)

		async def _forget_callback_credential(self, operation_id: UUID) -> None:
		if self._callback_registration_repo is None or self._callback_credential_repo is None:
		return
		registration = await self._callback_registration_repo.get_by_operation_id(operation_id)
		if registration is not None:
		await self._callback_credential_repo.delete_by_registration_id(registration.id)

src/open_exposure_gateway/main.py

+16 −1

Changes for src/open_exposure_gateway/main.py: 16 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		import asyncio
		from collections.abc import AsyncGenerator, Awaitable, Callable
		from contextlib import asynccontextmanager, suppress
		from datetime import timedelta
		from datetime import datetime, timedelta, timezone
		from typing import Optional

		import structlog
		@@ -56,6 +56,9 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.router import (
		router as federation_manager_router,
		)
		from open_exposure_gateway.api.platform.health import router as health_router
		from open_exposure_gateway.application.callback_secrets import (
		purge_expired_callback_credentials,
		)
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		)
		@@ -138,6 +141,18 @@ async def _run_reconciliation_pass(
		) -> None:
		# One transaction per operation so a single failure does not block the rest.
		logger = structlog.get_logger()
		async with session_maker() as session:
		try:
		purged = await purge_expired_callback_credentials(
		SqlCallbackCredentialRepository(session), datetime.now(timezone.utc)
		)
		await session.commit()
		if purged:
		logger.info("expired_callback_credentials_purged", count=purged)
		except Exception:
		await session.rollback()
		logger.exception("callback_credential_purge_failed")

		async with session_maker() as session:
		service = _build_eam_service(session, srm_client, callback_delivery_port)
		stale = await service.list_stale_operations(deadline)

tests/integration/test_postgres.py

+34 −0

Changes for tests/integration/test_postgres.py: 34 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import asyncio
		from datetime import datetime, timedelta, timezone
		from unittest.mock import AsyncMock
		from uuid import UUID, uuid4

		import pytest
		@@ -42,6 +43,7 @@ from open_exposure_gateway.domain.models import (
		QodSession,
		QodSessionState,
		)
		from open_exposure_gateway.main import _run_reconciliation_pass


		def _app_registration() -> AppRegistration:
		@@ -433,6 +435,38 @@ async def test_callback_credential_repo_delete_expired_removes_only_expired(
		assert await repo.get_by_registration_id(no_expiry.id) is not None


		async def test_reconciliation_pass_purges_expired_callback_credentials(
		db_engine: AsyncEngine,
		) -> None:
		session_maker = async_sessionmaker(db_engine, expire_on_commit=False)
		now = datetime.now(timezone.utc)
		async with session_maker() as session:
		expired = await _saved_callback_registration(session)
		live = await _saved_callback_registration(session)
		repo = SqlCallbackCredentialRepository(session)
		for registration, expires_at in (
		(expired, now - timedelta(seconds=1)),
		(live, now + timedelta(hours=1)),
		):
		await repo.save(
		CallbackCredential(
		callback_registration_id=registration.id,
		encrypted_token=b"ciphertext",
		expires_at=expires_at,
		)
		)
		await session.commit()

		await _run_reconciliation_pass(
		session_maker, AsyncMock(), AsyncMock(), deadline=timedelta(days=365)
		)

		async with session_maker() as session:
		repo = SqlCallbackCredentialRepository(session)
		assert await repo.get_by_registration_id(expired.id) is None
		assert await repo.get_by_registration_id(live.id) is not None


		async def test_callback_delivery_repo_persists_and_lists(db_session: AsyncSession) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		callback_registration = await SqlCallbackRegistrationRepository(db_session).save(