G3.2/G3.4: real EPT verification + wire the throttle, reopening agent auth

REOPENS the agent path — but only because possession is now actually proven.

EPT verification (api/app/ept.py): ES256 against Eternitas's published key set
at /.well-known/eternitas-keys. algorithms=['ES256'] makes alg:none and
algorithm confusion unrepresentable rather than merely unlikely; issuer and exp
are enforced by the library; an unknown kid is refused.

Order is deliberate: signature FIRST, trust lookup second. These EPTs live ~365
days and carry rev/tru baked in at issuance, so a year-old 'rev: false' proves
nothing — revocation and band still come from a live lookup on every request.

Found while building it: real EPTs put the passport in . The old code read
/, which no genuine EPT carries — so real agents were
never recognised and ONLY forged tokens ever authenticated. The bypass was not
just a hole, it was the only thing that worked.

Throttle (api/app/throttle.py): BAND_MULTIPLIER and rate_*_per_day were defined
and read by nothing. Now enforced on repo.create and grant.create, counted
against agent_actions (one source of truth, not a private counter that drifts
from the audit log). Fails CLOSED — a limiter that fails open protects you until
the moment something is wrong. Untrusted band is 403 read-only, not 429, because
'slow down' would be a lie.

Tests: 14 behavioral, signing real ES256 tokens with a locally-generated key so
they exercise the crypto path with no network dependency — genuine tokens
accepted, and alg:none / foreign key / tampered payload / expired / wrong issuer
/ unknown kid / missing claims all refused. 74 green.

Co-Authored-By: Claude (Fable 5) <noreply@anthropic.com>

api/app/auth.py

@@ -25,6 +25,7 @@ import httpx
 from fastapi import Header, Request
 
 from api.app.config import Settings
+from api.app.ept import EptInvalid, looks_like_ept, verify_ept
 from api.app.errors import RepairPointer, passport_unresolvable
 
 log = logging.getLogger(__name__)
@@ -159,49 +160,37 @@ async def get_caller(
     token = authorization.split(" ", 1)[1].strip()
 
     # --- agent (Eternitas EPT) --------------------------------------------
-    passport = _unverified_claim(token, "passport") or _unverified_claim(token, "sub_passport")
+    # Possession FIRST, reputation second. The signature proves the caller holds
-    if passport:
+    # this passport; the trust lookup then says what it may do. Doing only the
-        # ⚠️ SECURITY — trust is not authentication.
+    # second was the 2026-08-13 impersonation bypass.
-        #
+    if looks_like_ept(token):
-        # A trust lookup answers "is this passport reputable?". It does NOT
+        try:
-        # answer "does this caller actually hold this passport?". Skipping the
+            verified = verify_ept(token, settings.eternitas_base_url)
-        # second question is an authentication bypass: anyone who knows a
+        except EptInvalid as exc:
-        # passport number (they appear in logs, the lockbox and revocation
-        # messages) could present an UNSIGNED token naming it and be treated as
-        # that agent. Verified live 2026-08-13 — a forged `alg:none` token
-        # returned HTTP 200.
-        #
-        # ES256/JWKS verification of the EPT against Eternitas is not built yet
-        # (the G3.2/G9.1 verifier). Until it is, the agent path FAILS CLOSED in
-        # production — exactly as the human path below already does. This is not
-        # a downgrade of the "agents are citizens" design; it is refusing to
-        # seat a citizen whose ID we cannot yet check. It reopens automatically
-        # the moment `verify_ept_signature` exists and this gate consults it.
-        if settings.is_production and settings.require_verified_jwt:
             raise RepairPointer(
-                status_code=503,
+                status_code=401,
-                code="agent_signin_not_ready",
+                code="ept_invalid",
-                speak="Helper sign-in isn't switched on yet. Nothing you have is affected.",
+                speak="We couldn't confirm that helper's ID, so we didn't let it in.",
-                machine_cause=(
+                machine_cause=f"EPT verification failed: {exc}",
-                    "EPT signature verification (G3.2/G9.1) is not implemented; "
+                remediation_tool="windy_git.reissue_agent_token",
-                    "refusing an unverified agent token in production. A trust "
+            ) from exc
-                    "lookup proves reputation, not possession."
-                ),
-                remediation_tool=None,
-            )
 
