diff --git a/infra/govern/ed25519_auth.py b/infra/govern/ed25519_auth.py new file mode 100644 index 0000000..4fd885e --- /dev/null +++ b/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": "", "iat": , "exp": , "nonce": ""} + +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) diff --git a/tests/test_ed25519_auth.py b/tests/test_ed25519_auth.py new file mode 100644 index 0000000..ddeebfb --- /dev/null +++ b/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