Release 2.0.0

Sergio Gimenezrequested to merge
develop into main

Release 2.0.0

Merges develop into main for Release 2. FM is a full rewrite: a FastAPI service that speaks the GSMA OPG.04 E/WBI to partner operators and takes its work from SRM and OEG. The R1 Connexion stack is gone (!12 (merged)).

flowchart LR
  OEG["OEG"] -- "HTTP /internal<br/>partners, agreements, federation, identity" --> FM
  SRM["SRM"] -- "HTTP /internal<br/>onboard, deploy, terminate, instances, zones" --> FM
  FM["Federation Manager 2.0.0"] -- "E/WBI (OPG.04)<br/>OAuth2 client credentials" --> P["partner FM"]
  P -- "E/WBI + callbacks<br/>token must carry our audience" --> FM
  FM --> DB[("fm database<br/>Alembic migrations")]

/internal listens on its own port and never on the EWBI one (!33 (merged)).

What changed since v1.0.0-release

Federation lifecycle

  • FastAPI rewrite with the OPG.04 E/WBI surface, OAuth2 and the SRM command loop (!11 (merged)), aligned with GSMA and wired through SRM (!10 (merged)).
  • Partner registration admin API (!20 (merged)); the partner client secret is accepted by value and stored encrypted (!30 (merged)).
  • GetFederationDetails answered as the partner (!21 (merged)); onPartnerStatusEvent callbacks received (!22 (merged)).
  • Recovery from stale federated state (!26 (merged)).
  • Our identity on GET /internal/identity (!28 (merged)).

Zones and agreements

  • Our own zones offered to partners (!23 (merged)), the zone subscription flow (!24 (merged)), and sync of partner zones with clean-up of ended subscriptions (!36 (merged)).
  • Partner agreements through the internal API (!35 (merged)); one agreement per partner, edited in place (!40 (merged)).
  • Zone changes pushed on validity dates and retried, and withdrawn subscriptions released (!41 (merged)).

Applications

Security and operations

Tested

develop at 853bb20f (image ci-develop-853bb20f, pipeline green), deployed fresh with Helm 2.0 on the i2cat three-operator test environment on 2026-10-23 and federated pairwise:

  • Ansible R2 e2e suite: 23/23 passed. Covers EWBI auth (no token, wrong Keycloak, another partner's federation), federation through OEG, a federation that survives an FM restart, ending and re-establishing a federation, and partner zones.
  • Agreements and instances suite: 10/10 passed. Covers the second agreement refused with 409 and relayed by OEG, PATCH in place, end and start dates, a missed withdrawal resent, a token without our audience refused, a federated deploy from operator 1 onto operator 2's zone with the requested names kept, and a delete on both sides.

After merging

Tag v2.0.0 and publish the image as labs.etsi.org:5050/oop/code/federation-manager:2.0.0, which the Helm chart defaults to. CI doesn't push that tag on its own.

Merge request reports

keycloak/realm-import.json

+28 −0
Changes for keycloak/realm-import.json: 28 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -39,6 +39,20 @@ 
      ],

      "optionalClientScopes": [

        "fed-mgmt-notif"

      ],

      "protocolMappers": [

        {

          "name": "audience-federation-manager",

          "protocol": "openid-connect",

          "protocolMapper": "oidc-audience-mapper",

          "consentRequired": false,

          "config": {

            "included.custom.audience": "federation-manager",

            "access.token.claim": "true",

            "introspection.token.claim": "true",

            "id.token.claim": "false"

          }

        }

      ]

    },

    {
@@ -60,6 +74,20 @@ 
      ],

      "optionalClientScopes": [

        "fed-mgmt-notif"

      ],

      "protocolMappers": [

        {

          "name": "audience-federation-manager",

          "protocol": "openid-connect",

          "protocolMapper": "oidc-audience-mapper",

          "consentRequired": false,

          "config": {

            "included.custom.audience": "federation-manager",

            "access.token.claim": "true",

            "introspection.token.claim": "true",

            "id.token.claim": "false"

          }

        }

      ]

    }

  ]

src/federation_manager/adapters/database/agreement_repo.py

+82 −3
Changes for src/federation_manager/adapters/database/agreement_repo.py: 82 added lines, 3 removed lines.
Original line number Diff line number Diff line 
from datetime import datetime

