Harden Sigil: sigil/identity.py

sigil/identity.py

@@ -14,6 +14,7 @@ After step 2 the secret is gone. It is a bootstrap, not a credential.
 from __future__ import annotations
 
 import hashlib
+import re
 import secrets
 import time
 from typing import Optional
@@ -24,6 +25,22 @@ from .store import Store
 DEFAULT_NONCE_TTL = 120  # 2 minutes. Short: a nonce is a one-shot question.
 
 
+_AGENT_ID = re.compile(r"^[a-z0-9][a-z0-9_.-]{0,63}$")
+
+
+def _validate_agent_id(agent: str) -> None:
+    """Agent ids appear in log lines, URLs and filenames.
+
+    Constrain them to a safe alphabet and length rather than trusting callers,
+    so an id can never carry a newline into the audit trail or a path segment
+    into a file.
+    """
+    if not _AGENT_ID.match(agent or ""):
+        raise errors.SigilError(
+            f"invalid agent id {agent!r}: expected lowercase [a-z0-9._-], max 64 chars"
+        )
+
+
 def _hash_secret(secret: str) -> str:
     return hashlib.sha256(secret.encode("utf-8")).hexdigest()
 
@@ -34,6 +51,7 @@ def seed_bootstrap(store: Store, agent: str, secret: Optional[str] = None) -> st
     This is the only moment a shared secret exists. In production this replaces
     the agent's static passcode, issued once and never transmitted again.
     """
+    _validate_agent_id(agent)
     secret = secret or secrets.token_urlsafe(24)
     store.put_bootstrap(_hash_secret(secret), agent)
     store.record(
@@ -49,6 +67,9 @@ def enrol(
 
     Returns the key fingerprint, which is an identifier and safe to log.
     """
+    _validate_agent_id(agent)
+    if not store.allow(f"enrol:{agent}"):
+        raise errors.RateLimited(f"too many enrolment attempts for {agent}")
     if store.get_agent(agent) is not None:
         raise errors.EnrolmentFailed(f"{agent} is already enrolled")
     owner = store.spend_bootstrap(_hash_secret(secret))
@@ -68,8 +89,12 @@ def enrol(
     return fingerprint
 
 
-def challenge(store: Store, agent: str, ttl: int = DEFAULT_NONCE_TTL) -> dict:
+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."""
+    if not store.allow(f"challenge:{agent}", limit=limit, window=window):
+        raise errors.RateLimited(f"too many challenges for {agent}")
+
     row = store.get_agent(agent)
     if row is None:
         raise errors.UnknownAgent(f"{agent} is not enrolled")
@@ -96,6 +121,8 @@ def verify(
     nonce: str,
     signature_b64: str,
     ttl: int = sessions.DEFAULT_AGENT_TTL,
+    limit: Optional[int] = None,
+    window: Optional[int] = None,
 ) -> dict:
     """Check a signed nonce and, if it holds, mint an agent session.
 
@@ -108,12 +135,18 @@ def verify(
     if row is None:
         raise errors.NonceUnknown("no such challenge")
 
+    if not store.allow(f"verify:{row['agent']}", limit=limit, window=window):
+        raise errors.RateLimited("too many verification attempts")
+
     agent = row["agent"]
 
-    if row["used_at"] is not None:
+    # Claim the nonce before judging anything. Compare-and-set, so two
+    # concurrent verifies cannot both proceed, and a wrong signature still
+    # burns the nonce rather than leaving it open for another attempt.
+    if not store.claim_nonce(nonce):
         store.record(
             "auth.replay_refused", actor=agent, subject=agent,
-            detail={"reason": "nonce already used"},
+            detail={"reason": "nonce already used or claimed"},
         )
         raise errors.NonceUsed("challenge already answered")
 
@@ -126,10 +159,6 @@ def verify(
     if agent_row["revoked_at"] is not None:
         raise errors.UnknownAgent(f"{agent} is revoked")
 
-    # Consume before judging the signature. A wrong signature must not leave
-    # the nonce live for another attempt.
-    store.mark_nonce_used(nonce)
-
     public = keys.from_public_bytes(agent_row["pubkey"])
     try:
         signature = base64.b64decode(signature_b64.encode("ascii"))
@@ -155,20 +184,6 @@ def verify(
     )
 
 
-def rotate(store: Store, agent: str) -> str:
-    """Bind a new public key, keeping the same agent identity.
-
-    Used for key compromise or routine rotation. The old key stops working the
-    moment the row is replaced.
-    """
-    row = store.get_agent(agent)
-    if row is None:
-        raise errors.UnknownAgent(f"{agent} is not enrolled")
-    raise errors.EnrolmentFailed(
-        "rotate() requires a new public key: call enrol_rekey(agent, public_b64)"
-    )
-
-
 def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     """Re-bind a public key using a fresh one-time secret.
 
@@ -185,11 +200,14 @@ def enrol_rekey(store: Store, agent: str, public_b64: str, secret: str) -> str:
     fingerprint = keys.fingerprint(public)
     previous = row["fingerprint"]
     store.put_agent(agent, keys.public_bytes(public), fingerprint)
+    # A new key must not leave old sessions alive, or rotating after a
+    # compromise buys nothing.
+    killed = store.revoke_sessions_for_agent(agent, "key rotated")
     store.record(
         "identity.rekeyed",
         actor=agent,
         subject=agent,
         fingerprint=fingerprint,
-        detail={"previous_fingerprint": previous},
+        detail={"previous_fingerprint": previous, "sessions_revoked": killed},
     )
     return fingerprint