fix: align TI/QoD/EAM calls with SRM contract and validate app server addresses

George Papathanailrequested to merge
fix/critical-gaps into develop

What

Fixes four places where OEG's calls to SRM didn't match SRM's API, or where OEG accepted input that SRM later rejected.

Changes

GET /traffic-influences always returned 503 SRM's GET /internal/network-capabilities requires app_provider_id, but OEG didn't send it. OEG now passes the caller's app_provider_id.

Network-capability reads by ID expected one object, but SRM returns a list GET /internal/network-capabilities/{id} returns a list of capabilities for the instance. get_traffic_influence and get_network_capability now keep the items of the expected capability_type (traffic_influence / qod_session) and take the newest by created_at. If none match, the result is 404. One type per instance today, but SRM plans to allow several.

Duplicate POST /apps returned 503 instead of 409 submit_app now checks the local registration before calling SRM. An SRM 409 on catalog create (e.g. two concurrent requests) is mapped to AlreadyExistsException; other SRM errors are unchanged.

QoD applicationServer addresses weren't validated

  • ipv4Address / ipv6Address must parse as an address or address/mask; otherwise 400. Before, OEG returned 201 and the session then failed in SRM.
  • Masked values are sent to SRM as their network (10.0.0.5/24 → 10.0.0.0/24), which CAMARA allows but SRM's strict parsing rejected. Plain addresses are sent unchanged, and the API response still echoes the original value.

Tests

  • SRM client: app_provider_id sent; newest capability of the right type chosen; 404 when no item of that type; SRM 409 → AlreadyExistsException.
  • QoD flows: invalid addresses → 400 with no activate published; masked IPv4/IPv6 normalised in the SRM command.
  • EAM: duplicate appId → 409, and SRM is not called.
  • pytest tests/unit: 1190 passed. ruff, ruff format and mypy are clean.

Merge request reports

src/open_exposure_gateway/adapters/http/srm_client.py

+25 −3
Changes for src/open_exposure_gateway/adapters/http/srm_client.py: 25 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -7,6 +7,7 @@ from pydantic import ValidationError 


