feat(fm): answer inbound device-location queries from SRM (1fc3b751) · Commits · OOP / code / federation-manager

src/federation_manager/adapters/http/srm_client.py

0 → 100644

+25 −0

Original line number	Diff line number	Diff line
		import httpx

		from federation_manager.contracts.srm import LocationQueryRequestV1, LocationQueryResponseV1
		from federation_manager.domain.errors import NetworkQueryNotApplicable, SrmQueryFailed

		LOCATION_PATH = "/internal/network-queries/location"


		class HttpxSrmClient:
		def __init__(self, client: httpx.AsyncClient, base_url: str) -> None:
		self._client = client
		self._base_url = base_url.rstrip("/")

		async def query_location(self, request: LocationQueryRequestV1) -> LocationQueryResponseV1:
		try:
		response = await self._client.post(
		f"{self._base_url}{LOCATION_PATH}",
		json=request.model_dump(mode="json", exclude_none=True),
		)
		if response.status_code in (404, 422):
		raise NetworkQueryNotApplicable
		response.raise_for_status()
		return LocationQueryResponseV1.model_validate(response.json())
		except (httpx.HTTPError, ValueError) as error:
		raise SrmQueryFailed(str(error)[:200]) from None

src/federation_manager/api/errors.py

+37 −0

Original line number	Diff line number	Diff line
		@@ -13,6 +13,7 @@ from federation_manager.domain.errors import (
		FederationContextMissing,
		FederationContextUnknown,
		IdempotencyKeyReused,
		NetworkQueryNotApplicable,
		NoRouteMatched,
		PartnerEndpointConfigurationError,
		PartnerNotActive,
		@@ -21,6 +22,8 @@ from federation_manager.domain.errors import (
		PartnerTokenConfigurationError,
		PartnerTokenRequestFailed,
		PartnerUnknown,
		SrmQueryFailed,
		UnsupportedServiceApi,
		problem_type,
		)

		@@ -192,6 +195,40 @@ def register_exception_handlers(app: FastAPI) -> None:
		request.url.path,
		)

		@app.exception_handler(NetworkQueryNotApplicable)
		async def _query_not_applicable(
		request: Request, exc: NetworkQueryNotApplicable
		) -> JSONResponse:
		return problem(
		422,
		"service-not-applicable",
		"Service Not Applicable",
		"No local capability can serve the requested target.",
		request.url.path,
		)

		@app.exception_handler(SrmQueryFailed)
		async def _srm_query_failed(request: Request, exc: SrmQueryFailed) -> JSONResponse:
		return problem(
		500,
		"internal-error",
		"Internal Error",
		"The requested network information could not be retrieved.",
		request.url.path,
		)

		@app.exception_handler(UnsupportedServiceApi)
		async def _unsupported_service_api(
		request: Request, exc: UnsupportedServiceApi
		) -> JSONResponse:
		return problem(
		404,
		"service-api-unsupported",
		"Service API Not Supported",
		"This operator does not offer the requested Service API over EWBI.",
		request.url.path,
		)

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

src/federation_manager/api/ewbi/v1/service.py

0 → 100644

+29 −0

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

		from fastapi import APIRouter, Depends, Path

		from federation_manager.api.errors import EWBI_ERROR_RESPONSES
		from federation_manager.api.security import get_bearer_token
		from federation_manager.application.queries import InboundQueryService
		from federation_manager.contracts.ewbi import ApiForwardingRequest, ServiceApiResponse
		from federation_manager.dependencies import get_inbound_query_service
		from federation_manager.domain.ewbi import EWBI_BASE_PATH

		router = APIRouter(prefix=EWBI_BASE_PATH, tags=["ServiceAPIManagement"])

		FederationContextIdPath = Annotated[str, Path(pattern=r"^[A-Za-z0-9][A-Za-z0-9-]*$")]


		@router.post(
		"/{federationContextId}/apiservice/{serviceAPINameVal}",
		operation_id="APIForwarding",
		responses=EWBI_ERROR_RESPONSES,
		)
		async def api_forwarding(
		federationContextId: FederationContextIdPath, # noqa: N803 - GSMA path template name
		serviceAPINameVal: str, # noqa: N803 - GSMA path template name
		body: ApiForwardingRequest,
		service: Annotated[InboundQueryService, Depends(get_inbound_query_service)],
		token: Annotated[str, Depends(get_bearer_token)],
		) -> ServiceApiResponse:
		return await service.forward(token, federationContextId, serviceAPINameVal, body)

src/federation_manager/application/queries.py

0 → 100644

+152 −0

