A2A human principals: accept a controller-signed human identity on send (unified-chat slice 3)

taosmd/config.py

@@ -50,6 +50,10 @@
 _GENERATOR_PROFILE_KEY = "generator_profile"
 # Whether A2A registry auth runs in enforce mode (True) or verify-and-warn mode (False).
 _A2A_AUTH_ENFORCE_KEY = "a2a_auth_enforce"
+# Canonical IDs of human principals (controller sessions). These IDs skip the
+# registry revocation check and the grants check; a sub/from mismatch on a
+# human token is always rejected, even in verify-and-warn mode.
+_HUMAN_PRINCIPAL_IDS_KEY = "human_principal_ids"
 # Section under which collections settings live. ``allowed_roots`` is the
 # safety line of the collections contract: source paths must resolve inside
 # one of these directories. Empty (the default) means collections are off.
@@ -602,6 +606,60 @@ def set_a2a_auth_enforce(value: bool, data_dir=None) -> None:
     _write(data, data_dir)
 
 
+# ---------------------------------------------------------------------------
+# Human principal IDs (controller sessions)
+# ---------------------------------------------------------------------------
+
+def get_human_principal_ids(data_dir=None) -> list[str]:
+    """Return the configured human principal IDs, or [] if unset.
+
+    Resolution order (first non-empty wins):
+
+    1. ``TAOSMD_HUMAN_PRINCIPAL_IDS`` environment variable (comma-separated)
+    2. ``human_principal_ids`` list in ``~/.taosmd/config.json``
+
+    These IDs belong to human principals (controller sessions). They skip the
+    registry revocation check and the grants check; a sub/from mismatch on a
+    human token is always rejected, even in verify-and-warn mode.
+    """
+    env = os.environ.get("TAOSMD_HUMAN_PRINCIPAL_IDS")
+    if env and env.strip():
+        return [p.strip() for p in env.split(",") if p.strip()]
+    ids = _read(data_dir).get(_HUMAN_PRINCIPAL_IDS_KEY)
+    if isinstance(ids, list):
+        return [str(i) for i in ids if isinstance(i, str) and str(i).strip()]
+    return []
+
+
+def set_human_principal_ids(ids, clear: bool = False, data_dir=None) -> None:
+    """Persist the human principal IDs list (or clear it).
+
+    Args:
+        ids: List of human principal canonical ID strings. Ignored when
+            ``clear`` is True.
+        clear: when True, remove the setting.
+
+    Raises:
+        ValueError: when ``clear`` is False and ``ids`` is not a list of
+            non-empty strings.
+    """
+    data = _read(data_dir)
+    if clear:
+        data.pop(_HUMAN_PRINCIPAL_IDS_KEY, None)
+    else:
+        if not isinstance(ids, list):
+            raise ValueError("ids must be a list of strings (or pass clear=True)")
+        cleaned = []
+        for i in ids:
+            if not isinstance(i, str):
+                raise ValueError(f"human principal id must be a string, got {type(i).__name__}")
+            s = i.strip()
+            if s:
+                cleaned.append(s)
+        data[_HUMAN_PRINCIPAL_IDS_KEY] = cleaned
+    _write(data, data_dir)
+
+
 # ---------------------------------------------------------------------------
 # Collections: allowed roots
 # ---------------------------------------------------------------------------
@@ -689,6 +747,8 @@ def set_collections_allowed_roots(roots, clear: bool = False, data_dir=None) ->
     "set_serve_dashboard",
     "get_a2a_auth_enforce",
     "set_a2a_auth_enforce",
+    "get_human_principal_ids",
+    "set_human_principal_ids",
     "MANAGED_BY_STANDALONE",
     "MANAGED_BY_TAOS",
     "get_generator_profile",

taosmd/http_server.py

