fix: never send a delivery without auth once its token is gone (b1e7d9b2) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/callback_credentials.py

+13 −6

Changes for src/open_exposure_gateway/adapters/database/repos/callback_credentials.py: 13 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,7 +2,7 @@ from datetime import datetime
		from typing import Any, cast
		from uuid import UUID

		from sqlalchemy import CursorResult, delete, select
		from sqlalchemy import CursorResult, select, update
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import CallbackCredentialMapper
		@@ -35,16 +35,23 @@ class SqlCallbackCredentialRepository(CallbackCredentialRepository):
		raise RuntimeError("Saved callback credential could not be reloaded")
		return CallbackCredentialMapper.to_domain(row)

		async def delete_by_registration_id(self, callback_registration_id: UUID) -> None:
		stmt = delete(CallbackCredentialRow).where(
		CallbackCredentialRow.callback_registration_id == callback_registration_id
		async def revoke_by_registration_id(self, callback_registration_id: UUID) -> None:
		stmt = (
		update(CallbackCredentialRow)
		.where(CallbackCredentialRow.callback_registration_id == callback_registration_id)
		.values(encrypted_token=None)
		)
		await self._session.execute(stmt)

		async def delete_expired(self, now: datetime) -> int:
		stmt = delete(CallbackCredentialRow).where(
		async def revoke_expired(self, now: datetime) -> int:
		stmt = (
		update(CallbackCredentialRow)
		.where(
		CallbackCredentialRow.encrypted_token.is_not(None),
		CallbackCredentialRow.expires_at.is_not(None),
		CallbackCredentialRow.expires_at <= now,
		)
		.values(encrypted_token=None)
		)
		result = cast(CursorResult[Any], await self._session.execute(stmt))
		return result.rowcount

src/open_exposure_gateway/adapters/database/sql.py

+2 −2

Changes for src/open_exposure_gateway/adapters/database/sql.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -160,8 +160,8 @@ class CallbackCredentialRow(Base):
		callback_registration_id: Mapped[UUID] = mapped_column(
		ForeignKey("callback_registrations.id"), primary_key=True
		)
		# nonce + AES-GCM ciphertext; see application/callback_secrets.py.
		encrypted_token: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
		# nonce + AES-GCM ciphertext; NULL once revoked. See application/callback_secrets.py.
		encrypted_token: Mapped[bytes \| None] = mapped_column(LargeBinary)
		expires_at: Mapped[datetime \| None] = mapped_column(DateTime(timezone=True))
		created_at: Mapped[datetime] = mapped_column(
		DateTime(timezone=True),

src/open_exposure_gateway/application/callback_secrets.py

+13 −6

Changes for src/open_exposure_gateway/application/callback_secrets.py: 13 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -14,7 +14,11 @@ from uuid import UUID
		from cryptography.exceptions import InvalidTag
		from cryptography.hazmat.primitives.ciphers.aead import AESGCM

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

		_NONCE_BYTES = 12
		@@ -42,10 +46,11 @@ async def load_bearer_token(
		key: bytes \| None,
		now: datetime,
		) -> str \| None:
		"""The token to send on a delivery, or None to send without auth.
		"""The token to send on a delivery, or None when the client never supplied one.

		Raises CallbackCredentialExpiredError when the token has expired, and
		CallbackSecretError when it cannot be decrypted; the caller skips the delivery.
		Raises CallbackCredentialExpiredError when the token has expired,
		CallbackCredentialRevokedError when it was wiped, and CallbackSecretError when it
		cannot be decrypted; the caller skips the delivery.
		"""
		if repo is None:
		return None
		@@ -54,6 +59,8 @@ async def load_bearer_token(
		return None
		if credential.expires_at is not None and credential.expires_at <= now:
		raise CallbackCredentialExpiredError("callback credential has expired")
		if credential.encrypted_token is None:
		raise CallbackCredentialRevokedError("callback credential has been revoked")
		if key is None:
		raise CallbackSecretError("no callback secret key is configured")
		return decrypt_secret(credential.encrypted_token, key)
		@@ -62,5 +69,5 @@ async def load_bearer_token(
		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)
		"""Wipes tokens past their expiry; the rows stay so later deliveries are still skipped."""
		return await repo.revoke_expired(now)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+12 −2

Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 12 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -52,7 +52,11 @@ 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.errors import (
		CallbackCredentialExpiredError,
		CallbackCredentialRevokedError,
		CallbackSecretError,
		)
		from open_exposure_gateway.domain.models import (
		AppDeployment,
		AppDeploymentState,
		@@ -1310,6 +1314,12 @@ class EdgeApplicationManagementService:
		callback_registration_id=str(registration.id),
		)
		return
		except CallbackCredentialRevokedError:
		logger.info(
		"callback_skipped_credential_revoked",
		callback_registration_id=str(registration.id),
		)
		return
		except CallbackSecretError as exc:
		logger.error(
		"callback_skipped_credential_unusable",
		@@ -1372,7 +1382,7 @@ class EdgeApplicationManagementService:
		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)
		await self._callback_credential_repo.revoke_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

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+12 −2

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 12 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -28,7 +28,11 @@ from open_exposure_gateway.core.exceptions import (
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted
		from open_exposure_gateway.domain.errors import CallbackCredentialExpiredError, CallbackSecretError
		from open_exposure_gateway.domain.errors import (
		CallbackCredentialExpiredError,
		CallbackCredentialRevokedError,
		CallbackSecretError,
		)
		from open_exposure_gateway.domain.models import (
		CallbackCredential,
		CallbackDelivery,
		@@ -274,6 +278,12 @@ class QualityOnDemandService:
		callback_registration_id=str(registration.id),
		)
		return
		except CallbackCredentialRevokedError:
		logger.info(
		"qod_callback_skipped_credential_revoked",
		callback_registration_id=str(registration.id),
		)
		return
		except CallbackSecretError as exc:
		logger.error(
		"qod_callback_skipped_credential_unusable",
		@@ -532,4 +542,4 @@ class QualityOnDemandService:
		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)
		await self._callback_credential_repo.revoke_by_registration_id(registration.id)