from typing import Any

from uuid import UUID



from sqlalchemy import and_, func, or_, select

from sqlalchemy import Row, and_, func, insert, or_, select, update

from sqlalchemy.exc import IntegrityError

from sqlalchemy.ext.asyncio import AsyncSession



from federation_manager.adapters.database.tables import federation_agreements as agreements

from federation_manager.domain.errors import AgreementExists

from federation_manager.domain.models import Agreement, AppMapping



_ONE_PER_PARTNER = "uq_federation_agreements_one_per_partner"





class PostgresAgreementRepo:

    def __init__(self, session: AsyncSession) -> None:
@@ -26,8 +32,67 @@ class PostgresAgreementRepo: 
            .limit(1)

        )

        row = (await self._session.execute(stmt)).one_or_none()

        if row is None:

            return None

        return None if row is None else _to_agreement(row)



    async def find_by_id(self, agreement_id: UUID) -> Agreement | None:

        stmt = select(agreements).where(agreements.c.id == agreement_id)

        row = (await self._session.execute(stmt)).one_or_none()

        return None if row is None else _to_agreement(row)



    async def list_for_partner(self, partner_id: UUID) -> list[Agreement]:

        stmt = (

            select(agreements)

            .where(agreements.c.partner_op_id == partner_id)

            .order_by(agreements.c.created_at)

        )

        return [_to_agreement(row) for row in (await self._session.execute(stmt)).all()]



    async def next_validity_boundary(self, after: datetime) -> datetime | None:

        live = agreements.c.status == "active"

        starts = select(func.min(agreements.c.valid_from)).where(

            live, agreements.c.valid_from > after

        )

        ends = select(func.min(agreements.c.valid_until)).where(

            live, agreements.c.valid_until > after

        )

        found = [

            moment

            for moment in (

                (await self._session.execute(starts)).scalar_one_or_none(),

                (await self._session.execute(ends)).scalar_one_or_none(),

            )

            if moment is not None

        ]

        return min(found, default=None)



    async def add(self, agreement: Agreement) -> None:

        await self._write(

            agreement,

            insert(agreements).values(

                id=agreement.id, partner_op_id=agreement.partner_op_id, **_values(agreement)

            ),

        )



    async def update(self, agreement: Agreement) -> None:

        await self._write(

            agreement,

            update(agreements).where(agreements.c.id == agreement.id).values(**_values(agreement)),

        )



    async def _write(self, agreement: Agreement, stmt: Any) -> None:

        try:

            await self._session.execute(stmt)

            await self._session.commit()

        except IntegrityError as error:

            await self._session.rollback()

            # a concurrent create won, or an expired old agreement is being brought back

            # while the partner has a live one

            if _ONE_PER_PARTNER not in str(error.orig):

                raise

            raise AgreementExists(agreement.partner_op_id, None) from None





def _to_agreement(row: Row[Any]) -> Agreement:

    return Agreement(

        id=row.id,

        partner_op_id=row.partner_op_id,
@@ -50,4 +115,18 @@ class PostgresAgreementRepo: 
        valid_until=row.valid_until,

        status=row.status,

        zone_resources={UUID(str(k)): v for k, v in (row.zone_resources or {}).items()},

        usage_limits=dict(row.usage_limits or {}),

    )





def _values(agreement: Agreement) -> dict[str, Any]:

    return {

        "permitted_api_types": sorted(agreement.permitted_api_types),

        "permitted_zone_ids": sorted(str(z) for z in agreement.permitted_zone_ids),

        "service_spec_mappings": agreement.service_spec_mappings(),

        "usage_limits": agreement.usage_limits,

        "zone_resources": {str(k): v for k, v in agreement.zone_resources.items()},

        "valid_from": agreement.valid_from,

        "valid_until": agreement.valid_until,

        "status": agreement.status,

    }

src/federation_manager/adapters/database/federation_context_repo.py

+31 −2
Changes for src/federation_manager/adapters/database/federation_context_repo.py: 31 added lines, 2 removed lines.
Original line number Diff line number Diff line 
from typing import Any

from uuid import UUID



