Add support for external coordinator to tell two instances to federate their TAK servers

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "1.19.5+261028"
+current_version = "1.20.0+261027"
 parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)\\+(?P<build>\\d+)"
 serialize = ["{major}.{minor}.{patch}+{build}"]
 search = "{current_version}"

AGENTS.md

@@ -61,6 +61,8 @@ prek run --all-files
 - Router modules go in `src/rasenmaeher_api/routers/` named by domain (e.g., `enroll.py`).
 - SQLModel models in `src/rasenmaeher_api/models/`.
 - Lint/format with ruff (config in `pyproject.toml`); prek enforces this.
+- Do not use `assert` for runtime checks outside of tests
+- Do not add `#nosec` comments without user approval
 
 ## Architecture Notes
 

README.rst

@@ -309,7 +309,7 @@ is passed through the system.
 
 Export a previously generated key to the container::
 
-    export EXTERNAL_JWT_PUBKEY_B64=$(cat ~/p/jwt-test/private_key.key | base64)
+    export EXTERNAL_JWT_PUBKEY_B64=$(cat ~/p/jwt-test/public_key.pub | base64)
     rmlocal up --build
 
 Generate a token and set it in the environment `TOKEN`. The JWT should have
@@ -328,3 +328,91 @@ Versioning is handled with bump-my-version_. To increment, use ``bump-my-version
 You can use ``bump-my-version show-bump`` to see how each option would affect the version.
 
 .. _bump-my-version: https://github.com/callowayproject/bump-my-version
