Add internal endpoints for operator-triggered federation lifecycle (d1abcad5) · Commits · OOP / code / federation-manager

src/federation_manager/adapters/database/federation_context_repo.py

+14 −0

Original line number	Diff line number	Diff line
		@@ -46,6 +46,20 @@ class PostgresFederationContextRepo:
		)
		)

		async def find_outbound(
		self, partner_id: UUID, federation_context_id: str
		) -> FederationContext \| None:
		return await self._one(
		select(contexts)
		.where(
		contexts.c.partner_op_id == partner_id,
		contexts.c.direction == "outbound",
		contexts.c.federation_context_id == federation_context_id,
		)
		.order_by(contexts.c.created_at.desc())
		.limit(1)
		)

		async def add(self, context: FederationContext) -> None:
		await self._session.execute(
		insert(contexts).values(

src/federation_manager/adapters/http/ewbi_client.py

+22 −1

Original line number	Diff line number	Diff line
		@@ -20,14 +20,35 @@ class HttpxEwbiClient:
		self._allow_insecure = allow_insecure

		async def post(self, partner: PartnerOP, path: str, payload: dict[str, object]) -> EwbiResponse:
		return await self._send(partner, "POST", path, payload=payload)

		async def get(
		self, partner: PartnerOP, path: str, params: dict[str, str] \| None = None
		) -> EwbiResponse:
		return await self._send(partner, "GET", path, params=params)

		async def delete(self, partner: PartnerOP, path: str) -> EwbiResponse:
		return await self._send(partner, "DELETE", path)

		async def _send(
		self,
		partner: PartnerOP,
		method: str,
		path: str,
		*,
		payload: dict[str, object] \| None = None,
		params: dict[str, str] \| None = None,
		) -> EwbiResponse:
		url = self._url(partner, path)
		token = await self._token_provider.token_for(partner, scope="fed-mgmt")

		try:
		response = await self._client.post(
		response = await self._client.request(
		method,
		url,
		headers={"Accept": "application/json", "Authorization": f"Bearer {token}"},
		json=payload,
		params=params,
		)
		except httpx.HTTPError:
		raise PartnerRequestFailed(partner.id) from None

src/federation_manager/api/errors.py

+40 −1

Original line number	Diff line number	Diff line
		from typing import Any

		from fastapi import FastAPI, Request
		from fastapi import FastAPI, Request, Response
		from fastapi.responses import JSONResponse
		from pydantic.json_schema import models_json_schema

		@@ -13,11 +13,14 @@ from federation_manager.domain.errors import (
		FederationAlreadyExists,
		FederationContextMissing,
		FederationContextUnknown,
		FederationIdentityMismatch,
		IdempotencyKeyReused,
		NetworkQueryNotApplicable,
		NoRouteMatched,
		PartnerEndpointConfigurationError,
		PartnerNotActive,
		PartnerNotRegistered,
		PartnerRejectedRequest,
		PartnerRequestFailed,
		PartnerResponseInvalid,
		PartnerTokenConfigurationError,
		@@ -177,6 +180,42 @@ def register_exception_handlers(app: FastAPI) -> None:
		request.url.path,
		)

		@app.exception_handler(PartnerRejectedRequest)
		async def _partner_rejected(request: Request, exc: PartnerRejectedRequest) -> Response:
		if not 400 <= exc.status_code < 600:
		return problem(
		502,
		"partner-response-invalid",
		"Invalid Partner Response",
		"The partner operator's response did not match the OPG.04 contract.",
		request.url.path,
		)
		if exc.body is None:
		return Response(status_code=exc.status_code)
		return JSONResponse(
		exc.body, status_code=exc.status_code, media_type="application/problem+json"
		)

		@app.exception_handler(FederationIdentityMismatch)
		async def _identity_mismatch(request: Request, exc: FederationIdentityMismatch) -> JSONResponse:
		return problem(
		400,
		"federation-identity-mismatch",
		"Request Does Not Match This Operator",
		f"{exc.field} must be omitted or match this operator's own value.",
		request.url.path,
		)

		@app.exception_handler(PartnerNotRegistered)
		async def _partner_not_registered(request: Request, exc: PartnerNotRegistered) -> JSONResponse:
		return problem(
		404,
		"partner-not-registered",
		"Partner Not Registered",
		"No partner operator is registered with this identifier.",
		request.url.path,
		)

		@app.exception_handler(FederationAlreadyExists)
		async def _federation_exists(request: Request, exc: FederationAlreadyExists) -> JSONResponse:
		return problem(

src/federation_manager/api/internal/partner_federations.py

0 → 100644

+94 −0

Original line number	Diff line number	Diff line
		from typing import Annotated
		from uuid import UUID

		from fastapi import APIRouter, Depends, Query, Response
		from fastapi.responses import JSONResponse

		from federation_manager.application.partner_federations import PartnerFederationService
		from federation_manager.contracts.ewbi import (
		FederationContextIdResponse,
		FederationDetails,
		FederationRequestData,
		FederationResponseData,
		ZoneRegisteredData,
		ZoneRegistrationRequestData,
		ZoneRegistrationResponseData,
		)
		from federation_manager.dependencies import get_partner_federation_service

		router = APIRouter(
		prefix="/internal/partners/{partner_op_id}/federations",
		tags=["internal-partner-federations"],
		)

		Service = Annotated[PartnerFederationService, Depends(get_partner_federation_service)]


		@router.post("", responses={200: {"model": FederationResponseData}})
		async def create_federation(
		partner_op_id: UUID, body: FederationRequestData, service: Service
		) -> JSONResponse:
		return JSONResponse(await service.create(partner_op_id, body))


		@router.get("", response_model=FederationContextIdResponse)
		async def get_federation_context_id(
		partner_op_id: UUID, service: Service
		) -> FederationContextIdResponse:
		return FederationContextIdResponse(
		federation_context_id=await service.context_id(partner_op_id)
		)


		@router.get("/{federation_context_id}", responses={200: {"model": FederationDetails}})
		async def get_federation_details(
		partner_op_id: UUID, federation_context_id: str, service: Service
		) -> JSONResponse:
		return JSONResponse(await service.details(partner_op_id, federation_context_id))


		@router.delete("/{federation_context_id}")
		async def delete_federation(
		partner_op_id: UUID, federation_context_id: str, service: Service
		) -> Response:
		await service.delete(partner_op_id, federation_context_id)
		return Response(status_code=200)


		@router.get("/{federation_context_id}/zones", responses={200: {"model": ZoneRegisteredData}})
		async def get_zone_data(
		partner_op_id: UUID,
		federation_context_id: str,
		service: Service,
		zone_id: Annotated[str \| None, Query(alias="zoneId")] = None,
		) -> JSONResponse:
		return JSONResponse(await service.zone_data(partner_op_id, federation_context_id, zone_id))


		@router.post(
		"/{federation_context_id}/zones", responses={200: {"model": ZoneRegistrationResponseData}}
		)
		async def subscribe_zones(
		partner_op_id: UUID,
		federation_context_id: str,
		body: ZoneRegistrationRequestData,
		service: Service,
		) -> JSONResponse:
		return JSONResponse(await service.subscribe_zones(partner_op_id, federation_context_id, body))


		@router.get(
		"/{federation_context_id}/zones/{zone_id}", responses={200: {"model": ZoneRegisteredData}}
		)
		async def get_zone_details(
		partner_op_id: UUID, federation_context_id: str, zone_id: str, service: Service
		) -> JSONResponse:
		return JSONResponse(await service.zone_details(partner_op_id, federation_context_id, zone_id))


		@router.delete("/{federation_context_id}/zones/{zone_id}")
		async def unsubscribe_zone(
		partner_op_id: UUID, federation_context_id: str, zone_id: str, service: Service
		) -> Response:
		await service.unsubscribe_zone(partner_op_id, federation_context_id, zone_id)
		return Response(status_code=200)

src/federation_manager/application/federation.py

+65 −12

Original line number	Diff line number	Diff line
		@@ -9,6 +9,7 @@ from federation_manager.contracts.ewbi import (
		FederationRequestData,
		FederationResponseData,
		MobileNetworkIds,
		ZoneRegistrationRequestData,
		)
		from federation_manager.core.logging import get_logger
		from federation_manager.domain.errors import (
		@@ -16,6 +17,7 @@ from federation_manager.domain.errors import (
		FederationContextUnknown,
		FederationError,
		FederationEstablishmentFailed,
		FederationIdentityMismatch,
		PartnerNotActive,
		PartnerResponseInvalid,
		)
		@@ -47,6 +49,49 @@ class LocalOperator:
		mncs: tuple[str, ...]
		partner_status_link: str
		platform_caps: tuple[str, ...] = ("serviceAPIs",)
		avail_zone_notif_link: str \| None = None

		def federation_request(self, initial_date: datetime) -> FederationRequestData:
		return FederationRequestData(
		initial_date=initial_date,
		partner_status_link=self.partner_status_link,
		orig_op_federation_id=self.federation_id,
		orig_op_country_code=self.country_code,
		orig_op_mobile_network_codes=MobileNetworkIds(mcc=self.mcc, mncs=list(self.mncs)),
		)

		def complete_federation_request(self, request: FederationRequestData) -> FederationRequestData:
		_require_ours("origOPFederationId", request.orig_op_federation_id, self.federation_id)
		_require_ours("origOPCountryCode", request.orig_op_country_code, self.country_code)
		_require_ours("partnerStatusLink", request.partner_status_link, self.partner_status_link)
		codes = request.orig_op_mobile_network_codes
		# Federating only some of our networks is fine; naming one we don't operate is not.
		if codes is not None and (codes.mcc != self.mcc or not set(codes.mncs) <= set(self.mncs)):
		raise FederationIdentityMismatch("origOPMobileNetworkCodes")
		# This operator has no fixed network codes configured, so any claimed one isn't ours.
		if request.orig_op_fixed_network_codes:
		raise FederationIdentityMismatch("origOPFixedNetworkCodes")
		return request.model_copy(
		update={
		"orig_op_federation_id": self.federation_id,
		"orig_op_country_code": self.country_code,
		"orig_op_mobile_network_codes": codes
		or MobileNetworkIds(mcc=self.mcc, mncs=list(self.mncs)),
		}
		)

		def complete_zone_registration(
		self, request: ZoneRegistrationRequestData
		) -> ZoneRegistrationRequestData:
		_require_ours(
		"availZoneNotifLink", request.avail_zone_notif_link, self.avail_zone_notif_link
		)
		return request.model_copy(update={"avail_zone_notif_link": self.avail_zone_notif_link})


		def _require_ours(field: str, value: str \| None, ours: str \| None) -> None:
		if value is not None and value != ours:
		raise FederationIdentityMismatch(field)


		class FederationEstablishmentService:
		@@ -73,23 +118,31 @@ class FederationEstablishmentService:
		existing = await self._contexts.find_active_outbound(partner.id)
		if existing is not None:
		return existing

		request = FederationRequestData(
		initial_date=self._clock(),
		partner_status_link=self._local.partner_status_link,
		orig_op_federation_id=self._local.federation_id,
		orig_op_country_code=self._local.country_code,
		orig_op_mobile_network_codes=MobileNetworkIds(
		mcc=self._local.mcc, mncs=list(self._local.mncs)
		),
		context, _ = await self._request_federation(
		partner, self._local.federation_request(self._clock())
		)
		return context

		async def create(
		self, partner: PartnerOP, request: FederationRequestData
		) -> tuple[FederationContext, object]:
		completed = self._local.complete_federation_request(request)
		if not partner.is_active():
		raise PartnerNotActive(partner.status)
		if await self._contexts.find_active_outbound(partner.id) is not None:
		raise FederationAlreadyExists(partner.id)
		return await self._request_federation(partner, completed)

		async def _request_federation(
		self, partner: PartnerOP, request: FederationRequestData
		) -> tuple[FederationContext, object]:
		response = await self._ewbi_client.post(
		partner,
		CREATE_FEDERATION_PATH,
		request.model_dump(mode="json", by_alias=True, exclude_none=True),
		)
		if not response.is_success():
		raise FederationEstablishmentFailed(partner.id, response.status_code)
		raise FederationEstablishmentFailed(partner.id, response.status_code, response.body)
		try:
		accepted = FederationResponseData.model_validate(response.body)
		except ValidationError:
		@@ -102,10 +155,10 @@ class FederationEstablishmentService:
		federation_context_id=accepted.federation_context_id,
		status=AVAILABLE,
		created_at=self._clock(),
		status_callback_url=self._local.partner_status_link,
		status_callback_url=request.partner_status_link,
		)
		await self._contexts.add(context)
		return context
		return context, response.body

		async def establish_missing(self) -> list[FederationContext]:
		established: list[FederationContext] = []