from sqlalchemy import Select, insert, select, update
@@ -35,6 +36,14 @@ class PostgresFederationContextRepo: 
            .limit(1)

        )



    async def list_active_inbound(self) -> list[FederationContext]:

        stmt = (

            select(contexts)

            .where(contexts.c.direction == "inbound", contexts.c.status == "available")

            .order_by(contexts.c.created_at)

        )

        return [_to_context(row) for row in (await self._session.execute(stmt)).all()]



    async def find_inbound(

        self, partner_id: UUID, federation_context_id: str

    ) -> FederationContext | None:
@@ -69,6 +78,7 @@ class PostgresFederationContextRepo: 
                direction=context.direction,

                federation_context_id=context.federation_context_id,

                status_callback_url=context.status_callback_url,

                offered_zone_ids=_zone_ids(context.offered_zone_ids),

                status=context.status,

                created_at=context.created_at,

            )
@@ -81,10 +91,24 @@ class PostgresFederationContextRepo: 
        )

        await self._session.commit()



    async def set_offered_zones(self, context_id: UUID, zone_ids: frozenset[UUID]) -> None:

        await self._session.execute(

            update(contexts)

            .where(contexts.c.id == context_id)

            .values(offered_zone_ids=_zone_ids(zone_ids))

        )

        await self._session.commit()



    async def _one(self, stmt: Select[tuple[object, ...]]) -> FederationContext | None:

        row = (await self._session.execute(stmt)).one_or_none()

        if row is None:

            return None

        return None if row is None else _to_context(row)





def _zone_ids(zone_ids: frozenset[UUID] | None) -> list[str] | None:

    return None if zone_ids is None else sorted(str(z) for z in zone_ids)





def _to_context(row: Any) -> FederationContext:

    return FederationContext(

        id=row.id,

        partner_op_id=row.partner_op_id,
@@ -94,4 +118,9 @@ class PostgresFederationContextRepo: 
        created_at=row.created_at,

        agreement_id=row.agreement_id,

        status_callback_url=row.status_callback_url,

        offered_zone_ids=(

            None

            if row.offered_zone_ids is None

            else frozenset(UUID(str(z)) for z in row.offered_zone_ids)

        ),

    )

src/federation_manager/adapters/database/partner_lock.py

0 → 100644
+31 −0
Changes for src/federation_manager/adapters/database/partner_lock.py: 31 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import hashlib

from collections.abc import AsyncIterator

from contextlib import asynccontextmanager

from uuid import UUID



from sqlalchemy import text

from sqlalchemy.ext.asyncio import AsyncEngine





class PostgresPartnerLock:

    """A Postgres advisory lock per partner, so two FM replicas never send the same change."""



    def __init__(self, engine: AsyncEngine) -> None:

        self._engine = engine



    @asynccontextmanager

    async def hold(self, partner_id: UUID) -> AsyncIterator[None]:

        key = _key(partner_id)

        # a connection of its own: the lock lives as long as the connection holds it, whatever

        # the work inside commits or rolls back

        async with self._engine.connect() as connection:

            await connection.execute(text("SELECT pg_advisory_lock(:key)"), {"key": key})

            try:

                yield

            finally:

                await connection.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": key})





def _key(partner_id: UUID) -> int:

    digest = hashlib.blake2b(b"fm-zone-offer:" + partner_id.bytes, digest_size=8).digest()

    return int.from_bytes(digest, "big", signed=True)

src/federation_manager/adapters/database/partner_repo.py

+61 −14
Changes for src/federation_manager/adapters/database/partner_repo.py: 61 added lines, 14 removed lines.
Original line number Diff line number Diff line 
from typing import Any

from uuid import UUID



from sqlalchemy import Select, insert, select, update

from sqlalchemy import Select, delete, func, insert, select, update

from sqlalchemy.dialects.postgresql import insert as pg_insert

from sqlalchemy.exc import IntegrityError

from sqlalchemy.ext.asyncio import AsyncSession



from federation_manager.adapters.database.tables import partner_ops

from federation_manager.adapters.database.tables import partner_op_secrets, partner_ops

from federation_manager.domain.errors import PartnerRegistrationConflict

from federation_manager.domain.models import PartnerOP
@@ -13,43 +14,87 @@ from federation_manager.domain.models import PartnerOP 
_UNIQUE_FIELDS = ("mcc_mnc", "oauth2_client_id")