-        band, actions = await resolve_passport(settings, passport)
+        # The token is authentic. It is NOT evidence of current standing: these
+        # EPTs live ~365 days and carry `rev`/`tru` baked in at issuance, so a
+        # year-old `rev: false` proves nothing. Revocation and band come from a
+        # live lookup, every time.
+        band, actions = await resolve_passport(settings, verified.passport)
         if band.lower() == "untrusted":
             raise RepairPointer(
                 status_code=403,
                 code="agent_read_only",
                 speak="That helper can look, but it isn't allowed to make changes yet.",
-                machine_cause=f"passport {passport} band=untrusted is read-only",
+                machine_cause=f"passport {verified.passport} band=untrusted is read-only",
                 remediation_tool=None,
             )
         return Caller(
             actor_type=ActorType.agent,
-            passport=passport,
+            passport=verified.passport,
             band=band,
             allowed_actions=actions,
         )

api/app/ept.py

@@ -0,0 +1,140 @@
+"""EPT signature verification (G3.2 / G9.1) — the gate that makes an agent an agent.
+
+Trust is not authentication. A trust lookup answers *"is this passport
+reputable?"*; only a signature answers *"does this caller actually hold it?"*.
+Skipping the second question was a live impersonation bypass on 2026-08-13 — a
+forged `alg:none` token naming a passport read out of the logs returned HTTP 200.
+
+What this module refuses, deliberately and by construction:
+
+* **`alg: none`** — the original exploit. `algorithms=["ES256"]` makes it
+  unrepresentable rather than merely unlikely.
+* **Algorithm confusion.** Only ES256 is accepted. If an attacker presents an
+  HS256 token, PyJWT will not try to use an EC public key as an HMAC secret,
+  which is the classic way "verified" JWTs get forged.
+* **An unknown `kid`.** The key must be one Eternitas currently publishes.
+* **A wrong issuer or an expired token** — checked by the library, not by us.
+
+What it deliberately does NOT decide: whether the agent is *allowed* to act.
+The EPT carries `rev` and `tru` claims baked in at issuance, and these tokens
+live for a year (observed `exp` ≈ 365 days). A year-old `rev: false` is not
+evidence of anything. **Revocation and trust must come from a live lookup**, so
+this module returns only identity and the caller re-checks standing.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+
+import httpx
+import jwt
+from jwt import PyJWKClient
+
+log = logging.getLogger(__name__)
+
+ISSUER = "eternitas.ai"
+ALGORITHMS = ["ES256"]  # exactly one. Never widen this list.
+
+_jwks_client: PyJWKClient | None = None
+_jwks_url: str | None = None
+
+
+class EptInvalid(Exception):
+    """The token is not a valid, currently-signed Eternitas EPT."""
+
+
+@dataclass(frozen=True)
+class VerifiedEpt:
+    passport: str
+    operator: str | None
+    bot_name: str | None
+    issued_at: int | None
+    expires_at: int | None
+
+
+def _client(base_url: str) -> PyJWKClient:
+    """One cached JWKS client. PyJWKClient caches keys and refetches on an
+    unknown kid, so a key rotation heals itself without a redeploy."""
+    global _jwks_client, _jwks_url
+    url = f"{base_url.rstrip('/')}/.well-known/eternitas-keys"
+    if _jwks_client is None or _jwks_url != url:
+        _jwks_client = PyJWKClient(url, cache_keys=True, lifespan=300)
+        _jwks_url = url
+    return _jwks_client
+
+
+def verify_ept(token: str, eternitas_base_url: str) -> VerifiedEpt:
+    """Verify an EPT's signature and claims. Raises EptInvalid on ANY doubt.
+
+    There is no partial success and no "probably fine" path: every failure mode
+    below produces the same refusal, because a caller that cannot prove
+    possession is indistinguishable from an attacker.
+    """
+    try:
+        signing_key = _client(eternitas_base_url).get_signing_key_from_jwt(token)
+    except Exception as exc:  # noqa: BLE001 - unknown kid, unreachable JWKS, malformed
+        raise EptInvalid(f"no usable signing key: {type(exc).__name__}: {exc}") from exc
+
+    try:
+        claims = jwt.decode(
+            token,
+            signing_key.key,
+            algorithms=ALGORITHMS,   # ES256 only — closes alg:none and alg confusion
+            issuer=ISSUER,
+            options={
+                "require": ["sub", "iss", "exp"],
+                "verify_signature": True,
+                "verify_exp": True,
+                "verify_iss": True,
+            },
+        )
+    except jwt.PyJWTError as exc:
+        raise EptInvalid(f"{type(exc).__name__}: {exc}") from exc
+
+    # The REAL claim name. Eternitas puts the passport in `sub`; the previous
+    # code looked for `passport` / `sub_passport`, which no genuine EPT carries —
+    # so real agents were never recognised and only forged tokens ever "worked".
+    passport = claims.get("sub")
+    if not isinstance(passport, str) or not passport.strip():
+        raise EptInvalid("EPT carried no passport in `sub`")
+
+    return VerifiedEpt(
+        passport=passport,
+        operator=claims.get("ope"),
+        bot_name=claims.get("bot"),
+        issued_at=claims.get("iat"),
+        expires_at=claims.get("exp"),
+    )
+
+
+def looks_like_ept(token: str) -> bool:
+    """Cheap, unauthenticated triage: is this token even claiming to be an EPT?
+
+    Used ONLY to route a token to the right verifier. It decides nothing about
+    trust — an attacker controls every byte it reads.
+    """
+    try:
+        header = jwt.get_unverified_header(token)
+    except Exception:  # noqa: BLE001
+        return False
+    return header.get("typ") == "EPT" or header.get("alg") == "ES256"
+
+
+async def eternitas_reachable(base_url: str) -> bool:
+    """Whether the key set can be fetched at all. Used by /health/full so an
+    unreachable JWKS is reported rather than discovered during an outage."""
+    try:
+        async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as c:
+            r = await c.get(
+                f"{base_url.rstrip('/')}/.well-known/eternitas-keys",
+                headers={"User-Agent": "windy-git/1.0"},
+            )
+        return r.status_code == 200 and "keys" in r.json()
+    except Exception:  # noqa: BLE001
+        return False
+
+
+def seconds_until_expiry(ept: VerifiedEpt) -> int | None:
+    return None if ept.expires_at is None else int(ept.expires_at - time.time())

api/app/routes/repos.py

@@ -26,6 +26,7 @@ from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
+from api.app import throttle
 from api.app.auth import ActorType, Caller, get_caller
 from api.app.errors import RepairPointer
 from api.app.models.core import (
@@ -208,6 +209,11 @@ async def create_repo(
             remediation_tool=None,
         )
 
+    # Throttle before any side effect. Checking after would let a rate-limited
+    # agent still create the Gitea repo and only then be told no.
+    async with _sessionmaker(request)() as session:
+        await throttle.enforce(session, settings, caller, "repo.create")
+
     gitea = GiteaClient(settings)
     owner = _owner_login(caller)
     await gitea.ensure_user(owner, f"{owner}@windygit.com")
@@ -234,6 +240,8 @@ async def create_repo(
             ),
         )
         session.add(repo)
+        await session.flush()
+        await throttle.record(session, caller, "repo.create", repo_id=repo.id)
         await session.commit()
         await session.refresh(repo)
 
@@ -337,6 +345,7 @@ async def create_grant(
     """
     settings = request.app.state.settings
     async with _sessionmaker(request)() as session:
