Commit 8828b669 authored by George Papathanail's avatar George Papathanail
Browse files

Merge branch 'fix/eam-ownership-checks' into 'develop' 

fix: enforce app_provider_id ownership on EAM and TI app references

See merge request !38
parents 568078be 1cc2ac06

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)