SECURITY: enforce passport status — revocation now takes effect live
All checks were successful
check / gate (push) Successful in 21s
All checks were successful
check / gate (push) Successful in 21s
A revoked passport returns HTTP 200, status=revoked, band=unproven, allowed_actions=[] (verified live 2026-08-13). resolve_passport keyed refusal only on HTTP 4xx and band=="untrusted", so it returned band 'unproven' and the agent was seated. Revocation was NOT enforced on the live auth path at all — and now that agent auth actually works, a revoked agent could authenticate and act. Extracts decide_trust(body) -> (band, actions) | raise. Only status=="active" is allowed; revoked/suspended/frozen/unknown all refuse, fail-closed on the field that carries the most consequential fact about an identity. The agent call site turns that into a clean 403 passport_revoked. This is the REAL revocation gate — the token cannot be un-issued, but its standing is re-checked on every request, so revocation takes effect on the next call with no webhook required. The Eternitas webhook remains useful for invalidating locally-issued credentials/grants (G6.3, not built yet), but it was never the primary gate and its being unwired is no longer a live exposure. Behavioral tests: revoked body refused, active accepted, unknown/missing status fails closed. Co-Authored-By: Claude (Fable 5) <noreply@anthropic.com>
This commit is contained in:
@@ -73,6 +73,36 @@ BAND_MULTIPLIER: dict[str, float] = {
|
||||
}
|
||||
|
||||
|
||||
class PassportNotInGoodStanding(Exception):
|
||||
"""The passport resolved, but Eternitas does not list it as active
|
||||
(revoked / suspended / frozen / unknown status)."""
|
||||
|
||||
def __init__(self, passport: str, status: str) -> None:
|
||||
self.passport = passport
|
||||
self.status = status
|
||||
super().__init__(f"{passport} status={status!r}")
|
||||
|
||||
|
||||
def decide_trust(body: dict, passport: str) -> tuple[str, tuple[str, ...]]:
|
||||
"""The trust body -> (band, allowed_actions), or refuse.
|
||||
|
||||
THE GATE THAT WAS MISSING. A revoked passport returns HTTP 200 with
|
||||
`status: revoked`, `band: unproven`, `allowed_actions: []` — verified live
|
||||
2026-08-13. The previous code keyed refusal only on HTTP 4xx and on
|
||||
band=="untrusted", so a revoked agent (200, band unproven) authenticated and
|
||||
acted normally. Revocation was not enforced on the live path at all; the
|
||||
webhook that was supposed to be the backup was never the primary gate.
|
||||
|
||||
Only `status == "active"` is allowed. Anything else — including a status
|
||||
Eternitas invents tomorrow — refuses. Fail-closed on the field that carries
|
||||
the most consequential fact about an identity.
|
||||
"""
|
||||
status = str(body.get("status", "")).lower()
|
||||
if status != "active":
|
||||
raise PassportNotInGoodStanding(passport, status or "missing")
|
||||
return body.get("band", "unproven"), tuple(body.get("allowed_actions", ()))
|
||||
|
||||
|
||||
async def resolve_passport(settings: Settings, passport: str) -> tuple[str, tuple[str, ...]]:
|
||||
"""G3.6 — THE STATUS-CODE LAW.
|
||||
|
||||
@@ -106,8 +136,9 @@ async def resolve_passport(settings: Settings, passport: str) -> tuple[str, tupl
|
||||
continue
|
||||
last_status = r.status_code
|
||||
if r.status_code == 200:
|
||||
body = r.json()
|
||||
return body.get("band", "unproven"), tuple(body.get("allowed_actions", []))
|
||||
# decide_trust raises PassportNotInGoodStanding on a non-active
|
||||
# status; that propagates past the retry loop as a hard refusal.
|
||||
return decide_trust(r.json(), passport)
|
||||
if r.status_code in (400, 404):
|
||||
# Malformed or not-issued. Refuse immediately — retrying cannot help
|
||||
# and pretending it might is how a soft-allow gets written.
|
||||
@@ -179,7 +210,20 @@ async def get_caller(
|
||||
# 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)
|
||||
try:
|
||||
band, actions = await resolve_passport(settings, verified.passport)
|
||||
except PassportNotInGoodStanding as exc:
|
||||
# The signature is authentic, but the identity is no longer good.
|
||||
# Revocation takes effect here, live, on the next request — no
|
||||
# webhook required. That is the honest place for it: the token can't
|
||||
# be un-issued, but its standing is checked every time.
|
||||
raise RepairPointer(
|
||||
status_code=403,
|
||||
code="passport_revoked",
|
||||
speak="That helper's access has been turned off.",
|
||||
machine_cause=f"eternitas status for {verified.passport} is {exc.status!r}, not active",
|
||||
remediation_tool=None,
|
||||
) from exc
|
||||
if band.lower() == "untrusted":
|
||||
raise RepairPointer(
|
||||
status_code=403,
|
||||
|
||||
@@ -168,3 +168,32 @@ 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"
|
||||
|
||||
|
||||
# ---- revocation enforced on the live trust path (not just the webhook) ----
|
||||
def test_revoked_passport_is_refused_by_trust_decision():
|
||||
"""A revoked passport returns HTTP 200, status=revoked, band=unproven,
|
||||
allowed=[] (verified live 2026-08-13). The decision must refuse it — the
|
||||
old code returned band 'unproven' and seated the agent."""
|
||||
from api.app.auth import PassportNotInGoodStanding, decide_trust
|
||||
|
||||
revoked = {"status": "revoked", "band": "unproven", "allowed_actions": []}
|
||||
with pytest.raises(PassportNotInGoodStanding):
|
||||
decide_trust(revoked, "ET26-NJQT-QMR0")
|
||||
|
||||
|
||||
def test_active_passport_is_accepted_by_trust_decision():
|
||||
from api.app.auth import decide_trust
|
||||
|
||||
active = {"status": "active", "band": "gold", "allowed_actions": ["read", "send"]}
|
||||
band, actions = decide_trust(active, "ET26-1EF9-VJAN")
|
||||
assert band == "gold" and actions == ("read", "send")
|
||||
|
||||
|
||||
def test_unknown_or_missing_status_fails_closed():
|
||||
from api.app.auth import PassportNotInGoodStanding, decide_trust
|
||||
|
||||
for body in ({"band": "gold"}, {"status": "suspended"}, {"status": "frozen"},
|
||||
{"status": ""}, {}):
|
||||
with pytest.raises(PassportNotInGoodStanding):
|
||||
decide_trust(body, "ET26-X")
|
||||
|
||||
Reference in New Issue
Block a user