feat: validate federation request bodies for shape before forwarding (84fe0368) · Commits · OOP / code / open-exposure-gateway

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

+82 −21

Original line number	Diff line number	Diff line
		@@ -3,12 +3,14 @@
		Endpoints defined by ``API_definitions/federation-manager.yaml``. Every handler is a
		stateless synchronous relay to FM (ADR-0046, ADR-0048): the raw request body is
		forwarded to FM's matching internal endpoint unaltered, and FM's response body and
		status are relayed back unaltered. OEG does no request/response parsing or
		reshaping on this path -- the Pydantic schemas in ``schemas.py`` are referenced
		here only to document the response contract in the OpenAPI document (via
		``response_model``); they are not used to build, validate or serialize the actual
		request/response bodies at runtime, which would risk silently altering a payload
		OEG's hand-maintained models don't fully cover.
		status are relayed back unaltered. OEG does not reshape the request on this path --
		the two POST 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.

		This module exposes the ``FederationManagement`` tag plus ``zone_subscribe``,
		``get_zone_data`` and ``zone_unsubscribe`` from
		@@ -16,9 +18,11 @@ This module exposes the ``FederationManagement`` tag plus ``zone_subscribe``,
		(``PATCH /{federationContextId}/partner``) is out of scope.
		"""

		import json
		from typing import Annotated, Any

		from fastapi import APIRouter, Depends, 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
		@@ -26,10 +30,12 @@ from open_exposure_gateway.api.gsma.federation_manager.v1_2_0.schemas import (
		FederationContextId,
		FederationContextIdResponse,
		FederationDetails,
		FederationRequestData,
		FederationResponseData,
		ProblemDetails,
		ZoneIdentifier,
		ZoneRegisteredData,
		ZoneRegistrationRequestData,
		ZoneRegistrationResponseData,
		)
		from open_exposure_gateway.dependencies import get_fm_client
		@@ -51,6 +57,7 @@ _ERROR_RESPONSES: dict[int \| str, dict[str, Any]] = {
		409: {"model": ProblemDetails, "description": "Conflict"},
		422: {"model": ProblemDetails, "description": "Unprocessable entity"},
		500: {"model": ProblemDetails, "description": "Internal server error"},
		502: {"model": ProblemDetails, "description": "Bad gateway"},
		503: {"model": ProblemDetails, "description": "Service unavailable"},
		520: {"model": ProblemDetails, "description": "Unknown error"},
		}
		@@ -60,19 +67,62 @@ def _responses(*codes: int) -> dict[int \| str, dict[str, Any]]:
		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",
		)


		def _is_valid_json(content: bytes) -> bool:
		try:
		json.loads(content)
		except ValueError:
		return False
		return True


		async def _relay(
		fm_client: FmClient,
		method: str,
		path: str,
		request: Request,
		x_correlator: str \| None,
		request_schema: type[BaseModel] \| None = None,
		) -> Response:
		"""Forward one federation-lifecycle operation to FM and relay its answer unaltered.

		``path`` is the operator-facing suffix after ``BASE_PATH`` (e.g. ``/partner``),
		which ``FmClient`` maps onto FM's internal endpoint.
		which ``FmClient`` maps onto FM's internal endpoint. 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.

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

		try:
		fm_response = await fm_client.relay(
		method,
		@@ -82,11 +132,13 @@ async def _relay(
		x_correlator=x_correlator,
		)
		except FmUnavailableError as exc:
		problem = ProblemDetails(title="Federation Manager unavailable", detail=str(exc))
		return Response(
		content=problem.model_dump_json(exclude_none=True),
		status_code=503,
		media_type="application/problem+json",
		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",
		)

		return Response(
		@@ -102,12 +154,14 @@ async def _relay(
		summary="Create a one-direction federation with a partner operator platform",
		operation_id="create_federation",
		response_model=FederationResponseData,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def create_federation(
		request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None
		) -> Response:
		return await _relay(fm_client, "POST", "/partner", request, x_correlator)
		return await _relay(
		fm_client, "POST", "/partner", request, x_correlator, request_schema=FederationRequestData
		)


		@router.get(
		@@ -116,7 +170,7 @@ async def create_federation(
		summary="Retrieve details about the federation context with the partner OP",
		operation_id="get_federation_details",
		response_model=FederationDetails,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def get_federation_details(
		federationContextId: FederationContextId,
		@@ -133,7 +187,7 @@ async def get_federation_details(
		summary="Remove an existing federation with the partner OP",
		operation_id="delete_federation_details",
		status_code=200,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def delete_federation_details(
		federationContextId: FederationContextId,
		@@ -152,7 +206,7 @@ async def delete_federation_details(
		summary="Retrieve the existing federationContextId with the partner operator platform",
		operation_id="get_federation_context_id",
		response_model=FederationContextIdResponse,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def get_federation_context_id(
		request: Request, fm_client: FmClientDep, x_correlator: XCorrelatorHeader = None
		@@ -166,7 +220,7 @@ async def get_federation_context_id(
		summary="Subscribe to partner OP availability zones and reserve zone resources",
		operation_id="zone_subscribe",
		response_model=ZoneRegistrationResponseData,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def zone_subscribe(
		federationContextId: FederationContextId,
		@@ -174,7 +228,14 @@ async def zone_subscribe(
		fm_client: FmClientDep,
		x_correlator: XCorrelatorHeader = None,
		) -> Response:
		return await _relay(fm_client, "POST", f"/{federationContextId}/zones", request, x_correlator)
		return await _relay(
		fm_client,
		"POST",
		f"/{federationContextId}/zones",
		request,
		x_correlator,
		request_schema=ZoneRegistrationRequestData,
		)


		@router.get(
		@@ -186,7 +247,7 @@ async def zone_subscribe(
		),
		operation_id="get_zone_data",
		response_model=ZoneRegisteredData,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def get_zone_data(
		federationContextId: FederationContextId,
		@@ -209,7 +270,7 @@ async def get_zone_data(
		),
		operation_id="zone_unsubscribe",
		status_code=200,
		responses=_responses(400, 401, 404, 409, 422, 500, 503, 520),
		responses=_responses(400, 401, 404, 409, 422, 500, 502, 503, 520),
		)
		async def zone_unsubscribe(
		federationContextId: FederationContextId,

tests/unit/test_federation_manager_endpoints.py

+84 −2

Original line number	Diff line number	Diff line
		@@ -21,6 +21,15 @@ from open_exposure_gateway.main import app
		_CTX = "fed-ctx-1"
		_ZONE = "zone-a"

		_VALID_CREATE_BODY = (
		b'{"origOPFederationId": "orig-op-1", "initialDate": "2026-09-04T12:00:00Z", '
		b'"partnerStatusLink": "https://orig-op.example.com/status"}'
		)
		_VALID_ZONE_SUBSCRIBE_BODY = (
		b'{"acceptedAvailabilityZones": ["zone-a"], '
		b'"availZoneNotifLink": "https://orig-op.example.com/zone-notif"}'
		)


		class _FakeFmClient:
		def __init__(self) -> None:
		@@ -68,11 +77,11 @@ def client() -> TestClient:


		_ENDPOINTS = [
		("post", f"{BASE_PATH}/partner", "POST", "/partner", b'{"a": 1}'),
		("post", f"{BASE_PATH}/partner", "POST", "/partner", _VALID_CREATE_BODY),
		("get", f"{BASE_PATH}/{_CTX}/partner", "GET", f"/{_CTX}/partner", None),
		("delete", f"{BASE_PATH}/{_CTX}/partner", "DELETE", f"/{_CTX}/partner", None),
		("get", f"{BASE_PATH}/fed-context-id", "GET", "/fed-context-id", None),
		("post", f"{BASE_PATH}/{_CTX}/zones", "POST", f"/{_CTX}/zones", b'{"a": 1}'),
		("post", f"{BASE_PATH}/{_CTX}/zones", "POST", f"/{_CTX}/zones", _VALID_ZONE_SUBSCRIBE_BODY),
		("get", f"{BASE_PATH}/{_CTX}/zones/{_ZONE}", "GET", f"/{_CTX}/zones/{_ZONE}", None),
		("delete", f"{BASE_PATH}/{_CTX}/zones/{_ZONE}", "DELETE", f"/{_CTX}/zones/{_ZONE}", None),
		]
		@@ -141,6 +150,79 @@ def test_fm_unavailable_returns_problem_details_503(
		assert response.json()["title"] == "Federation Manager unavailable"


		def test_malformed_request_body_returns_problem_details_400(
		client: TestClient, fake_fm_client: _FakeFmClient
		) -> None:
		response = client.post(
		f"{BASE_PATH}/partner",
		content=b"{not valid json",
		headers={"content-type": "application/json"},
		)

		assert response.status_code == 400
		assert response.headers["content-type"] == "application/problem+json"
		assert response.json()["title"] == "Malformed request body"
		assert fake_fm_client.calls == []


		def test_wrong_shape_request_body_returns_problem_details_400(
		client: TestClient, fake_fm_client: _FakeFmClient
		) -> None:
		"""Valid JSON, but missing FederationRequestData's required fields."""
		response = client.post(
		f"{BASE_PATH}/partner",
		content=b'{"a": 1}',
		headers={"content-type": "application/json"},
		)

		assert response.status_code == 400
		assert response.headers["content-type"] == "application/problem+json"
		assert response.json()["title"] == "Request body does not match the expected schema"
		assert fake_fm_client.calls == []


		def test_wrong_shape_zone_subscribe_body_returns_problem_details_400(
		client: TestClient, fake_fm_client: _FakeFmClient
		) -> None:
		"""acceptedAvailabilityZones must be non-empty."""
		response = client.post(
		f"{BASE_PATH}/{_CTX}/zones",
		content=b'{"acceptedAvailabilityZones": [], "availZoneNotifLink": "https://x.example.com"}',
		headers={"content-type": "application/json"},
		)

		assert response.status_code == 400
		assert fake_fm_client.calls == []


		def test_valid_shape_request_body_is_still_forwarded_as_raw_bytes(
		client: TestClient, fake_fm_client: _FakeFmClient
		) -> None:
		"""Shape validation is a gate, not a reshape -- FM gets the original bytes."""
		response = client.post(
		f"{BASE_PATH}/partner",
		content=_VALID_CREATE_BODY,
		headers={"content-type": "application/json"},
		)

		assert response.status_code == 200
		assert fake_fm_client.calls[0]["content"] == _VALID_CREATE_BODY


		def test_unparseable_fm_response_returns_problem_details_502(
		client: TestClient, fake_fm_client: _FakeFmClient
		) -> None:
		fake_fm_client.response = httpx.Response(
		200, content=b"{not valid json", headers={"content-type": "application/json"}
		)

		response = client.get(f"{BASE_PATH}/fed-context-id")

		assert response.status_code == 502
		assert response.headers["content-type"] == "application/problem+json"
		assert response.json()["title"] == "Federation Manager returned an unusable response"


		def test_openapi_exposes_federation_management_operations() -> None:
		paths = app.openapi()["paths"]
		seen = {