Commit cfd3146a authored by George Papathanail's avatar George Papathanail
Browse files

fix: require operator role on federation routes

parent f7d72080

src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/problems.py

+7 −3
Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/problems.py: 7 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -161,9 +161,11 @@ def invalid_params(errors: Iterable[Mapping[str, Any]]) -> list[InvalidParam]: 
    ``loc``) is reported as ``body``. A leading FastAPI location (``path``,

    ``query``, ...) is dropped, so a bad path segment is reported by its name alone

    (e.g. ``partnerOpId``). The rest of the Pydantic error (``input``, ``ctx``,

    ``url``, ``type``) is deliberately dropped.

    ``url``, ``type``) is deliberately dropped. Exact duplicates are collapsed: a

    header declared by both the handler and an auth dependency (``x-correlator``)

    is validated, and so reported, once per declaration.

    """

    params = []

    params: list[InvalidParam] = []

    for error in errors:

        loc: Sequence[str | int] = error["loc"]

        if len(loc) > 1 and loc[0] in _REQUEST_LOCATIONS:
@@ -171,5 +173,7 @@ def invalid_params(errors: Iterable[Mapping[str, Any]]) -> list[InvalidParam]: 
        path = ""

        for part in loc:

            path += f"[{part}]" if isinstance(part, int) else f".{part}"

        params.append(InvalidParam(param=path.lstrip(".") or "body", reason=error["msg"]))

        param = InvalidParam(param=path.lstrip(".") or "body", reason=error["msg"])

        if param not in params:

            params.append(param)

    return params

src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py

+2 −1
Changes for src/open_exposure_gateway/api/gsma/federation_manager/v1_4_0/router.py: 2 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -58,6 +58,7 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.schemas import ( 
    ZoneRegistrationRequestData,

    ZoneRegistrationResponseData,

)

from open_exposure_gateway.core.authorization import require_role

from open_exposure_gateway.dependencies import get_fm_client



PARTNER_PREFIX = "/partners/{partnerOpId}/federations"
@@ -66,7 +67,7 @@ logger = structlog.get_logger(__name__) 


# Tags are set per route (not on the router) so each endpoint carries only its own

# GSMA operation's tag.

router = APIRouter(prefix=BASE_PATH)

router = APIRouter(prefix=BASE_PATH, dependencies=[Depends(require_role)])



FmClientDep = Annotated[FmClient, Depends(get_fm_client)]

src/open_exposure_gateway/core/authorization.py

+26 −1
Changes for src/open_exposure_gateway/core/authorization.py: 26 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -84,6 +84,31 @@ OPERATION_REQUIREMENTS: dict[str, OperationRequirements] = { 
    "POST /location-retrieval/v0.5/retrieve": OperationRequirements(

        "app-provider", scope="location-retrieval:read"

    ),

    # Federation Manager (platform namespace, ADR-0051): operator-only, no CAMARA scope.

    "POST /platform/v1/federation/partners/{partnerOpId}/federations": (

        OperationRequirements("operator")

    ),

    "GET /platform/v1/federation/partners/{partnerOpId}/federations": (

        OperationRequirements("operator")

    ),

    "GET /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}": (

        OperationRequirements("operator")

    ),

    "DELETE /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}": (

        OperationRequirements("operator")

    ),

    "GET /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}/zones": (

        OperationRequirements("operator")

    ),

    "POST /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}/zones": (

        OperationRequirements("operator")

    ),

    "GET /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}/zones/{zoneId}": (  # noqa: E501

        OperationRequirements("operator")

    ),

    "DELETE /platform/v1/federation/partners/{partnerOpId}/federations/{federationContextId}/zones/{zoneId}": (  # noqa: E501

        OperationRequirements("operator")

    ),

}
@@ -98,7 +123,7 @@ def _requirements_for(request: Request) -> OperationRequirements: 
    requirements = OPERATION_REQUIREMENTS.get(key)

    if requirements is None:

        # Fail closed: an unlisted route is a bug, not something to let through.

        raise RuntimeError(f"No CAMARA operation requirements registered for {key!r}")

        raise RuntimeError(f"No operation requirements registered for {key!r}")

    return requirements

tests/unit/conftest.py

+1 −1
Changes for tests/unit/conftest.py: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -57,7 +57,7 @@ def fake_caller_context(request: Request, x_correlator: XCorrelatorHeader = None 
        x_correlator=x_correlator_header(request),

        tenant_id="test_tenant",

        app_provider_id="test_app_provider",

        roles=["app-provider"],

        roles=["app-provider", "operator"],

        # Every CAMARA scope, not just the ones exercised by a given test: this fixture

        # stands in for a fully-authorized caller so business-logic tests don't also have

        # to know about require_scope's per-operation table.

tests/unit/test_authorization.py

+34 −0
Changes for tests/unit/test_authorization.py: 34 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -8,9 +8,11 @@ from typing import Optional 


import pytest

from fastapi import Request

from fastapi.routing import APIRoute

from fastapi.testclient import TestClient



from open_exposure_gateway.core.authorization import (

    OPERATION_REQUIREMENTS,

    require_role,

    require_scope,

)
@@ -29,6 +31,7 @@ _LOCATION_RETRIEVAL_ROUTE = ( 
)  # scope: location-retrieval:read

# EAM's openId scheme requires its scope -- unlike QoD's routes above.

_SCOPE_ALWAYS_REQUIRED_ROUTE = ("GET", "/edge-application-management/v0.1alpha1/apps")

_FEDERATION_ROUTE = ("POST", "/platform/v1/federation/partners/{partnerOpId}/federations")





def _request(method: str, path: str) -> Request:
@@ -84,6 +87,37 @@ class TestRequireRole: 
        result = require_role(request, caller, _settings(enforcement_enabled=False))

        assert result is caller



    def test_app_provider_is_rejected_on_a_federation_route(self) -> None:

        request = _request(*_FEDERATION_ROUTE)

        caller = _caller(roles=["app-provider"])

        with pytest.raises(ForbiddenException):

            require_role(request, caller, _settings())



    def test_operator_is_allowed_on_a_federation_route(self) -> None:

        request = _request(*_FEDERATION_ROUTE)

        caller = _caller(roles=["operator"])

        assert require_role(request, caller, _settings()) is caller





class TestEveryRouteIsRegistered:

    def test_every_guarded_route_has_requirements(self) -> None:

        # Catches a route added under a guarded router but forgotten in the table,

        # which would otherwise only surface as a 500 at request time.

        guarded_prefixes = (

            "/qod/",

            "/edge-application-management/",

            "/location-retrieval/",

            "/platform/v1/federation/",

        )

        missing = [

            f"{method} {route.path}"

            for route in app.routes

            if isinstance(route, APIRoute) and route.path.startswith(guarded_prefixes)

            for method in route.methods or ()

            if f"{method} {route.path}" not in OPERATION_REQUIREMENTS

        ]

        assert missing == []





class TestRequireScope:

    def test_missing_scope_is_rejected(self) -> None: