Add optional Attestix attestation for agent registration

nanda_adapter/core/agent_bridge.py

@@ -15,6 +15,7 @@
 import asyncio
 from mcp_utils import MCPClient
 import base64
+import attestation
 
 import sys
 sys.stdout.reconfigure(line_buffering=True)
@@ -89,6 +90,9 @@ def register_with_registry(agent_id, agent_url, api_url):
             "agent_url": agent_url,
             "api_url": api_url
         }
+        proof = attestation.attest_registration(agent_id, agent_url)
+        if proof:
+            data["proof"] = proof
         print(f"Registering agent {agent_id} with URL {agent_url} at registry {registry_url}...")
         response = requests.post(f"{registry_url}/register", json=data)
         if response.status_code == 200:
@@ -109,6 +113,10 @@ def lookup_agent(agent_id):
         response = requests.get(f"{registry_url}/lookup/{agent_id}")
         if response.status_code == 200:
             agent_url = response.json().get("agent_url")
+            proof = response.json().get("proof")
+            if not attestation.verify_registration(agent_id, agent_url, proof):
+                print(f"WARNING: Attestix proof for agent {agent_id} failed verification - refusing URL")
+                return None
             print(f"Found agent {agent_id} at URL: {agent_url}")
             return agent_url
         print(f"Agent {agent_id} not found in registry")

nanda_adapter/core/attestation.py

@@ -0,0 +1,90 @@
+# attestation.py
+"""Optional Attestix-backed attestation for NANDA agent registration.
+
+NANDA's registry binds an agent_id to a URL with no proof: any caller can POST
+any agent_id -> any URL, and lookups trust whatever the registry returns. This
+module lets an agent sign that binding (agent_id -> agent_url) with an Ed25519
+key, and lets peers verify it before trusting a looked-up URL.
+
+It is entirely opt-in:
+  * No ATTESTIX_AGENT_KEY set  -> attest_registration() returns None; behaviour
+    is byte-for-byte identical to today (unsigned registration).
+  * Key set but `attestix` not installed -> degrades to unsigned, with a notice.
+
+Enable: pip install nanda-adapter[attestix]
+Generate a key seed:
+    python -c "import os,base64; print(base64.urlsafe_b64encode(os.urandom(32)).decode())"
+    export ATTESTIX_AGENT_KEY=<that value>
+"""
+import base64
+import os
+
+SUITE = "ed25519-jcs-2026"
+
+# ponytail: fail-open verification (missing proof -> accepted) so a signed agent
+# can still talk to the existing unsigned network. Set ATTESTIX_STRICT=1 to reject
+# any agent that presents no valid proof, once your peers all sign.
+STRICT = os.getenv("ATTESTIX_STRICT", "").lower() in ("1", "true", "yes")
+
+
+def _binding(agent_id, agent_url):
+    return {"agent_id": agent_id, "agent_url": agent_url}
+
+
+def _signing_key():
+    """Ed25519 private key from ATTESTIX_AGENT_KEY (base64url 32-byte seed), or None."""
+    raw = os.environ.get("ATTESTIX_AGENT_KEY")
+    if not raw:
+        return None
+    from attestix.auth.crypto import private_key_from_bytes
+    return private_key_from_bytes(base64.urlsafe_b64decode(raw))
+
+
+def attest_registration(agent_id, agent_url):
+    """Return a proof dict binding agent_id -> agent_url, or None if unconfigured."""
+    try:
+        key = _signing_key()
+        if key is None:
+            return None
+        from attestix.auth.crypto import public_key_to_did_key, sign_json_payload
+        return {
+            "suite": SUITE,
+            "issuer": public_key_to_did_key(key.public_key()),
+            "signature": sign_json_payload(key, _binding(agent_id, agent_url)),
+        }
+    except Exception as e:  # missing lib / bad key -> stay unsigned, never crash registration
+        print(f"Attestix: skipping registration proof ({e})")
+        return None
+
+
+def verify_registration(agent_id, agent_url, proof):
+    """True if a peer's (agent_id -> agent_url) binding is trustworthy.
+
+    No proof -> accepted unless ATTESTIX_STRICT (see STRICT above). A present
+    proof must be cryptographically valid AND bind exactly this agent_id+url.
+    """
+    if not proof:
+        return not STRICT
+    if proof.get("suite") != SUITE:
+        return False  # unknown suite -> can't verify it -> don't trust it
+    try:
+        from attestix.auth.crypto import did_key_to_public_key, verify_json_signature
+        pub = did_key_to_public_key(proof["issuer"])
+        return verify_json_signature(pub, _binding(agent_id, agent_url), proof["signature"])
+    except Exception as e:
+        print(f"Attestix: proof verification error ({e})")
+        return False
+
+
+if __name__ == "__main__":
+    # Runnable self-check of the security path (no attestix import needed if installed).
+    seed = base64.urlsafe_b64encode(bytes(range(32))).decode()
+    os.environ["ATTESTIX_AGENT_KEY"] = seed
+    p = attest_registration("agent-1", "https://a.example/a2a")
+    assert p and p["suite"] == SUITE, "should produce a proof when key is set"
+    assert verify_registration("agent-1", "https://a.example/a2a", p), "valid proof must verify"
+    assert not verify_registration("agent-1", "https://evil.example/a2a", p), "swapped URL must fail"
+    assert not verify_registration("agent-2", "https://a.example/a2a", p), "swapped id must fail"
+    assert not verify_registration("agent-1", "https://a.example/a2a", {**p, "suite": "other"}), "unknown suite must fail"
+    assert verify_registration("x", "y", None) is True, "no proof accepted in default (non-strict)"
+    print("attestation self-check OK")

