diff --git a/docs/AGENT_VANITY_WALLETS.md b/docs/AGENT_VANITY_WALLETS.md new file mode 100644 index 000000000..b1106c742 --- /dev/null +++ b/docs/AGENT_VANITY_WALLETS.md @@ -0,0 +1,95 @@ +# Agent Vanity Wallets + +This document describes the first implementation slice for +rustchain-bounties#30: deterministic AI-agent vanity wallet generation and +local registration. + +## Scope + +The current slice covers the milestone boundary: + +- Validate agent vanity names. +- Generate `RTC--` wallet IDs from agent identity plus a + hardware fingerprint. +- Optionally mine the hash portion for a hex prefix or suffix. +- Bind one registered agent to one hardware fingerprint hash in SQLite. +- Store the optional Ed25519 public key that later attestation work can use for + signed agent proofs. +- Provide a small CLI for generation, registration, and listing. + +It does not yet extend `/attest/submit` or implement useful-work proofs. Those +belong to the later bounty milestones. + +## Address Scheme + +The canonical wallet format is: + +```text +RTC--<10 hex chars> +``` + +The hash is derived from canonical JSON containing: + +- normalized `agent_name` +- `hardware_fingerprint_hash` +- optional `public_key_hex` +- nonce +- scheme tag `rustchain-agent-vanity-v1` + +Because the hardware fingerprint hash is part of the seed, the same agent name +on different hardware generates a different wallet. + +## CLI Examples + +Generate without saving: + +```bash +python -m node.agent_vanity_wallets generate claude-code \ + --fingerprint '{"cpu":"IBM POWER8","clock_skew_ppm":18.4}' +``` + +Register in a local node database: + +```bash +python -m node.agent_vanity_wallets --db rustchain_v2.db register claude-code \ + --fingerprint ./fingerprint.json \ + --pubkey 0000000000000000000000000000000000000000000000000000000000000000 +``` + +Mine a short vanity hash prefix: + +```bash +python -m node.agent_vanity_wallets generate sophia \ + --fingerprint ./fingerprint.json \ + --hash-prefix 00 +``` + +List registrations: + +```bash +python -m node.agent_vanity_wallets --db rustchain_v2.db list +``` + +## Registration Guarantees + +The SQLite table has unique constraints on: + +- `agent_name` +- `wallet` +- `hardware_fingerprint_hash` + +That enforces the first version of the "one agent per physical machine" rule. +Re-registering the same agent on the same fingerprint is idempotent. Registering +a second agent against the same fingerprint is rejected with +`hardware_already_bound_to_agent`. + +## Next Milestone + +The next slice should wire this identity into attestation: + +1. Add a node route such as `POST /api/agents/vanity/register`. +2. Extend `/attest/submit` with `agent_type`, `agent_version`, and + `agent_proof`. +3. Verify a signed challenge with the stored Ed25519 public key. +4. Record useful-work proofs such as merged PRs, bounty completions, served API + calls, or inference-token counters. diff --git a/node/agent_vanity_wallets.py b/node/agent_vanity_wallets.py new file mode 100644 index 000000000..6f98a0503 --- /dev/null +++ b/node/agent_vanity_wallets.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: MIT +""" +Agent vanity wallets for RustChain. + +This module implements the first milestone of rustchain-bounties#30: +deterministic agent vanity wallet generation plus local registration storage. +It deliberately keeps attestation integration separate so the wallet identity +primitive can be tested without a live node. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sqlite3 +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Iterable, Mapping, Optional + + +DEFAULT_DB_PATH = "/root/rustchain/rustchain_v2.db" +AGENT_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,19}$") +RESERVED_AGENT_NAMES = { + "admin", + "bank", + "burn", + "coinbase", + "escrow", + "founder", + "genesis", + "operator", + "root", + "system", + "treasury", +} +MAX_VANITY_ATTEMPTS = 250_000 + + +class AgentVanityError(ValueError): + """Raised when an agent vanity wallet request is invalid.""" + + +@dataclass(frozen=True) +class VanityWallet: + agent_name: str + wallet: str + hardware_fingerprint_hash: str + nonce: int + vanity_digest: str + public_key_hex: Optional[str] = None + + +def normalize_agent_name(agent_name: str) -> str: + """Return the canonical vanity-name form or raise ``AgentVanityError``.""" + if not isinstance(agent_name, str): + raise AgentVanityError("agent_name_must_be_text") + normalized = agent_name.strip().lower().replace("_", "-") + if not AGENT_NAME_RE.fullmatch(normalized): + raise AgentVanityError("agent_name_must_be_3_to_20_alnum_dash") + if normalized in RESERVED_AGENT_NAMES or normalized.startswith("rtc-"): + raise AgentVanityError("agent_name_reserved") + return normalized + + +def canonical_hardware_fingerprint(hardware_fingerprint: Mapping[str, Any] | str) -> str: + """Return a stable SHA-256 hash for a hardware fingerprint payload.""" + if isinstance(hardware_fingerprint, Mapping): + encoded = json.dumps( + hardware_fingerprint, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + elif isinstance(hardware_fingerprint, str): + text = hardware_fingerprint.strip() + if not text: + raise AgentVanityError("hardware_fingerprint_required") + encoded = text.encode("utf-8") + else: + raise AgentVanityError("hardware_fingerprint_must_be_mapping_or_text") + return hashlib.sha256(encoded).hexdigest() + + +def validate_public_key_hex(public_key_hex: Optional[str]) -> Optional[str]: + """Validate an optional Ed25519 public key encoded as 32-byte hex.""" + if public_key_hex in (None, ""): + return None + if not isinstance(public_key_hex, str): + raise AgentVanityError("public_key_must_be_hex_text") + value = public_key_hex.strip().lower() + try: + raw = bytes.fromhex(value) + except ValueError as exc: + raise AgentVanityError("public_key_must_be_hex") from exc + if len(raw) != 32: + raise AgentVanityError("public_key_must_be_32_bytes") + return value + + +def _seed_material( + agent_name: str, + hardware_hash: str, + public_key_hex: Optional[str], + nonce: int, +) -> bytes: + payload = { + "agent_name": agent_name, + "hardware_fingerprint_hash": hardware_hash, + "public_key_hex": public_key_hex or "", + "nonce": nonce, + "scheme": "rustchain-agent-vanity-v1", + } + return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def generate_vanity_wallet( + agent_name: str, + hardware_fingerprint: Mapping[str, Any] | str, + *, + public_key_hex: Optional[str] = None, + hash_prefix: str = "", + hash_suffix: str = "", + max_attempts: int = MAX_VANITY_ATTEMPTS, +) -> VanityWallet: + """ + Generate ``RTC--`` deterministically from agent identity + and hardware fingerprint. + + Optional ``hash_prefix`` / ``hash_suffix`` constraints mine over a nonce + against the derived hash portion. With no constraints, nonce 0 is returned. + """ + normalized = normalize_agent_name(agent_name) + hardware_hash = canonical_hardware_fingerprint(hardware_fingerprint) + key_hex = validate_public_key_hex(public_key_hex) + prefix = hash_prefix.lower().strip() + suffix = hash_suffix.lower().strip() + if not re.fullmatch(r"[0-9a-f]*", prefix + suffix): + raise AgentVanityError("vanity_constraints_must_be_hex") + if max_attempts < 1: + raise AgentVanityError("max_attempts_must_be_positive") + + for nonce in range(max_attempts): + digest = hashlib.sha256(_seed_material(normalized, hardware_hash, key_hex, nonce)).hexdigest() + vanity_part = digest[:10] + if vanity_part.startswith(prefix) and vanity_part.endswith(suffix): + return VanityWallet( + agent_name=normalized, + wallet=f"RTC-{normalized}-{vanity_part}", + hardware_fingerprint_hash=hardware_hash, + nonce=nonce, + vanity_digest=digest, + public_key_hex=key_hex, + ) + raise AgentVanityError("vanity_pattern_not_found") + + +SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS agent_vanity_wallets ( + agent_name TEXT PRIMARY KEY, + wallet TEXT NOT NULL UNIQUE, + hardware_fingerprint_hash TEXT NOT NULL UNIQUE, + public_key_hex TEXT, + nonce INTEGER NOT NULL, + vanity_digest TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_agent_vanity_wallet ON agent_vanity_wallets(wallet); +CREATE INDEX IF NOT EXISTS idx_agent_vanity_hw ON agent_vanity_wallets(hardware_fingerprint_hash); +""" + + +def init_agent_vanity_tables(conn: sqlite3.Connection) -> None: + """Create agent vanity registration tables.""" + conn.executescript(SCHEMA_SQL) + + +def register_agent_vanity_wallet( + conn: sqlite3.Connection, + agent_name: str, + hardware_fingerprint: Mapping[str, Any] | str, + *, + public_key_hex: Optional[str] = None, + hash_prefix: str = "", + hash_suffix: str = "", + now_ts: Optional[int] = None, +) -> VanityWallet: + """Generate and persist a one-agent-per-machine vanity wallet registration.""" + init_agent_vanity_tables(conn) + wallet = generate_vanity_wallet( + agent_name, + hardware_fingerprint, + public_key_hex=public_key_hex, + hash_prefix=hash_prefix, + hash_suffix=hash_suffix, + ) + now = int(time.time()) if now_ts is None else int(now_ts) + existing = conn.execute( + """ + SELECT agent_name, wallet, hardware_fingerprint_hash, public_key_hex, nonce, vanity_digest + FROM agent_vanity_wallets + WHERE agent_name = ? OR hardware_fingerprint_hash = ? OR wallet = ? + """, + (wallet.agent_name, wallet.hardware_fingerprint_hash, wallet.wallet), + ).fetchone() + if existing: + row = _row_to_wallet(existing) + if row == wallet: + return row + if row.agent_name == wallet.agent_name: + raise AgentVanityError("agent_name_already_registered") + if row.hardware_fingerprint_hash == wallet.hardware_fingerprint_hash: + raise AgentVanityError("hardware_already_bound_to_agent") + raise AgentVanityError("wallet_already_registered") + + conn.execute( + """ + INSERT INTO agent_vanity_wallets ( + agent_name, wallet, hardware_fingerprint_hash, public_key_hex, + nonce, vanity_digest, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + wallet.agent_name, + wallet.wallet, + wallet.hardware_fingerprint_hash, + wallet.public_key_hex, + wallet.nonce, + wallet.vanity_digest, + now, + now, + ), + ) + conn.commit() + return wallet + + +def _row_to_wallet(row: Iterable[Any]) -> VanityWallet: + values = list(row) + return VanityWallet( + agent_name=values[0], + wallet=values[1], + hardware_fingerprint_hash=values[2], + public_key_hex=values[3], + nonce=int(values[4]), + vanity_digest=values[5], + ) + + +def get_agent_vanity_wallet(conn: sqlite3.Connection, agent_name: str) -> Optional[VanityWallet]: + """Fetch a registration by agent name.""" + init_agent_vanity_tables(conn) + normalized = normalize_agent_name(agent_name) + row = conn.execute( + """ + SELECT agent_name, wallet, hardware_fingerprint_hash, public_key_hex, nonce, vanity_digest + FROM agent_vanity_wallets + WHERE agent_name = ? + """, + (normalized,), + ).fetchone() + return _row_to_wallet(row) if row else None + + +def list_agent_vanity_wallets(conn: sqlite3.Connection) -> list[VanityWallet]: + """List all registered agent vanity wallets in stable order.""" + init_agent_vanity_tables(conn) + rows = conn.execute( + """ + SELECT agent_name, wallet, hardware_fingerprint_hash, public_key_hex, nonce, vanity_digest + FROM agent_vanity_wallets + ORDER BY created_at ASC, agent_name ASC + """ + ).fetchall() + return [_row_to_wallet(row) for row in rows] + + +def _load_fingerprint_arg(value: str) -> Mapping[str, Any] | str: + candidate = Path(value) + if candidate.exists(): + return json.loads(candidate.read_text(encoding="utf-8")) + try: + parsed = json.loads(value) + except json.JSONDecodeError: + return value + return parsed if isinstance(parsed, Mapping) else value + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="RustChain agent vanity wallet registration") + parser.add_argument("--db", default=DEFAULT_DB_PATH, help="SQLite DB path") + sub = parser.add_subparsers(dest="cmd", required=True) + + gen = sub.add_parser("generate", help="Generate a vanity wallet without saving it") + gen.add_argument("agent_name") + gen.add_argument("--fingerprint", required=True, help="JSON string, text seed, or JSON file path") + gen.add_argument("--pubkey", dest="public_key_hex") + gen.add_argument("--hash-prefix", default="") + gen.add_argument("--hash-suffix", default="") + + reg = sub.add_parser("register", help="Generate and save a vanity wallet registration") + reg.add_argument("agent_name") + reg.add_argument("--fingerprint", required=True, help="JSON string, text seed, or JSON file path") + reg.add_argument("--pubkey", dest="public_key_hex") + reg.add_argument("--hash-prefix", default="") + reg.add_argument("--hash-suffix", default="") + + sub.add_parser("list", help="List registered agent vanity wallets") + return parser + + +def main(argv: Optional[list[str]] = None) -> int: + args = build_parser().parse_args(argv) + if args.cmd == "generate": + wallet = generate_vanity_wallet( + args.agent_name, + _load_fingerprint_arg(args.fingerprint), + public_key_hex=args.public_key_hex, + hash_prefix=args.hash_prefix, + hash_suffix=args.hash_suffix, + ) + print(json.dumps(asdict(wallet), sort_keys=True)) + return 0 + + with sqlite3.connect(args.db) as conn: + if args.cmd == "register": + wallet = register_agent_vanity_wallet( + conn, + args.agent_name, + _load_fingerprint_arg(args.fingerprint), + public_key_hex=args.public_key_hex, + hash_prefix=args.hash_prefix, + hash_suffix=args.hash_suffix, + ) + print(json.dumps(asdict(wallet), sort_keys=True)) + return 0 + if args.cmd == "list": + wallets = [asdict(wallet) for wallet in list_agent_vanity_wallets(conn)] + print(json.dumps(wallets, sort_keys=True)) + return 0 + raise AssertionError(f"unhandled command: {args.cmd}") + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/test_agent_vanity_wallets.py b/tests/test_agent_vanity_wallets.py new file mode 100644 index 000000000..ff1dc1ab7 --- /dev/null +++ b/tests/test_agent_vanity_wallets.py @@ -0,0 +1,94 @@ +import json +import sqlite3 +import subprocess +import sys + +import pytest + +from node.agent_vanity_wallets import ( + AgentVanityError, + generate_vanity_wallet, + get_agent_vanity_wallet, + list_agent_vanity_wallets, + normalize_agent_name, + register_agent_vanity_wallet, +) + + +FINGERPRINT = { + "cpu": "IBM POWER8", + "clock_skew_ppm": 18.4, + "cache_signature": "l2:stable:l3:wide", + "thermal_curve": [31.2, 34.8, 37.1], +} + + +def test_agent_name_normalization_and_reserved_names(): + assert normalize_agent_name("Claude_Code") == "claude-code" + with pytest.raises(AgentVanityError, match="reserved"): + normalize_agent_name("treasury") + with pytest.raises(AgentVanityError, match="3_to_20"): + normalize_agent_name("ab") + + +def test_vanity_wallet_is_deterministic_for_same_agent_and_hardware(): + first = generate_vanity_wallet("Sophia", FINGERPRINT) + second = generate_vanity_wallet("sophia", dict(reversed(list(FINGERPRINT.items())))) + + assert first == second + assert first.wallet.startswith("RTC-sophia-") + assert len(first.wallet.rsplit("-", 1)[-1]) == 10 + + +def test_vanity_constraints_mine_hash_prefix(): + wallet = generate_vanity_wallet("g4agent", FINGERPRINT, hash_prefix="00") + + assert wallet.wallet.startswith("RTC-g4agent-00") + assert wallet.nonce > 0 + + +def test_public_key_must_be_valid_ed25519_length(): + with pytest.raises(AgentVanityError, match="32_bytes"): + generate_vanity_wallet("agentx", FINGERPRINT, public_key_hex="abcd") + + +def test_register_persists_wallet_and_is_idempotent(): + conn = sqlite3.connect(":memory:") + + created = register_agent_vanity_wallet(conn, "powerbot", FINGERPRINT, now_ts=123) + loaded = get_agent_vanity_wallet(conn, "powerbot") + repeated = register_agent_vanity_wallet(conn, "powerbot", FINGERPRINT, now_ts=456) + + assert loaded == created + assert repeated == created + assert list_agent_vanity_wallets(conn) == [created] + + +def test_registration_rejects_second_agent_on_same_hardware(): + conn = sqlite3.connect(":memory:") + register_agent_vanity_wallet(conn, "agent-one", FINGERPRINT) + + with pytest.raises(AgentVanityError, match="hardware_already_bound"): + register_agent_vanity_wallet(conn, "agent-two", FINGERPRINT) + + +def test_cli_generate_outputs_json(): + proc = subprocess.run( + [ + sys.executable, + "-m", + "node.agent_vanity_wallets", + "generate", + "demo-agent", + "--fingerprint", + json.dumps(FINGERPRINT), + ], + check=True, + capture_output=True, + text=True, + ) + + payload = json.loads(proc.stdout) + assert payload["agent_name"] == "demo-agent" + assert payload["wallet"].startswith("RTC-demo-agent-") +