Original line number	Diff line number	Diff line
		from collections.abc import Callable
		from datetime import datetime, timezone
		from uuid import UUID, uuid4

		from federation_manager.application.authentication import PartnerAuthenticator
		from federation_manager.application.authorization import AgreementChecker
		from federation_manager.contracts.ewbi import (
		ApiForwardingRequest,
		ApiResponseBody,
		ServiceApiResponse,
		)
		from federation_manager.contracts.srm import (
		CapabilityParameters,
		CapabilityTarget,
		DeviceTarget,
		LocationQueryRequestV1,
		SourceSpecificationV1,
		)
		from federation_manager.domain.errors import (
		AgreementViolation,
		FederationContextUnknown,
		UnsupportedServiceApi,
		)
		from federation_manager.domain.models import FederationTransaction
		from federation_manager.domain.ports import (
		FederationContextRepositoryPort,
		SrmQueryPort,
		TransactionRepositoryPort,
		)

		INBOUND = "inbound"
		DEVICE_LOCATION = "DeviceLocation"
		API_TYPES = {DEVICE_LOCATION: "device-location"}
		OPG_SOURCE = SourceSpecificationV1(family="gsma_opg", api="APIForwarding", version="1.4.0")


		def _utcnow() -> datetime:
		return datetime.now(timezone.utc)


		class InboundQueryService:
		def __init__(
		self,
		authenticator: PartnerAuthenticator,
		contexts: FederationContextRepositoryPort,
		agreements: AgreementChecker,
		transactions: TransactionRepositoryPort,
		srm: SrmQueryPort,
		*,
		clock: Callable[[], datetime] = _utcnow,
		id_factory: Callable[[], UUID] = uuid4,
		) -> None:
		self._authenticator = authenticator
		self._contexts = contexts
		self._agreements = agreements
		self._transactions = transactions
		self._srm = srm
		self._clock = clock
		self._new_id = id_factory

		async def forward(
		self,
		token: str,
		federation_context_id: str,
		service_api: str,
		request: ApiForwardingRequest,
		) -> ServiceApiResponse:
		api_type = API_TYPES.get(service_api)
		if api_type is None:
		raise UnsupportedServiceApi(service_api)

		partner = await self._authenticator.authenticate(token)
		context = await self._contexts.find_inbound(partner.id, federation_context_id)
		if context is None or context.is_terminated():
		raise FederationContextUnknown(partner.id)
		agreement = await self._agreements.require(partner, api_type)
		specification_id = agreement.resolve_api_spec(api_type)
		if specification_id is None:
		raise AgreementViolation

		correlation_id = self._new_id()
		transaction = FederationTransaction(
		id=self._new_id(),
		partner_op_id=partner.id,
		agreement_id=agreement.id,
		federation_context_row_id=context.id,
		direction=INBOUND,
		correlation_id=correlation_id,
		external_txn_id=request.txn_identifier,
		api_type=api_type,
		status="pending",
		request_summary={"api_type": api_type, "service_api": service_api},
		started_at=self._clock(),
		)
		await self._transactions.add(transaction)

		try:
		result = await self._srm.query_location(
		LocationQueryRequestV1(
		correlation_id=str(correlation_id),
		app_provider_id=str(partner.id),
		service_specification_id=specification_id,
		target=CapabilityTarget(device=_device(request)),
		parameters=_parameters(request),
		source_spec=OPG_SOURCE,
		)
		)
		except Exception:
		await self._transactions.record_outcome(
		transaction.id, status="failed", completed_at=self._clock()
		)
		raise

		await self._transactions.record_outcome(
		transaction.id, status="completed", completed_at=self._clock()
		)
		return ServiceApiResponse(
		customer_id=request.customer_id,
		txn_identifier=request.txn_identifier,
		api_response=ApiResponseBody(
		media_type="application/json",
		response_content=result.model_dump(mode="json", exclude_none=True),
		),
		)


		def _device(request: ApiForwardingRequest) -> DeviceTarget:
		device = request.service_api_body.api_content.get("device", {})
		if not isinstance(device, dict):
		raise AgreementViolation
		target = DeviceTarget(
		phone_number=_text(device.get("phoneNumber")),
		ipv4=_text(
		(device.get("ipv4Address") or {}).get("publicAddress")
		if isinstance(device.get("ipv4Address"), dict)
		else None
		),
		ipv6=_text(device.get("ipv6Address")),
		network_access_id=_text(device.get("networkAccessIdentifier")),
		)
		if not any((target.phone_number, target.ipv4, target.ipv6, target.network_access_id)):
		raise AgreementViolation
		return target


		def _parameters(request: ApiForwardingRequest) -> CapabilityParameters \| None:
		max_age = request.service_api_body.api_content.get("maxAge")
		return CapabilityParameters(max_age_seconds=max_age) if isinstance(max_age, int) else None


		def _text(value: object) -> str \| None:
		return value if isinstance(value, str) and value else None

src/federation_manager/contracts/srm.py

+39 −1

Original line number	Diff line number	Diff line
		# Vendored from srm/interface-contract.md §B/§C — hand-synced; SRM's spec wins.
		from datetime import datetime
		from typing import Literal
		from typing import Any, Literal
		from uuid import UUID

		from pydantic import BaseModel, ConfigDict, Field
		@@ -88,3 +88,41 @@ class SrmOperationCompletedV1(BaseModel):
		error: dict[str, object] \| None = None
		correlation_id: str
		completed_at: datetime


		class DeviceTarget(BaseModel):
		phone_number: str \| None = None
		ipv4: str \| None = None
		ipv6: str \| None = None
		network_access_id: str \| None = None


		class CapabilityTarget(BaseModel):
		device: DeviceTarget \| None = None


		class CapabilityParameters(BaseModel):
		max_age_seconds: int \| None = None


		class SourceSpecificationV1(BaseModel):
		family: Literal["camara", "tmf", "gsma_opg", "internal"]
		api: str
		version: str


		class LocationQueryRequestV1(BaseModel):
		correlation_id: str
		app_provider_id: str
		service_specification_id: UUID
		target: CapabilityTarget
		parameters: CapabilityParameters \| None = None
		source_spec: SourceSpecificationV1 \| None = None


		class LocationQueryResponseV1(BaseModel):
		model_config = ConfigDict(extra="ignore")

		last_location_time: datetime
		area: dict[str, Any]
		srm_resolved_identifier: str \| None = None