Commit 125c798e authored by George Papathanail's avatar George Papathanail
Browse files

feat(qod): store and send notificationAuthToken

parent 03a6c4d7

src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/schemas.py

+3 −1
Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v0_10_1/schemas.py: 3 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -88,7 +88,9 @@ class PortsSpec(BaseModel): 


class Webhook(BaseModel):

    notificationUrl: str

    notificationAuthToken: Optional[str] = Field(default=None, min_length=20, max_length=256)

    notificationAuthToken: Optional[str] = Field(

        default=None, min_length=20, max_length=256, repr=False

    )





class Message(BaseModel):

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+61 −6
Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 61 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -10,6 +10,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import ( 
    QosStatus,

    SessionInfo,

)

from open_exposure_gateway.application.callback_secrets import encrypt_secret, load_bearer_token

from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (

    build_activate_command,

    build_deactivate_command,
@@ -24,9 +25,12 @@ from open_exposure_gateway.core.exceptions import ( 
    BadRequestException,

    DownstreamServiceException,

    NotFoundException,

    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.models import (

    CallbackCredential,

    CallbackDelivery,

    CallbackRegistration,

    Operation,
@@ -37,6 +41,7 @@ from open_exposure_gateway.domain.models import ( 
)

from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus, Subject

from open_exposure_gateway.ports.database.callbacks import (

    CallbackCredentialRepository,

    CallbackDeliveryRepository,

    CallbackRegistrationRepository,

)
@@ -75,6 +80,8 @@ class QualityOnDemandService: 
        callback_delivery_repo: Optional[CallbackDeliveryRepository] = None,

        callback_delivery_port: Optional[QodCallbackDeliveryPort] = None,

        service_specification_id: UUID = DEFAULT_QOD_SERVICE_SPECIFICATION_ID,

        callback_credential_repo: Optional[CallbackCredentialRepository] = None,

        callback_secret_key: Optional[bytes] = None,

    ) -> None:

        self.srm_client = srm_client

        self._publisher = publisher
@@ -84,6 +91,8 @@ class QualityOnDemandService: 
        self._callback_delivery_repo = callback_delivery_repo

        self._callback_delivery_port = callback_delivery_port

        self._service_specification_id = service_specification_id

        self._callback_credential_repo = callback_credential_repo

        self._callback_secret_key = callback_secret_key



    def _parse_session_id(self, session_id: str) -> UUID:

        try:
@@ -115,6 +124,18 @@ class QualityOnDemandService: 
        if self._operation_repo is None or self._qod_session_repo is None:

            raise RuntimeError("Operation/QodSession repositories are not available")



        # Before anything is saved: without a key, OEG would accept a token it cannot

        # store and the webhook would never get an authenticated notification.

        notification_auth_token = (

            request.webhook.notificationAuthToken if request.webhook is not None else None

        )

        if notification_auth_token is not None and self._callback_secret_key is None:

            raise NotImplementedException(

                "webhook.notificationAuthToken is not supported by this gateway: no "

                "callback secret key is configured, so an authenticated webhook would "

                "receive no notifications"

            )



        operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)

        session_id = uuid4()

        # The platform-seeded QoD specification, identical on every session (ADR-0035).
@@ -171,7 +192,7 @@ class QualityOnDemandService: 
        if request.webhook is not None:

            if self._callback_registration_repo is None:

                raise RuntimeError("CallbackRegistrationRepository is not available")

            await self._callback_registration_repo.save(

            registration = await self._callback_registration_repo.save(

                CallbackRegistration(

                    id=uuid4(),

                    operation_id=operation_id,
@@ -179,10 +200,21 @@ class QualityOnDemandService: 
                    api_family="quality-on-demand",

                    sink=request.webhook.notificationUrl,

                    event_types=["org.camaraproject.qod.v0.qos-status-changed"],

                    sink_credential_ref=(

                        f"secret://oeg/{operation_id}/notification-auth-token"

                        if request.webhook.notificationAuthToken

                        else None

                    sink_credential_ref=None,

                    expires_at=None,

                )

            )

            if notification_auth_token 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"

                    )

                # CAMARA QoD tokens carry no expiry.

                await self._callback_credential_repo.save(

                    CallbackCredential(

                        callback_registration_id=registration.id,

                        encrypted_token=encrypt_secret(

                            notification_auth_token, self._callback_secret_key

                        ),

                        expires_at=None,

                    )
@@ -225,6 +257,27 @@ class QualityOnDemandService: 
        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(

                "qod_callback_skipped_credential_expired",

                callback_registration_id=str(registration.id),

            )

            return

        except CallbackSecretError as exc:

            logger.error(

                "qod_callback_skipped_credential_unusable",

                callback_registration_id=str(registration.id),

                error=str(exc),

            )

            return



        source = f"{get_settings().public_base_url}/qod/v0/sessions/{session_id}"

        cloud_event = build_qos_status_changed_event(

            session_id=session_id,
@@ -239,7 +292,9 @@ class QualityOnDemandService: 
        )

        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"

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
@@ -266,6 +266,7 @@ def get_qod_service( 
    ),

    callback_delivery_repo: CallbackDeliveryRepository = Depends(get_callback_delivery_repo),

    callback_delivery_port: QodCallbackDeliveryPort = Depends(get_qod_callback_client),

    callback_credential_repo: CallbackCredentialRepository = Depends(get_callback_credential_repo),

) -> QualityOnDemandService:

    return QualityOnDemandService(

        srm,
@@ -276,6 +277,8 @@ def get_qod_service( 
        callback_delivery_repo,

        callback_delivery_port,

        get_settings().qod_settings.service_specification_id,

        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
@@ -191,6 +191,8 @@ def _build_qod_service( 
        callback_delivery_repo=SqlCallbackDeliveryRepository(session),

        callback_delivery_port=qod_callback_client,

        service_specification_id=get_settings().qod_settings.service_specification_id,

        callback_credential_repo=SqlCallbackCredentialRepository(session),

        callback_secret_key=get_settings().callback_settings.secret_key_bytes,

    )

tests/unit/conftest.py

+8 −0
Changes for tests/unit/conftest.py: 8 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -235,6 +235,8 @@ def live_qod( 
    callback_registration_repo: FakeCallbackRegistrationRepository,

    callback_delivery_repo: FakeCallbackDeliveryRepository,

    qod_callback_delivery_port: FakeQodCallbackDeliveryPort,

    callback_credential_repo: FakeCallbackCredentialRepository,

    callback_secret_key: bytes | None,

) -> None:

    """Wires QoD's real completion/status handlers behind the fake bus, sharing the

    same repo/port instances the api_client's qod_service writes to and delivers with."""
@@ -245,6 +247,8 @@ def live_qod( 
        callback_registration_repo,

        callback_delivery_repo,

        qod_callback_delivery_port,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=callback_secret_key,

    )
@@ -257,6 +261,8 @@ def qod_service( 
    callback_registration_repo: FakeCallbackRegistrationRepository,

    callback_delivery_repo: FakeCallbackDeliveryRepository,

    qod_callback_delivery_port: FakeQodCallbackDeliveryPort,

    callback_credential_repo: FakeCallbackCredentialRepository,

    callback_secret_key: bytes | None,

) -> QualityOnDemandService:

    return QualityOnDemandService(

        srm_client=fake_srm,
@@ -266,6 +272,8 @@ def qod_service( 
        callback_registration_repo=callback_registration_repo,

        callback_delivery_repo=callback_delivery_repo,

        callback_delivery_port=qod_callback_delivery_port,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=callback_secret_key,

    )