+        await throttle.enforce(session, settings, caller, "grant.create")
         repo = await _load_repo(session, repo_id, caller)
         is_owner = (caller.identity_id and repo.identity_id == caller.identity_id) or (
             caller.passport and repo.passport == caller.passport
@@ -364,6 +373,8 @@ async def create_grant(
             expires_at=expires,
         )
         session.add(grant)
+        await session.flush()
+        await throttle.record(session, caller, "grant.create", repo_id=repo.id)
         await session.commit()
         await session.refresh(grant)
         grant_id, role = grant.id, grant.role

api/app/throttle.py

@@ -0,0 +1,169 @@
+"""EI-band velocity limits (G3.4) — throttle by trust, never by omission.
+
+`BAND_MULTIPLIER` and the `rate_*_per_day` settings existed since G0 and were
+**read by nothing**: a documented invariant with no implementation, which is the
+exact failure pattern the ecosystem audits kept finding. This module is the
+missing consumer.
+
+The doctrine (§0.6) is *capability-completeness*: an agent is never denied a
+capability it should have, it is **rate-limited by how much it has proven**.
+Platinum gets 10x, gold 4x, standard 1x, watch 0.5x, and untrusted is read-only.
+
+Counting is done against `agent_actions`, which is already the append-only record
+of every agent write. That is deliberate: a limiter with its own private counter
+disagrees with the audit log the moment either is restarted, and then nobody can
+say what actually happened. One source of truth, queried.
+
+**Fail-closed.** If the count cannot be taken, the action is refused. A limiter
+that fails open is decoration — it protects you right up until the moment
+something is wrong, which is the only moment it matters.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.app.auth import BAND_MULTIPLIER, ActorType, Caller
+from api.app.config import Settings
+from api.app.errors import RepairPointer
+from api.app.models.core import AgentAction
+
+log = logging.getLogger(__name__)
+
+WINDOW = timedelta(days=1)
+
+# action name -> the settings field holding its per-day base for a standard band
+ACTION_BASE: dict[str, str] = {
+    "repo.create": "rate_repo_creates_per_day",
+    "grant.create": "rate_grants_per_day",
+    "push": "rate_pushes_per_day",
+    "push.force": "rate_force_pushes_per_day",
+}
+
+
+def limit_for(settings: Settings, action: str, band: str | None) -> int:
+    """Effective per-day allowance. Unknown bands get the standard rate.
+
+    Unknown-band handling is a real decision, not a default. Eternitas began
+    emitting `unproven` on 2026-07-30 without it appearing in the documented
+    enum. Treating an unrecognised band as untrusted would lock out every freshly
+    hatched agent the day Eternitas adds a name; treating it as platinum would be
+    a hole. Standard-with-no-bonus is the honest middle.
+    """
+    base = getattr(settings, ACTION_BASE[action])
+    mult = BAND_MULTIPLIER.get((band or "").lower(), 1.0)
+    return int(base * mult)
+
+
+async def enforce(
+    session: AsyncSession,
+    settings: Settings,
+    caller: Caller,
+    action: str,
+) -> None:
+    """Raise 429 if this agent has spent its allowance. No-op for humans.
+
+    Humans are governed by account tier and their own session; this is the
+    agent-velocity control specifically (I-6: parity in capability, asymmetry in
+    throttle).
+    """
+    if caller.actor_type != ActorType.agent or not caller.passport:
+        return
+    if action not in ACTION_BASE:  # unknown action = unlimited would be a hole
+        raise RepairPointer(
+            status_code=500,
+            code="throttle_unknown_action",
+            speak="Something went wrong on our side. Nothing was changed.",
+            machine_cause=f"no rate base configured for action {action!r}",
+            remediation_tool=None,
+        )
+
+    band = (caller.band or "").lower()
+
+    # Untrusted is read-only. This is a capability decision, not a rate: it gets
+    # a 403 with a different explanation, because "slow down" would be a lie.
+    if BAND_MULTIPLIER.get(band, 1.0) <= 0:
+        raise RepairPointer(
+            status_code=403,
+            code="agent_read_only",
+            speak="That helper can look, but it isn't allowed to make changes yet.",
+            machine_cause=f"passport {caller.passport} band={band!r} is read-only",
+            remediation_tool=None,
+        )
+
+    allowed = limit_for(settings, action, band)
+    since = datetime.now(UTC) - WINDOW
+
+    try:
+        used = (
+            await session.execute(
+                select(func.count())
+                .select_from(AgentAction)
+                .where(
+                    AgentAction.passport == caller.passport,
+                    AgentAction.action == action,
+                    AgentAction.result == "ok",
+                    AgentAction.ts >= since,
+                )
+            )
+        ).scalar_one()
+    except Exception as exc:  # noqa: BLE001
+        # FAIL CLOSED. A limiter that fails open protects you until the moment
+        # something is wrong, which is the only moment it matters.
+        log.warning("throttle count failed for %s/%s: %s", caller.passport, action, exc)
+        raise RepairPointer(
+            status_code=503,
+            code="throttle_unavailable",
+            speak="We couldn't check that helper's limits, so we didn't make the change.",
+            machine_cause=f"agent_actions count failed: {type(exc).__name__}",
+            remediation_tool=None,
+        ) from exc
+
+    if used >= allowed:
+        raise RepairPointer(
+            status_code=429,
+            code="agent_rate_limited",
+            speak=(
+                "That helper has done a lot in the last day, so we've paused it. "
+                "It'll be able to continue shortly."
+            ),
+            machine_cause=(
+                f"passport {caller.passport} used {used}/{allowed} of {action} "
+                f"in 24h (band={band or 'unknown'})"
+            ),
+            remediation_tool=None,
+            used=used,
+            allowed=allowed,
+            band=band or "unknown",
+        )
+
+
+async def record(
+    session: AsyncSession,
+    caller: Caller,
+    action: str,
+    result: str = "ok",
+    repo_id=None,
+) -> None:
+    """Append the action that was just allowed.
+
+    Written AFTER the work succeeds, on purpose: counting attempts would let a
+    failing agent exhaust its own allowance by retrying, turning a transient
+    error into a lockout.
+    """
+    if caller.actor_type != ActorType.agent or not caller.passport:
+        return
+    session.add(
+        AgentAction(
+            passport=caller.passport,
+            repo_id=repo_id,
+            action=action,
+            ei_at_action=caller.band,
+            result=result,
+        )
+    )
+    await session.flush()

api/tests/test_ept_and_throttle.py

@@ -0,0 +1,170 @@
+"""Behavioral tests for EPT verification and EI throttling.
+
+These sign real ES256 tokens with a locally-generated key and verify against a
+locally-served key set, so they exercise the ACTUAL crypto path with no network
+dependency and no reliance on Eternitas being reachable.
+
+This file exists because the suite it joins was ~86 "does the source contain
+this string" assertions and zero that ran the auth decision — which is how a
+live impersonation bypass passed every test on 2026-08-13.
+"""
+
+from __future__ import annotations
+
+import time
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives.asymmetric import ec
+
+from api.app import ept as ept_mod
+from api.app.auth import BAND_MULTIPLIER
+from api.app.config import Settings
+from api.app.ept import EptInvalid, verify_ept
+from api.app.throttle import ACTION_BASE, limit_for
+
+ISSUER = "eternitas.ai"
+KID = "test-key-1"
+
+
+@pytest.fixture
+def signing(monkeypatch):
+    """A real EC keypair; point the verifier's JWKS lookup at its public half."""
+    key = ec.generate_private_key(ec.SECP256R1())
+
+    class _FakeJWK:
+        def __init__(self, k):
+            self.key = k
+
+    class _FakeClient:
+        def __init__(self, *a, **kw):
+            pass
+
+        def get_signing_key_from_jwt(self, token):
+            header = jwt.get_unverified_header(token)
+            if header.get("kid") != KID:
+                raise Exception(f"unknown kid {header.get('kid')!r}")
+            return _FakeJWK(key.public_key())
+
+    monkeypatch.setattr(ept_mod, "_jwks_client", None)
+    monkeypatch.setattr(ept_mod, "PyJWKClient", _FakeClient)
+    return key
+
+
+def _sign(key, claims, alg="ES256", kid=KID):
+    return jwt.encode(claims, key, algorithm=alg, headers={"kid": kid, "typ": "EPT"})
+
+
+def _claims(**over):
+    c = {
+        "sub": "ET26-TEST-0001",
+        "iss": ISSUER,
+        "iat": int(time.time()) - 10,
+        "exp": int(time.time()) + 3600,
+    }
+    c.update(over)
+    return c
+
+
+# ---- the property that was broken ----------------------------------------
+def test_genuine_ept_is_accepted(signing):
+    v = verify_ept(_sign(signing, _claims()), "https://api.eternitas.ai")
+    assert v.passport == "ET26-TEST-0001"
+
+
+def test_passport_comes_from_sub_not_a_passport_claim(signing):
+    """Real EPTs put the passport in `sub`. The pre-fix code read `passport` /
+    `sub_passport`, which no genuine EPT carries — so real agents were never
+    recognised and only forged tokens ever worked."""
+    tok = _sign(signing, _claims(sub="ET26-REAL-9999", passport="ET26-LIES-0000"))
+    assert verify_ept(tok, "https://api.eternitas.ai").passport == "ET26-REAL-9999"
+
+
+def test_alg_none_is_refused(signing):
+    """The exact 2026-08-13 exploit."""
+    import base64
+    import json as _j
+
+    def seg(d):
+        return base64.urlsafe_b64encode(_j.dumps(d).encode()).rstrip(b"=").decode()
+
+    forged = f"{seg({'alg':'none','typ':'EPT','kid':KID})}.{seg(_claims())}."
+    with pytest.raises(EptInvalid):
+        verify_ept(forged, "https://api.eternitas.ai")
+
+
+def test_signature_from_a_different_key_is_refused(signing):
+    attacker = ec.generate_private_key(ec.SECP256R1())
+    with pytest.raises(EptInvalid):
+        verify_ept(_sign(attacker, _claims()), "https://api.eternitas.ai")
+
+
+def test_tampered_payload_is_refused(signing):
+    tok = _sign(signing, _claims())
+    h, _p, s = tok.split(".")
+    import base64
+    import json as _j
+
+    evil = base64.urlsafe_b64encode(
+        _j.dumps(_claims(sub="ET26-EVIL-0000")).encode()
+    ).rstrip(b"=").decode()
+    with pytest.raises(EptInvalid):
+        verify_ept(f"{h}.{evil}.{s}", "https://api.eternitas.ai")
+
+
+def test_expired_token_is_refused(signing):
+    """Genuinely signed, genuinely expired — proves exp is enforced rather than
+    the token merely failing to parse."""
+    tok = _sign(signing, _claims(exp=int(time.time()) - 5, iat=int(time.time()) - 100))
+    with pytest.raises(EptInvalid):
+        verify_ept(tok, "https://api.eternitas.ai")
+
+
+def test_wrong_issuer_is_refused(signing):
+    """Correctly signed by a trusted key but claiming another issuer."""
+    with pytest.raises(EptInvalid):
+        verify_ept(_sign(signing, _claims(iss="evil.example.com")), "https://api.eternitas.ai")
+
+
+def test_unknown_kid_is_refused(signing):
+    with pytest.raises(EptInvalid):
+        verify_ept(_sign(signing, _claims(), kid="attacker-key"), "https://api.eternitas.ai")
+
+
+def test_missing_required_claims_are_refused(signing):
+    for missing in ("sub", "exp"):
+        c = _claims()
+        c.pop(missing)
+        with pytest.raises(EptInvalid):
+            verify_ept(_sign(signing, c), "https://api.eternitas.ai")
+
+
+def test_only_es256_is_ever_accepted():
+    """Widening this list reopens algorithm confusion."""
+    assert ept_mod.ALGORITHMS == ["ES256"]
+
+
+# ---- the throttle that used to be dead code ------------------------------
+def test_band_multiplier_is_actually_consumed():
+    """BAND_MULTIPLIER was defined and read by nothing before this."""
+    s = Settings()
+    assert limit_for(s, "repo.create", "platinum") == s.rate_repo_creates_per_day * 10
+    assert limit_for(s, "repo.create", "gold") == s.rate_repo_creates_per_day * 4
+    assert limit_for(s, "repo.create", "standard") == s.rate_repo_creates_per_day
+    assert limit_for(s, "repo.create", "watch") == s.rate_repo_creates_per_day // 2
+
+
+def test_unknown_band_gets_standard_not_unlimited_and_not_zero():
+    s = Settings()
+    assert limit_for(s, "repo.create", "a-band-invented-tomorrow") == s.rate_repo_creates_per_day
+    assert limit_for(s, "repo.create", None) == s.rate_repo_creates_per_day
+
+
+def test_untrusted_band_is_read_only():
+    assert BAND_MULTIPLIER["untrusted"] == 0
+
+
+def test_every_throttled_action_has_a_configured_base():
+    s = Settings()
+    for action, field in ACTION_BASE.items():
+        assert getattr(s, field) > 0, f"{action} has no positive base rate"

api/tests/test_invariants.py

@@ -658,23 +658,27 @@ def _fake_request(settings):
 
 @_pytest.mark.asyncio
 async def test_security_forged_agent_token_is_refused_in_production():
-    """A token with alg:none naming a real passport must NOT authenticate.
+    """The 2026-08-13 exploit: an alg:none token naming a real passport returned
-    This is the exploit that returned HTTP 200 on 2026-08-13, exercised through
+    HTTP 200 as that agent. It must now be refused whichever gate catches it —
-    the real get_caller decision rather than by grepping for a string."""
+    an EPT-shaped forgery by signature verification, a JWT-shaped one by the
+    human gate. What is asserted is REFUSAL, not a particular error code."""
     from api.app.auth import get_caller
     from api.app.config import Settings
     from api.app.errors import RepairPointer
 
     settings = Settings(environment="production", require_verified_jwt=True,
-                        eternitas_platform_api_key="x", eternitas_base_url="https://api.eternitas.ai")
+                        eternitas_platform_api_key="x")
     req = _fake_request(settings)
 