from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.exceptions import (

    AlreadyExistsException,

    DownstreamServiceException,

    ErrorCode,

    NotFoundException,
@@ -70,6 +71,17 @@ def _problem_document(response: httpx.Response) -> dict[str, Any] | None: 
    return body if isinstance(body, dict) else None





def _latest_capability(data: Any, capability_type: str) -> dict[str, Any]:

    matching = [

        capability

        for capability in (data if isinstance(data, list) else [data])

        if isinstance(capability, dict) and capability.get("capability_type") == capability_type

    ]

    if not matching:

        raise NotFoundException(message="Resource not found")

    return max(matching, key=lambda capability: capability.get("created_at") or "")





class SRMClient:

    def __init__(self) -> None:

        settings = get_settings()
@@ -198,7 +210,7 @@ class SRMClient: 
        data = await self._request(

            "GET", f"/internal/network-capabilities/{service_instance_id}", headers=headers

        )

        return SRMNetworkCapability.model_validate(data)

        return SRMNetworkCapability.model_validate(_latest_capability(data, "qod_session"))



    async def get_traffic_influence(

        self,
@@ -209,14 +221,17 @@ class SRMClient: 
        data = await self._request(

            "GET", f"/internal/network-capabilities/{traffic_influence_id}", headers=headers

        )

        return SRMTrafficInfluenceCapability.model_validate(data)

        return SRMTrafficInfluenceCapability.model_validate(

            _latest_capability(data, "traffic_influence")

        )



    async def get_traffic_influences(

        self,

        app_provider_id: str,

        service_specification_id: UUID | None = None,

        x_correlator: str | None = None,

    ) -> list[SRMTrafficInfluenceCapability]:

        params = {"capability_type": "traffic_influence"}

        params = {"app_provider_id": app_provider_id, "capability_type": "traffic_influence"}

        if service_specification_id is not None:

            params["service_specification_id"] = str(service_specification_id)

        headers = {"x-correlator": x_correlator} if x_correlator else None
@@ -248,9 +263,16 @@ class SRMClient: 
        self, payload: dict[str, Any], x_correlator: str | None = None

    ) -> SRMCatalogServiceSpecificationCreated:

        headers = {"x-correlator": x_correlator} if x_correlator else None

        try:

            data = await self._request(

                "POST", "/internal/catalog/service-specifications", json=payload, headers=headers

            )

        except DownstreamServiceException as exc:

            if isinstance(exc.details, dict) and exc.details.get("status_code") == 409:

                raise AlreadyExistsException(

                    message="An app with this appId is already registered"

                ) from exc

            raise

        return SRMCatalogServiceSpecificationCreated.model_validate(data)



    async def delete_app(self, app_id: UUID, x_correlator: str | None = None) -> None:

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

+18 −1
Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/schemas.py: 18 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -5,7 +5,12 @@ from uuid import UUID 


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



from open_exposure_gateway.api.camara.common import validated_ipv4, validated_ipv6

from open_exposure_gateway.api.camara.common import (

    normalized_ipv4_network,

    normalized_ipv6_network,

    validated_ipv4,

    validated_ipv6,

)



SOURCE_API = "quality-on-demand"

SOURCE_API_VERSION = "1.1.0"
@@ -90,6 +95,18 @@ class ApplicationServer(BaseModel): 
    ipv4Address: Optional[str] = None

    ipv6Address: Optional[str] = None



    @field_validator("ipv4Address")

    @classmethod

    def _validate_ipv4_network(cls, value: Optional[str]) -> Optional[str]:

        normalized_ipv4_network(value)

        return value



    @field_validator("ipv6Address")

    @classmethod

    def _validate_ipv6_network(cls, value: Optional[str]) -> Optional[str]:

        normalized_ipv6_network(value)

        return value



    @model_validator(mode="after")

    def _require_at_least_one_address(self) -> "ApplicationServer":

        if not any((self.ipv4Address, self.ipv6Address)):

src/open_exposure_gateway/api/camara/common.py

+15 −1
Changes for src/open_exposure_gateway/api/camara/common.py: 15 added lines, 1 removed line.
Original line number Diff line number Diff line 
import re

from ipaddress import IPv4Address, IPv6Address

from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network

from typing import Annotated, Optional



from fastapi import Header
@@ -29,3 +29,17 @@ def validated_ipv6(value: Optional[str]) -> Optional[str]: 
    if value is not None:

        IPv6Address(value)

    return value





def normalized_ipv4_network(value: Optional[str]) -> Optional[str]:

    if value is None:

        return None

    network = IPv4Network(value, strict=False)

    return str(network) if "/" in value else value





def normalized_ipv6_network(value: Optional[str]) -> Optional[str]:

    if value is None:

        return None

    network = IPv6Network(value, strict=False)

    return str(network) if "/" in value else value

src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py

+6 −2
Changes for src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py: 6 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -2,6 +2,10 @@ from datetime import datetime, timedelta, timezone 
from typing import Literal

from uuid import UUID, uuid4



from open_exposure_gateway.api.camara.common import (

    normalized_ipv4_network,

    normalized_ipv6_network,

)

from open_exposure_gateway.api.camara.quality_on_demand.v1_1_0.schemas import (

    SOURCE_API,

    SOURCE_API_VERSION,
@@ -82,8 +86,8 @@ def build_activate_command( 
            ports=_to_capability_ports(request.devicePorts),

        ),

        application_server=NetworkCapabilityTargetApplicationServer(

            ipv4=application_server.ipv4Address,

            ipv6=application_server.ipv6Address,

            ipv4=normalized_ipv4_network(application_server.ipv4Address),

            ipv6=normalized_ipv6_network(application_server.ipv6Address),

            ports=_to_capability_ports(request.applicationServerPorts),

        ),

    )

src/open_exposure_gateway/application/services/edge_application_management_service.py

+5 −2
Changes for src/open_exposure_gateway/application/services/edge_application_management_service.py: 5 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -388,6 +388,11 @@ class EdgeApplicationManagementService: 
                "is not supported in this release"

            )



        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        if await self._app_registration_repo.get_by_app_id(app_id) is not None:

            raise AlreadyExistsException(message=f"App {app_id} is already registered")



        app_registration_id = uuid4()

        translation = build_app_registration_translation(

            manifest, app_id, app_registration_id, tenant_id, app_provider_id
@@ -406,8 +411,6 @@ class EdgeApplicationManagementService: 
                },

            )



        if self._app_registration_repo is None:

            raise RuntimeError("AppRegistrationRepository is not available")

        try:

            await self._app_registration_repo.save(

                AppRegistration(