Follow-up fixes to the federation-lifecycle proxy

"""OEG-authored ``ProblemDetails`` for the federation-lifecycle surface (§K.6a, ADR-0050).

Lives apart from ``router.py`` so ``api/error_handlers.py`` can use it too: the router

imports ``error_handlers``, so ``error_handlers`` importing the router would be circular.

"""

from collections.abc import Iterable, Mapping, Sequence

from http import HTTPStatus

from typing import Any, NamedTuple

from fastapi import Response

from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.schemas import (

    InvalidParam,

    ProblemDetails,

)

# OEG's own platform namespace, mirrored 1:1 at FM's `/internal` (FmClient does a

# prefix swap, nothing else -- see fm_client.py).

BASE_PATH = "/platform/v1/federation"

class OegProblem(NamedTuple):

    """Fixed template for one kind of OEG-authored error (§K.6a)."""

    status: int

    type: str

    title: str

    detail: str

    cause: str

# Every error OEG itself raises on this surface. `detail` is a fixed sentence per case

# -- never exception text or upstream diagnostics (§K.6a).

MALFORMED_BODY = OegProblem(

    400,

    "urn:oop:ewbi:error:malformed-request-body",

    "Malformed request body",

    "Request body is not valid JSON.",

    "MALFORMED_REQUEST_BODY",

)

INVALID_BODY = OegProblem(

    400,

    "urn:oop:ewbi:error:invalid-request-body",

    "Request body does not match the expected schema",

    "One or more request body fields are invalid; see invalidParams.",

    "INVALID_REQUEST_BODY",

)

INVALID_PARAMETERS = OegProblem(

    400,

    "urn:oop:ewbi:error:invalid-request-parameters",

    "Invalid request parameters",

    "One or more path, query or header parameters are invalid; see invalidParams.",

    "INVALID_REQUEST_PARAMETERS",

)

UPSTREAM_UNAVAILABLE = OegProblem(

    503,

    "urn:oop:ewbi:error:upstream-unavailable",

    "Federation Manager unavailable",

    "The federation-lifecycle request could not be forwarded to FM.",

    "UPSTREAM_UNAVAILABLE",

)

UPSTREAM_BAD_RESPONSE = OegProblem(

    502,

    "urn:oop:ewbi:error:upstream-bad-response",

    "Federation Manager returned an unusable response",

    "FM's response could not be relayed.",

    "UPSTREAM_BAD_RESPONSE",

)

INTERNAL_ERROR = OegProblem(

    500,

    "urn:oop:ewbi:error:internal-error",

    "Internal server error",

    "The request could not be processed.",

    "INTERNAL_ERROR",

)

# Framework-level errors (raised before or outside a route handler), by status.

_BY_STATUS = {

    401: OegProblem(

        401,

        "urn:oop:ewbi:error:unauthenticated",

        "Unauthenticated",

        "The request lacks valid authentication credentials.",

        "UNAUTHENTICATED",

    ),

    403: OegProblem(

        403,

        "urn:oop:ewbi:error:permission-denied",

        "Permission denied",

        "The caller is not allowed to perform this operation.",

        "PERMISSION_DENIED",

    ),

    404: OegProblem(

        404,

        "urn:oop:ewbi:error:not-found",

        "Not found",

        "No federation-lifecycle resource matches the request path.",

        "NOT_FOUND",

    ),

    405: OegProblem(

        405,

        "urn:oop:ewbi:error:method-not-allowed",

        "Method not allowed",

        "The request method is not supported on this path.",

        "METHOD_NOT_ALLOWED",

    ),

    500: INTERNAL_ERROR,

}

def is_federation_path(path: str) -> bool:

    return path == BASE_PATH or path.startswith(f"{BASE_PATH}/")

def problem_for_status(status: int) -> OegProblem:

    """The template for a framework-level error with this status.

    Statuses without a dedicated template get a generic one built from the HTTP reason

    phrase, so a new status can never fall back to exception text.

    """

    if status in _BY_STATUS:

        return _BY_STATUS[status]

    try:

        phrase = HTTPStatus(status).phrase

    except ValueError:

        return INTERNAL_ERROR._replace(status=status)

    slug = phrase.lower().replace(" ", "-").replace("'", "")

    return OegProblem(

        status,

        f"urn:oop:ewbi:error:{slug}",

        phrase,

        "The request could not be processed.",

        slug.replace("-", "_").upper(),

    )

def problem_response(

    problem: OegProblem,

    invalid_params: list[InvalidParam] | None = None,

    headers: Mapping[str, str] | None = None,

) -> Response:

    body = ProblemDetails(**problem._asdict(), invalidParams=invalid_params)

    return Response(

        content=body.model_dump_json(exclude_none=True),

        status_code=problem.status,

        media_type="application/problem+json",

        headers=headers,

    )

# FastAPI prefixes request-parameter errors with where the value came from.

_REQUEST_LOCATIONS = {"path", "query", "header", "cookie", "body"}

def invalid_params(errors: Iterable[Mapping[str, Any]]) -> list[InvalidParam]:

    """One ``invalidParams`` entry per Pydantic error: field path plus message only.

    ``loc`` becomes a dotted path with list indices in brackets (e.g.

    ``origOPMobileNetworkCodes.mncs[0]``); an error on the body as a whole (empty

    ``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.

    """

    params = []

    for error in errors:

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

        if len(loc) > 1 and loc[0] in _REQUEST_LOCATIONS:

            loc = loc[1:]

        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"]))

    return params

"""FastAPI routes for the platform-namespace federation-lifecycle relay (ADR-0046, ADR-0048).

"""FastAPI routes for the platform-namespace federation-lifecycle relay (ADR-0051, ADR-0050).

Bodies are raw GSMA (OPG.04 v6.0 / artifact v1.4.0), forwarded unaltered, but paths

are OEG's own platform namespace rather than the GSMA EWBI paths: GSMA addresses the

bodies (``FederationRequestData``, ``ZoneRegistrationRequestData``) are validated for

shape before being forwarded, but the *validated* object is only used as a gate; the

original raw bytes are what's actually sent to FM, so a field OEG's model doesn't

happen to cover is still forwarded unchanged rather than dropped. The remaining

schemas in ``schemas.py`` are referenced here only to document the response contract

in the OpenAPI document (via ``response_model``); FM's response is never parsed

against them.

happen to cover is still forwarded unchanged rather than dropped. FM's response is

gated the same way: each route passes the statuses FM may return for it and its

success schema, and anything outside that contract becomes OEG's 502 (§K.6a). A

response that passes is still relayed byte-for-byte.

This module exposes the ``FederationManagement`` tag plus ``zone_subscribe``,

``get_zone_data``, ``get_zone_details`` and ``zone_unsubscribe`` from

import json

from typing import Annotated, Any

import httpx

import structlog

from fastapi import APIRouter, Depends, Query, Request, Response

from pydantic import BaseModel, ValidationError

from open_exposure_gateway.adapters.http.fm_client import FmClient, FmUnavailableError

from open_exposure_gateway.api.camara.common import XCorrelatorHeader

from open_exposure_gateway.api.error_handlers import x_correlator_header

from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.problems import (

    BASE_PATH,

    INVALID_BODY,

    MALFORMED_BODY,

    UPSTREAM_BAD_RESPONSE,

    UPSTREAM_UNAVAILABLE,

    invalid_params,

    problem_response,

)

from open_exposure_gateway.api.gsma.federation_manager.v1_4_0.schemas import (

    FederationContextId,

    FederationContextIdResponse,

)

from open_exposure_gateway.dependencies import get_fm_client

# OEG's own platform namespace, mirrored 1:1 at FM's `/internal` (FmClient does a

# prefix swap, nothing else -- see fm_client.py).

BASE_PATH = "/platform/v1/federation"

PARTNER_PREFIX = "/partners/{partnerOpId}/federations"

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)

    return {code: _ERROR_RESPONSES[code] for code in codes}

def _problem_response(status_code: int, title: str, detail: str) -> Response:

    problem = ProblemDetails(title=title, detail=detail)

    return Response(

        content=problem.model_dump_json(exclude_none=True),

        status_code=status_code,

        media_type="application/problem+json",

    )

# Every status FM's profiled OPG.04 v1.4.0 artifact declares for these eight operations

# -- all of them declare the same set. The spec's bodiless `default` response is not

# counted: FM only uses the statuses each operation declares (FM REQ-FED-14). 502 is

# OEG's own and never FM's.

_FM_STATUSES = frozenset({200, 400, 401, 404, 409, 422, 500, 503, 520})

_PROBLEM_JSON = "application/problem+json"

def _is_valid_json(content: bytes) -> bool:

    return True

def _is_valid(schema: type[BaseModel], content: bytes) -> bool:

    try:

        schema.model_validate_json(content)

    except ValidationError:

        return False

    return True

