feat(identity): offline Ed25519 verifier — issuer-free identity for the auth seam

infra/govern/ed25519_auth.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""infra/govern/ed25519_auth.py — offline, issuer-free identity for the auth seam.
+
+A verifier for `principals.register_verifier` that authenticates a PERSON (or an agent) by an Ed25519
+signature instead of a bearer secret. It exists because the fleet is tailnet-scoped and deliberately
+dependency-free: an OIDC/OAuth verifier can check a token offline, but *obtaining* one needs the internet
+and the IdP up, and shortening token lifetimes to tighten revocation makes that dependency worse. This has
+neither problem — no issuer, no JWKS, no network, in either direction.
+
+THE SHAPE. The bearer is a self-contained, short-lived ASSERTION: a DSSE envelope (the same
+`infra.cwp.sign` surface that signs grants and exod results — no new crypto) over
+
+    {"pub": "<hex raw ed25519 public key>", "iat": <unix>, "exp": <unix>, "nonce": "<random>"}
+
+base64url-encoded so it fits an `Authorization: Bearer` header. Verification is self-contained: the
+assertion carries the public key, the signature is checked AGAINST THAT KEY, and the verifier returns
+`sign.keyid(pub)` — `"ed25519:<16 hex>"` — as the SUBJECT.
+
+That is not circular, and the distinction matters: anyone can mint a well-formed assertion with a key they
+generated, and it will verify. What they cannot do is make it resolve to a principal — `resolve_principal`
+maps a subject to a principal only if some entry in the registry DECLARES that exact subject:
+
+    "alice": {"subject": "ed25519:9f2c…", "acl": {...}}
+
+So the signature proves possession of a key, and the mounted registry decides whether that key is anybody.
+Revocation is deleting the line — instant, offline, on a file already bind-mounted read-only into every node,
+with no expiry window to wait out and no issuer to consult.
+
+REPLAY. An assertion is bearer-shaped: whoever holds it can present it until `exp`. Two bounded defences —
+a short TTL, and a nonce cache that refuses any nonce seen twice inside its own validity window. The cache is
+per-process and self-pruning; it does not need to persist, because an entry can only be replayed while the
+assertion is still valid, and a restart shortens rather than extends that window.
+
+CLOCK SKEW is a fail-closed input, not an afterthought: an assertion from the future is refused beyond
+`_SKEW`, so a client with a wildly wrong clock cannot mint a long-lived credential by post-dating `exp`.
+"""
+from __future__ import annotations
+
+import base64
+import json
+import secrets
+import time
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
+
+from infra.cwp import sign as _sign
+
+PAYLOAD_TYPE = "application/cwp-auth+json"
+
+_SKEW = 60                 # seconds of tolerated clock skew, both directions
+_MAX_TTL = 15 * 60         # ceiling on an assertion's own claimed lifetime — a client cannot mint a long one
+_DEFAULT_TTL = 300
+_NONCE_CACHE_MAX = 4096    # bounded; entries self-expire at their assertion's exp
+
+
+def _b64u(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).decode().rstrip("=")
+
+
+def _unb64u(s: str) -> bytes:
+    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
+
+
+def mint_assertion(private_key: Ed25519PrivateKey, *, ttl: int = _DEFAULT_TTL, now: int = None) -> str:
+    """Produce a bearer assertion for `private_key`. Client-side helper — govd never calls this."""
+    now = int(now if now is not None else time.time())
+    ttl = max(1, min(int(ttl), _MAX_TTL))
+    body = {"pub": _sign.public_raw(private_key).hex(), "iat": now, "exp": now + ttl,
+            "nonce": secrets.token_urlsafe(12)}
+    env = _sign.sign(body, private_key, payload_type=PAYLOAD_TYPE)
+    return _b64u(json.dumps(env, separators=(",", ":"), sort_keys=True).encode())
+
+
+class Verifier:
+    """A `bearer -> subject|None` callable for principals.register_verifier.
+
+    EVERY failure path returns None. It never raises and never distinguishes *why* to the caller: a caller
+    that could tell "bad signature" from "expired" from "replayed" would leak an oracle, and govd's answer
+    is 401 either way.
+    """
+
+    def __init__(self, *, skew: int = _SKEW, max_ttl: int = _MAX_TTL, cache_max: int = _NONCE_CACHE_MAX):
+        self.skew, self.max_ttl, self.cache_max = int(skew), int(max_ttl), int(cache_max)
+        self._seen: dict = {}                                    # nonce -> exp
+
+    def _replayed(self, nonce: str, exp: int, now: int) -> bool:
+        """True iff this nonce was already spent inside its own validity window. Prunes as it goes."""
+        if len(self._seen) >= self.cache_max:                    # prune expired first; only then refuse
+            for k, e in [(k, e) for k, e in self._seen.items() if e <= now]:
+                self._seen.pop(k, None)
+            if len(self._seen) >= self.cache_max:
+                return True                                      # cache full of LIVE nonces -> fail CLOSED
+        if self._seen.get(nonce, 0) > now:
+            return True
+        self._seen[nonce] = exp
+        return False
+
+    def __call__(self, bearer: str, *, now: int = None):
+        now = int(now if now is not None else time.time())
+        try:
+            env = json.loads(_unb64u(str(bearer)).decode())
+            if not isinstance(env, dict):
+                return None
+            body = json.loads(_unb64u(env["payload"]).decode()) if isinstance(env.get("payload"), str) \
+                else env.get("payload")
+            if not isinstance(body, dict):
+                return None
+            pub_hex, iat, exp = body.get("pub"), int(body.get("iat", 0)), int(body.get("exp", 0))
+            nonce = str(body.get("nonce") or "")
+            if not pub_hex or not nonce:
+                return None
+            pub_raw = bytes.fromhex(str(pub_hex))
+            if len(pub_raw) != 32:                               # not an ed25519 public key
+                return None
+            if exp <= now or exp - iat > self.max_ttl:           # expired, or claims a lifetime past the ceiling
+                return None
+            if iat > now + self.skew:                            # minted in the future -> refuse
+                return None
+            if not _sign.verify(env, Ed25519PublicKey.from_public_bytes(pub_raw)):
+                return None
+            if self._replayed(nonce, exp, now):
+                return None
+            return _sign.keyid(pub_raw)                          # "ed25519:<16 hex>" == the declared `subject`
+        except Exception:
+            return None                                          # malformed / undecodable / anything at all
+
+
+def install(name: str = "ed25519", **kw) -> Verifier:
+    """Register a verifier under `name` and return it (so a caller can inspect or reset it in tests)."""
+    from infra.govern import principals
+    v = Verifier(**kw)
+    principals.register_verifier(name, v)
+    return v
+
+
+def subject_for(public_raw: bytes) -> str:
+    """The `subject` string to put in principals.json for this public key."""
+    return _sign.keyid(public_raw)

tests/test_ed25519_auth.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Tests for infra/govern/ed25519_auth.py — the offline, issuer-free identity verifier.
+
+Every case here is an IDENTITY decision, so each asserts a fail-CLOSED outcome explicitly. The property
+that matters most is the last one: a cryptographically VALID assertion from a key nobody declared must
+resolve to nobody. Possession of a key proves possession of a key; the mounted registry decides whether
+that key is anyone.
+"""
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from infra.govern import ed25519_auth as EA
+from infra.govern import principals as P
+
+
+@pytest.fixture()
+def key():
+    return Ed25519PrivateKey.generate()
+
+
+@pytest.fixture()
+def verifier():
+    return EA.Verifier()
+
+
+def _reg(key, name="alice"):
+    return {name: {"subject": EA.subject_for(EA._sign.public_raw(key)), "acl": {}},
+            "bot": {"token_sha": P.token_sha("s3cret")}}
+
+
+# ───────────────────────── the happy path ─────────────────────────
+
+def test_valid_assertion_resolves_to_the_declaring_principal(key):
+    EA.install("t_ed", )
+    assert P.resolve_principal(EA.mint_assertion(key), _reg(key), "t_ed") == "alice"
+
+
+def test_subject_is_the_stable_keyid(key):
+    pub = EA._sign.public_raw(key)
+    assert EA.subject_for(pub) == EA._sign.keyid(pub)
+    assert EA.subject_for(pub).startswith("ed25519:")
+
+
+# ───────────────────────── fail-closed ─────────────────────────
+
+def test_valid_signature_from_an_UNDECLARED_key_is_nobody(key, verifier):
+    """THE property. A stranger can mint a perfectly valid assertion with their own key — and it must
+    resolve to no principal, because the registry never declared that subject."""
+    stranger = Ed25519PrivateKey.generate()
+    a = EA.mint_assertion(stranger)
+    # the assertion verifies on its own terms — the signature IS valid
+    assert verifier(a) == EA.subject_for(EA._sign.public_raw(stranger))
+    # …and resolves to NO principal, because the registry declares only alice's key
+    EA.install("t_ed2")
+    assert P.resolve_principal(a, _reg(key), "t_ed2") is None
+
+
+def test_undeclared_key_resolves_to_nobody_through_the_seam(key):
+    EA.install("t_ed3")
+    stranger = Ed25519PrivateKey.generate()
+    assert P.resolve_principal(EA.mint_assertion(stranger), _reg(key), "t_ed3") is None
+
+
+@pytest.mark.parametrize("bearer", ["", "garbage", "!!!not-base64!!!", "e30", "null"])
+def test_malformed_bearers_are_refused(verifier, bearer):
+    assert verifier(bearer) is None
+
+
+def test_expired_assertion_refused(key, verifier):
+    assert verifier(EA.mint_assertion(key, ttl=1, now=int(time.time()) - 100)) is None
+
+
+def test_future_dated_assertion_refused(key, verifier):
+    # a client with a wrong (or lying) clock must not be able to post-date its way to a long credential
+    assert verifier(EA.mint_assertion(key, now=int(time.time()) + 9999)) is None
+
+
+def test_minter_clamps_ttl_to_the_ceiling(key, verifier):
+    a = EA.mint_assertion(key, ttl=99999)
+    env = json.loads(EA._unb64u(a).decode())
+    body = json.loads(EA._unb64u(env["payload"]).decode())
+    assert body["exp"] - body["iat"] <= EA._MAX_TTL
+    assert verifier(a) is not None
+
+
+def test_oversized_ttl_in_a_HAND_ROLLED_assertion_refused(key, verifier):
+    """The minter clamps, but the verifier must not TRUST the minter — a hand-rolled assertion claiming a
+    year-long lifetime is refused on its own terms."""
+    now = int(time.time())
+    body = {"pub": EA._sign.public_raw(key).hex(), "iat": now, "exp": now + 365 * 24 * 3600, "nonce": "x1"}
+    env = EA._sign.sign(body, key, payload_type=EA.PAYLOAD_TYPE)
+    assert verifier(EA._b64u(json.dumps(env, separators=(",", ":"), sort_keys=True).encode())) is None
+
+
+def test_tampered_payload_refused(key, verifier):
+    """Swapping the embedded pubkey invalidates the signature — the assertion is self-verifying."""
+    other = Ed25519PrivateKey.generate()
+    env = json.loads(EA._unb64u(EA.mint_assertion(key)).decode())
+    body = json.loads(EA._unb64u(env["payload"]).decode())
+    body["pub"] = EA._sign.public_raw(other).hex()
+    env["payload"] = EA._b64u(json.dumps(body, separators=(",", ":"), sort_keys=True).encode())
+    assert verifier(EA._b64u(json.dumps(env, separators=(",", ":"), sort_keys=True).encode())) is None
+
+
+def test_non_ed25519_pubkey_length_refused(key, verifier):
+    now = int(time.time())
+    body = {"pub": "aa" * 16, "iat": now, "exp": now + 60, "nonce": "n"}   # 16 bytes, not 32
+    env = EA._sign.sign(body, key, payload_type=EA.PAYLOAD_TYPE)
+    assert verifier(EA._b64u(json.dumps(env, separators=(",", ":"), sort_keys=True).encode())) is None
+
+
+# ───────────────────────── replay ─────────────────────────
+
+def test_replay_within_the_validity_window_refused(key, verifier):
+    a = EA.mint_assertion(key)
+    assert verifier(a) is not None
+    assert verifier(a) is None, "a spent nonce must not be reusable while still unexpired"
+
+
+def test_distinct_assertions_both_accepted(key, verifier):
+    assert verifier(EA.mint_assertion(key)) is not None
+    assert verifier(EA.mint_assertion(key)) is not None   # different nonce
+
+
+def test_nonce_cache_prunes_expired_and_stays_bounded(key):
+    v = EA.Verifier(cache_max=8)
+    now = int(time.time())
+    for _ in range(20):                                    # far more than cache_max, all short-lived
+        assert v(EA.mint_assertion(key, ttl=1, now=now), now=now) is not None
+        now += 2                                           # each expires before the next
+    assert len(v._seen) <= 8
+
+
+def test_cache_full_of_LIVE_nonces_fails_closed(key):
+    v = EA.Verifier(cache_max=2)
+    now = int(time.time())
+    assert v(EA.mint_assertion(key, ttl=600, now=now), now=now) is not None
+    assert v(EA.mint_assertion(key, ttl=600, now=now), now=now) is not None
+    # cache is full and nothing is prunable — refuse rather than evict a live nonce and permit a replay
+    assert v(EA.mint_assertion(key, ttl=600, now=now), now=now) is None
+
+
+# ───────────────────────── the seam contract ─────────────────────────
+
+def test_verifier_never_raises(verifier):
+    for junk in (None, 123, b"bytes", {"a": 1}, "\x00\xff"):
+        assert verifier(junk) is None
+
+
+def test_bearer_secret_does_not_authenticate_under_this_scheme(key):
+    EA.install("t_ed4")
+    assert P.resolve_principal("s3cret", _reg(key), "t_ed4") is None