@@ -617,10 +617,12 @@ def _make_handler(data_dir, runner: _ServiceLoop, verifier=None,
             # The revoked and grants feeds are admin-gated (#710/#719): send the
             # configured taOS local token on them; pin the issuer.
             _registry_admin_token = _config.get_registry_token(data_dir)
+            _human_principal_ids = set(_config.get_human_principal_ids(data_dir))
             _registry_verifier = registry_auth.verifier_from_url(
                 _registry_url,
                 revoked_token=_registry_admin_token,
                 expected_iss=registry_auth.REGISTRY_ISS,
+                human_principal_ids=_human_principal_ids,
             )
             _grants_verifier = registry_auth.grants_verifier_from_url(
                 _registry_url,
@@ -1465,22 +1467,28 @@ def _handle_a2a_send(self) -> None:
                 else:
                     try:
                         _registry_verifier.authorize(token, from_)
+                    except registry_auth.HumanAuthError as exc:
+                        self._send_json(403, {"error": f"registry auth: {exc}"})
+                        return
                     except registry_auth.AuthError as exc:
                         warn_reason = str(exc)
                         _reject_status = 403
                         _reject_msg = f"registry auth: {exc}"
 
                 # Grant check: token proves identity; grant proves permission.
+                # Human principals (controller sessions) have no registry grant,
+                # so the grants check is skipped for them.
                 if warn_reason is None and _grants_verifier is not None:
-                    try:
-                        if not _grants_verifier.has_grant(from_):
-                            warn_reason = "no a2a_send grant"
+                    if not _registry_verifier.is_human(from_):
+                        try:
+                            if not _grants_verifier.has_grant(from_):
+                                warn_reason = "no a2a_send grant"
+                                _reject_status = 403
+                                _reject_msg = f"registry auth: no active grant for {from_!r}"
+                        except registry_auth.AuthError as exc:
+                            warn_reason = str(exc)
                             _reject_status = 403
-                            _reject_msg = f"registry auth: no active grant for {from_!r}"
-                    except registry_auth.AuthError as exc:
-                        warn_reason = str(exc)
-                        _reject_status = 403
-                        _reject_msg = f"registry auth: {exc}"
+                            _reject_msg = f"registry auth: {exc}"
 
                 if warn_reason is not None:
                     enforce = _config.get_a2a_auth_enforce(data_dir)

taosmd/registry_auth.py

@@ -35,6 +35,10 @@ class AuthError(Exception):
     """Raised when a token fails verification or the auth policy."""
 
 
+class HumanAuthError(AuthError):
+    """Raised when a human principal's sub does not match the claimed from."""
+
+
 def _require_jwt():
     try:
         import jwt  # noqa: PLC0415
@@ -58,23 +62,32 @@ def decode_and_verify(token: str, public_key: str) -> dict:
 
 
 def authorize_sender(token: str, claimed_from: str, *, public_key: str,
-                     revoked: set[str], expected_iss: str | None = None) -> dict:
+                     revoked: set[str], expected_iss: str | None = None,
+                     human_principal_ids: set[str] | None = None) -> dict:
     """Authorise a bus sender. Returns the verified claims or raises AuthError.
 
     Policy (after the EdDSA signature check):
-      * the token must carry a ``sub`` (the agent canonical_id);
+      * the token must carry a ``sub`` (the principal canonical_id);
       * ``sub`` must equal the message ``from`` (no impersonation);
-      * ``sub`` must not be in the registry revocation set;
+      * if the principal is a human (sub in ``human_principal_ids``), a
+        sub/from mismatch raises :class:`HumanAuthError` (always rejected,
+        even in verify-and-warn mode);
+      * for agent principals, ``sub`` must not be in the registry revocation set;
       * when ``expected_iss`` is set, ``iss`` must match it (issuer pinning).
     """
     claims = decode_and_verify(token, public_key)
     sub = claims.get("sub")
     if not sub:
         raise AuthError("token has no 'sub' (canonical_id) claim")
     if sub != claimed_from:
+        if human_principal_ids and sub in human_principal_ids:
+            raise HumanAuthError(
+                f"human token sub {sub!r} does not match from {claimed_from!r}"
+            )
         raise AuthError(f"token sub {sub!r} does not match from {claimed_from!r}")
-    if sub in revoked:
-        raise AuthError(f"canonical_id {sub!r} is revoked")
+    if not (human_principal_ids and sub in human_principal_ids):
+        if sub in revoked:
+            raise AuthError(f"canonical_id {sub!r} is revoked")
     if expected_iss is not None and claims.get("iss") != expected_iss:
         raise AuthError(f"token iss {claims.get('iss')!r} != expected {expected_iss!r}")
     return claims
@@ -96,20 +109,34 @@ class RegistryVerifier:
     ``pubkey_loader`` and ``revoked_loader`` are injected so the network layer
     can be supplied by the caller (and stubbed in tests). ``clock`` defaults to
     wall-clock ``time.time``; an injected clock makes refresh timing testable.
+
+    ``human_principal_ids`` is the set of canonical_ids that belong to human
+    principals (controller sessions). Human principals skip the registry
+    revocation check (they are not in the registry) and skip the grants check.
+    A human principal whose token ``sub`` does not match the message ``from``
+    raises :class:`HumanAuthError`, which the bus always rejects (even in
+    verify-and-warn mode) so a human cannot impersonate another human or an
+    agent handle.
     """
 
     def __init__(self, *, pubkey_loader, revoked_loader,
                  refresh_interval: float = 300.0, clock=time.time,
-                 expected_iss: str | None = None):
+                 expected_iss: str | None = None,
+                 human_principal_ids: set[str] | None = None):
         self._pubkey_loader = pubkey_loader
         self._revoked_loader = revoked_loader
         self._refresh_interval = refresh_interval
         self._clock = clock
         self._expected_iss = expected_iss
+        self._human_principal_ids = human_principal_ids or set()
         self._pubkey: str | None = None
         self._revoked: set[str] = set()
         self._revoked_fetched_at: float | None = None
 
+    def is_human(self, canonical_id: str) -> bool:
+        """Return True if ``canonical_id`` is a known human principal."""
+        return canonical_id in self._human_principal_ids
+
     def _get_pubkey(self) -> str:
         if self._pubkey is None:
             self._pubkey = self._pubkey_loader()
@@ -125,23 +152,32 @@ def _get_revoked(self) -> set[str]:
                 self._revoked_fetched_at = now
             except Exception as exc:  # noqa: BLE001
                 if self._revoked_fetched_at is None:
-                    # Never loaded: we cannot prove an agent is unrevoked, so
-                    # fail CLOSED rather than fall through to an empty allowlist.
                     raise AuthError(
                         f"revocation feed unavailable, refusing to authorise: {exc}"
                     ) from exc
-                # Already have a known-good set: keep it across a transient
-                # refresh failure (fail-safe, never silently un-revokes).
                 logger.warning("registry revocation refresh failed, "
                                "using last-good set: %s", exc)
         return self._revoked
 
     def authorize(self, token: str, claimed_from: str) -> dict:
-        """Authorise a sender; return verified claims or raise AuthError."""
+        """Authorise a sender; return verified claims or raise AuthError.
+
+        Human principals skip the revocation feed fetch entirely so the
+        fail-closed revocation check never blocks controller-signed humans.
+        """
+        try:
+            import jwt  # noqa: PLC0415
+            raw = jwt.decode(token, options={"verify_signature": False})
+        except Exception:  # noqa: BLE001
+            raw = {}
+        sub = raw.get("sub")
+        is_human = bool(sub and sub in self._human_principal_ids)
+        revoked = set() if is_human else self._get_revoked()
         return authorize_sender(
             token, claimed_from,
-            public_key=self._get_pubkey(), revoked=self._get_revoked(),
+            public_key=self._get_pubkey(), revoked=revoked,
             expected_iss=self._expected_iss,
+            human_principal_ids=self._human_principal_ids,
         )
 
 
@@ -337,7 +373,8 @@ def grants_verifier_from_url(base_url: str, *, refresh_interval: float = 300.0,
 def verifier_from_url(base_url: str, *, refresh_interval: float = 300.0,
                       opener=_http_get, clock=time.time,
                       expected_iss: str | None = REGISTRY_ISS,
-                      revoked_token: str | None = None) -> "RegistryVerifier":
+                      revoked_token: str | None = None,
+                      human_principal_ids: set[str] | None = None) -> "RegistryVerifier":
     """Build a :class:`RegistryVerifier` that fetches from a registry base URL.
 
     The HTTP getter is injectable (``opener``) so callers/tests can supply
@@ -346,11 +383,16 @@ def verifier_from_url(base_url: str, *, refresh_interval: float = 300.0,
     ``revoked_token`` is the taOS local/admin token sent as a Bearer header on
     the revoked-feed poll (the #710 contract moved it behind admin auth). The
     pubkey endpoint stays public and is fetched without a token.
+
+    ``human_principal_ids`` is the set of canonical_ids that belong to human
+    principals (controller sessions). These principals skip the registry
+    revocation check and their sub/from mismatch is always rejected.
     """
     base = base_url.rstrip("/")
     return RegistryVerifier(
         pubkey_loader=lambda: parse_pubkey_response(opener(base + PUBKEY_PATH)),
         revoked_loader=lambda: parse_revoked_response(
             opener(base + REVOKED_PATH, token=revoked_token)),
         refresh_interval=refresh_interval, clock=clock, expected_iss=expected_iss,
+        human_principal_ids=human_principal_ids,
     )