Commit 67c86c1b authored by George Papathanail's avatar George Papathanail
Browse files

feat(eam): store app-instance sinkCredential encrypted

parent f2b35e49

src/open_exposure_gateway/application/services/edge_application_management_service.py

+52 −25
Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 52 added lines, 25 removed lines.
Original line number Diff line number Diff line
@@ -19,8 +19,9 @@ from open_exposure_gateway.api.camara.edge_application_management.v0_1_0_alpha_1 
    EdgeCloudZone,

    KubernetesResources,

    SubmittedApp,

    SubscriptionRequest,

)

from open_exposure_gateway.application.callback_secrets import load_bearer_token

from open_exposure_gateway.application.callback_secrets import encrypt_secret, load_bearer_token

from open_exposure_gateway.application.mappers.edge_application_mapper import (

    build_app_deployment_info,

    build_app_deployment_status_change_events,
@@ -59,6 +60,7 @@ from open_exposure_gateway.domain.models import ( 
    AppInstanceState,

    AppRegistration,

    AppRegistrationStatus,

    CallbackCredential,

    CallbackDelivery,

    CallbackRegistration,

    Operation,
@@ -477,6 +479,8 @@ class EdgeApplicationManagementService: 
        if self._operation_repo is None or self._app_instance_repo is None:

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



        self._require_callback_secret_key_for(request.subscriptionRequest)



        if idempotency_key is not None:

            existing_operation = await self._operation_repo.get_by_idempotency_key(

                tenant_id, idempotency_key
@@ -552,30 +556,7 @@ class EdgeApplicationManagementService: 
        )



        if request.subscriptionRequest is not None:

            if self._callback_registration_repo is None:

                raise RuntimeError("CallbackRegistrationRepository is not available")

            # TODO: the raw subscription.sinkCredential token is dropped here — the

            # secret:// ref below points at a store that does not exist yet, and

            # HttpCallbackClient sends no Authorization header, so any sink requiring

            # auth rejects every callback (ADR-0008 requires a sinkCredential bearer

            # token). Either persist the credential or reject sinkCredential with 501.

            subscription = request.subscriptionRequest

            await self._callback_registration_repo.save(

                CallbackRegistration(

                    id=uuid4(),

                    operation_id=operation_id,

                    tenant_id=tenant_id,

                    api_family="edge-application-management",

                    sink=subscription.sink,

                    event_types=list(subscription.types or []),

                    sink_credential_ref=f"secret://oeg/{operation_id}/sink-credential"

                    if subscription.sinkCredential is not None

                    else None,

                    expires_at=subscription.config.subscriptionExpireTime

                    if subscription.config

                    else None,

                )

            )

            await self._save_subscription(request.subscriptionRequest, operation_id, tenant_id)



        _command = build_deploy_command(translation, requested_at)

        await self._publish(
@@ -1398,6 +1379,52 @@ class EdgeApplicationManagementService: 
                    )

                )



    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

        cannot store and the sink would never get an authenticated notification."""

        if (

            subscription is not None

            and subscription.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"

            )



    async def _save_subscription(

        self, subscription: SubscriptionRequest, operation_id: UUID, tenant_id: str

    ) -> None:

        if self._callback_registration_repo is None:

            raise RuntimeError("CallbackRegistrationRepository is not available")

        registration = await self._callback_registration_repo.save(

            CallbackRegistration(

                id=uuid4(),

                operation_id=operation_id,

                tenant_id=tenant_id,

                api_family="edge-application-management",

                sink=subscription.sink,

                event_types=list(subscription.types or []),

                sink_credential_ref=None,

                expires_at=subscription.config.subscriptionExpireTime

                if subscription.config

                else None,

            )

        )

        credential = subscription.sinkCredential

        if credential is None:

            return

        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,

            )

        )



    async def _resolve_app_id(self, app_registration_id: UUID) -> Optional[UUID]:

        if self._app_registration_repo is None:

            return 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
@@ -241,6 +241,7 @@ def get_edge_app_service( 
        get_callback_registration_repo

    ),

    app_deployment_repo: AppDeploymentRepository = Depends(get_app_deployment_repo),

    callback_credential_repo: CallbackCredentialRepository = Depends(get_callback_credential_repo),

) -> EdgeApplicationManagementService:

    return EdgeApplicationManagementService(

        srm,
@@ -250,6 +251,8 @@ def get_edge_app_service( 
        app_instance_repo,

        callback_registration_repo,

        app_deployment_repo=app_deployment_repo,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=get_settings().callback_settings.secret_key_bytes,

    )

tests/unit/conftest.py

+21 −0
Changes for tests/unit/conftest.py: 21 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -35,6 +35,7 @@ from tests.unit.fakes import ( 
    FakeAppDeploymentRepository,

    FakeAppInstanceRepository,

    FakeAppRegistrationRepository,

    FakeCallbackCredentialRepository,

    FakeCallbackDeliveryPort,

    FakeCallbackDeliveryRepository,

    FakeCallbackRegistrationRepository,
@@ -148,6 +149,18 @@ def callback_delivery_repo() -> FakeCallbackDeliveryRepository: 
    return FakeCallbackDeliveryRepository()





@pytest.fixture()

def callback_credential_repo() -> FakeCallbackCredentialRepository:

    return FakeCallbackCredentialRepository()





@pytest.fixture()

def callback_secret_key() -> bytes | None:

    """Unset by default, as in a deployment without CALLBACK_SETTINGS__SECRET_KEY;

    override in a test class to exercise authenticated callbacks."""

    return None





@pytest.fixture()

def live_srm(

    fake_bus: FakeDataBus,
@@ -159,6 +172,8 @@ def live_srm( 
    callback_delivery_port: FakeCallbackDeliveryPort,

    callback_delivery_repo: FakeCallbackDeliveryRepository,

    app_deployment_repo: FakeAppDeploymentRepository,

    callback_credential_repo: FakeCallbackCredentialRepository,

    callback_secret_key: bytes | None,

) -> FakeSRMClient:

    """Fake SRM with its async side running: consumes commands, publishes completions."""

    wire_operation_consumer(
@@ -170,6 +185,8 @@ def live_srm( 
        callback_delivery_port=callback_delivery_port,

        callback_delivery_repo=callback_delivery_repo,

        app_deployment_repo=app_deployment_repo,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=callback_secret_key,

    )

    wire_srm_worker(fake_bus, fake_srm)

    return fake_srm
@@ -194,6 +211,8 @@ def eam_service( 
    app_instance_repo: FakeAppInstanceRepository,

    callback_registration_repo: FakeCallbackRegistrationRepository,

    app_deployment_repo: FakeAppDeploymentRepository,

    callback_credential_repo: FakeCallbackCredentialRepository,

    callback_secret_key: bytes | None,

) -> EdgeApplicationManagementService:

    return EdgeApplicationManagementService(

        srm_client=fake_srm,
@@ -203,6 +222,8 @@ def eam_service( 
        app_instance_repo=app_instance_repo,

        callback_registration_repo=callback_registration_repo,

        app_deployment_repo=app_deployment_repo,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=callback_secret_key,

    )

tests/unit/fakes.py

+4 −0
Changes for tests/unit/fakes.py: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -334,6 +334,8 @@ def wire_operation_consumer( 
    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,

) -> NatsOperationConsumer:

    """Wires OEG's real completion handler (EdgeApplicationManagementService.handle_completed)

    behind the fake bus -- pass the same repo/port instances used to build the service under
@@ -349,6 +351,8 @@ def wire_operation_consumer( 
        callback_delivery_port=callback_delivery_port,

        callback_delivery_repo=callback_delivery_repo,

        app_deployment_repo=app_deployment_repo,

        callback_credential_repo=callback_credential_repo,

        callback_secret_key=callback_secret_key,

    )

    consumer = NatsOperationConsumer(

        client=AsyncMock(), subject=Subject.OPERATION_COMPLETED, handler=service.handle_completed

tests/unit/test_eam_flows.py

+57 −0
Changes for tests/unit/test_eam_flows.py: 57 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -41,6 +41,7 @@ from tests.unit.fakes import ( 
    FakeAppDeploymentRepository,

    FakeAppInstanceRepository,

    FakeAppRegistrationRepository,

    FakeCallbackCredentialRepository,

    FakeCallbackDeliveryPort,

    FakeCallbackRegistrationRepository,

    FakeDataBus,
@@ -431,6 +432,62 @@ class TestCreateAppInstanceFlow: 
        assert stored_instance.state == AppInstanceState.READY





class TestCreateAppInstanceSinkCredentialFlow:

    @pytest.fixture(autouse=True)

    def _registered_app(self, register_app: Callable[[Any], None]) -> None:

        register_app(APP_ID)



    BODY: dict[str, Any] = {

        **CREATE_INSTANCE_BODY,

        "subscriptionRequest": {

            "sink": "https://client.example.com/callback",

            "sinkCredential": VALID_SINK_CREDENTIAL,

            "types": [

                "org.camaraproject.edge-application-management.v0.app-instance-status-change"

            ],

        },

    }



    def test_returns_501_and_saves_nothing_without_a_key(

        self,

        api_client: TestClient,

        fake_bus: FakeDataBus,

        operation_repo: FakeOperationRepository,

        callback_registration_repo: FakeCallbackRegistrationRepository,

    ) -> None:

        response = api_client.post(f"{EAM_BASE}/appinstances", json=self.BODY)

        assert response.status_code == 501

        assert operation_repo.rows == {}

        assert callback_registration_repo.rows == {}

        assert fake_bus.published == []





class TestCreateAppInstanceSinkCredentialWithKeyFlow:

    @pytest.fixture(autouse=True)

    def _registered_app(self, register_app: Callable[[Any], None]) -> None:

        register_app(APP_ID)



    @pytest.fixture()

    def callback_secret_key(self) -> bytes:

        return bytes(range(32))



    def test_token_is_sent_as_bearer_on_callback(

        self,

        api_client: TestClient,

        live_srm: FakeSRMClient,

        callback_delivery_port: FakeCallbackDeliveryPort,

        callback_credential_repo: FakeCallbackCredentialRepository,

    ) -> None:

        response = api_client.post(

            f"{EAM_BASE}/appinstances", json=TestCreateAppInstanceSinkCredentialFlow.BODY

        )



        assert response.status_code == 202

        (credential,) = list(callback_credential_repo.rows.values())

        assert VALID_SINK_CREDENTIAL["accessToken"].encode() not in credential.encrypted_token

        assert callback_delivery_port.bearer_tokens == [VALID_SINK_CREDENTIAL["accessToken"]]





class TestCreateAppInstanceUnregisteredAppFlow:

    """No autouse app registration here — this class exists to prove the

    unregistered-appId path specifically."""