feat(fm): EWBI heartbeat endpoint with RFC 7807 errors

from fastapi import FastAPI, Request

from fastapi.responses import JSONResponse

from federation_manager.domain.errors import (

    AuthenticationFailed,

    PartnerNotActive,

    PartnerUnknown,

)

_TYPE_BASE = "urn:oop:ewbi:error:"

def problem(status: int, code: str, title: str, detail: str, instance: str) -> JSONResponse:

    return JSONResponse(

        status_code=status,

        media_type="application/problem+json",

        content={

            "type": f"{_TYPE_BASE}{code}",

            "title": title,

            "status": status,

            "detail": detail,

            "instance": instance,

        },

    )

def register_exception_handlers(app: FastAPI) -> None:

    # Details stay generic: nothing about our internals crosses the operator boundary.

    @app.exception_handler(AuthenticationFailed)

    async def _auth_failed(request: Request, exc: AuthenticationFailed) -> JSONResponse:

        return problem(

            401,

            "authentication-failed",

            "Authentication Failed",

            "Client certificate missing or unreadable.",

            request.url.path,

        )

    @app.exception_handler(PartnerUnknown)

    async def _partner_unknown(request: Request, exc: PartnerUnknown) -> JSONResponse:

        return problem(

            401,

            "partner-unknown",

            "Unknown Partner",

            "Client certificate is not registered with this operator.",

            request.url.path,

        )

    @app.exception_handler(PartnerNotActive)

    async def _partner_not_active(request: Request, exc: PartnerNotActive) -> JSONResponse:

        return problem(

            403,

            "partner-not-active",

            "Partner Not Active",

            "Federation with this partner is not currently active.",

            request.url.path,

        )

from typing import Annotated

from fastapi import APIRouter, Depends, Header

from federation_manager.application.authentication import PartnerAuthenticator

from federation_manager.dependencies import get_partner_authenticator

from federation_manager.domain.errors import AuthenticationFailed

router = APIRouter(prefix="/ewbi/v1/management", tags=["ewbi-management"])

@router.post("/heartbeat")

async def heartbeat(

    auth: Annotated[PartnerAuthenticator, Depends(get_partner_authenticator)],

    x_client_certificate_thumbprint: Annotated[str | None, Header()] = None,

) -> dict[str, str]:

    if not x_client_certificate_thumbprint:

        raise AuthenticationFailed

    await auth.authenticate(x_client_certificate_thumbprint)

    return {"status": "ALIVE"}

from collections.abc import AsyncIterator

from typing import Annotated

from fastapi import Depends, Request

from sqlalchemy.ext.asyncio import AsyncSession

from federation_manager.adapters.database.partner_repo import PostgresPartnerRepo

from federation_manager.application.authentication import PartnerAuthenticator

from federation_manager.domain.ports import PartnerRepositoryPort

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:

    async with request.app.state.session_maker() as session:

        yield session

def get_partner_repo(

    session: Annotated[AsyncSession, Depends(get_session)],

) -> PartnerRepositoryPort:

    return PostgresPartnerRepo(session)

def get_partner_authenticator(

    repo: Annotated[PartnerRepositoryPort, Depends(get_partner_repo)],

) -> PartnerAuthenticator:

    return PartnerAuthenticator(repo)