+    for typ, expected in (("JWT", "human_signin_not_ready"), ("EPT", "ept_invalid")):
+        def seg(d):
+            return _b64.urlsafe_b64encode(_json.dumps(d).encode()).rstrip(b"=").decode()
+        forged = (f"{seg({'alg':'none','typ':typ})}"
+                  f".{seg({'passport':'ET26-1EF9-VJAN','sub':'ET26-1EF9-VJAN'})}.sig")
         with _pytest.raises(RepairPointer) as exc:
-        await get_caller(req, authorization=f"Bearer {_forged_bearer('ET26-1EF9-VJAN')}",
+            await get_caller(req, authorization=f"Bearer {forged}", x_service_token=None)
-                         x_service_token=None)
+        assert exc.value.status_code in (401, 403, 503), f"{typ} was not refused"
-    # Must be refused, and must be refused BEFORE any trust lookup could seat it.
+        assert exc.value.code == expected, f"{typ} -> {exc.value.code}"
-    assert exc.value.status_code in (401, 503)
-    assert exc.value.code == "agent_signin_not_ready"
 
 
 @_pytest.mark.asyncio

pyproject.toml

@@ -25,6 +25,9 @@ dependencies = [
     "alembic>=1.14",
     "httpx>=0.27",
     "boto3>=1.35",
+    # ES256 verification of Eternitas EPTs (G3.2/G9.1). Without crypto extras
+    # PyJWT cannot verify EC signatures and silently offers no protection.
+    "pyjwt[crypto]>=2.9",
 ]
 
 [project.optional-dependencies]