fix: enforce app_provider_id ownership on EAM and TI app references

George Papathanailrequested to merge
fix/eam-ownership-checks into develop

What

Adds app_provider_id ownership checks to the Edge App Management endpoints that act on an app, instance or deployment by ID, and to the appId check in Traffic Influence create. Before this, any authenticated caller could read, delete, deploy or reference resources owned by another app provider.

Changes

Endpoint Owner source On mismatch
GET /apps/{appId} service_specification.app_provider_id from the SRM catalog 404
DELETE /apps/{appId} same, checked before the SRM delete 404, nothing deleted
POST /appinstances, POST /deployments (appId) same 400 "App … is not registered"
POST /traffic-influences (appId) same (replaces the previous tenant_id comparison) 400
DELETE /appinstances/{id}, DELETE /deployments/{id} app_provider_id of the deploy operation 404, no terminate published
  • Reads and deletes return 404 rather than 403, so callers can't probe which IDs exist (same pattern as QoD sessions and TI).
  • Ownership is based on app_provider_id only; tenant_id is not used.
  • New helper _get_owned_app in EdgeApplicationManagementService; the router now passes the caller to get_app.

Behaviour notes

  • An instance or deployment whose deploy operation row is missing now returns 404 on delete. This matches how GET /deployments already filters.
  • The app checks make one SRM catalog read. GET /apps/{appId} already did this; for the other paths it's one extra call.

Tests

  • New tests per endpoint: another provider is rejected, and nothing is deleted, persisted or published.
  • Test fixtures now seed an owner: the SRM catalog entry, and the deploy operation for delete tests.
  • The TI test for the old tenant_id rule is replaced by test_other_providers_app_id_is_rejected.
  • pytest tests/unit: 1146 passed. ruff, ruff format and mypy are clean.

Merge request reports

src/open_exposure_gateway/api/camara/edge_application_management/v0_1_0_alpha_1/router.py

+3 −1
Changes for src/open_exposure_gateway/api/camara/edge_application_management/v0_1_0_alpha_1/router.py: 3 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -145,7 +145,9 @@ async def get_app( 
    service: EdgeAppService,

    caller: Caller,

) -> Any:

    return await service.get_app(app_id=appId, x_correlator=caller.x_correlator)

    return await service.get_app(

        app_id=appId, app_provider_id=caller.app_provider_id, x_correlator=caller.x_correlator

    )





