Close the second round of review findings: revoke, reinstate, expiry

sigil/identity.py

@@ -89,6 +89,58 @@ def enrol(
     return fingerprint
 
 
+def is_revoked(row, now: Optional[float] = None) -> bool:
+    """Is this agent row currently revoked?
+
+    A revocation with an end time has lapsed and the agent is back in force.
+    Anything still valid reads as revoked, so a malformed row fails closed.
+    """
+    if row["revoked_at"] is None:
+        return False
+    until = row["revoked_until"] if "revoked_until" in row.keys() else None
+    if until is None:
+        return True
+    return (now if now is not None else time.time()) < until
+
+
+def revoke(store: Store, agent: str, until: Optional[int] = None,
+           reason: Optional[str] = None, actor: str = "system") -> int:
+    """Revoke an agent, optionally until a time.
+
+    `until=None` is permanent. A timestamp suspends the identity until then,
+    which is a real distinction: an incident response that ends should not need
+    a re-enrolment, and a permanent death should not be undone by a clock.
+
+    Revocation is an administrative act, so it is recorded with `actor` rather
+    than pretending an agent revoked itself.
+    """
+    _validate_agent_id(agent)
+    if store.get_agent(agent) is None:
+        raise errors.UnknownAgent(f"{agent} is not enrolled")
+    killed = store.revoke_agent(agent, until, reason)
+    store.record(
+        "identity.revoked",
+        actor=actor,
+        subject=agent,
+        detail={
+            "until": until,
+            "permanent": until is None,
+            "reason": reason,
+            "sessions_revoked": killed,
+        },
+    )
+    return killed
+
+
+def reinstate(store: Store, agent: str, actor: str = "system") -> None:
+    """Lift a revocation. Audited, because it undoes a security decision."""
+    _validate_agent_id(agent)
+    if store.get_agent(agent) is None:
+        raise errors.UnknownAgent(f"{agent} is not enrolled")
+    store.reinstate(agent)
+    store.record("identity.reinstated", actor=actor, subject=agent)
+
+
 def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
               limit: Optional[int] = None, window: Optional[int] = None) -> dict:
     """Issue a single-use nonce for an agent to sign."""
@@ -98,7 +150,7 @@ def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL,
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
-    if row["revoked_at"] is not None:
+    if is_revoked(row):
         raise errors.UnknownAgent(f"{agent} is revoked")
     nonce = secrets.token_urlsafe(24)
     expires = int(time.time()) + ttl
@@ -156,7 +208,7 @@ def verify(
     agent_row = store.get_agent(agent)
     if agent_row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
-    if agent_row["revoked_at"] is not None:
+    if is_revoked(agent_row):
         raise errors.UnknownAgent(f"{agent} is revoked")
 
     public = keys.from_public_bytes(agent_row["pubkey"])