Feat/callback sink credential (!33) · Merge requests · OOP / code / open-exposure-gateway

Authenticated callback delivery (EAM `sinkCredential`, QoD `notificationAuthToken`)

Why

Clients can give OEG a bearer token to send when it POSTs status notifications to their callback URL. OEG never sent it:

create_app_instance and QoD session creation saved a fake secret://oeg/... reference and threw the token away.
create_app_deployment rejected the token with 501.
Neither HTTP callback client sent an Authorization header.

So any client whose callback endpoint requires auth never received a notification.

These tokens are issued by the client and protect only OEG→client calls. They are not Keycloak tokens.

What this MR does

OEG now encrypts the token, stores it, and sends it as Authorization: Bearer <token> on every notification until the token expires. Then it deletes the token.

Storage is optional and controlled by one new setting:

`CALLBACK_SETTINGS__SECRET_KEY`	Request carrying a token	Request without a token
unset (default)	501, checked before anything is saved or published	unchanged
set (base64url of 32 bytes)	accepted (EAM 202, QoD 201); token stored encrypted and sent on each callback	unchanged

When the key is unset, OEG logs one warning at startup. A key that is set but malformed stops startup with a clear error.

How

Encryption: AES-GCM with a fresh 12-byte nonce for every token, stored as nonce + ciphertext. This follows the portal's session_crypto.py, with one difference: the portal uses a per-session key from a cookie, but OEG uses one app-wide key. OEG decrypts in the background (NATS consumer) when no client request is in flight to carry a key.
Storage: a new callback_credentials table, one row per callback registration:
- callback_registration_id (primary key and foreign key)
- encrypted_token
- expires_at, not encrypted, so the purge can query it
- created_at
It's a new table rather than a new column because create_all never alters existing tables.
Delivery: both EAM and QoD use one shared helper, load_bearer_token:
- no credential stored → send without auth (behaviour as before);
- token expired → skip the delivery and log a warning with the registration id;
- token can't be decrypted, or the key is missing → log, skip that delivery, and keep the consumer running.
Expiry:
- EAM: accessTokenExpiresUtc from the request.
- QoD: CAMARA gives no expiry, so OEG uses the session's end time plus a 1 h grace period (NOTIFICATION_AUTH_TOKEN_GRACE). That way the final DURATION_EXPIRED / NETWORK_TERMINATED notification still carries the token.
Cleanup:
- EAM: the credential is deleted once every instance created by the subscribing operation is TERMINATED. The final notification is still sent with the token first.
- QoD: DELETE /sessions/{id} deletes the credential right after the DELETE_REQUESTED notification.
- Expired credentials are purged at the start of each reconciliation pass, in a separate transaction. If the purge fails, reconciliation still runs.

API changes

EAM sinkCredential is typed: it's now an AccessTokenCredential instead of a dict. The fields are:
- credentialType: ACCESSTOKEN
- accessToken (up to 8192 characters)
- accessTokenType: bearer
- accessTokenExpiresUtc, which must include a time zone
All four are required and unknown fields are rejected, as in the vendored CAMARA spec. An expiry already in the past returns 400. Payloads in the old untyped shape now get 400.
400 validation messages no longer include the submitted values. Pydantic puts the offending input into each error, so a malformed credential copied its token into the 400 response. This applies to all CAMARA endpoints.
QoD notificationAuthToken and EAM app-instance sinkCredential: without a key these now return 501, where they used to return 201/202 with the token silently dropped.

Database

New table callback_credentials, created by create_all.
callback_registrations.sink_credential_ref is removed from the ORM model and domain model. Databases created before this MR still have the column. It's nullable and no longer written, so no migration is needed.

Secrets hygiene

Tokens never reach the database unencrypted. Tests check every saved operation, instance, deployment, session and registration row.
Tokens are never logged. Tests capture logs on the skip paths.
Token fields use repr=False, and the key is a SecretStr.
Settings validation errors don't echo their input, so a bad key never reaches the startup error log.
Changing the key makes every stored token unreadable. Those deliveries are skipped and logged.