nanda_adapter/examples/attestix_verified.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Attestix-verified NANDA agent.
+
+NANDA's registry maps an agent_id to a URL with no proof of who owns it. This
+example shows how to make that binding verifiable: the agent signs its
+(agent_id -> agent_url) with an Ed25519 key, peers verify the signature before
+trusting a looked-up URL, and a spoofed entry is rejected.
+
+The attestation is opt-in and lives in nanda_adapter.core.attestation. Enable it
+by setting ATTESTIX_AGENT_KEY before starting any NANDA agent - no code change to
+your improvement logic is required.
+
+Run this file directly for an offline sign/verify demo (no API key, no registry):
+    pip install nanda-adapter[attestix]
+    python attestix_verified.py
+"""
+import base64
+import os
+
+# attestation is the new sibling module in nanda_adapter/core
+from nanda_adapter.core import attestation
+
+
+def generate_agent_key() -> str:
+    """Return a fresh base64url Ed25519 seed for ATTESTIX_AGENT_KEY."""
+    return base64.urlsafe_b64encode(os.urandom(32)).decode()
+
+
+def offline_demo():
+    """Prove the security property without a server or registry."""
+    # In production you persist this once and export it; here we make one on the fly.
+    os.environ["ATTESTIX_AGENT_KEY"] = generate_agent_key()
+
+    agent_id, agent_url = "pirate-agent-7", "https://pirate.example.com/a2a"
+
+    proof = attestation.attest_registration(agent_id, agent_url)
+    print(f"Signed registration proof:\n  issuer: {proof['issuer']}\n  suite:  {proof['suite']}\n")
+
+    # A peer who looks this agent up verifies the proof before trusting the URL.
+    print("honest lookup           ->", attestation.verify_registration(agent_id, agent_url, proof))
+    print("spoofed URL (attacker)  ->", attestation.verify_registration(agent_id, "https://evil.example/a2a", proof))
+    print("spoofed agent_id        ->", attestation.verify_registration("admin-agent", agent_url, proof))
+
+
+def run_verified_agent():
+    """Start a real NANDA agent whose registration is Attestix-signed.
+
+    Identical to any other example - the only difference is ATTESTIX_AGENT_KEY in
+    the environment, which makes register_with_registry() attach a proof and
+    lookup_agent() verify peers' proofs automatically.
+    """
+    from nanda_adapter import NANDA
+
+    if not os.getenv("ATTESTIX_AGENT_KEY"):
+        os.environ["ATTESTIX_AGENT_KEY"] = generate_agent_key()
+        print("No ATTESTIX_AGENT_KEY set - generated an ephemeral one for this run.")
+
+    def echo_logic(message_text: str) -> str:
+        return message_text
+
+    nanda = NANDA(echo_logic)
+    print("Starting Attestix-verified NANDA agent (registration will be signed)...")
+
+    domain = os.getenv("DOMAIN_NAME", "localhost")
+    if domain != "localhost":
+        nanda.start_server_api(os.getenv("ANTHROPIC_API_KEY"), domain)
+    else:
+        nanda.start_server()
+
+
+def main():
+    if os.getenv("RUN_AGENT"):
+        run_verified_agent()
+    else:
+        offline_demo()
+
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -47,7 +47,8 @@ def read_readme():
     extras_require={
         "langchain": ["langchain-core", "langchain-anthropic"],
         "crewai": ["crewai", "langchain-anthropic"],
-        "all": ["langchain-core", "langchain-anthropic", "crewai"]
+        "attestix": ["attestix>=0.4.0"],
+        "all": ["langchain-core", "langchain-anthropic", "crewai", "attestix>=0.4.0"]
     },
     entry_points={
         "console_scripts": [

tests/test_attestation.py

@@ -0,0 +1,71 @@
+"""Unit tests for the optional Attestix attestation layer.
+
+Auto-skips when the optional `attestix` dependency is not installed. Exercises
+attest_registration / verify_registration directly (no network, no HTTP); the
+security-critical paths are: a valid proof verifies, tampering on id/url/signature
+is rejected, an unknown suite is rejected, and the unconfigured (no-key) path stays
+backward-compatible.
+"""
+import base64
+import os
+
+import pytest
+
+pytest.importorskip("attestix")  # skip the whole module if the optional dep is absent
+
+from nanda_adapter.core import attestation
+
+AGENT_ID = "pay-bot"
+AGENT_URL = "https://pay.example/a2a"
+
+
+@pytest.fixture
+def agent_key(monkeypatch):
+    """Set ATTESTIX_AGENT_KEY to a fresh 32-byte seed (same shape the registry path uses)."""
+    seed = base64.urlsafe_b64encode(os.urandom(32)).decode()
+    monkeypatch.setenv("ATTESTIX_AGENT_KEY", seed)
+    return seed
+
+
+def test_attest_then_verify_roundtrip(agent_key):
+    proof = attestation.attest_registration(AGENT_ID, AGENT_URL)
+    assert proof is not None
+    assert proof["suite"] == attestation.SUITE
+    assert proof["signature"]
+    assert attestation.verify_registration(AGENT_ID, AGENT_URL, proof) is True
+
+
+def test_tampered_url_rejected(agent_key):
+    proof = attestation.attest_registration(AGENT_ID, AGENT_URL)
+    # proof was signed for AGENT_URL; verifying against a swapped URL must fail
+    assert attestation.verify_registration(AGENT_ID, "https://attacker.example/a2a", proof) is False
+
+
+def test_tampered_agent_id_rejected(agent_key):
+    proof = attestation.attest_registration(AGENT_ID, AGENT_URL)
+    assert attestation.verify_registration("other-bot", AGENT_URL, proof) is False
+
+
+def test_garbled_signature_rejected(agent_key):
+    proof = attestation.attest_registration(AGENT_ID, AGENT_URL)
+    bad = {**proof, "signature": proof["signature"][:-4] + "AAAA"}
+    assert attestation.verify_registration(AGENT_ID, AGENT_URL, bad) is False
+
+
+def test_unknown_suite_rejected(agent_key):
+    proof = attestation.attest_registration(AGENT_ID, AGENT_URL)
+    assert attestation.verify_registration(AGENT_ID, AGENT_URL, {**proof, "suite": "nope-1"}) is False
+
+
+def test_no_key_returns_none_and_backward_compat(monkeypatch):
+    monkeypatch.delenv("ATTESTIX_AGENT_KEY", raising=False)
+    # ensure non-strict (fail-open) default for this case
+    monkeypatch.delenv("ATTESTIX_STRICT", raising=False)
+    monkeypatch.setattr(attestation, "STRICT", False)
+    assert attestation.attest_registration(AGENT_ID, AGENT_URL) is None
+    # an unsigned peer (proof=None) is accepted when not strict -> existing network keeps working
+    assert attestation.verify_registration(AGENT_ID, AGENT_URL, None) is True
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))