Commit 5a80ea19 authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

feat(fm): accept the partner client secret by value, stored encrypted 

The portal registers partners from a browser and can't mount a secret file
into FM's container. POST/PATCH /internal/partners now take our_client_secret:
FM encrypts it with AES-256-GCM under FM_PARTNER_SECRET_KEY into
partner_op_secrets and only reports our_client_secret_set. our_client_secret_ref
still works; a partner uses one or the other. Without a key, a request
carrying a secret gets 501.
parent 2fb00308

README.md

+1 −0
Changes for README.md: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -97,6 +97,7 @@ Settings are environment variables with the `FM_` prefix, read by `federation_ma 
| `FM_BOOTSTRAP_FEDERATION` | `false` | Federate with every active partner at startup (ADR-0043) |

| `FM_EVENT_CONSUMER_DURABLE` | `fm-event-worker` | JetStream durable name; must differ per deployment |

| `FM_ALLOW_INSECURE_PARTNER_ENDPOINTS` | `false` | Permit plain-HTTP partner endpoints. **Development only** |

| `FM_PARTNER_SECRET_KEY` | unset | Base64url 32-byte key that encrypts partner secrets sent as `our_client_secret`. Unset: such requests get 501 and partners use `our_client_secret_ref` files. Changing it makes stored secrets unreadable |



## API

pyproject.toml

+1 −0
Changes for pyproject.toml: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -11,6 +11,7 @@ requires-python = ">=3.12" 
dependencies = [

    "alembic>=1.13",

    "asyncpg>=0.30",

    "cryptography>=43",

    "fastapi[standard]>=0.115",

    "httpx>=0.27",

    "nats-py>=2.6",

src/federation_manager/adapters/database/partner_repo.py

+51 −14
Changes for src/federation_manager/adapters/database/partner_repo.py: 51 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,78 @@ 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 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 +125,5 @@ class PostgresPartnerRepo: 
            token_endpoint=row.token_endpoint,

            base_url=row.base_url,

            name=row.name,

            our_client_secret_set=row.our_client_secret_set,

        )

src/federation_manager/adapters/database/tables.py

+23 −0
Changes for src/federation_manager/adapters/database/tables.py: 23 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -7,6 +7,7 @@ from sqlalchemy import ( 
    ForeignKey,

    Index,

    Integer,

    LargeBinary,

    MetaData,

    String,

    Table,
@@ -59,6 +60,28 @@ partner_ops = Table( 
    ),

)



# Kept apart from partner_ops so the ciphertext is read only when a token is requested.

partner_op_secrets = Table(

    "partner_op_secrets",

    metadata,

    Column(

        "partner_op_id",

        PGUUID(as_uuid=True),

        ForeignKey("partner_ops.id", ondelete="CASCADE"),

        primary_key=True,

    ),

    # AES-256-GCM: 12-byte nonce followed by ciphertext and tag

    Column("encrypted_secret", LargeBinary, nullable=False),

    Column("created_at", DateTime(timezone=True), nullable=False, server_default=func.now()),

    Column(

        "updated_at",

        DateTime(timezone=True),

        nullable=False,

        server_default=func.now(),

        onupdate=func.now(),

    ),

)



federation_agreements = Table(

    "federation_agreements",

    metadata,

src/federation_manager/adapters/security/client_secret_token_provider.py

+29 −3
Changes for src/federation_manager/adapters/security/client_secret_token_provider.py: 29 added lines, 3 removed lines.
Original line number Diff line number Diff line 
import asyncio

import time

from collections.abc import Callable

from collections.abc import Awaitable, Callable

from dataclasses import dataclass

from pathlib import Path

from typing import TypeAlias, cast
@@ -11,10 +11,14 @@ import httpx 
from federation_manager.domain.errors import (

    PartnerTokenConfigurationError,

    PartnerTokenRequestFailed,

    SecretUndecryptable,

)

from federation_manager.domain.models import PartnerOP

from federation_manager.domain.ports import SecretCipherPort



_CacheKey: TypeAlias = tuple[UUID, str, str, str, str]

StoredSecretLoader: TypeAlias = Callable[[UUID], Awaitable[bytes | None]]

_STORED = "stored"





@dataclass(frozen=True)
@@ -23,16 +27,20 @@ class _CachedToken: 
    expires_at: float





class FileClientSecretTokenProvider:

class ClientSecretTokenProvider:

    def __init__(

        self,

        client: httpx.AsyncClient,

        *,

        stored_secrets: StoredSecretLoader | None = None,

        cipher: SecretCipherPort | None = None,

        refresh_skew_seconds: float = 30.0,

        clock: Callable[[], float] = time.monotonic,

        allow_insecure: bool = False,

    ) -> None:

        self._client = client

        self._stored_secrets = stored_secrets

        self._cipher = cipher

        self._allow_insecure = allow_insecure

        self._refresh_skew_seconds = refresh_skew_seconds

        self._clock = clock
@@ -41,8 +49,12 @@ class FileClientSecretTokenProvider: 


    async def token_for(self, partner: PartnerOP, scope: str = "fed-mgmt") -> str:

        client_id = self._required(partner, "client id", partner.our_client_id)

        secret_ref = self._required(partner, "client secret ref", partner.our_client_secret_ref)

        secret_ref = partner.our_client_secret_ref or (

            _STORED if partner.our_client_secret_set else None

        )

        secret_ref = self._required(partner, "client secret", secret_ref)

        token_endpoint = self._token_endpoint(partner)

        # a replaced stored secret is picked up at the next refresh, as a rewritten file is

        cache_key = (partner.id, scope, client_id, secret_ref, token_endpoint)



        cached = self._cache.get(cache_key)
@@ -55,6 +67,9 @@ class FileClientSecretTokenProvider: 
            if cached is not None and self._is_usable(cached):

                return cached.access_token



            if secret_ref == _STORED:

                secret = await self._stored_secret(partner)

            else:

                secret = self._read_secret(partner, secret_ref)

            access_token, expires_in = await self._request_token(

                partner,
@@ -90,6 +105,17 @@ class FileClientSecretTokenProvider: 
            raise PartnerTokenConfigurationError(partner.id, "token endpoint")

        return value



    async def _stored_secret(self, partner: PartnerOP) -> str:

        if self._stored_secrets is None or self._cipher is None:

            raise PartnerTokenConfigurationError(partner.id, "client secret")

        blob = await self._stored_secrets(partner.id)

        if blob is None:

            raise PartnerTokenConfigurationError(partner.id, "client secret")

        try:

            return self._cipher.decrypt(blob)

        except SecretUndecryptable:

            raise PartnerTokenConfigurationError(partner.id, "client secret") from None



    @staticmethod

    def _read_secret(partner: PartnerOP, secret_ref: str) -> str:

        try: