feat: store sinkCredential and send it as a bearer on webhooks (b016faf4) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/traffic_influence_callback_client.py

+10 −6

Changes for src/open_exposure_gateway/adapters/http/traffic_influence_callback_client.py: 10 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -9,11 +9,15 @@ class HttpTrafficInfluenceCallbackClient:
		settings = get_settings()
		self.timeout = settings.callback_settings.timeout

		async def deliver(self, sink: str, event: TrafficInfluenceChangedCloudEvent) -> None:
		async def deliver(
		self,
		sink: str,
		event: TrafficInfluenceChangedCloudEvent,
		bearer_token: str \| None = None,
		) -> None:
		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=event.model_dump_json(),
		headers={"Content-Type": "application/cloudevents+json"},
		)
		response = await client.post(sink, content=event.model_dump_json(), headers=headers)
		response.raise_for_status()

src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py

+19 −3

Changes for src/open_exposure_gateway/api/camara/traffic_influence/v0_10_0/schemas.py: 19 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from datetime import datetime, timezone
		from enum import StrEnum
		from typing import Any, Literal, Optional
		from uuid import UUID

		from pydantic import BaseModel, Field
		from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator

		SOURCE_API = "traffic-influence"
		SOURCE_API_VERSION = "0.10.0"
		@@ -37,10 +37,26 @@ class Config(BaseModel):
		initialEvent: Optional[bool] = None


		class AccessTokenCredential(BaseModel):
		model_config = ConfigDict(extra="forbid")

		credentialType: Literal["ACCESSTOKEN"]
		accessToken: str = Field(min_length=1, max_length=8192, repr=False)
		accessTokenType: Literal["bearer"]
		accessTokenExpiresUtc: AwareDatetime

		@field_validator("accessTokenExpiresUtc")
		@classmethod
		def _expires_in_future(cls, value: datetime) -> datetime:
		if value <= datetime.now(timezone.utc):
		raise ValueError("accessTokenExpiresUtc must be in the future")
		return value


		class SubscriptionRequest(BaseModel):
		protocol: SubscriptionProtocol
		sink: str = Field(pattern=r"^https:\/\/.+$")
		sinkCredential: Optional[dict[str, Any]] = None
		sinkCredential: Optional[AccessTokenCredential] = None
		types: list[SubscriptionEventType] = Field(min_length=1, max_length=1)
		config: Config

src/open_exposure_gateway/application/services/traffic_influence_service.py

