Merge branch 'feat/partner-typed-secret' into 'develop' (916f35d8) · Commits · OOP / code / federation-manager

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: