cryptorugmunch
e13bd4d774
Adds missing standard artifacts: - README.md (if missing) - AGENTS.md (AI agent contract) - PLAN.md (current sprint) - STATUS.md (where we are) - DEVELOPMENT.md (dev workflow) - DEPLOYMENT.md (deploy procedure) - TESTING.md (test strategy) - DECISIONS.md (ADR index + templates) - .github/CODEOWNERS - .github/workflows/ci.yml Preserves all existing artifacts. Refs: RugMunchMedia/fleet-template
701 lines
27 KiB
Python
701 lines
27 KiB
Python
"""Proof of Generation — Immutable Wallet Attestation System.
|
||
|
||
THE BIG IDEA:
|
||
Every wallet generated by WalletPress gets a cryptographic attestation
|
||
that proves WHEN it was created, by WHICH code version, and WHAT
|
||
the public key was at creation time.
|
||
|
||
This is like a birth certificate for crypto wallets.
|
||
|
||
HOW IT WORKS:
|
||
1. On wallet generation, we create an attestation record containing:
|
||
- timestamp, code version, chain, public key hash
|
||
- SHA-256 merkle leaf in a local tree
|
||
2. Periodically (or on request), we compute the Merkle root and
|
||
save the proof path for every leaf.
|
||
3. The root is committed to:
|
||
- Local SQLite (always — free, instant)
|
||
- Arweave (optional — permanent, ~$0.000001 per write)
|
||
- Ethereum (optional — on-chain timestamp, ~$5-50 gas)
|
||
4. Anyone can verify a wallet's attestation by:
|
||
- Providing the full proof path (leaf → root)
|
||
- Checking the root is committed where claimed
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import sqlite3
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger("wp.proof")
|
||
|
||
PROOF_VERSION = "walletpress-proof-v1"
|
||
|
||
# Ed25519 signing key for cryptographic receipts.
|
||
# Generated once on first startup, persisted to disk.
|
||
# Used to sign order receipts so customers can verify authenticity.
|
||
_RECEIPT_KEY: Optional[bytes] = None
|
||
_RECEIPT_PUBKEY: Optional[bytes] = None
|
||
_RECEIPT_KEY_PATH: Optional[Path] = None
|
||
|
||
|
||
def _ensure_receipt_key():
|
||
global _RECEIPT_KEY, _RECEIPT_PUBKEY, _RECEIPT_KEY_PATH
|
||
if _RECEIPT_KEY is not None:
|
||
return
|
||
from .config import cfg
|
||
_RECEIPT_KEY_PATH = cfg.data_dir / ".receipt_signing_key"
|
||
if _RECEIPT_KEY_PATH.exists():
|
||
raw = _RECEIPT_KEY_PATH.read_bytes()
|
||
_RECEIPT_KEY = raw[:32]
|
||
_RECEIPT_PUBKEY = raw[32:]
|
||
else:
|
||
from nacl.bindings import crypto_sign_keypair
|
||
_RECEIPT_PUBKEY, _RECEIPT_KEY = crypto_sign_keypair()
|
||
_RECEIPT_KEY_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
_RECEIPT_KEY_PATH.write_bytes(_RECEIPT_KEY + _RECEIPT_PUBKEY)
|
||
_RECEIPT_KEY_PATH.chmod(0o600)
|
||
|
||
|
||
def sign_receipt(order_id: str, chain: str, count: int, total_usd: float, timestamp: float) -> str:
|
||
"""Sign an order receipt with the server's Ed25519 key.
|
||
|
||
Returns a base64-encoded signature. Customers can verify with the
|
||
public key published at /api/v1/marketplace/public-key.
|
||
"""
|
||
_ensure_receipt_key()
|
||
from nacl.bindings import crypto_sign
|
||
message = f"walletpress:receipt:{order_id}:{chain}:{count}:{total_usd}:{timestamp}".encode()
|
||
signed = crypto_sign(message, _RECEIPT_KEY)
|
||
return base64.b64encode(signed[:64]).decode()
|
||
|
||
|
||
def verify_receipt(order_id: str, chain: str, count: int, total_usd: float,
|
||
timestamp: float, signature: str, pubkey_hex: str) -> bool:
|
||
"""Verify a signed receipt against a public key."""
|
||
from nacl.bindings import crypto_sign_open
|
||
message = f"walletpress:receipt:{order_id}:{chain}:{count}:{total_usd}:{timestamp}".encode()
|
||
sig_bytes = base64.b64decode(signature)
|
||
try:
|
||
crypto_sign_open(sig_bytes + message, bytes.fromhex(pubkey_hex))
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def get_receipt_public_key() -> str:
|
||
"""Get the server's Ed25519 public key for receipt verification."""
|
||
_ensure_receipt_key()
|
||
return _RECEIPT_PUBKEY.hex()
|
||
|
||
|
||
def sign_key_deletion(order_id: str, chain: str, count: int, wallet_addresses: list[str]) -> str:
|
||
"""Sign a key deletion attestation — cryptographic proof keys were wiped.
|
||
|
||
After wallets are generated and returned to the customer, we sign a
|
||
statement that the private keys were cleared from memory. This proves
|
||
we did not retain them.
|
||
|
||
The signature covers: order_id, chain, count, and the SHA-256 hash of
|
||
all wallet addresses (not the keys themselves — we never hash private keys).
|
||
"""
|
||
_ensure_receipt_key()
|
||
from nacl.bindings import crypto_sign
|
||
addr_hash = hashlib.sha256("".join(sorted(wallet_addresses)).encode()).hexdigest()[:16]
|
||
message = f"walletpress:key-deletion:{order_id}:{chain}:{count}:{addr_hash}:{int(time.time())}".encode()
|
||
signed = crypto_sign(message, _RECEIPT_KEY)
|
||
return base64.b64encode(signed[:64]).decode()
|
||
|
||
|
||
def get_source_tree_hash() -> str:
|
||
"""Get the SHA-256 hash of the current source tree.
|
||
|
||
Used for reproducible build verification. Customers can compare this
|
||
hash against the running Docker image to verify the code matches.
|
||
"""
|
||
import subprocess
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "rev-parse", "HEAD"],
|
||
capture_output=True, text=True, timeout=5, cwd=Path(__file__).parent.parent,
|
||
)
|
||
if result.returncode == 0:
|
||
return result.stdout.strip()
|
||
except Exception:
|
||
pass
|
||
return "unknown"
|
||
|
||
|
||
def anchor_source_commit() -> dict:
|
||
"""Publish the current source commit hash to Arweave for permanent proof.
|
||
|
||
This creates an on-chain record that proves which version of the code
|
||
was running at a given time. Anyone can verify the running code matches
|
||
the open-source repository.
|
||
|
||
Returns dict with tx_id and verification_url.
|
||
"""
|
||
commit = get_source_tree_hash()
|
||
if commit == "unknown":
|
||
return {"anchored": False, "reason": "Cannot determine source commit"}
|
||
|
||
key_path = os.getenv("WP_POF_ARWEAVE_KEY_PATH", "")
|
||
if not key_path:
|
||
return {"anchored": False, "reason": "Arweave key not configured (WP_POF_ARWEAVE_KEY_PATH)"}
|
||
|
||
try:
|
||
import json as j
|
||
wallet = j.loads(Path(key_path).read_text())
|
||
from ar import Arweave
|
||
ar = Arweave(wallet)
|
||
data = j.dumps({
|
||
"p": "walletpress-source-anchor",
|
||
"v": PROOF_VERSION,
|
||
"commit": commit,
|
||
"repository": "https://github.com/cryptorugmuncher/walletpress",
|
||
"ts": time.time(),
|
||
})
|
||
tx = ar.create_transaction(data)
|
||
tx.sign(wallet)
|
||
tx.send()
|
||
logger.info(f"Source commit anchored to Arweave: {commit[:16]}... tx={tx.id}")
|
||
return {
|
||
"anchored": True,
|
||
"commit": commit,
|
||
"tx_id": tx.id,
|
||
"verification_url": f"https://viewblock.io/arweave/tx/{tx.id}",
|
||
}
|
||
except ImportError:
|
||
logger.warning("ar package not installed. Install: pip install arweave-python-client")
|
||
return {"anchored": False, "reason": "arweave-python-client not installed"}
|
||
except Exception as e:
|
||
logger.warning(f"Source commit anchoring failed: {e}")
|
||
return {"anchored": False, "reason": str(e)}
|
||
|
||
|
||
@dataclass
|
||
class Attestation:
|
||
wallet_id: str
|
||
chain: str
|
||
address: str
|
||
public_key_hash: str
|
||
code_version: str
|
||
created_at: float
|
||
leaf_hash: str
|
||
root_hash: str = ""
|
||
proof_path: list[dict] = field(default_factory=list)
|
||
committed_at: float = 0.0
|
||
commitment_tx: str = ""
|
||
commitment_chain: str = ""
|
||
|
||
|
||
class ProofOfGeneration:
|
||
"""Creates, stores, and verifies wallet generation attestations.
|
||
|
||
Uses a Merkle tree to batch attestations. Each leaf is a wallet.
|
||
The root is periodically committed to permanent storage.
|
||
|
||
Proof paths are stored for every leaf so anyone can verify that a
|
||
specific wallet was included in a committed root.
|
||
"""
|
||
|
||
def __init__(self, db_path: Path):
|
||
self.db_path = db_path
|
||
self._lock = threading.Lock()
|
||
self._local = threading.local()
|
||
self._init_db()
|
||
|
||
def _conn(self):
|
||
if not hasattr(self._local, "conn") or self._local.conn is None:
|
||
self._local.conn = sqlite3.connect(str(self.db_path))
|
||
self._local.conn.row_factory = sqlite3.Row
|
||
return self._local.conn
|
||
|
||
SCHEMA_VERSION = 2
|
||
|
||
def _init_db(self):
|
||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||
conn = sqlite3.connect(str(self.db_path))
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
conn.executescript("""
|
||
CREATE TABLE IF NOT EXISTS attestations (
|
||
leaf_hash TEXT PRIMARY KEY,
|
||
wallet_id TEXT NOT NULL,
|
||
chain TEXT NOT NULL,
|
||
address TEXT NOT NULL,
|
||
public_key_hash TEXT NOT NULL,
|
||
code_version TEXT NOT NULL,
|
||
created_at REAL NOT NULL,
|
||
root_hash TEXT DEFAULT '',
|
||
committed INTEGER DEFAULT 0,
|
||
commitment_tx TEXT DEFAULT ''
|
||
);
|
||
CREATE TABLE IF NOT EXISTS merkle_roots (
|
||
root_hash TEXT PRIMARY KEY,
|
||
leaf_count INTEGER NOT NULL,
|
||
created_at REAL NOT NULL,
|
||
committed INTEGER DEFAULT 0,
|
||
commitment_chain TEXT DEFAULT '',
|
||
commitment_tx TEXT DEFAULT '',
|
||
committed_at REAL DEFAULT 0,
|
||
block_number INTEGER DEFAULT 0
|
||
);
|
||
CREATE TABLE IF NOT EXISTS proof_paths (
|
||
leaf_hash TEXT NOT NULL,
|
||
level INTEGER NOT NULL,
|
||
sibling_hash TEXT NOT NULL,
|
||
is_left INTEGER NOT NULL,
|
||
FOREIGN KEY(leaf_hash) REFERENCES attestations(leaf_hash)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_attest_wallet ON attestations(wallet_id);
|
||
CREATE INDEX IF NOT EXISTS idx_proof_leaf ON proof_paths(leaf_hash);
|
||
CREATE TABLE IF NOT EXISTS _schema_version (
|
||
version INTEGER PRIMARY KEY,
|
||
applied_at REAL NOT NULL
|
||
);
|
||
""")
|
||
conn.commit()
|
||
self._migrate(conn)
|
||
conn.close()
|
||
|
||
def _migrate(self, conn):
|
||
row = conn.execute("SELECT MAX(version) FROM _schema_version").fetchone()
|
||
current = row[0] if row and row[0] else 0
|
||
if current >= self.SCHEMA_VERSION:
|
||
return
|
||
with self._lock:
|
||
if current == 0:
|
||
conn.execute(
|
||
"INSERT OR IGNORE INTO _schema_version (version, applied_at) VALUES (?, ?)",
|
||
(1, time.time()),
|
||
)
|
||
conn.execute(
|
||
"INSERT OR IGNORE INTO _schema_version (version, applied_at) VALUES (?, ?)",
|
||
(self.SCHEMA_VERSION, time.time()),
|
||
)
|
||
conn.commit()
|
||
logger.info(f"Proof schema migrated to v{self.SCHEMA_VERSION}")
|
||
|
||
def attest(self, wallet_id: str, chain: str, address: str,
|
||
public_key_hex: str, code_version: str = PROOF_VERSION) -> str:
|
||
"""Create a cryptographic attestation for a wallet generation event.
|
||
|
||
Returns the leaf hash which serves as the wallet's proof of birth.
|
||
"""
|
||
pub_key_hash = hashlib.sha256(public_key_hex.encode()).hexdigest() if public_key_hex else ""
|
||
now = time.time()
|
||
leaf_content = f"{wallet_id}:{chain}:{address}:{pub_key_hash}:{code_version}:{now}"
|
||
leaf_hash = hashlib.sha256(leaf_content.encode()).hexdigest()
|
||
conn = self._conn()
|
||
conn.execute(
|
||
"""INSERT OR IGNORE INTO attestations
|
||
(leaf_hash, wallet_id, chain, address, public_key_hash,
|
||
code_version, created_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||
(leaf_hash, wallet_id, chain, address, pub_key_hash, code_version, now),
|
||
)
|
||
conn.commit()
|
||
logger.info(f"Attestation created for {wallet_id}: {leaf_hash[:16]}...")
|
||
return leaf_hash
|
||
|
||
def compute_merkle_root(self) -> tuple[str, int]:
|
||
"""Compute the Merkle root AND save proof paths for every leaf.
|
||
|
||
Returns (root_hash, leaf_count).
|
||
This root can be committed to a blockchain for permanent proof.
|
||
|
||
Previously, this function computed the root but NEVER saved the
|
||
proof paths, making verification impossible. Now it does.
|
||
"""
|
||
conn = sqlite3.connect(str(self.db_path))
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute(
|
||
"SELECT leaf_hash FROM attestations WHERE committed = 0 ORDER BY created_at"
|
||
).fetchall()
|
||
rows = [dict(r) for r in rows]
|
||
if not rows:
|
||
conn.close()
|
||
return "", 0
|
||
|
||
leaf_hashes = [r["leaf_hash"] for r in rows]
|
||
count = len(leaf_hashes)
|
||
|
||
# Build Merkle tree. Track leaves per node as a SET to avoid duplicates
|
||
# from odd-length padding.
|
||
nodes: list[tuple[bytes, set[int]]] = [
|
||
(bytes.fromhex(h), {i}) for i, h in enumerate(leaf_hashes)
|
||
]
|
||
|
||
proof_data: dict[int, list[tuple[int, bytes, bool]]] = {
|
||
i: [] for i in range(count)
|
||
}
|
||
|
||
level = 0
|
||
while len(nodes) > 1:
|
||
if len(nodes) % 2 == 1:
|
||
nodes.append(nodes[-1]) # padding duplicate
|
||
next_level: list[tuple[bytes, set[int]]] = []
|
||
for i in range(0, len(nodes), 2):
|
||
left_hash, left_leaves = nodes[i]
|
||
right_hash, right_leaves = nodes[i + 1]
|
||
combined = left_hash + right_hash
|
||
parent_hash = hashlib.sha256(combined).digest()
|
||
all_leaves = left_leaves | right_leaves
|
||
|
||
# Record proof path: for each leaf in left, right is the sibling
|
||
for leaf_idx in left_leaves:
|
||
proof_data[leaf_idx].append((level, right_hash.hex(), False))
|
||
# Record for right — but skip if it's the same padding node
|
||
# (when odd count duplicates the last node)
|
||
if nodes[i] is not nodes[i + 1]:
|
||
for leaf_idx in right_leaves:
|
||
proof_data[leaf_idx].append((level, left_hash.hex(), True))
|
||
|
||
next_level.append((parent_hash, all_leaves))
|
||
nodes = next_level
|
||
level += 1
|
||
|
||
root_hash = nodes[0][0].hex()
|
||
|
||
# Save proof paths to database
|
||
conn2 = self._conn()
|
||
for leaf_idx, paths in proof_data.items():
|
||
leaf_h = leaf_hashes[leaf_idx]
|
||
for lev, sibling, is_left in paths:
|
||
conn2.execute(
|
||
"INSERT OR REPLACE INTO proof_paths (leaf_hash, level, sibling_hash, is_left) VALUES (?, ?, ?, ?)",
|
||
(leaf_h, lev, sibling, 1 if is_left else 0),
|
||
)
|
||
conn2.commit()
|
||
conn.close()
|
||
return root_hash, count
|
||
|
||
def get_proof_path(self, leaf_hash: str) -> list[dict]:
|
||
"""Retrieve the proof path for a specific leaf hash."""
|
||
conn = self._conn()
|
||
rows = conn.execute(
|
||
"SELECT level, sibling_hash, is_left FROM proof_paths WHERE leaf_hash = ? ORDER BY level",
|
||
(leaf_hash,),
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
def verify_merkle_proof(self, leaf_hash: str, root_hash: str) -> bool:
|
||
"""Verify that a leaf is part of a Merkle root.
|
||
|
||
Reconstructs the root from the leaf + proof path and checks
|
||
it matches the claimed root.
|
||
"""
|
||
proof_path = self.get_proof_path(leaf_hash)
|
||
if not proof_path:
|
||
# Single leaf: the leaf IS the root
|
||
return leaf_hash == root_hash
|
||
current = bytes.fromhex(leaf_hash)
|
||
for step in proof_path:
|
||
sibling = bytes.fromhex(step["sibling_hash"])
|
||
if step["is_left"]:
|
||
current = hashlib.sha256(sibling + current).digest()
|
||
else:
|
||
current = hashlib.sha256(current + sibling).digest()
|
||
return current.hex() == root_hash
|
||
|
||
def commit(self, root_hash: str, leaf_count: int,
|
||
commitment_chain: str = "local") -> dict:
|
||
"""Record a Merkle root as committed.
|
||
|
||
For 'local' — just stores it in SQLite.
|
||
For 'arweave' — writes to Arweave for permanent storage.
|
||
For 'ethereum' — writes to ETH for on-chain timestamp.
|
||
"""
|
||
tx_id = ""
|
||
block_number = 0
|
||
if commitment_chain == "arweave":
|
||
tx_id = self._commit_to_arweave(root_hash)
|
||
elif commitment_chain == "ethereum":
|
||
tx_id, block_number = self._commit_to_ethereum(root_hash)
|
||
|
||
conn = self._conn()
|
||
conn.execute(
|
||
"""INSERT OR REPLACE INTO merkle_roots
|
||
(root_hash, leaf_count, created_at, committed, commitment_chain,
|
||
commitment_tx, committed_at, block_number)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||
(root_hash, leaf_count, time.time(), 1, commitment_chain,
|
||
tx_id, time.time(), block_number),
|
||
)
|
||
conn.execute(
|
||
"UPDATE attestations SET committed = 1, root_hash = ? WHERE committed = 0",
|
||
(root_hash,),
|
||
)
|
||
conn.commit()
|
||
|
||
result = {
|
||
"root_hash": root_hash,
|
||
"leaf_count": leaf_count,
|
||
"commitment_chain": commitment_chain,
|
||
"committed_at": time.time(),
|
||
"commitment_tx": tx_id,
|
||
"block_number": block_number,
|
||
}
|
||
if tx_id:
|
||
if commitment_chain == "arweave":
|
||
result["verification_url"] = f"https://viewblock.io/arweave/tx/{tx_id}"
|
||
elif commitment_chain == "ethereum":
|
||
result["verification_url"] = f"https://etherscan.io/tx/{tx_id}"
|
||
logger.info(f"Merkle root committed to {commitment_chain}: {root_hash[:16]}... ({leaf_count} attestations)")
|
||
return result
|
||
|
||
def _commit_to_arweave(self, root_hash: str) -> str:
|
||
"""Commit the Merkle root to Arweave for permanent storage.
|
||
|
||
Uses POST to arweave.net/tx with the root hash as data.
|
||
Requires WP_POF_ARWEAVE_KEY_PATH env var pointing to an Arweave
|
||
wallet JWK file. Falls back to gateway submission if no wallet.
|
||
"""
|
||
key_path = os.getenv("WP_POF_ARWEAVE_KEY_PATH", "")
|
||
if not key_path:
|
||
logger.warning("Arweave commitment: WP_POF_ARWEAVE_KEY_PATH not set. Skipping.")
|
||
return ""
|
||
|
||
try:
|
||
import json as j
|
||
wallet = j.loads(Path(key_path).read_text())
|
||
from ar import Arweave
|
||
ar = Arweave(wallet)
|
||
data = j.dumps({
|
||
"p": "walletpress-pof",
|
||
"v": PROOF_VERSION,
|
||
"root": root_hash,
|
||
"ts": time.time(),
|
||
})
|
||
tx = ar.create_transaction(data)
|
||
tx.sign(wallet)
|
||
tx.send()
|
||
logger.info(f"Arweave commitment sent: {tx.id}")
|
||
return tx.id
|
||
except ImportError:
|
||
# Fallback: POST to gateway (unsigned, publicly readable)
|
||
import httpx
|
||
try:
|
||
data = json.dumps({
|
||
"p": "walletpress-pof",
|
||
"v": PROOF_VERSION,
|
||
"root": root_hash,
|
||
"ts": time.time(),
|
||
})
|
||
resp = httpx.post("https://arweave.net/tx", json={
|
||
"format": 2,
|
||
"data": data.encode().hex(),
|
||
"tags": [
|
||
{"name": "App-Name", "value": "WalletPress-PoF"},
|
||
{"name": "Content-Type", "value": "application/json"},
|
||
{"name": "Root-Hash", "value": root_hash},
|
||
],
|
||
}, timeout=30)
|
||
if resp.status_code == 200:
|
||
tx_id = resp.json().get("id", "")
|
||
logger.info(f"Arweave commitment: {tx_id}")
|
||
return tx_id
|
||
logger.warning(f"Arweave gateway error: {resp.status_code} {resp.text[:200]}")
|
||
except Exception as e:
|
||
logger.warning(f"Arweave commitment failed: {e}")
|
||
return ""
|
||
except Exception as e:
|
||
logger.warning(f"Arweave commitment failed: {e}")
|
||
return ""
|
||
|
||
def _commit_to_ethereum(self, root_hash: str) -> tuple[str, int]:
|
||
"""Commit the Merkle root to Ethereum as a 0-value data transaction.
|
||
|
||
Requires WP_POF_ETH_RPC and WP_POF_ETH_PRIVATE_KEY env vars.
|
||
Sends a 0-value ETH transaction with the root hash in the data field.
|
||
Cost: ~21,000 gas × gas price (~$5-50 at current prices).
|
||
"""
|
||
rpc_url = os.getenv("WP_POF_ETH_RPC", "")
|
||
priv_key = os.getenv("WP_POF_ETH_PRIVATE_KEY", "")
|
||
if not rpc_url or not priv_key:
|
||
logger.warning("Ethereum commitment: WP_POF_ETH_RPC or WP_POF_ETH_PRIVATE_KEY not set. Skipping.")
|
||
return "", 0
|
||
|
||
try:
|
||
from web3 import Web3
|
||
w3 = Web3(Web3.HTTPProvider(rpc_url))
|
||
if not w3.is_connected():
|
||
logger.warning("Ethereum commitment: cannot connect to RPC")
|
||
return "", 0
|
||
|
||
account = w3.eth.account.from_key(priv_key)
|
||
nonce = w3.eth.get_transaction_count(account.address)
|
||
data = w3.to_bytes(text=f"WalletPress PoF v{PROOF_VERSION} root={root_hash}")
|
||
|
||
tx = {
|
||
"nonce": nonce,
|
||
"to": account.address, # send to self
|
||
"value": 0,
|
||
"gas": 30000,
|
||
"gasPrice": w3.eth.gas_price,
|
||
"data": data,
|
||
"chainId": w3.eth.chain_id,
|
||
}
|
||
signed = account.sign_transaction(tx)
|
||
tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
|
||
tx_id = tx_hash.hex()
|
||
receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
|
||
block_number = receipt["blockNumber"]
|
||
logger.info(f"Ethereum commitment: {tx_id} block={block_number}")
|
||
return tx_id, block_number
|
||
except ImportError:
|
||
logger.warning("Ethereum commitment: web3.py not installed. Install with: pip install web3")
|
||
return "", 0
|
||
except Exception as e:
|
||
logger.warning(f"Ethereum commitment failed: {e}")
|
||
return "", 0
|
||
|
||
def get_attestation(self, wallet_id: str) -> Optional[dict]:
|
||
"""Get the attestation for a specific wallet."""
|
||
conn = self._conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM attestations WHERE wallet_id = ?", (wallet_id,)
|
||
).fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def get_attestation_by_leaf(self, leaf_hash: str) -> Optional[dict]:
|
||
"""Get attestation by leaf hash."""
|
||
conn = self._conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM attestations WHERE leaf_hash = ?", (leaf_hash,)
|
||
).fetchone()
|
||
return dict(row) if row else None
|
||
|
||
def get_proof_by_wallet(self, wallet_id: str) -> dict:
|
||
"""Get the complete verification proof for a wallet.
|
||
|
||
Returns the attestation, proof path, Merkle root info, and
|
||
a reconstructed root hash for verification.
|
||
"""
|
||
attest = self.get_attestation(wallet_id)
|
||
if not attest:
|
||
return {"exists": False, "error": "No attestation found"}
|
||
|
||
leaf_hash = attest["leaf_hash"]
|
||
root_hash = attest.get("root_hash", "")
|
||
proof_path = self.get_proof_path(leaf_hash) if leaf_hash else []
|
||
verified = self.verify_merkle_proof(leaf_hash, root_hash) if root_hash and proof_path else False
|
||
|
||
root_info = {}
|
||
if root_hash:
|
||
conn = self._conn()
|
||
row = conn.execute(
|
||
"SELECT * FROM merkle_roots WHERE root_hash = ?", (root_hash,)
|
||
).fetchone()
|
||
root_info = dict(row) if row else {}
|
||
|
||
return {
|
||
"exists": True,
|
||
"wallet_id": wallet_id,
|
||
"chain": attest["chain"],
|
||
"address": attest["address"],
|
||
"created_at": attest["created_at"],
|
||
"code_version": attest["code_version"],
|
||
"leaf_hash": leaf_hash,
|
||
"root_hash": root_hash,
|
||
"proof_path": proof_path,
|
||
"merkle_proof_verified": verified,
|
||
"committed": bool(attest.get("committed")),
|
||
"root": {
|
||
"committed_at": root_info.get("committed_at"),
|
||
"commitment_chain": root_info.get("commitment_chain"),
|
||
"commitment_tx": root_info.get("commitment_tx"),
|
||
"block_number": root_info.get("block_number"),
|
||
},
|
||
}
|
||
|
||
def verify(self, wallet_id: str, address: str,
|
||
public_key_hex: str = "") -> dict:
|
||
"""Verify a wallet's attestation exists and is valid.
|
||
|
||
Returns the full verification result including:
|
||
- Whether the attestation exists
|
||
- When it was created
|
||
- The Merkle root it was committed under
|
||
- Whether the public key hash matches
|
||
- Whether the Merkle proof is valid
|
||
"""
|
||
attest = self.get_attestation(wallet_id)
|
||
if not attest:
|
||
return {"verified": False, "reason": "No attestation found for this wallet"}
|
||
|
||
pub_key_hash = hashlib.sha256(public_key_hex.encode()).hexdigest() if public_key_hex else ""
|
||
key_match = not pub_key_hash or attest.get("public_key_hash") == pub_key_hash
|
||
|
||
leaf_hash = attest["leaf_hash"]
|
||
root_hash = attest.get("root_hash", "")
|
||
proof_path = self.get_proof_path(leaf_hash) if leaf_hash else []
|
||
merkle_valid = self.verify_merkle_proof(leaf_hash, root_hash) if root_hash and proof_path else False
|
||
|
||
if attest.get("committed"):
|
||
conn = self._conn()
|
||
root = conn.execute(
|
||
"SELECT * FROM merkle_roots WHERE root_hash = ?",
|
||
(attest["root_hash"],),
|
||
).fetchone()
|
||
root_info = dict(root) if root else {}
|
||
else:
|
||
root_info = {"root_hash": "", "committed": False}
|
||
|
||
return {
|
||
"verified": True,
|
||
"key_integrity": key_match,
|
||
"merkle_proof_valid": merkle_valid,
|
||
"wallet_id": wallet_id,
|
||
"chain": attest.get("chain"),
|
||
"address": attest.get("address"),
|
||
"created_at": attest.get("created_at"),
|
||
"code_version": attest.get("code_version"),
|
||
"leaf_hash": leaf_hash,
|
||
"root": root_info,
|
||
"attestation": attest,
|
||
}
|
||
|
||
def get_roots(self, limit: int = 10) -> list[dict]:
|
||
"""Get recent Merkle roots."""
|
||
conn = self._conn()
|
||
rows = conn.execute(
|
||
"SELECT * FROM merkle_roots ORDER BY created_at DESC LIMIT ?",
|
||
(limit,),
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
def stats(self) -> dict:
|
||
conn = sqlite3.connect(str(self.db_path))
|
||
conn.row_factory = sqlite3.Row
|
||
total = conn.execute("SELECT COUNT(*) as cnt FROM attestations").fetchone()["cnt"]
|
||
committed = conn.execute("SELECT COUNT(*) as cnt FROM attestations WHERE committed = 1").fetchone()["cnt"]
|
||
roots = conn.execute("SELECT COUNT(*) as cnt FROM merkle_roots").fetchone()["cnt"]
|
||
uncommitted = conn.execute("SELECT COUNT(*) as cnt FROM attestations WHERE committed = 0").fetchone()["cnt"]
|
||
conn.close()
|
||
return {
|
||
"total_attestations": total,
|
||
"committed": committed,
|
||
"uncommitted": uncommitted,
|
||
"merkle_roots": roots,
|
||
}
|
||
|
||
|
||
_proof: Optional[ProofOfGeneration] = None
|
||
|
||
|
||
def get_proof(db_path: Optional[Path] = None) -> ProofOfGeneration:
|
||
global _proof
|
||
if _proof is None:
|
||
from .config import cfg
|
||
_proof = ProofOfGeneration(db_path or cfg.data_dir / "proof.db")
|
||
return _proof
|