Commit 0d52047e authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

fix(fm): answer the OEG-proxied federation routes only with OPG.04 statuses

On /internal/partners/{partnerOpId}/federations/... OEG relays only the
statuses OPG.04 v1.4.0 declares (200/400/401/404/409/422/500/503/520)
and
turns anything else into its own 502 (OEG §K.6a), so a partner outage
reached the operator as "FM is broken". FM now maps onto that set there,
keeping the cause in `type`:

- partner unreachable 502 -> 503, partner response invalid 502 -> 520
- partner-not-active / agreement-expired 403 -> 409,
  agreement-violation 403 -> 422
- partner rejections: 400/404/409/422 relayed as problem+json (FM writes
  the body if the partner sent none); any other status -> 520
  partner-rejected, so a partner's 401/5xx never reads as ours
- request validation -> 400 problem+json with invalidParams
- unhandled exceptions -> 500 problem+json, no exception text

Inbound EWBI routes are unchanged.
parent d7a075d5
Loading
Loading
Loading
Loading
Loading
+91 −20
Changes for src/federation_manager/api/errors.py: 91 added lines, 20 removed lines.
Original line number Diff line number Diff line
import re
from typing import Any

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, PlainTextResponse
from pydantic.json_schema import models_json_schema

from federation_manager.contracts.ewbi import InvalidParam, ProblemDetails
@@ -89,21 +92,53 @@ def problem(
    detail: str,
    instance: str,
    headers: dict[str, str] | None = None,
    invalid_params: list[InvalidParam] | None = None,
) -> JSONResponse:
    return JSONResponse(
        status_code=status,
        media_type="application/problem+json",
        headers=headers,
        content={
    content: dict[str, Any] = {
        "type": problem_type(code),
        "title": title,
        "status": status,
        "detail": detail,
        "instance": instance,
        },
    }
    if invalid_params:
        content["invalidParams"] = [p.model_dump(exclude_none=True) for p in invalid_params]
    return JSONResponse(
        status_code=status,
        media_type="application/problem+json",
        headers=headers,
        content=content,
    )


# The operator-facing federation lifecycle OEG proxies onto /internal (§L.3b). OEG relays only
# the statuses OPG.04 v1.4.0 declares for those operations (_PROBLEM_STATUSES) and turns
# anything else into its own 502, so on these routes FM maps its EWBI-side 403/502 onto that
# set; `type` still says which failure it was.
_PROXIED_PATH = re.compile(r"^/internal/partners/[^/]+/federations(/|$)")

# Partner rejections that describe the operator's own request and are relayed as they are.
# Anything else the partner answers (its 401/403 is about our token, its 5xx about itself)
# must not look like FM's or the operator's own fault.
_RELAYED_PARTNER_STATUSES = frozenset({400, 404, 409, 422})


def _proxied(request: Request) -> bool:
    return _PROXIED_PATH.match(request.url.path) is not None


def _invalid_params(exc: RequestValidationError) -> list[InvalidParam]:
    params = []
    for error in exc.errors():
        # Drop FastAPI's leading "body"/"path"/"query" so the param is named as the caller sent it.
        loc = list(error["loc"])
        if len(loc) > 1 and loc[0] in {"body", "path", "query", "header"}:
            loc = loc[1:]
        param = "".join(f"[{part}]" if isinstance(part, int) else f".{part}" for part in loc)
        params.append(InvalidParam(param=param.lstrip(".") or "body", reason=error["msg"]))
    return params


def register_exception_handlers(app: FastAPI) -> None:
    # Details stay generic: nothing about our internals crosses the operator boundary.
    @app.exception_handler(AuthenticationFailed)
@@ -131,7 +166,7 @@ def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(PartnerNotActive)
    async def _partner_not_active(request: Request, exc: PartnerNotActive) -> JSONResponse:
        return problem(
            403,
            409 if _proxied(request) else 403,
            "partner-not-active",
            "Partner Not Active",
            "Federation with this partner is not currently active.",
@@ -141,7 +176,7 @@ def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AgreementExpired)
    async def _agreement_expired(request: Request, exc: AgreementExpired) -> JSONResponse:
        return problem(
            403,
            409 if _proxied(request) else 403,
            "agreement-expired",
            "Federation Agreement Expired",
            "No federation agreement with this partner is currently valid.",
@@ -151,7 +186,7 @@ def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AgreementViolation)
    async def _agreement_violation(request: Request, exc: AgreementViolation) -> JSONResponse:
        return problem(
            403,
            422 if _proxied(request) else 403,
            "agreement-violation",
            "Federation Agreement Violation",
            "Request is outside the scope of the federation agreement with this partner.",
@@ -172,33 +207,42 @@ def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(PartnerTokenRequestFailed)
    async def _partner_unreachable(request: Request, exc: Exception) -> JSONResponse:
        return problem(
            502,
            503 if _proxied(request) else 502,
            "partner-unreachable",
            "Partner Unreachable",
            "The partner operator did not return a usable response.",
            request.url.path,
        )

    @app.exception_handler(PartnerResponseInvalid)
    async def _partner_response_invalid(
        request: Request, exc: PartnerResponseInvalid
    ) -> JSONResponse:
    def _partner_response_invalid_problem(request: Request) -> JSONResponse:
        return problem(
            502,
            520 if _proxied(request) else 502,
            "partner-response-invalid",
            "Invalid Partner Response",
            "The partner operator's response did not match the OPG.04 contract.",
            request.url.path,
        )

    @app.exception_handler(PartnerResponseInvalid)
    async def _partner_response_invalid(
        request: Request, exc: PartnerResponseInvalid
    ) -> JSONResponse:
        return _partner_response_invalid_problem(request)

    @app.exception_handler(PartnerRejectedRequest)
    async def _partner_rejected(request: Request, exc: PartnerRejectedRequest) -> Response:
        if not 400 <= exc.status_code < 600:
            return _partner_response_invalid_problem(request)
        if _proxied(request):
            if exc.status_code in _RELAYED_PARTNER_STATUSES and isinstance(exc.body, dict):
                return JSONResponse(
                    exc.body, status_code=exc.status_code, media_type="application/problem+json"
                )
            return problem(
                502,
                "partner-response-invalid",
                "Invalid Partner Response",
                "The partner operator's response did not match the OPG.04 contract.",
                exc.status_code if exc.status_code in _RELAYED_PARTNER_STATUSES else 520,
                "partner-rejected",
                "Partner Rejected The Request",
                f"The partner operator answered {exc.status_code}.",
                request.url.path,
            )
        if exc.body is None:
@@ -445,3 +489,30 @@ def register_exception_handlers(app: FastAPI) -> None:
            "You have not subscribed to this zone.",
            request.url.path,
        )

    # FastAPI's own 422 is plain application/json; on the proxied routes it is OPG.04's 400.
    @app.exception_handler(RequestValidationError)
    async def _request_invalid(request: Request, exc: RequestValidationError) -> Response:
        if not _proxied(request):
            return await request_validation_exception_handler(request, exc)
        return problem(
            400,
            "invalid-request",
            "Invalid Request",
            "One or more request parameters or body fields are invalid; see invalidParams.",
            request.url.path,
            invalid_params=_invalid_params(exc),
        )

    # Starlette still re-raises after this, so the traceback is logged as before.
    @app.exception_handler(Exception)
    async def _unexpected(request: Request, exc: Exception) -> Response:
        if not _proxied(request):
            return PlainTextResponse("Internal Server Error", status_code=500)
        return problem(
            500,
            "internal-error",
            "Internal Error",
            "The request could not be processed.",
            request.url.path,
        )
+102 −10
Changes for tests/test_partner_federations.py: 102 added lines, 10 removed lines.
Original line number Diff line number Diff line
@@ -17,6 +17,7 @@ from federation_manager.dependencies import (
    get_local_operator,
    get_partner_repo,
)
from federation_manager.domain.errors import PartnerRequestFailed
from federation_manager.domain.models import EwbiResponse, FederationContext, PartnerOP
from federation_manager.main import create_app
from tests.fakes import (
@@ -73,10 +74,17 @@ def _context(partner: PartnerOP, status: str = "available") -> FederationContext


class Harness:
    def __init__(self, response: EwbiResponse | None = None, *, federated: bool = True) -> None:
        self.partner = _partner()
    def __init__(
        self,
        response: EwbiResponse | None = None,
        *,
        federated: bool = True,
        error: Exception | None = None,
        partner_status: str = "active",
    ) -> None:
        self.partner = replace(_partner(), status=partner_status)
        self.contexts = InMemoryFederationContextRepo([_context(self.partner)] if federated else [])
        self.ewbi = FakeEwbiClient(response or EwbiResponse(200, None))
        self.ewbi = FakeEwbiClient(response or EwbiResponse(200, None), error=error)
        app = create_app(lifespan=_no_infra)
        app.dependency_overrides[get_partner_repo] = lambda: InMemoryPartnerRepo([self.partner])
        app.dependency_overrides[get_federation_context_repo] = lambda: self.contexts
@@ -84,7 +92,7 @@ class Harness:
        app.dependency_overrides[get_local_operator] = lambda: LOCAL
        self.catalogue = InMemoryCatalogueRepo()
        app.dependency_overrides[get_catalogue_repo] = lambda: self.catalogue
        self.client = TestClient(app)
        self.client = TestClient(app, raise_server_exceptions=False)

    @property
    def base(self) -> str:
@@ -191,14 +199,14 @@ def test_partner_rejection_is_relayed_unchanged_and_nothing_is_stored() -> None:
    assert h.contexts.contexts == []


def test_a_partner_answer_outside_the_contract_is_a_bad_gateway() -> None:
def test_a_partner_answer_outside_the_contract_is_an_unknown_error() -> None:
    h = Harness(EwbiResponse(200, {"platformCaps": ["serviceAPIs"]}), federated=False)

    response = h.client.post(
        h.base, json={"initialDate": "2026-09-15T10:00:00Z", "partnerStatusLink": STATUS_LINK}
    )

    assert response.status_code == 502
    assert response.status_code == 520
    assert _problem_type(response) == "partner-response-invalid"
    assert h.contexts.contexts == []

@@ -266,17 +274,22 @@ def test_a_partner_that_no_longer_knows_the_context_has_terminated_it() -> None:


@pytest.mark.parametrize(
    ("partner_status", "local_status"),
    [(200, "terminated"), (404, "terminated"), (409, "terminated"), (500, "available")],
    ("partner_status", "our_status", "local_status"),
    [
        (200, 200, "terminated"),
        (404, 404, "terminated"),
        (409, 409, "terminated"),
        (500, 520, "available"),
    ],
)
def test_delete_terminates_locally_only_once_the_partner_has_let_go(
    partner_status: int, local_status: str
    partner_status: int, our_status: int, local_status: str
) -> None:
    h = Harness(EwbiResponse(partner_status, PROBLEM if partner_status >= 400 else None))

    response = h.client.delete(f"{h.base}/{CONTEXT_ID}")

    assert response.status_code == partner_status
    assert response.status_code == our_status
    assert h.ewbi.requests == [
        ("DELETE", f"/operatorplatform/federation/v1/{CONTEXT_ID}/partner", None)
    ]
@@ -343,3 +356,82 @@ def test_an_unknown_zone_does_not_end_the_federation() -> None:

    assert response.status_code == 404
    assert h.context_status() == "available"


# --- Only OPG.04 v1.4.0's declared statuses reach OEG (§L.3b) -------------------------------

_DECLARED = {400, 401, 404, 409, 422, 500, 503, 520}


def _assert_problem(response: Any, status: int, code: str) -> None:
    assert response.status_code == status
    assert status in _DECLARED
    assert response.headers["content-type"] == "application/problem+json"
    assert _problem_type(response) == code


def test_an_unreachable_partner_is_service_unavailable() -> None:
    h = Harness(error=PartnerRequestFailed(uuid4()))

    response = h.client.get(f"{h.base}/{CONTEXT_ID}")

    _assert_problem(response, 503, "partner-unreachable")


def test_an_inactive_partner_is_a_conflict() -> None:
    h = Harness(federated=False, partner_status="suspended")

    response = h.client.post(
        h.base, json={"initialDate": "2026-09-15T10:00:00Z", "partnerStatusLink": STATUS_LINK}
    )

    _assert_problem(response, 409, "partner-not-active")
    assert h.ewbi.requests == []


@pytest.mark.parametrize("partner_status", [401, 403, 429, 500, 502])
def test_a_partner_rejection_about_the_partner_itself_is_an_unknown_error(
    partner_status: int,
) -> None:
    h = Harness(EwbiResponse(partner_status, PROBLEM))

    response = h.client.get(f"{h.base}/{CONTEXT_ID}")

    _assert_problem(response, 520, "partner-rejected")
    assert str(partner_status) in response.json()["detail"]


def test_a_bodiless_partner_rejection_still_gets_a_problem_body() -> None:
    h = Harness(EwbiResponse(409, None))

    response = h.client.delete(f"{h.base}/{CONTEXT_ID}/zones/zone-madrid-1")

    _assert_problem(response, 409, "partner-rejected")


def test_an_invalid_body_is_a_bad_request_naming_the_field() -> None:
    h = Harness(federated=False)

    response = h.client.post(h.base, json={"partnerStatusLink": STATUS_LINK})

    _assert_problem(response, 400, "invalid-request")
    assert [p["param"] for p in response.json()["invalidParams"]] == ["initialDate"]
    assert h.ewbi.requests == []


def test_a_non_uuid_partner_is_a_bad_request() -> None:
    h = Harness()

    response = h.client.get("/internal/partners/partner-a/federations")

    _assert_problem(response, 400, "invalid-request")
    assert [p["param"] for p in response.json()["invalidParams"]] == ["partner_op_id"]


def test_an_unexpected_failure_is_still_a_problem() -> None:
    h = Harness(error=RuntimeError("boom"))

    response = h.client.get(f"{h.base}/{CONTEXT_ID}")

    _assert_problem(response, 500, "internal-error")
    assert "boom" not in response.text