feat(eam): send stored bearer token on eam callbacks (c4256247) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/callback_secrets.py

+27 −1

Changes for src/open_exposure_gateway/application/callback_secrets.py: 27 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -8,11 +8,14 @@ in flight to carry a per-session key.
		"""

		import os
		from datetime import datetime
		from uuid import UUID

		from cryptography.exceptions import InvalidTag
		from cryptography.hazmat.primitives.ciphers.aead import AESGCM

		from open_exposure_gateway.domain.errors import CallbackSecretError
		from open_exposure_gateway.domain.errors import CallbackCredentialExpiredError, CallbackSecretError
		from open_exposure_gateway.ports.database.callbacks import CallbackCredentialRepository

		_NONCE_BYTES = 12

		@@ -31,3 +34,26 @@ def decrypt_secret(blob: bytes, key: bytes) -> str:
		except (InvalidTag, ValueError) as exc:
		raise CallbackSecretError("callback secret could not be decrypted") from exc
		return plaintext.decode()


		async def load_bearer_token(
		repo: CallbackCredentialRepository \| None,
		callback_registration_id: UUID,
		key: bytes \| None,
		now: datetime,
		) -> str \| None:
		"""The token to send on a delivery, or None to send without auth.

		Raises CallbackCredentialExpiredError when the token has expired, and
		CallbackSecretError when it cannot be decrypted; the caller skips the delivery.
		"""
		if repo is None:
		return None
		credential = await repo.get_by_registration_id(callback_registration_id)
		if credential is None:
		return None
		if credential.expires_at is not None and credential.expires_at <= now:
		raise CallbackCredentialExpiredError("callback credential has expired")
		if key is None:
		raise CallbackSecretError("no callback secret key is configured")
		return decrypt_secret(credential.encrypted_token, key)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+31 −1

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 31 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -20,6 +20,7 @@ from open_exposure_gateway.api.camara.edge_application_management.v0_1_0_alpha_1
		KubernetesResources,
		SubmittedApp,
		)
		from open_exposure_gateway.application.callback_secrets import load_bearer_token
		from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_app_deployment_info,
		build_app_deployment_status_change_events,
		@@ -50,6 +51,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMOperationCompleted,
		Subject,
		)
		from open_exposure_gateway.domain.errors import CallbackCredentialExpiredError, CallbackSecretError
		from open_exposure_gateway.domain.models import (
		AppDeployment,
		AppDeploymentState,
		@@ -70,6 +72,7 @@ from open_exposure_gateway.ports.callback_delivery_port import (
		CallbackEvent,
		)
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackCredentialRepository,
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		@@ -292,6 +295,8 @@ class EdgeApplicationManagementService:
		callback_delivery_port: CallbackDeliveryPort \| None = None,
		callback_delivery_repo: CallbackDeliveryRepository \| None = None,
		app_deployment_repo: AppDeploymentRepository \| None = None,
		callback_credential_repo: CallbackCredentialRepository \| None = None,
		callback_secret_key: bytes \| None = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		@@ -302,6 +307,8 @@ class EdgeApplicationManagementService:
		self._callback_delivery_port = callback_delivery_port
		self._callback_delivery_repo = callback_delivery_repo
		self._app_deployment_repo = app_deployment_repo
		self._callback_credential_repo = callback_credential_repo
		self._callback_secret_key = callback_secret_key

		async def get_edge_cloud_zones(
		self,
		@@ -1323,6 +1330,27 @@ class EdgeApplicationManagementService:
		if registration is None or not registration.is_active:
		continue

		try:
		bearer_token = await load_bearer_token(
		self._callback_credential_repo,
		registration.id,
		self._callback_secret_key,
		datetime.now(timezone.utc),
		)
		except CallbackCredentialExpiredError:
		logger.warning(
		"callback_skipped_credential_expired",
		callback_registration_id=str(registration.id),
		)
		continue
		except CallbackSecretError as exc:
		logger.error(
		"callback_skipped_credential_unusable",
		callback_registration_id=str(registration.id),
		error=str(exc),
		)
		continue

		app_id = await self._resolve_app_id(group[0].app_registration_id)
		if app_id is None:
		logger.warning(
		@@ -1348,7 +1376,9 @@ class EdgeApplicationManagementService:
		for event in deliveries:
		last_error: Optional[str] = None
		try:
		await self._callback_delivery_port.deliver(registration.sink, event)
		await self._callback_delivery_port.deliver(
		registration.sink, event, bearer_token=bearer_token
		)
		state = "delivered"
		except Exception as exc:
		state = "failed"

src/open_exposure_gateway/domain/errors.py

+4 −0

Changes for src/open_exposure_gateway/domain/errors.py: 4 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		class CallbackSecretError(Exception):
		"""A stored callback secret could not be decrypted. Carries no secret material."""


		class CallbackCredentialExpiredError(CallbackSecretError):
		"""The client's callback token is past its expiry; the delivery must be skipped."""

src/open_exposure_gateway/main.py

+5 −0

Changes for src/open_exposure_gateway/main.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,9 @@ from open_exposure_gateway.adapters.database.repos.app_instances import SqlAppIn
		from open_exposure_gateway.adapters.database.repos.app_registrations import (
		SqlAppRegistrationRepository,
		)
		from open_exposure_gateway.adapters.database.repos.callback_credentials import (
		SqlCallbackCredentialRepository,
		)
		from open_exposure_gateway.adapters.database.repos.callback_deliveries import (
		SqlCallbackDeliveryRepository,
		)
		@@ -86,6 +89,8 @@ def _build_eam_service(
		callback_registration_repo=SqlCallbackRegistrationRepository(session),
		callback_delivery_port=callback_delivery_port,
		callback_delivery_repo=SqlCallbackDeliveryRepository(session),
		callback_credential_repo=SqlCallbackCredentialRepository(session),
		callback_secret_key=get_settings().callback_settings.secret_key_bytes,
		)

tests/unit/fakes.py

+31 −0

Changes for tests/unit/fakes.py: 31 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -55,6 +55,7 @@ from open_exposure_gateway.domain.models import (
		AppInstanceState,
		AppRegistration,
		AppRegistrationStatus,
		CallbackCredential,
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		@@ -74,6 +75,7 @@ from open_exposure_gateway.ports.callback_delivery_port import (
		CallbackEvent,
		)
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackCredentialRepository,
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		@@ -605,6 +607,35 @@ class FakeCallbackRegistrationRepository(CallbackRegistrationRepository):
		return stored.model_copy(deep=True)


		class FakeCallbackCredentialRepository(CallbackCredentialRepository):
		def __init__(self) -> None:
		self.rows: dict[UUID, CallbackCredential] = {}

		async def get_by_registration_id(
		self, callback_registration_id: UUID
		) -> CallbackCredential \| None:
		row = self.rows.get(callback_registration_id)
		return row.model_copy(deep=True) if row is not None else None

		async def save(self, callback_credential: CallbackCredential) -> CallbackCredential:
		stored = callback_credential.model_copy(deep=True)
		self.rows[stored.callback_registration_id] = stored
		return stored.model_copy(deep=True)

		async def delete_by_registration_id(self, callback_registration_id: UUID) -> None:
		self.rows.pop(callback_registration_id, None)

		async def delete_expired(self, now: datetime) -> int:
		expired = [
		registration_id
		for registration_id, row in self.rows.items()
		if row.expires_at is not None and row.expires_at <= now
		]
		for registration_id in expired:
		del self.rows[registration_id]
		return len(expired)


		class FakeCallbackDeliveryRepository(CallbackDeliveryRepository):
		def __init__(self) -> None:
		self.rows: dict[UUID, CallbackDelivery] = {}