+111 −4

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 111 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		from open_exposure_gateway.api.camara.traffic_influence.v0_10_0.schemas import (
		TrafficInfluenceState as ApiTrafficInfluenceState,
		)
		from open_exposure_gateway.application.callback_secrets import encrypt_secret, load_bearer_token
		from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		build_activate_command,
		build_deactivate_command,
		@@ -28,9 +29,16 @@ from open_exposure_gateway.core.exceptions import (
		ConflictException,
		DownstreamServiceException,
		NotFoundException,
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted
		from open_exposure_gateway.domain.errors import (
		CallbackCredentialExpiredError,
		CallbackCredentialRevokedError,
		CallbackSecretError,
		)
		from open_exposure_gateway.domain.models import (
		CallbackCredential,
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		@@ -46,6 +54,7 @@ from open_exposure_gateway.domain.models import (
		from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus
		from open_exposure_gateway.domain.traffic_influence import SRMTrafficInfluenceCapability, Subject
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackCredentialRepository,
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		@@ -92,6 +101,8 @@ class TrafficInfluenceService:
		callback_delivery_port: Optional[TrafficInfluenceCallbackDeliveryPort] = None,
		app_registration_repo: Optional[AppRegistrationRepository] = None,
		app_instance_repo: Optional[AppInstanceRepository] = None,
		callback_credential_repo: Optional[CallbackCredentialRepository] = None,
		callback_secret_key: Optional[bytes] = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		@@ -102,6 +113,8 @@ class TrafficInfluenceService:
		self._callback_delivery_port = callback_delivery_port
		self._app_registration_repo = app_registration_repo
		self._app_instance_repo = app_instance_repo
		self._callback_credential_repo = callback_credential_repo
		self._callback_secret_key = callback_secret_key

		def _parse_traffic_influence_id(self, traffic_influence_id: str) -> UUID:
		try:
		@@ -212,6 +225,17 @@ class TrafficInfluenceService:
		message="Only HTTP notification delivery is supported for subscriptionRequest"
		)

		if (
		request.subscriptionRequest is not None
		and request.subscriptionRequest.sinkCredential is not None
		and self._callback_secret_key is None
		):
		raise NotImplementedException(
		"subscriptionRequest.sinkCredential is not supported by this gateway: "
		"no callback secret key is configured, so an authenticated sink would "
		"receive no notifications"
		)

		if idempotency_key is not None:
		existing = await self._operation_repo.get_by_idempotency_key(tenant_id, idempotency_key)
		if existing is not None:
		@@ -248,7 +272,9 @@ class TrafficInfluenceService:
		idempotency_key=idempotency_key,
		metadata={
		"app_id": str(request.appId),
		"request": request.model_dump(mode="json"),
		"request": request.model_dump(
		mode="json", exclude={"subscriptionRequest": {"sinkCredential"}}
		),
		},
		)
		try:
		@@ -284,7 +310,7 @@ class TrafficInfluenceService:
		# the generic callback table instead. Delivery happens later, keyed on this
		# operation_id, from handle_completed/handle_status_changed/delete_traffic_influence.
		if request.subscriptionRequest is not None and self._callback_registration_repo is not None:
		await self._callback_registration_repo.save(
		registration = await self._callback_registration_repo.save(
		CallbackRegistration(
		id=uuid4(),
		operation_id=operation_id,
		@@ -295,6 +321,21 @@ class TrafficInfluenceService:
		expires_at=request.subscriptionRequest.config.subscriptionExpireTime,
		)
		)
		credential = request.subscriptionRequest.sinkCredential
		if credential is not None:
		if self._callback_credential_repo is None or self._callback_secret_key is None:
		raise RuntimeError(
		"CallbackCredentialRepository/callback secret key is not available"
		)
		await self._callback_credential_repo.save(
		CallbackCredential(
		callback_registration_id=registration.id,
		encrypted_token=encrypt_secret(
		credential.accessToken, self._callback_secret_key
		),
		expires_at=credential.accessTokenExpiresUtc,
		)
		)

		await self._publish(
		Subject.TASK_ACTIVATE,
		@@ -312,7 +353,11 @@ class TrafficInfluenceService:
		edgeCloudZoneId=request.edgeCloudZoneId,
		sourceTrafficFilters=request.sourceTrafficFilters,
		destinationTrafficFilters=request.destinationTrafficFilters,
		subscriptionRequest=request.subscriptionRequest,
		subscriptionRequest=(
		request.subscriptionRequest.model_copy(update={"sinkCredential": None})
		if request.subscriptionRequest is not None
		else None
		),
		)

		async def _deliver_traffic_influence_changed(
		@@ -331,6 +376,42 @@ class TrafficInfluenceService:
		if self._callback_delivery_port is None or self._callback_delivery_repo is None:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		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(
		"traffic_influence_callback_skipped_credential_expired",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(
		registration.id, registration_operation_id, "credential_expired"
		)
		return
		except CallbackCredentialRevokedError:
		logger.info(
		"traffic_influence_callback_skipped_credential_revoked",
		callback_registration_id=str(registration.id),
		)
		await self._record_skipped_delivery(
		registration.id, registration_operation_id, "credential_revoked"
		)
		return
		except CallbackSecretError as exc:
		logger.error(
		"traffic_influence_callback_skipped_credential_unusable",
		callback_registration_id=str(registration.id),
		error=str(exc),
		)
		await self._record_skipped_delivery(
		registration.id, registration_operation_id, "credential_unusable"
		)
		return

		source = (
		f"{get_settings().public_base_url}/traffic-influence/v0.10/traffic-influences/"
		f"{traffic_influence.traffic_influence_id}"
		@@ -363,7 +444,9 @@ class TrafficInfluenceService:
		)
		last_error: Optional[str] = None
		try:
		await self._callback_delivery_port.deliver(registration.sink, cloud_event)
		await self._callback_delivery_port.deliver(
		registration.sink, cloud_event, bearer_token=bearer_token
		)
		state = "delivered"
		except Exception as exc:
		state = "failed"
		@@ -383,6 +466,28 @@ class TrafficInfluenceService:
		)
		)

		async def _record_skipped_delivery(
		self, callback_registration_id: UUID, operation_id: UUID, reason: str
		) -> None:
		assert self._callback_delivery_repo is not None
		await self._callback_delivery_repo.save(
		CallbackDelivery(
		id=uuid4(),
		callback_registration_id=callback_registration_id,
		operation_id=operation_id,
		attempt=1,
		state="skipped",
		last_error=reason,
		)
		)

		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.revoke_by_registration_id(registration.id)

		async def handle_completed(self, event: SRMOperationCompleted) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")
		@@ -480,6 +585,8 @@ class TrafficInfluenceService:
		# report (delete_session already notifies synchronously going UNAVAILABLE), but
		# TI's CAMARA spec requires 'deleted' itself to be an observable state.
		await self._deliver_traffic_influence_changed(saved.operation_id, saved, event.completed_at)
		if saved.state == RecordTrafficInfluenceState.DELETED:
		await self._forget_callback_credential(saved.operation_id)

		async def handle_status_changed(self, event: SRMOperationStatus) -> None:
		if self._operation_repo is None or self._traffic_influence_repo is None:

src/open_exposure_gateway/dependencies.py

+3 −0

Changes for src/open_exposure_gateway/dependencies.py: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -347,6 +347,7 @@ def get_traffic_influence_service(
		),
		app_registration_repo: AppRegistrationRepository = Depends(get_app_registration_repo),
		app_instance_repo: AppInstanceRepository = Depends(get_app_instance_repo),
		callback_credential_repo: CallbackCredentialRepository = Depends(get_callback_credential_repo),
		) -> TrafficInfluenceService:
		return TrafficInfluenceService(
		srm,
		@@ -358,4 +359,6 @@ def get_traffic_influence_service(
		callback_delivery_port,
		app_registration_repo,
		app_instance_repo,
		callback_credential_repo=callback_credential_repo,
		callback_secret_key=get_settings().callback_settings.secret_key_bytes,
		)

src/open_exposure_gateway/main.py

+2 −0

Changes for src/open_exposure_gateway/main.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -242,6 +242,8 @@ def _build_traffic_influence_service(
		callback_delivery_port=traffic_influence_callback_client,
		app_registration_repo=SqlAppRegistrationRepository(session),
		app_instance_repo=SqlAppInstanceRepository(session),
		callback_credential_repo=SqlCallbackCredentialRepository(session),
		callback_secret_key=get_settings().callback_settings.secret_key_bytes,
		)