feat(a2a): verify-and-warn mode for bus registry-auth (#138 Phase 2)

.gitignore

@@ -15,3 +15,4 @@ models/cross-encoder-onnx/
 .env
 *.log
 eval/reports/
+.superpowers/

taosmd/config.py

@@ -44,6 +44,8 @@
 _SERVE_DASHBOARD_KEY = "serve_dashboard"
 # Key under which the active global generator-profile id is stored.
 _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"
 
 MANAGED_BY_STANDALONE = "standalone"
 MANAGED_BY_TAOS = "taos"
@@ -512,6 +514,40 @@ def set_serve_dashboard(value: bool, data_dir=None) -> None:
     _write(data, data_dir)
 
 
+# ---------------------------------------------------------------------------
+# A2A auth enforce mode
+# ---------------------------------------------------------------------------
+
+def get_a2a_auth_enforce(data_dir=None) -> bool:
+    """Return whether A2A registry auth is in enforce mode.
+
+    Resolution order:
+
+    1. ``TAOSMD_A2A_AUTH_ENFORCE`` env var (``"1"`` or ``"true"`` = True)
+    2. ``a2a_auth_enforce`` bool in ``~/.taosmd/config.json``
+    3. Default: ``False`` (verify-and-warn mode)
+
+    When False (default), the A2A bus logs a warning when a post fails
+    registry auth checks but still accepts the message (verify-and-warn).
+    When True, failing posts are rejected with 401/403 as they were before
+    verify-and-warn was introduced.
+    """
+    env = os.environ.get("TAOSMD_A2A_AUTH_ENFORCE")
+    if env is not None:
+        return env.strip().lower() in ("1", "true", "yes")
+    data = _read(data_dir)
+    if _A2A_AUTH_ENFORCE_KEY in data:
+        return bool(data[_A2A_AUTH_ENFORCE_KEY])
+    return False
+
+
+def set_a2a_auth_enforce(value: bool, data_dir=None) -> None:
+    """Persist the a2a_auth_enforce flag."""
+    data = _read(data_dir)
+    data[_A2A_AUTH_ENFORCE_KEY] = bool(value)
+    _write(data, data_dir)
+
+
 __all__ = [
     "get_memory_model",
     "set_memory_model",
@@ -530,6 +566,8 @@ def set_serve_dashboard(value: bool, data_dir=None) -> None:
     "set_managed_by",
     "get_serve_dashboard",
     "set_serve_dashboard",
+    "get_a2a_auth_enforce",
+    "set_a2a_auth_enforce",
     "MANAGED_BY_STANDALONE",
     "MANAGED_BY_TAOS",
     "get_generator_profile",

taosmd/http_server.py

@@ -20,9 +20,19 @@
 
 Security note
 -------------
-When a registry verifier is configured, ``POST /a2a/send`` requires a valid
-registry-minted EdDSA-JWT Bearer token and an active grant; the token's
-``sub`` is matched against the message ``from`` to prevent impersonation.
+When a registry verifier is configured, ``POST /a2a/send`` runs in one of
+two modes controlled by the ``a2a_auth_enforce`` config key:
+
+* **verify-and-warn** (default, ``a2a_auth_enforce=false``): the registry
+  EdDSA-JWT and grant check are performed; on failure a ``WARNING`` is logged
+  (including the sender handle and reason) but the message is still accepted.
+  This lets a deployment observe auth violations before enabling hard enforcement.
+* **enforce** (``a2a_auth_enforce=true``): failure returns ``401`` (missing
+  token) or ``403`` (bad token / no grant) and the message is dropped.
+
+In both modes the token's ``sub`` claim is matched against the ``from`` field
+to prevent impersonation, and an active grant in the grants feed is required.
+
 Data endpoints (ingest, search, tasks) additionally support *optional* token
 binding: when a Bearer token is present and the registry verifier is
 configured, the token is verified and any ``project_id`` claim in it overrides
@@ -1155,30 +1165,57 @@ def _handle_a2a_send(self) -> None:
                 raise _BadRequest("'from' (non-empty string) is required")
             if not isinstance(body_text, str) or not body_text:
                 raise _BadRequest("'body' (non-empty string) is required")
-            # Registry auth (opt-in): when a verifier is configured, the sender
-            # must present a registry-minted EdDSA-JWT whose sub matches 'from',
-            # AND must have an active permission grant in the grants feed.
+            # Registry auth (opt-in): when a verifier is configured, run the
+            # identity + grant checks and collect any failure reason.
+            # In enforce mode (a2a_auth_enforce=true) failures are rejected with
+            # 401/403. In verify-and-warn mode (default) failures are logged as
+            # a WARNING but the message is accepted, allowing operators to observe
+            # violations before enabling hard enforcement.
             if _registry_verifier is not None:
                 from . import registry_auth  # noqa: PLC0415 - optional path
                 auth = self.headers.get("Authorization", "")
                 token = auth[len("Bearer "):].strip() if auth.startswith("Bearer ") else ""
+
+                # Compute warn_reason (None = auth passed) and the status/message
+                # to use in enforce mode. We collect these without returning early
+                # so the enforce vs. warn decision is made in one place below.
+                warn_reason: str | None = None
+                _reject_status: int = 403
+                _reject_msg: str = ""
+
                 if not token:
-                    self._send_json(401, {"error": "registry auth: Bearer token required"})
-                    return
-                try:
-                    _registry_verifier.authorize(token, from_)
-                except registry_auth.AuthError as exc:
-                    self._send_json(403, {"error": f"registry auth: {exc}"})
-                    return
+                    warn_reason = "missing Bearer token"
+                    _reject_status = 401
+                    _reject_msg = "registry auth: Bearer token required"
+                else:
+                    try:
+                        _registry_verifier.authorize(token, from_)
+                    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.
-                if _grants_verifier is not None:
+                if warn_reason is None and _grants_verifier is not None:
                     try:
                         if not _grants_verifier.has_grant(from_):
-                            self._send_json(403, {"error": f"registry auth: no active grant for {from_!r}"})
-                            return
+                            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:
-                        self._send_json(403, {"error": f"registry auth: {exc}"})
+                        warn_reason = str(exc)
+                        _reject_status = 403
+                        _reject_msg = f"registry auth: {exc}"
+
+                if warn_reason is not None:
+                    enforce = _config.get_a2a_auth_enforce(data_dir)
+                    if enforce:
+                        self._send_json(_reject_status, {"error": _reject_msg})
                         return
+                    logger.warning(
+                        "a2a verify-and-warn: accepting unverified post from %r: %s",
+                        from_, warn_reason,
+                    )
             result = runner.run(
                 service.a2a_send(
                     sender=from_, body=body_text,

tests/test_http_server_registry_auth.py

@@ -21,7 +21,7 @@
 from cryptography.hazmat.primitives import serialization
 
 from taosmd import api as taosmd_api
-from taosmd import http_server, registry_auth
+from taosmd import config as cfg, http_server, registry_auth
 
 
 def _keypair():
@@ -57,11 +57,18 @@ def _post_send(base_url, from_, body, token=None):
 
 @pytest.fixture
 def authed_server(tmp_path, monkeypatch):
-    """Live server built with a registry verifier (fake opener, no network)."""
+    """Live server built with a registry verifier (fake opener, no network).
+
+    Runs in enforce mode (a2a_auth_enforce=True) so that auth failures return
+    401/403 rather than being logged and accepted.
+    """
     data_dir = tmp_path / "taosmd-data"
     data_dir.mkdir()
     monkeypatch.setattr(taosmd_api, "_stores_cache", {})
 
+    # Enforce mode required so these rejection tests return 401/403.
+    cfg.set_a2a_auth_enforce(True, str(data_dir))
+
     def fake_opener(url, token=None):
         if url.endswith(registry_auth.PUBKEY_PATH):
             return json.dumps({"pubkey": PUB_PEM})
@@ -113,11 +120,17 @@ def test_send_with_token_from_wrong_key_is_rejected(authed_server):
 
 @pytest.fixture
 def iss_pinned_server(tmp_path, monkeypatch):
-    """Live server whose verifier pins iss to the taOS registry value."""
+    """Live server whose verifier pins iss to the taOS registry value.
+
+    Runs in enforce mode so that issuer-mismatch rejections fire as 403.
+    """
     data_dir = tmp_path / "taosmd-data"
     data_dir.mkdir()
     monkeypatch.setattr(taosmd_api, "_stores_cache", {})
 
+    # Enforce mode required so the wrong-iss rejection returns 403.
+    cfg.set_a2a_auth_enforce(True, str(data_dir))
+
     def fake_opener(url, token=None):
         if url.endswith(registry_auth.PUBKEY_PATH):
             return json.dumps({"public_key": PUB_PEM})

tests/test_http_server_trust_enforcement.py

@@ -1,11 +1,14 @@
 """Trust & Comms enforcement: grant check on A2A bus + dashboard gating.
 
-Tests the two new enforcement layers added on top of the existing registry
-verifier (test_http_server_registry_auth.py covers token verification alone):
-
-1. Grant check: a valid token + an active grant is required; a valid token
-   without a grant is rejected with 403.
-2. Dashboard gating: when managed_by=taos and serve_dashboard is not overridden,
+Tests the enforcement layers added on top of the existing registry verifier
+(test_http_server_registry_auth.py covers token verification alone):
+
+1. Verify-and-warn mode (default, a2a_auth_enforce=False): a verifier is
+   configured but auth failures are logged as WARNING and the message is
+   accepted. See warn_server fixture and the warn-mode test block.
+2. Enforce mode (a2a_auth_enforce=True): auth failures return 401/403.
+   See enforced_server fixture and the enforce-mode test block.
+3. Dashboard gating: when managed_by=taos and serve_dashboard is not overridden,
    GET / and GET /ui return 404; API routes stay up.
 """
 from __future__ import annotations
@@ -102,13 +105,43 @@ def fake_opener(url, timeout=5.0, token=None):
     return verifier, gv
 
 
+@pytest.fixture
+def warn_server(tmp_path, monkeypatch):
+    """Server with verifiers wired in but a2a_auth_enforce NOT set (default=False).
+
+    Auth failures are logged as WARNING and the message is still accepted.
+    """
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    monkeypatch.setattr(taosmd_api, "_stores_cache", {})
+
+    verifier, gv = _make_verifiers([{"canonical_id": "agent-allowed"}])
+    # Do NOT set a2a_auth_enforce -- default is False (warn mode).
+    httpd = http_server.make_server(
+        "127.0.0.1", 0, data_dir=str(data_dir),
+        verifier=verifier, grants_verifier=gv,
+    )
+    httpd.service_loop.run(taosmd_api._ensure_stores(str(data_dir)))
+    host, port = httpd.server_address[:2]
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://{host}:{port}"
+    finally:
+        httpd.shutdown()
+        httpd.service_loop.close()
+
+
 @pytest.fixture
 def enforced_server(tmp_path, monkeypatch):
-    """Server with both token verifier AND grants verifier wired in."""
+    """Server with both token verifier AND grants verifier wired in, enforce=True."""
     data_dir = tmp_path / "data"
     data_dir.mkdir()
     monkeypatch.setattr(taosmd_api, "_stores_cache", {})
 
+    # Explicitly enable enforce mode so failures return 401/403.
+    cfg.set_a2a_auth_enforce(True, str(data_dir))
+
     verifier, gv = _make_verifiers([{"canonical_id": "agent-allowed"}])
     httpd = http_server.make_server(
         "127.0.0.1", 0, data_dir=str(data_dir),
@@ -126,7 +159,53 @@ def enforced_server(tmp_path, monkeypatch):
 
 
 # ---------------------------------------------------------------------------
-# Grant check tests
+# Verify-and-warn mode tests (verifier configured, a2a_auth_enforce=False)
+# ---------------------------------------------------------------------------
+
+def test_warn_no_token_accepted(warn_server, caplog):
+    """No token: message accepted (200) and warning logged in warn mode."""
+    import logging
+    with caplog.at_level(logging.WARNING, logger="taosmd.http_server"):
+        status, body = _post_send(warn_server, "any-agent", "hello")
+    assert status == 200, body
+    assert any("verify-and-warn" in r.message and "missing Bearer token" in r.message
+               for r in caplog.records)
+
+
+def test_warn_invalid_token_accepted(warn_server, caplog):
+    """Invalid token: message accepted and warning logged in warn mode."""
+    import logging
+    with caplog.at_level(logging.WARNING, logger="taosmd.http_server"):
+        status, body = _post_send(warn_server, "any-agent", "hello", token="not-a-jwt")
+    assert status == 200, body
+    assert any("verify-and-warn" in r.message for r in caplog.records)
+
+
+def test_warn_valid_token_no_grant_accepted(warn_server, caplog):
+    """Valid token but no grant: message accepted and warning logged in warn mode."""
+    import logging
+    token = _mint("agent-no-grant")
+    with caplog.at_level(logging.WARNING, logger="taosmd.http_server"):
+        status, body = _post_send(warn_server, "agent-no-grant", "hello", token=token)
+    assert status == 200, body
+    assert any(
+        "verify-and-warn" in r.message and "no a2a_send grant" in r.message
+        for r in caplog.records
+    )
+
+
+def test_warn_valid_token_and_grant_no_warning(warn_server, caplog):
+    """Valid token + active grant: 200 with no verify-and-warn warning."""
+    import logging
+    token = _mint("agent-allowed")
+    with caplog.at_level(logging.WARNING, logger="taosmd.http_server"):
+        status, body = _post_send(warn_server, "agent-allowed", "hello", token=token)
+    assert status == 200, body
+    assert not any("verify-and-warn" in r.message for r in caplog.records)
+
+
+# ---------------------------------------------------------------------------
+# Enforce mode tests (a2a_auth_enforce=True)
 # ---------------------------------------------------------------------------
 
 def test_send_allowed_with_valid_token_and_grant(enforced_server):
@@ -149,12 +228,15 @@ def test_send_rejected_valid_token_no_grant(enforced_server):
 
 
 def test_send_rejected_expired_grant(tmp_path, monkeypatch):
-    """An expired grant is treated the same as no grant."""
+    """An expired grant is treated the same as no grant (enforce mode)."""
     import time
     data_dir = tmp_path / "data"
     data_dir.mkdir()
     monkeypatch.setattr(taosmd_api, "_stores_cache", {})
 
+    # Enforce mode required so the expired-grant rejection actually fires.
+    cfg.set_a2a_auth_enforce(True, str(data_dir))
+
     past = time.time() - 10.0
     verifier, gv = _make_verifiers(
         [{"canonical_id": "agent-expired", "expires_at": past}]