Configuration

.env.example documents CALLBACK_SETTINGS__SECRET_KEY and how to generate one:

python -c "import os,base64;print(base64.urlsafe_b64encode(os.urandom(32)).decode())"

Also in this MR

cryptography is now a direct dependency. It used to come in only through pyjwt[crypto]. Relocked with uv 0.12.19.

Known limitations

The purge runs inside the reconciliation loop, so it only runs while reconciliation is enabled (the default). Delivery skips expired tokens either way.
Nothing in OEG deactivates or deletes a callback registration today, so cleanup relies on the final notification and the purge.
Skipped deliveries are logged but don't add a callback_deliveries row.
The QoD POST /sessions response returns webhook.notificationAuthToken to the caller as received, because CAMARA's SessionInfo reuses the request's Webhook schema.
QoD extend (POST /sessions/{id}/extend) is still a 501 stub. A TODO notes that it must move the token's expires_at too.

Testing

Unit:
- config: key unset, valid, malformed, and masked in output;
- AES-GCM: encrypt/decrypt round trip, wrong key, tampered data, fresh nonce each time;
- repository;
- delivery adapters: header present or absent;
- EAM and QoD delivery: no credential, valid, expired, undecryptable, missing key;
- storage with and without a key, including 501 before an idempotent replay;
- sinkCredential validation (400s, token not echoed);
- cleanup on terminate and delete, and purge.
HTTP flows: create with a token, the fake SRM completes, and the callback arrives with Authorization: Bearer <token>. Covered for app instances, deployments, and QoD (completion plus status change).
Postgres integration (testcontainers): repository CRUD and delete_expired; a real reconciliation pass removes only expired rows.
Results: pytest tests/unit 943 passed, 1 skipped. tests/integration/test_postgres.py 24 passed. ruff check, ruff format --check and mypy --strict all clean.

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

0 → 100644

+57 −0