def _partners() -> Select[tuple[object, ...]]:

    secret_set = partner_op_secrets.c.partner_op_id.is_not(None).label("our_client_secret_set")

    return select(partner_ops, secret_set).outerjoin(

        partner_op_secrets, partner_op_secrets.c.partner_op_id == partner_ops.c.id

    )





class PostgresPartnerRepo:

    def __init__(self, session: AsyncSession) -> None:

        self._session = session



    async def find_by_oauth2_client_id(self, client_id: str) -> PartnerOP | None:

        return await self._one(

            select(partner_ops).where(partner_ops.c.oauth2_client_id == client_id)

        )

        return await self._one(_partners().where(partner_ops.c.oauth2_client_id == client_id))



    async def find_by_id(self, partner_id: UUID) -> PartnerOP | None:

        return await self._one(select(partner_ops).where(partner_ops.c.id == partner_id))

        return await self._one(_partners().where(partner_ops.c.id == partner_id))



    async def list_active(self) -> list[PartnerOP]:

        stmt = select(partner_ops).where(partner_ops.c.status == "active")

        stmt = _partners().where(partner_ops.c.status == "active")

        rows = (await self._session.execute(stmt)).all()

        return [self._to_partner(row) for row in rows]



    async def find_by_mcc_mnc(self, mcc_mnc: str) -> PartnerOP | None:

        return await self._one(select(partner_ops).where(partner_ops.c.mcc_mnc == mcc_mnc))

        return await self._one(_partners().where(partner_ops.c.mcc_mnc == mcc_mnc))



    async def list_all(self) -> list[PartnerOP]:

        stmt = select(partner_ops).order_by(partner_ops.c.created_at)

        stmt = _partners().order_by(partner_ops.c.created_at)

        rows = (await self._session.execute(stmt)).all()

        return [self._to_partner(row) for row in rows]



    async def add(self, partner: PartnerOP) -> None:

        await self._write(insert(partner_ops).values(id=partner.id, **self._values(partner)))

    async def find_encrypted_secret(self, partner_id: UUID) -> bytes | None:

        stmt = select(partner_op_secrets.c.encrypted_secret).where(

            partner_op_secrets.c.partner_op_id == partner_id

        )

        blob: bytes | None = (await self._session.execute(stmt)).scalar_one_or_none()

        return blob



    async def set_supported_apis(self, partner_id: UUID, apis: tuple[str, ...]) -> None:

        # written only from what the partner tells us, never through the registry

        await self._session.execute(

            update(partner_ops)

            .where(partner_ops.c.id == partner_id)

            .values(supported_apis=list(apis))

        )

        await self._session.commit()



    async def add(self, partner: PartnerOP, encrypted_secret: bytes | None = None) -> None:

        await self._write(

            insert(partner_ops).values(id=partner.id, **self._values(partner)),

            *self._secret_writes(partner, encrypted_secret),

        )



    async def update(self, partner: PartnerOP) -> None:

    async def update(self, partner: PartnerOP, encrypted_secret: bytes | None = None) -> None:

        await self._write(

            update(partner_ops)

            .where(partner_ops.c.id == partner.id)

            .values(**self._values(partner))

            .values(**self._values(partner)),

            *self._secret_writes(partner, encrypted_secret),

        )



    @staticmethod

    def _secret_writes(partner: PartnerOP, encrypted_secret: bytes | None) -> list[Any]:

        if not partner.our_client_secret_set:

            return [

                delete(partner_op_secrets).where(partner_op_secrets.c.partner_op_id == partner.id)

            ]

        if encrypted_secret is None:

            return []

        upsert = pg_insert(partner_op_secrets).values(

            partner_op_id=partner.id, encrypted_secret=encrypted_secret

        )

        return [

            upsert.on_conflict_do_update(

                index_elements=[partner_op_secrets.c.partner_op_id],

                set_={"encrypted_secret": encrypted_secret, "updated_at": func.now()},

            )

        ]



    async def _write(self, stmt: Any) -> None:

    async def _write(self, *stmts: Any) -> None:

        try:

            for stmt in stmts:

                await self._session.execute(stmt)

            await self._session.commit()

        except IntegrityError as error:
@@ -89,4 +134,6 @@ class PostgresPartnerRepo: 
            token_endpoint=row.token_endpoint,

            base_url=row.base_url,

            name=row.name,

            our_client_secret_set=row.our_client_secret_set,

            supported_apis=tuple(row.supported_apis or ()),

        )