+
+TAK federation orchestration
+----------------------------
+
+Federation orchestration is specified only for TAK Servers because there
+is urgent need for it and this compose based deployment is being deprecated.
+A generic federation orchestration system will be specified for https://github.com/opendefence/core
+
+A service whose JWT signing public key RMAPI trusts can call:
+
+* ``POST /api/v1/federation/tak/trust``
+* ``POST /api/v1/federation/tak/outgoing``
+
+Both require ``Authorization: Bearer <JWT>`` and accept exactly
+``{"remote_base": "fast-flounder.solution.dev.pvarki.fi"}``. The peer is a full
+DNS hostname with an optional HTTPS port (``:4439`` for development). Schemes,
+paths, credentials, IP literals, ``tak.`` prefixes, unknown fields and
+self-federation are rejected. Hostnames are lowercased and port 443 is omitted.
+The first label must fit TAK's 30-character display-name limit. Cross-domain
+peers are allowed.
+
+The caller runs trust on both servers, waits for both successful responses,
+and calls outgoing on **exactly one**. The caller owns this coordination;
+RMAPI does not elect a server or initiate a reverse connection.
+
+Use multikeyjwt 1.8 or later to issue a fresh token for each request::
+
+    from uuid import uuid4
+    from multikeyjwt import Issuer
+
+    token = Issuer.singleton().issue(
+        {
+            "iss": "federation-orchestrator",
+            "sub": "federation-orchestrator",
+            "aud": "rasenmaeher:tak-federation:localmaeher.dev.pvarki.fi",
+            "nonce": str(uuid4()),
+            "federation_action": "trust",
+            "remote_base": "fast-flounder.solution.dev.pvarki.fi",
+        },
+        lifetime=60,
+    )
+
+The signing key belongs to the external service. Provision only its PEM public
+key through ``JWT_PUBKEY_PATH`` or ``EXTERNAL_JWT_PUBKEY_B64`` (Compose maps
+``RMAPI_EXTERNAL_JWT_PUBKEY_B64``). Restart workers after key changes; removing
+the environment variable alone does not remove a previously persisted
+``external.pub``. Issuer and subject are attribution; signing-key trust is not
+bound to a particular issuer name.
+
+The audience must be one exact string for the target deployment's manifest
+``dns``; it contains no HTTPS port. Audience arrays are rejected. ``iss`` and
+``sub`` must be nonempty strings. Integer ``iat``, ``nbf`` and ``exp`` are all
+required, with ``iat <= nbf < exp`` and at most 300 seconds from issue to expiry.
+Clocks must be synchronized; verification uses zero clock leeway. The signed
+action must match the route, and the canonical signed peer must match the body,
+including any nondefault HTTPS port. Browser cookies and ordinary session
+tokens cannot authorize these routes.
+
+Every nonce is a UUID, normalized before consumption in the existing database.
+Replays and concurrent duplicate requests are rejected before dispatch. After
+an authorized attempt, retry with a fresh token and nonce, even if the product
+request failed or timed out. TAK's idempotent readbacks make the desired-state
+retry safe.
+
+RMAPI synchronously calls only manifest product ``tak`` using its product mTLS
+identity, a fixed internal federation path, no redirects and a 30-second budget.
+It never forwards the external JWT. The TAK integration must provide both
+internal federation endpoints; deploy that implementation before using these
+public routes. Local development HTTPS uses port 4439 and the development CA;
+use a verified CA bundle when calling it.
+
+HTTP 200 confirms a validated configuration readback, including no-ops;
+``changed`` distinguishes changes from existing configuration. Trust returns the
+CA fingerprint and both default group directions. Outgoing returns the observed
+connection status (possibly null); success does not prove traffic delivery or
+persistence across restart.
+
+Failures use ``detail`` with ``code``, ``stage``, ``message`` and
+``may_have_changed``. Statuses are 403 for JWT/replay rejection, 422 for input,
+404 when TAK is absent, 409 for prerequisite/configuration conflicts, 502 for
+upstream or invalid responses, 503 for unavailable local resources/lock state,
+and 504 for timeouts. A timeout may follow a successful write. Verified actor,
+issuer, operation, peer, nonce and result are audited without logging the JWT.
+
+Upgrading to multikeyjwt 1.8 also changes existing authentication: any
+Authorization header takes precedence over cookies, including invalid/empty
+headers. Cookie clients must omit unusable headers. Optional JWT authentication
+returns ``None`` on invalid credentials; existing mTLS fallback remains usable.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "rasenmaeher_api"
-version = "1.19.5+261028"
+version = "1.20.0+261027"
 description = "python-rasenmaeher-api"
 authors = [
     { name = "Aciid", email = "703382+Aciid@users.noreply.github.com" },
@@ -17,7 +17,7 @@ dependencies = [
     "pydantic>=2.0,<3.0",
     "pydantic-settings>=2.0,<3.0",
     "requests>=2.31,<3.0",
-    "multikeyjwt>=1.7,<2.0",
+    "multikeyjwt>=1.8,<2.0",
     "uvicorn[standard]>=0.47,<1.0",
     "gunicorn>=26.0",
     "libpvarki>=2.4,<3.0",
@@ -95,7 +95,6 @@ extend-immutable-calls = [
 ]
 
 [tool.mypy]
-python_version="3.12"
 strict=true
 
 [[tool.mypy.overrides]]

src/rasenmaeher_api/__init__.py

@@ -1,3 +1,3 @@
 """python-rasenmaeher-api"""
 
-__version__ = "1.19.5+261028"  # NOTE Use `bump-my-version` to bump versions correctly
+__version__ = "1.20.0+261027"  # NOTE Use `bump-my-version` to bump versions correctly

src/rasenmaeher_api/db/base.py

@@ -23,9 +23,11 @@ class ORMBaseModel(SQLModel, table=False):
     __table_args__ = {"schema": "raesenmaeher"}
 
     pk: uuid.UUID = Field(primary_key=True, default_factory=uuid.uuid4)
-    created: datetime.datetime = Field(sa_column_kwargs={"default": utcnow}, nullable=False)
-    updated: datetime.datetime = Field(sa_column_kwargs={"default": utcnow, "onupdate": utcnow}, nullable=False)
-    deleted: datetime.datetime = Field(nullable=True)
+    created: datetime.datetime | None = Field(default=None, sa_column_kwargs={"default": utcnow}, nullable=False)
+    updated: datetime.datetime | None = Field(
+        default=None, sa_column_kwargs={"default": utcnow, "onupdate": utcnow}, nullable=False
+    )
+    deleted: datetime.datetime | None = Field(default=None, nullable=True)
 
     @classmethod
     async def by_pk(cls, pkin: str | uuid.UUID, allow_deleted: bool = False) -> Self:

src/rasenmaeher_api/db/config.py

@@ -24,7 +24,9 @@ class DBConfig:
     """DB config dataclass, functools etc used to avoid import-time side-effects"""
 
     driver: str = field(
-        default_factory=cast(Callable[..., str], functools.partial(config, "RM_DATABASE_DRIVER", default="postgresql"))
+        default_factory=cast(
+            Callable[..., str], functools.partial(config, "RM_DATABASE_DRIVER", default="postgresql+psycopg2")
+        )
     )
     host: str | None = field(default_factory=functools.partial(config, "RM_DATABASE_HOST", default=None))
     port: int = field(
@@ -84,7 +86,8 @@ def singleton(cls, **kwargs: Any) -> "DBConfig":
         """Get a singleton"""
         if DBConfig._singleton is None:
             DBConfig._singleton = DBConfig(**kwargs)
-        assert DBConfig._singleton is not None  # nosec B101
+        if DBConfig._singleton is None:
+            raise RuntimeError("DBConfig singleton is not initialized")
         return DBConfig._singleton
 
     def __post_init__(self) -> None:

src/rasenmaeher_api/db/dbinit.py

@@ -30,7 +30,8 @@ async def init_db() -> None:
     lockpath = Path(tempfile.gettempdir()) / "dbinit.lock"
     lock = filelock.FileLock(lockpath)
     wrapper = EngineWrapper.singleton()
-    assert wrapper.engine  # nosec B101
+    if not wrapper.engine:
+        raise RuntimeError("Database engine is not initialized")
     engine = wrapper.engine
     try:
         await asyncio.sleep(random.random() * 2)  # nosec B311

src/rasenmaeher_api/db/engine.py

@@ -26,12 +26,14 @@ def singleton(cls, **kwargs: Any) -> "EngineWrapper":
         """Get a singleton"""
         if EngineWrapper._singleton is None:
             EngineWrapper._singleton = EngineWrapper(**kwargs)
-        assert EngineWrapper._singleton is not None  # nosec B101
+        if EngineWrapper._singleton is None:
+            raise RuntimeError("EngineWrapper singleton is not initialized")
         return EngineWrapper._singleton
 
     def __post_init__(self) -> None:
         """create one engine"""
-        assert self.config.dsn  # nosec B101
+        if not self.config.dsn:
+            raise RuntimeError("Database DSN is not configured")
         if self.config.pool_min_size and self.config.pool_max_size:
             self.engine = create_engine(
                 self.config.dsn,

src/rasenmaeher_api/db/enrollments.py

@@ -42,7 +42,9 @@ class EnrollmentPool(ORMBaseModel, table=True):
 
     owner: uuid.UUID = Field(f"{ORMBaseModel.__table_args__['schema']}.users.pk", nullable=False)
     active: bool = Field(nullable=False, default=True)
-    extra: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
+    extra: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
     invitecode: str = Field(nullable=False, index=True, unique=True)
 
     @classmethod
@@ -112,7 +114,7 @@ async def create_for_owner(cls, person: Person, extra: dict[str, Any] | None = N
                 invitecode=code,
                 active=True,
                 owner=person.pk,
-                extra=extra,
+                extra=extra or {},
             )
             session.add(obj)
             session.commit()
@@ -156,18 +158,20 @@ class Enrollment(ORMBaseModel, table=True):
 
     approvecode: str = Field(nullable=False, index=True, unique=True)
     callsign: str = Field(nullable=False, index=True, unique=True)
-    decided_on: datetime.datetime = Field(nullable=True, default=None)
-    decided_by: uuid.UUID = Field(
+    decided_on: datetime.datetime | None = Field(nullable=True, default=None)
+    decided_by: uuid.UUID | None = Field(
         foreign_key=f"{ORMBaseModel.__table_args__['schema']}.users.pk", nullable=True, default=None
     )
-    person: uuid.UUID = Field(
+    person: uuid.UUID | None = Field(
         foreign_key=f"{ORMBaseModel.__table_args__['schema']}.users.pk", nullable=True, default=None
     )
-    pool: uuid.UUID = Field(
+    pool: uuid.UUID | None = Field(
         foreign_key=f"{ORMBaseModel.__table_args__['schema']}.enrollmentpools.pk", nullable=True, default=None
     )
     state: int = Field(nullable=False, index=False, unique=False, default=EnrollmentState.PENDING)
-    extra: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
+    extra: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
     csr: str | None = Field(default=None, nullable=True)
 
     @classmethod
@@ -299,7 +303,7 @@ async def create_for_callsign(
                 approvecode=code,
                 callsign=callsign,
                 state=EnrollmentState.PENDING,
-                extra=extra,
+                extra=extra or {},
                 pool=poolpk,
                 csr=csr,
             )

src/rasenmaeher_api/db/logincodes.py

@@ -26,9 +26,13 @@ class LoginCode(ORMBaseModel, table=True):
     __tablename__ = "logincodes"
 
     code: str = Field(nullable=False, index=True, unique=True)
-    auditmeta: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
-    used_on: datetime.datetime = Field(nullable=True)
-    claims: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
+    auditmeta: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
+    used_on: datetime.datetime | None = Field(default=None, nullable=True)
+    claims: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
 
     @classmethod
     async def use_code(cls, code: str, auditmeta: dict[str, Any] | None = None) -> str:

src/rasenmaeher_api/db/nonces.py

@@ -4,6 +4,7 @@
 from typing import Any
 
 from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.exc import IntegrityError
 from sqlmodel import Field, select
 
 from .base import ORMBaseModel
@@ -19,7 +20,9 @@ class SeenToken(ORMBaseModel, table=True):
     __tablename__ = "seentokens"
 
     token: str = Field(nullable=False, index=True, unique=True)
-    auditmeta: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
+    auditmeta: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
 
     @classmethod
     async def use_token(cls, token: str, auditmeta: dict[str, Any] | None = None) -> None:
@@ -33,10 +36,18 @@ async def use_token(cls, token: str, auditmeta: dict[str, Any] | None = None) ->
             pass
         if not auditmeta:
             auditmeta = {}
-        with EngineWrapper.get_session() as session:
-            dbtoken = SeenToken(token=token, auditmeta=auditmeta)
-            session.add(dbtoken)
-            session.commit()
+        try:
+            with EngineWrapper.get_session() as session:
+                dbtoken = SeenToken(token=token, auditmeta=auditmeta)
+                session.add(dbtoken)
+                session.commit()
+        except IntegrityError as exc:
+            # Another worker can insert the nonce after the initial read.
+            try:
+                await SeenToken.by_token(token)
+            except NotFound:
+                raise exc
+            raise TokenReuse() from exc
 
     @classmethod
     async def by_token(cls, token: str) -> "SeenToken":

src/rasenmaeher_api/db/people.py

@@ -46,8 +46,10 @@ class Person(ORMBaseModel, table=True):
     callsign: str = Field(nullable=False, index=True, unique=True)
     # Directory with the key, cert and pfx
     certspath: str = Field(nullable=False, index=False, unique=True)
-    extra: dict[str, Any] = Field(sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"})
-    revoke_reason: str = Field(nullable=True, index=False)
+    extra: dict[str, Any] = Field(
+        default_factory=dict, sa_type=JSONB, nullable=False, sa_column_kwargs={"server_default": "{}"}
+    )
+    revoke_reason: str | None = Field(default=None, nullable=True, index=False)
     cert_serial: str | None = Field(default=None, nullable=True, index=True, unique=True)
 
     @classmethod
@@ -113,7 +115,7 @@ async def create_with_cert(
             certspath.mkdir(parents=True)
             certspath.chmod(PRIVDIR_MODE)
             try:
-                newperson = Person(pk=puuid, callsign=callsign, certspath=str(certspath), extra=extra)
+                newperson = Person(pk=puuid, callsign=callsign, certspath=str(certspath), extra=extra or {})
                 session.add(newperson)
                 session.commit()
                 if csrpem:
@@ -386,8 +388,10 @@ class Role(SQLModel, table=True):
     __table_args__ = ORMBaseModel.__table_args__
 
     pk: uuid.UUID = Field(primary_key=True, default_factory=uuid.uuid4)
-    created: datetime.datetime = Field(sa_column_kwargs={"default": utcnow}, nullable=False)
-    updated: datetime.datetime = Field(sa_column_kwargs={"default": utcnow, "onupdate": utcnow}, nullable=False)
+    created: datetime.datetime | None = Field(default=None, sa_column_kwargs={"default": utcnow}, nullable=False)
+    updated: datetime.datetime | None = Field(
+        default=None, sa_column_kwargs={"default": utcnow, "onupdate": utcnow}, nullable=False
+    )
     #    user: uuid.UUID = Field(foreign_key="users.pk")
     user: uuid.UUID = Field(foreign_key=f"{ORMBaseModel.__table_args__['schema']}.users.pk")
     role: str = Field(nullable=False, index=True)

src/rasenmaeher_api/mtlsinit.py

@@ -53,8 +53,10 @@ def check_mtls_init() -> bool:
     """Check if we have the cert and key"""
     check_settings_clientpaths()
     config = RMSettings.singleton()
-    assert config.mtls_client_cert_path is not None  # nosec B101
-    assert config.mtls_client_key_path is not None  # nosec B101
+    if config.mtls_client_cert_path is None:
+        raise RuntimeError("mTLS client certificate path is not configured")
+    if config.mtls_client_key_path is None:
+        raise RuntimeError("mTLS client key path is not configured")
     cert_path = Path(config.mtls_client_cert_path)
     key_path = Path(config.mtls_client_key_path)
     LOGGER.debug(f"cert_path={cert_path}  exits={cert_path.exists()}")
@@ -71,8 +73,10 @@ async def mtls_init() -> None:
     )
     check_settings_clientpaths()
     config = RMSettings.singleton()
-    assert config.mtls_client_key_path is not None  # nosec B101
-    assert config.mtls_client_cert_path is not None  # nosec B101
+    if config.mtls_client_key_path is None:
+        raise RuntimeError("mTLS client key path is not configured")
+    if config.mtls_client_cert_path is None:
+        raise RuntimeError("mTLS client certificate path is not configured")
     if (pth := Path(config.mtls_client_key_path)) != privkeypath:
         privkeypath = pth
     certpath = pubkeypath.parent / f"{CERT_NAME_PREFIX}.pem"
@@ -120,8 +124,10 @@ async def get_session_winit() -> aiohttp.ClientSession:
     await mtls_init()
     check_settings_clientpaths()
     config = RMSettings.singleton()
-    assert config.mtls_client_cert_path is not None  # nosec B101
-    assert config.mtls_client_key_path is not None  # nosec B101
+    if config.mtls_client_cert_path is None:
+        raise RuntimeError("mTLS client certificate path is not configured")
+    if config.mtls_client_key_path is None:
+        raise RuntimeError("mTLS client key path is not configured")
     cert_path = Path(config.mtls_client_cert_path)
     key_path = Path(config.mtls_client_key_path)
     return libsession((cert_path, key_path))