def _breaks_fm_contract(

    fm_response: httpx.Response,

    fm_statuses: frozenset[int],

    response_schema: type[BaseModel] | None,

) -> str | None:

    """Why FM's response is outside its contract for this operation, or ``None`` if it isn't.

    §K.6a's 502 case: a status the operation doesn't declare, an error without an

    ``application/problem+json`` ``ProblemDetails`` body, or a success body that doesn't

    match the operation's response schema. The response is only inspected, never

    changed -- one that passes is relayed byte-for-byte.

    """

    status = fm_response.status_code

    content = fm_response.content

    if status not in fm_statuses:

        return "undeclared status"

    if status >= 400:

        media_type = fm_response.headers.get("content-type", "").split(";")[0].strip().lower()

        if media_type != _PROBLEM_JSON:

            return "error without application/problem+json"

        if not _is_valid(ProblemDetails, content):

            return "error body is not ProblemDetails"

        return None

    if response_schema is not None:

        if not _is_valid(response_schema, content):

            return "success body does not match the response schema"

        return None

    if content and not _is_valid_json(content):

        return "success body is not valid JSON"

    return None

async def _relay(

    fm_client: FmClient,

    method: str,

    path: str,

    request: Request,

    x_correlator: str | None,

    fm_statuses: frozenset[int],

    request_schema: type[BaseModel] | None = None,

    response_schema: type[BaseModel] | None = None,

) -> Response:

    """Forward one federation-lifecycle operation to FM and relay its answer unaltered.

    matching internal endpoint. The request's query string is forwarded as-is

    (e.g. ``GetZoneData``'s ``zoneId`` filter). A body that isn't even

    syntactically JSON can't reach FM meaningfully, so that's rejected here

    (ADR-0047: OEG-authored 400); the mirror case on the way back -- FM answering

    with a body it labelled JSON that isn't actually parseable -- is ADR-0047's 502.

    (ADR-0050: OEG-authored 400); the mirror case on the way back -- FM answering

    outside its contract for the operation (``fm_statuses`` and ``response_schema``;

    see ``_breaks_fm_contract``) -- is ADR-0050's 502.

    When ``request_schema`` is given, the parsed body is also validated against it

    as a shape check before being forwarded. The validated object itself is

    discarded afterwards -- what's sent to FM is always the original raw bytes, so

    this is a gate, not a reshape.

    this is a gate, not a reshape. The same holds for FM's response.

    """

    body = await request.body()

    if body:

        try:

            parsed = json.loads(body)

        except ValueError:

            return _problem_response(

                400, "Malformed request body", "Request body is not valid JSON"

            )

            return problem_response(MALFORMED_BODY)

        if request_schema is not None:

            try:

                request_schema.model_validate(parsed)

            except ValidationError as exc:

                return _problem_response(

                    400,

                    "Request body does not match the expected schema",

                    str(exc.errors())[:512],

                )

                return problem_response(INVALID_BODY, invalid_params(exc.errors()))

    try:

        fm_response = await fm_client.relay(

            path,

            content=body or None,

            content_type=request.headers.get("content-type"),

            x_correlator=x_correlator,

            x_correlator=x_correlator_header(request),

            params=tuple(request.query_params.multi_items()),

        )

    except FmUnavailableError as exc:

        return _problem_response(503, "Federation Manager unavailable", str(exc))

    if fm_response.content and not _is_valid_json(fm_response.content):

        return _problem_response(

            502,

            "Federation Manager returned an unusable response",

            "Response body is not valid JSON",

    except FmUnavailableError:

        # FmClient has already logged the underlying cause.

        return problem_response(UPSTREAM_UNAVAILABLE)

    reason = _breaks_fm_contract(fm_response, fm_statuses, response_schema)

    if reason is not None:

        # Status and reason only -- never the body (§N.7: bodies aren't logged).

        logger.warning(

            "FM response outside its contract",

            method=method,

            path=path,

            fm_status=fm_response.status_code,

            reason=reason,

            x_correlator=x_correlator_header(request),

        )

        return problem_response(UPSTREAM_BAD_RESPONSE)

    return Response(

        content=fm_response.content,

        "POST",

        f"/partners/{partnerOpId}/federations",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

        request_schema=FederationRequestData,

        response_schema=FederationResponseData,

    )

    x_correlator: XCorrelatorHeader = None,

) -> Response:

    return await _relay(

        fm_client, "GET", f"/partners/{partnerOpId}/federations", request, x_correlator

        fm_client,

        "GET",

        f"/partners/{partnerOpId}/federations",

        request,

        fm_statuses=_FM_STATUSES,

        response_schema=FederationContextIdResponse,

    )

        "GET",

        f"/partners/{partnerOpId}/federations/{federationContextId}",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

        response_schema=FederationDetails,

    )

        "DELETE",

        f"/partners/{partnerOpId}/federations/{federationContextId}",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

    )

        "GET",

        f"/partners/{partnerOpId}/federations/{federationContextId}/zones",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

        response_schema=ZoneRegisteredData,

    )

        "POST",

        f"/partners/{partnerOpId}/federations/{federationContextId}/zones",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

        request_schema=ZoneRegistrationRequestData,

        response_schema=ZoneRegistrationResponseData,

    )

        "GET",

        f"/partners/{partnerOpId}/federations/{federationContextId}/zones/{zoneId}",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

        response_schema=ZoneRegisteredData,

    )

        "DELETE",

        f"/partners/{partnerOpId}/federations/{federationContextId}/zones/{zoneId}",

        request,

        x_correlator,

        fm_statuses=_FM_STATUSES,

    )

from datetime import datetime

from enum import StrEnum

from typing import Annotated, Optional

from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator

from pydantic import BaseModel, Field, StringConstraints, model_validator

# --- Scalar types (OpenAPI: string/integer with pattern or format) -------------

ZoneIdentifier = Annotated[str, StringConstraints(pattern=_ID_PATTERN)]

PartnerOpId = Annotated[str, StringConstraints(pattern=_ID_PATTERN)]

PartnerOpId = UUID

"""Our own registered id for the partner operator platform (not part of GSMA -- this

is the platform-namespace path param that says which partner an EWBI call is for,

since the GSMA body itself never names the callee)."""

since the GSMA body itself never names the callee). FM assigns it as a UUID when the

partner is registered, and its ``/internal`` routes accept nothing else."""

CountryCode = Annotated[str, StringConstraints(pattern=r"^[A-Z]{2}$")]

"""ISO 3166-1 Alpha-2 country code of the partner operator."""

class ProblemDetails(BaseModel):

    """RFC 7807 error body used by every Federation Manager error response."""

    """RFC 7807 error body used by every Federation Manager error response.

    ``type`` and ``status`` aren't in OPG.04's ``ProblemDetails`` but are allowed as

    additive RFC 7807 fields (FM §L.5); OEG-authored errors always set them (§K.6a).

    """

    type: Optional[str] = None

    title: Optional[str] = None

    status: Optional[int] = None

    detail: Optional[str] = None

    cause: Optional[str] = None

    invalidParams: Optional[Annotated[list[InvalidParam], Field(min_length=1)]] = None

    ``partnerCallbackCredentials`` no longer exists as a field.

    """

    model_config = ConfigDict(extra="forbid")

    initialDate: datetime

    partnerStatusLink: Uri

    origOPFederationId: Optional[FederationIdentifier] = None

class ZoneRegistrationRequestData(BaseModel):

    """Body of ``ZoneSubscribe``."""

    """Body of ``ZoneSubscribe``.

    model_config = ConfigDict(extra="forbid")

    v1.4.0 requires only ``acceptedAvailabilityZones``; FM fills in its own

    ``availZoneNotifLink`` when the caller omits it.

    """

    acceptedAvailabilityZones: Annotated[list[ZoneIdentifier], Field(min_length=1)]

    availZoneNotifLink: Uri

    availZoneNotifLink: Optional[Uri] = None

class ZoneRegistrationResponseData(BaseModel):

import uuid

from collections.abc import Mapping

import structlog

from fastapi import FastAPI, Request

from fastapi.exceptions import RequestValidationError

from fastapi.responses import JSONResponse

from fastapi.responses import JSONResponse, Response

from starlette.exceptions import HTTPException as StarletteHTTPException

from open_exposure_gateway.api.camara.common import X_CORRELATOR_PATTERN

from open_exposure_gateway.api.gsma.federation_manager.v1_4_0 import problems

from open_exposure_gateway.core.exceptions import (

    BadRequestException,

    ConflictException,

    return x_correlator

def _error_response(

    request: Request,

    status: int,

    code: ErrorCode,

    message: str,

    headers: Mapping[str, str] | None = None,

) -> Response:

    """CAMARA ``ErrorInfo``, or on the federation-lifecycle routes a GSMA ``ProblemDetails``.

    §K.6a: every error on those routes is ``application/problem+json``, including the

    ones raised before or outside the route handler that land here. ``message`` is never

    used there -- ProblemDetails carries a fixed ``detail`` per status (§K.6a).

    """

    response_headers = {"x-correlator": x_correlator_header(request), **(headers or {})}

    if problems.is_federation_path(request.url.path):