@router.post(

src/open_exposure_gateway/application/services/edge_application_management_service.py

+40 −20
Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 40 added lines, 20 removed lines.
Original line number Diff line number Diff line
@@ -48,6 +48,7 @@ from open_exposure_gateway.core.exceptions import ( 
    NotImplementedException,

)

from open_exposure_gateway.domain.edge_application_management import (

    SRMCatalogPayload,

    SRMCompletedInstance,

    SRMOperationCompleted,

    Subject,
@@ -350,17 +351,12 @@ class EdgeApplicationManagementService: 
            ) from exc



    async def get_app(

        self, app_id: UUID, x_correlator: Optional[str] = None

        self, app_id: UUID, app_provider_id: str, x_correlator: Optional[str] = None

    ) -> AppManifestEnvelope:

        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        app_registration = await self._app_registration_repo.get_by_app_id(app_id)

        if app_registration is None:

        owned = await self._get_owned_app(app_id, app_provider_id, x_correlator)

        if owned is None:

            raise NotFoundException(message=f"App {app_id} not found")



        catalog = await self.srm_client.get_app(

            app_id=app_registration.app_registration_id, x_correlator=x_correlator

        )

        _, catalog = owned

        try:

            manifest = build_app_manifest(catalog)

        except (ValueError, TypeError, IndexError) as exc:
@@ -452,9 +448,10 @@ class EdgeApplicationManagementService: 
        if self._app_instance_repo is None:

            raise RuntimeError("AppInstanceRepository is not available")



        app_registration = await self._app_registration_repo.get_by_app_id(app_id)

        if app_registration is None:

        owned = await self._get_owned_app(app_id, app_provider_id, x_correlator)

        if owned is None:

            raise NotFoundException(message=f"App {app_id} not found")

        app_registration, _ = owned



        has_instances = await self._app_instance_repo.exists_for_app_registration(

            app_registration.app_registration_id
@@ -506,9 +503,10 @@ class EdgeApplicationManagementService: 
                    edgeCloudZoneId=request.edgeCloudZoneId,

                )



        app_registration = await self._app_registration_repo.get_by_app_id(request.appId)

        if app_registration is None:

        owned = await self._get_owned_app(request.appId, app_provider_id, x_correlator)

        if owned is None:

            raise BadRequestException(message=f"App {request.appId} is not registered")

        app_registration, _ = owned



        if await self._app_instance_repo.exists_in_zone(

            app_registration.app_registration_id, request.edgeCloudZoneId
@@ -610,9 +608,10 @@ class EdgeApplicationManagementService: 
                    )

                return AppDeploymentId(appDeploymentId=existing_deployment.app_deployment_id)



        app_registration = await self._app_registration_repo.get_by_app_id(request.appId)

        if app_registration is None:

        owned = await self._get_owned_app(request.appId, app_provider_id, x_correlator)

        if owned is None:

            raise BadRequestException(message=f"App {request.appId} is not registered")

        app_registration, _ = owned



        if len(set(request.edgeCloudZones)) != len(request.edgeCloudZones):

            raise BadRequestException(message="edgeCloudZones must not contain duplicates")
@@ -788,12 +787,12 @@ class EdgeApplicationManagementService: 
        ):

            raise NotFoundException(message=f"App deployment {app_deployment_id} not found")



        # Close an in-flight deploy so its late completion is ignored.

        deploy_operation = await self._operation_repo.get_by_id(app_deployment.operation_id)

        if (

            deploy_operation is not None

            and deploy_operation.status not in _TERMINAL_OPERATION_STATUSES

        ):

        if deploy_operation is None or deploy_operation.app_provider_id != app_provider_id:

            raise NotFoundException(message=f"App deployment {app_deployment_id} not found")



        # Close an in-flight deploy so its late completion is ignored.

        if deploy_operation.status not in _TERMINAL_OPERATION_STATUSES:

            await self._operation_repo.save(

                deploy_operation.model_copy(

                    update={
@@ -968,6 +967,9 @@ class EdgeApplicationManagementService: 
        app_instance = await self._app_instance_repo.get_by_id(app_instance_id)

        if app_instance is None or app_instance.state == AppInstanceState.TERMINATED:

            raise NotFoundException(message=f"App instance {app_instance_id} not found")

        deploy_operation = await self._operation_repo.get_by_id(app_instance.operation_id)

        if deploy_operation is None or deploy_operation.app_provider_id != app_provider_id:

            raise NotFoundException(message=f"App instance {app_instance_id} not found")



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

        command = build_terminate_instance_command(
@@ -1456,6 +1458,24 @@ class EdgeApplicationManagementService: 
            )

        )



    async def _get_owned_app(

        self, app_id: UUID, app_provider_id: str, x_correlator: Optional[str]

    ) -> Optional[tuple[AppRegistration, SRMCatalogPayload]]:

        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        app_registration = await self._app_registration_repo.get_by_app_id(app_id)

        if app_registration is None:

            return None

        try:

            catalog = await self.srm_client.get_app(

                app_id=app_registration.app_registration_id, x_correlator=x_correlator

            )

        except NotFoundException:

            return None

        if catalog.service_specification.app_provider_id != app_provider_id:

            return None

        return app_registration, catalog



    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/application/services/traffic_influence_service.py

+17 −4
Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 17 added lines, 4 removed lines.
Original line number Diff line number Diff line
@@ -177,12 +177,23 @@ class TrafficInfluenceService: 
            return instances[0].zone_id

        return None



    async def _resolve_app_registration_id(self, app_id: UUID, tenant_id: str) -> UUID:

    async def _resolve_app_registration_id(

        self, app_id: UUID, app_provider_id: str, x_correlator: Optional[str]

    ) -> UUID:

        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        not_registered = BadRequestException(message=f"App {app_id} is not registered")

        app_registration = await self._app_registration_repo.get_by_app_id(app_id)

        if app_registration is None or app_registration.tenant_id != tenant_id:

            raise BadRequestException(message=f"App {app_id} is not registered")

        if app_registration is None:

            raise not_registered

        try:

            catalog = await self.srm_client.get_app(

                app_id=app_registration.app_registration_id, x_correlator=x_correlator

            )

        except NotFoundException:

            raise not_registered from None

        if catalog.service_specification.app_provider_id != app_provider_id:

            raise not_registered

        return app_registration.app_registration_id



    async def _replay_create(self, operation: Operation) -> TrafficInfluenceResponse:
@@ -242,7 +253,9 @@ class TrafficInfluenceService: 
            if existing is not None:

                return await self._replay_create(existing)



        app_registration_id = await self._resolve_app_registration_id(request.appId, tenant_id)

        app_registration_id = await self._resolve_app_registration_id(

            request.appId, app_provider_id, x_correlator

        )



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

        traffic_influence_id = uuid4()

tests/unit/test_eam_service.py

+164 −40

File changed.

Preview size limit exceeded, changes collapsed.

tests/unit/test_traffic_influence_flows.py

+20 −7
Changes for tests/unit/test_traffic_influence_flows.py: 20 added lines, 7 removed lines.
Original line number Diff line number Diff line
@@ -80,7 +80,9 @@ TI_APP_REGISTRATION_ID = UUID("11111111-2222-3333-4444-555555555555") 




@pytest.fixture(autouse=True)

def registered_app(app_registration_repo: FakeAppRegistrationRepository) -> None:

def registered_app(

    app_registration_repo: FakeAppRegistrationRepository, fake_srm: FakeSRMClient

) -> None:

    """A Traffic Influence request always targets an already-registered app."""

    app_registration_repo.rows[TI_APP_REGISTRATION_ID] = AppRegistration(

        app_registration_id=TI_APP_REGISTRATION_ID,
@@ -91,6 +93,18 @@ def registered_app(app_registration_repo: FakeAppRegistrationRepository) -> None 
        package_type=PackageType.HELM,

        status=AppRegistrationStatus.REGISTERED,

    )

    fake_srm.catalog[str(TI_APP_REGISTRATION_ID)] = {

        "service_specification": {

            "id": str(TI_APP_REGISTRATION_ID),

            "ref": TI_BODY["appId"],

            "name": "traffic-influence-app",

            "version": "1.0.0",

            "app_provider_id": "test_app_provider",

            "descriptor": {},

        },

        "service_deployment_units": [],

        "service_capability_requirements": [],

    }





def _ready_app_instance(app_instance_repo: FakeAppInstanceRepository, zone_id: UUID) -> AppInstance:
@@ -277,17 +291,16 @@ class TestTrafficInfluenceCreateFlow: 
        assert not [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE]

        assert traffic_influence_repo.rows == {}



    def test_other_tenants_app_id_is_rejected(

    def test_other_providers_app_id_is_rejected(

        self,

        api_client: TestClient,

        app_registration_repo: FakeAppRegistrationRepository,

        fake_srm: FakeSRMClient,

        fake_bus: FakeDataBus,

        traffic_influence_repo: FakeTrafficInfluenceRepository,

    ) -> None:

        registration = app_registration_repo.rows[TI_APP_REGISTRATION_ID]

        app_registration_repo.rows[TI_APP_REGISTRATION_ID] = registration.model_copy(

            update={"tenant_id": "other-tenant"}

        )

        fake_srm.catalog[str(TI_APP_REGISTRATION_ID)]["service_specification"][

            "app_provider_id"

        ] = "other-provider"



        response = api_client.post(f"{TI_BASE}/traffic-influences", json=TI_BODY)