Changes for src/open_exposure_gateway/adapters/database/repos/callback_credentials.py: 57 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from typing import Any, cast
		from uuid import UUID

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

		from open_exposure_gateway.adapters.database.mappers import CallbackCredentialMapper
		from open_exposure_gateway.adapters.database.sql import CallbackCredentialRow
		from open_exposure_gateway.domain.models import CallbackCredential
		from open_exposure_gateway.ports.database.callbacks import CallbackCredentialRepository


		class SqlCallbackCredentialRepository(CallbackCredentialRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_registration_id(
		self, callback_registration_id: UUID
		) -> CallbackCredential \| None:
		stmt = select(CallbackCredentialRow).where(
		CallbackCredentialRow.callback_registration_id == callback_registration_id
		)
		row = await self._session.scalar(stmt)
		return CallbackCredentialMapper.to_domain(row) if row is not None else None

		async def save(self, callback_credential: CallbackCredential) -> CallbackCredential:
		merged = await self._session.merge(CallbackCredentialMapper.to_row(callback_credential))
		await self._session.flush()
		stmt = select(CallbackCredentialRow).where(
		CallbackCredentialRow.callback_registration_id == merged.callback_registration_id
		)
		row = await self._session.scalar(stmt)
		if row is None:
		raise RuntimeError("Saved callback credential could not be reloaded")
		return CallbackCredentialMapper.to_domain(row)

		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 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/mappers.py

+21 −2

Changes for src/open_exposure_gateway/adapters/database/mappers.py: 21 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@ from open_exposure_gateway.adapters.database.sql import (
		AppDeploymentRow,
		AppInstanceRow,
		AppRegistrationRow,
		CallbackCredentialRow,
		CallbackDeliveryRow,
		CallbackRegistrationRow,
		OperationRow,
		@@ -13,6 +14,7 @@ from open_exposure_gateway.domain.models import (
		AppDeployment,
		AppInstance,
		AppRegistration,
		CallbackCredential,
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		@@ -182,7 +184,6 @@ class CallbackRegistrationMapper:
		api_family=row.api_family,
		sink=row.sink,
		event_types=row.event_types,
		sink_credential_ref=row.sink_credential_ref,
		expires_at=row.expires_at,
		is_active=row.is_active,
		created_at=row.created_at,
		@@ -197,12 +198,30 @@ class CallbackRegistrationMapper:
		api_family=domain.api_family,
		sink=domain.sink,
		event_types=domain.event_types,
		sink_credential_ref=domain.sink_credential_ref,
		expires_at=domain.expires_at,
		is_active=domain.is_active,
		)


		class CallbackCredentialMapper:
		@staticmethod
		def to_domain(row: CallbackCredentialRow) -> CallbackCredential:
		return CallbackCredential(
		callback_registration_id=row.callback_registration_id,
		encrypted_token=row.encrypted_token,
		expires_at=row.expires_at,
		created_at=row.created_at,
		)

		@staticmethod
		def to_row(domain: CallbackCredential) -> CallbackCredentialRow:
		return CallbackCredentialRow(
		callback_registration_id=domain.callback_registration_id,
		encrypted_token=domain.encrypted_token,
		expires_at=domain.expires_at,
		)


		class CallbackDeliveryMapper:
		@staticmethod
		def to_domain(row: CallbackDeliveryRow) -> CallbackDelivery:

src/open_exposure_gateway/adapters/database/sql.py

+20 −1

Changes for src/open_exposure_gateway/adapters/database/sql.py: 20 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -9,6 +9,7 @@ from sqlalchemy import (
		ForeignKey,
		Index,
		Integer,
		LargeBinary,
		MetaData,
		String,
		Text,
		@@ -140,7 +141,6 @@ class CallbackRegistrationRow(Base):
		tenant_id: Mapped[str] = mapped_column(String(255), nullable=False)
		api_family: Mapped[str] = mapped_column(String(64), nullable=False)
		sink: Mapped[str] = mapped_column(Text, nullable=False)
		sink_credential_ref: Mapped[str \| None] = mapped_column(String(255))
		event_types: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
		expires_at: Mapped[datetime \| None] = mapped_column(DateTime(timezone=True))
		is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
		@@ -151,6 +151,25 @@ class CallbackRegistrationRow(Base):
		)


		class CallbackCredentialRow(Base):
		# Separate table rather than a column on callback_registrations: create_all never
		# alters an existing table.
		__tablename__ = "callback_credentials"
		__table_args__ = (Index("idx_callback_credentials_expires_at", "expires_at"),)

		callback_registration_id: Mapped[UUID] = mapped_column(
		ForeignKey("callback_registrations.id"), primary_key=True
		)
		# 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),
		nullable=False,
		server_default=func.now(),
		)


		class CallbackDeliveryRow(Base):
		__tablename__ = "callback_deliveries"

src/open_exposure_gateway/adapters/http/callback_client.py

+7 −6

Changes for src/open_exposure_gateway/adapters/http/callback_client.py: 7 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -18,16 +18,17 @@ class HttpCallbackClient:
		settings = get_settings()
		self.timeout = settings.callback_settings.timeout

		async def deliver(self, sink: str, event: CallbackEvent) -> None:
		async def deliver(
		self, sink: str, event: CallbackEvent, bearer_token: str \| None = None
		) -> None:
		content: bytes = (
		_DEPLOYMENT_EVENTS_ADAPTER.dump_json(event)
		if isinstance(event, list)
		else event.model_dump_json().encode()
		)
		headers = {"Content-Type": "application/cloudevents+json"}
		if bearer_token is not None:
		headers["Authorization"] = f"Bearer {bearer_token}"
		async with httpx.AsyncClient(timeout=self.timeout) as client:
		response = await client.post(
		sink,
		content=content,
		headers={"Content-Type": "application/cloudevents+json"},
		)
		response = await client.post(sink, content=content, headers=headers)
		response.raise_for_status()

src/open_exposure_gateway/adapters/http/qod_callback_client.py