diff --git a/nanda_adapter/core/agent_bridge.py b/nanda_adapter/core/agent_bridge.py index 26361d6..89fe523 100644 --- a/nanda_adapter/core/agent_bridge.py +++ b/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") diff --git a/nanda_adapter/core/attestation.py b/nanda_adapter/core/attestation.py new file mode 100644 index 0000000..42d382b --- /dev/null +++ b/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= +""" +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") diff --git a/nanda_adapter/examples/attestix_verified.py b/nanda_adapter/examples/attestix_verified.py new file mode 100644 index 0000000..2e3cb27 --- /dev/null +++ b/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() diff --git a/setup.py b/setup.py index a320742..f6d34cc 100644 --- a/setup.py +++ b/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": [ diff --git a/tests/test_attestation.py b/tests/test_attestation.py new file mode 100644 index 0000000..714c6ee --- /dev/null +++ b/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"]))