Rug Munch Media LLC
59e00c3bcc
P3-1 — ruff config: 93 → 1 error (one legit unused import)
Added [tool.ruff] section to backend/pyproject.toml with:
- target-version = py310
- line-length = 100
- Per-file ignores for intentional E402 (mcp_server, main.py,
routers/airdrop.py, tests/) and F401 on plugin re-exports.
Fixed all auto-fixable issues via 'ruff check . --fix'. The
remaining 1 F401 is coincurve.ecdsa.recover — imported in
core/smart_wallet.py for a feature that's not yet wired up.
Wave 6 (pen test prep):
- bandit: 3 High (all pycryptodome false positives — pyCrypto the
original is deprecated, but pycryptodome is the active fork we use
for Keccak-256 in Ethereum address generation).
- bandit: 7 Medium (mostly bind-all-interfaces 0.0.0.0 which is
intentional for container deploys).
- bandit: 85 Low (mostly asserts on test code — acceptable).
- pip-audit: zero vulnerabilities in our actual dependencies.
Code hygiene improvements:
- core/pdf.py: properly imports qrcode (was missing).
- cli/verify_receipt.py: confirmed false-positive (timestamp is
a function parameter, ignored).
- routers/wallet_admin.py: removed duplicate class definitions
introduced during Wave 5 split.
Refs: AUDIT.md P3-1 (final), Wave 6 prep
1447 lines
54 KiB
Python
1447 lines
54 KiB
Python
"""Chain Vault Router — Complete wallet generation & management API.
|
|
|
|
Every endpoint is documented, authenticated, and audited.
|
|
This implements the full ~49 endpoint API that the WordPress plugin expects.
|
|
|
|
Trust: All crypto operations are deterministic. Same input = same output,
|
|
ALWAYS, on ANY software. Verify with your own wallet software.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
import time
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Request
|
|
from pydantic import BaseModel, Field
|
|
|
|
from core.auth import get_key_store
|
|
from core.audit import get_audit
|
|
from core.config import cfg
|
|
from core.proof import get_proof
|
|
from .tx_broadcaster import broadcast as broadcast_tx
|
|
from core.vault import WalletEntry, get_vault
|
|
from wallet_engine.chains import CHAINS, CHAIN_GROUPS, get_chains_for_tier
|
|
from wallet_engine.generator import get_generator
|
|
|
|
logger = logging.getLogger("wp.chain_vault")
|
|
|
|
|
|
router = APIRouter(prefix="/api/v1/chain-vault", tags=["Chain Vault"])
|
|
|
|
|
|
def _require_totp(request: Request) -> None:
|
|
"""Require valid TOTP code for admin operations.
|
|
|
|
Shared with wallet_admin.py — kept here so chain_vault.py can use
|
|
it independently. If you change the logic, update both.
|
|
"""
|
|
from core.totp import get_totp_manager
|
|
mgr = get_totp_manager()
|
|
if not mgr.is_enabled():
|
|
return
|
|
code = request.headers.get("X-TOTP-Code", "")
|
|
if not mgr.verify(code):
|
|
raise HTTPException(status_code=401, detail="TOTP code required or invalid. Set X-TOTP-Code header.")
|
|
|
|
# ── Models ───────────────────────────────────────────────────────
|
|
|
|
|
|
class GenerateRequest(BaseModel):
|
|
chain: str = Field(..., description="Chain key (eth, sol, btc, etc.)")
|
|
count: int = Field(default=1, ge=1, le=100, description="Number of wallets")
|
|
label: str = Field(default="", description="Optional label")
|
|
tags: list[str] = Field(default_factory=list)
|
|
passphrase: str = Field(default="", description="BIP39 passphrase for hidden wallets")
|
|
|
|
|
|
class BatchGenerateRequest(BaseModel):
|
|
chains: list[str] = Field(..., description="List of chain keys")
|
|
label_prefix: str = Field(default="", description="Prefix for all labels")
|
|
|
|
|
|
class GenerateAllRequest(BaseModel):
|
|
label_prefix: str = Field(default="")
|
|
|
|
|
|
class ImportRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
private_key: str = Field(..., description="Hex-encoded private key")
|
|
label: str = Field(default="")
|
|
tags: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class FromMnemonicRequest(BaseModel):
|
|
mnemonic: str = Field(..., description="BIP39 mnemonic phrase")
|
|
passphrase: str = Field(default="", description="Optional BIP39 passphrase")
|
|
chains: list[str] = Field(default_factory=list, description="Chains to derive (empty = all)")
|
|
|
|
|
|
class DeriveAddressRequest(BaseModel):
|
|
mnemonic: str = Field(...)
|
|
chain: str = Field(...)
|
|
path: str = Field(default="", description="Custom derivation path (optional)")
|
|
|
|
|
|
class HDWalletRequest(BaseModel):
|
|
mnemonic: str = Field(...)
|
|
passphrase: str = Field(default="")
|
|
label: str = Field(default="")
|
|
|
|
|
|
class RotateRequest(BaseModel):
|
|
wallet_id: str = Field(...)
|
|
reason: str = Field(default="scheduled_rotation")
|
|
|
|
|
|
class RotateSweepRequest(BaseModel):
|
|
wallet_id: str = Field(...)
|
|
to_address: str = Field(...)
|
|
|
|
|
|
class DistributeRequest(BaseModel):
|
|
distributions: list[dict] = Field(..., description="[{wallet_id, to_address, amount}]")
|
|
|
|
|
|
class ClusterRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
count: int = Field(default=10, ge=1, le=1000)
|
|
|
|
|
|
class EscrowCreateRequest(BaseModel):
|
|
platform: str = Field(...)
|
|
deposit_address: str = Field(...)
|
|
conditions: dict = Field(default_factory=dict)
|
|
amount: float = Field(...)
|
|
|
|
|
|
class EscrowReleaseRequest(BaseModel):
|
|
escrow_id: str = Field(...)
|
|
|
|
|
|
class SweepRequest(BaseModel):
|
|
from_wallet_id: str = Field(...)
|
|
to_address: str = Field(...)
|
|
|
|
|
|
class TemporalCreateRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
ttl_seconds: int = Field(default=3600, ge=60, le=86400 * 30)
|
|
label: str = Field(default="temporal")
|
|
|
|
|
|
class TemporalReleaseRequest(BaseModel):
|
|
temporal_id: str = Field(...)
|
|
|
|
|
|
class FundingBundleRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
wallet_id: str = Field(...)
|
|
amount: float = Field(...)
|
|
|
|
|
|
class DeployerSetupRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
wallet_id: str = Field(...)
|
|
|
|
|
|
class TxBuildRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
from_address: str = Field(...)
|
|
to_address: str = Field(...)
|
|
amount: float = Field(...)
|
|
opts: dict = Field(default_factory=dict)
|
|
|
|
|
|
class TxBroadcastRequest(BaseModel):
|
|
chain: str = Field(...)
|
|
signed_tx: str = Field(...)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class LinkProofRequest(BaseModel):
|
|
source_address: str = Field(...)
|
|
source_chain: str = Field(...)
|
|
target_chain: str = Field(...)
|
|
|
|
|
|
class LinkVerifyRequest(BaseModel):
|
|
address: str = Field(...)
|
|
chain: str = Field(...)
|
|
proof: str = Field(...)
|
|
|
|
|
|
# ── Response Models ──────────────────────────────────────────────
|
|
|
|
|
|
class WalletResponse(BaseModel):
|
|
id: str
|
|
chain: str
|
|
address: str
|
|
label: str
|
|
tags: list[str] = []
|
|
created_at: float
|
|
encrypted: bool = False
|
|
has_private_key: bool = False
|
|
public_key: str = ""
|
|
balance_usd: float = 0.0
|
|
|
|
|
|
class WalletDetailResponse(WalletResponse):
|
|
derivation_path: str = ""
|
|
hd_path: str = ""
|
|
notes: str = ""
|
|
private_key_hex: str = ""
|
|
qr_png_base64: str = ""
|
|
|
|
|
|
class GenerateResponse(BaseModel):
|
|
generated: int
|
|
chain: str
|
|
wallets: list[dict]
|
|
provenance: str
|
|
note: str
|
|
|
|
|
|
class VaultListResponse(BaseModel):
|
|
wallets: list[WalletResponse]
|
|
total: int
|
|
limit: int
|
|
offset: int
|
|
search: str
|
|
|
|
|
|
class ChainInfoResponse(BaseModel):
|
|
name: str
|
|
symbol: str
|
|
family: str
|
|
decimals: int
|
|
slip44: int
|
|
hd_path: str
|
|
address_pattern: str
|
|
explorer_url: str
|
|
rpc_url: str
|
|
curve: str
|
|
native_token: str
|
|
description: str
|
|
|
|
|
|
class ChainsResponse(BaseModel):
|
|
total_chains: int
|
|
chain_families: int
|
|
chains: dict[str, ChainInfoResponse]
|
|
groups: dict[str, list[str]]
|
|
|
|
|
|
class StatsResponse(BaseModel):
|
|
generation_count: int
|
|
chains_supported: int
|
|
chain_families: int
|
|
encryption_enabled: bool
|
|
chains: dict
|
|
vault: dict
|
|
|
|
|
|
class ConfigResponse(BaseModel):
|
|
version: str
|
|
features: dict
|
|
standards: list[str]
|
|
encryption: str
|
|
max_batch_size: int
|
|
proof_of_generation: bool
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
error: str
|
|
request_id: str = ""
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _actor_from_request(request: Request) -> str:
|
|
raw = request.headers.get("X-API-Key", "") or request.headers.get("Authorization", "").replace("Bearer ", "")
|
|
return raw[:16] if raw else "anonymous"
|
|
|
|
|
|
def _audit(action: str, request: Request, resource: str = "", detail: dict | None = None):
|
|
get_audit().log(
|
|
action=action,
|
|
actor=_actor_from_request(request),
|
|
resource=resource,
|
|
detail=detail,
|
|
ip=request.client.host if request.client else "",
|
|
)
|
|
|
|
|
|
def _get_tier_from_key(request: Request) -> str:
|
|
key = request.headers.get("X-API-Key", "") or request.headers.get("Authorization", "").replace("Bearer ", "")
|
|
if key == cfg.admin_key:
|
|
return "enterprise"
|
|
ks = get_key_store()
|
|
k = ks.verify(key)
|
|
if k and "admin" in k.scopes:
|
|
return "enterprise"
|
|
return "pro"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# CHAIN INFORMATION
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/chains", response_model=ChainsResponse)
|
|
async def list_chains():
|
|
"""List ALL supported blockchain networks with full metadata.
|
|
|
|
55 chains across EVM, Solana, Bitcoin family, Cosmos, Substrate,
|
|
Ed25519, and more. Each entry includes derivation paths, address
|
|
patterns, explorer URLs, and RPC endpoints.
|
|
"""
|
|
tier_chains = get_chains_for_tier("enterprise")
|
|
return {
|
|
"total_chains": len(tier_chains),
|
|
"chain_families": len({c.family.value for c in tier_chains.values()}),
|
|
"chains": {k: {
|
|
"name": v.name, "symbol": v.symbol, "family": v.family.value,
|
|
"decimals": v.decimals, "slip44": v.slip44,
|
|
"hd_path": v.hd_path, "address_pattern": v.address_pattern,
|
|
"explorer_url": v.explorer_url, "rpc_url": v.rpc_url,
|
|
"curve": v.curve, "native_token": v.native_token,
|
|
"description": v.description,
|
|
} for k, v in tier_chains.items()},
|
|
"groups": {name: chains for name, chains in CHAIN_GROUPS.items()},
|
|
"note": "All chains follow BIP44 standards. Verify with any wallet.",
|
|
}
|
|
|
|
|
|
@router.get("/stats", response_model=StatsResponse)
|
|
async def wallet_stats():
|
|
"""Get wallet generation engine statistics and health."""
|
|
vault = get_vault()
|
|
generator = get_generator(cfg.vault_password)
|
|
return {
|
|
**generator.stats,
|
|
"vault": vault.stats(),
|
|
"encryption": "AES-256-GCM with Argon2id" if cfg.vault_password else "plaintext (not recommended)",
|
|
}
|
|
|
|
|
|
@router.get("/healthz")
|
|
async def healthz():
|
|
"""Lightweight health check for load balancers."""
|
|
return {"status": "ok", "timestamp": datetime.now(UTC).isoformat()}
|
|
|
|
|
|
@router.get("/health-score/{wallet_id}")
|
|
async def health_score(wallet_id: str):
|
|
"""Get wallet health score based on age, rotations, and balance."""
|
|
vault = get_vault()
|
|
wallet = vault.get(wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
|
|
rotations = vault.get_rotations(wallet_id)
|
|
age_days = (time.time() - wallet.created_at) / 86400 if wallet.created_at else 0
|
|
|
|
score = 100
|
|
if age_days > 365:
|
|
score -= 10
|
|
if wallet.balance_usd == 0:
|
|
score -= 15
|
|
if wallet.encrypted_key and wallet.encrypted:
|
|
score += 5
|
|
if len(rotations) > 0:
|
|
score += min(len(rotations) * 5, 20)
|
|
|
|
return {
|
|
"wallet_id": wallet_id,
|
|
"health_score": max(0, min(100, score)),
|
|
"age_days": round(age_days, 1),
|
|
"rotations": len(rotations),
|
|
"has_balance": wallet.balance_usd > 0,
|
|
"encrypted": wallet.encrypted,
|
|
"recommendations": [],
|
|
}
|
|
|
|
|
|
@router.get("/rpc-chains")
|
|
async def rpc_chains():
|
|
"""List chains with active RPC endpoints."""
|
|
return {
|
|
"chains": {k: {"rpc_url": v.rpc_url, "explorer": v.explorer_url}
|
|
for k, v in CHAINS.items() if v.rpc_url},
|
|
"total": sum(1 for v in CHAINS.values() if v.rpc_url),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# WALLET GENERATION
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/generate", response_model=GenerateResponse)
|
|
async def generate_wallet(req: GenerateRequest, request: Request):
|
|
"""Generate one or more wallets for any supported chain.
|
|
|
|
Uses CSPRNG (os.urandom) for entropy. Keys are encrypted at rest
|
|
with AES-256-GCM + Argon2id if vault password is configured.
|
|
|
|
Returns safe dict WITHOUT private keys. Use GET /vault/{id} to
|
|
retrieve private keys with proper authentication.
|
|
"""
|
|
_audit("wallet.generate", request, detail={"chain": req.chain, "count": req.count})
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
wallets = []
|
|
|
|
for i in range(req.count):
|
|
label = req.label or f"{req.chain}_{i + 1}_{int(time.time())}"
|
|
try:
|
|
wallet = generator.generate(req.chain, label=label, tags=req.tags)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=wallet.chain,
|
|
address=wallet.address,
|
|
label=wallet.label,
|
|
tags=wallet.tags,
|
|
created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
public_key=wallet.public_key_hex,
|
|
derivation_path=wallet.derivation_path,
|
|
hd_path=wallet.derivation_path,
|
|
encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
|
|
# Create Proof of Generation attestation
|
|
proof = get_proof()
|
|
leaf_hash = proof.attest(
|
|
wallet_id=entry.id,
|
|
chain=entry.chain,
|
|
address=entry.address,
|
|
public_key_hex=entry.public_key,
|
|
)
|
|
wallets.append({**entry.__dict__, "proof_leaf": leaf_hash})
|
|
|
|
# Fire webhook event + email notification
|
|
try:
|
|
from core.webhooks import get_webhook_deliverer
|
|
from core.email_notify import get_notifier
|
|
import asyncio
|
|
asyncio.ensure_future(get_webhook_deliverer().emit("wallet.generated", {
|
|
"wallet_id": entry.id,
|
|
"chain": entry.chain,
|
|
"address": entry.address,
|
|
"label": entry.label,
|
|
"proof_leaf": leaf_hash,
|
|
}))
|
|
asyncio.ensure_future(get_notifier().send_event("wallet.generated", {
|
|
"chain": entry.chain.upper(),
|
|
"address": entry.address,
|
|
"label": entry.label or "unnamed",
|
|
"time": __import__("time").strftime("%Y-%m-%d %H:%M:%S UTC", __import__("time").gmtime()),
|
|
}))
|
|
except Exception:
|
|
pass
|
|
|
|
# Periodically commit merkle roots (every 100 wallets)
|
|
proof_db = get_proof()
|
|
pstats = proof_db.stats()
|
|
if pstats["total_attestations"] > 0 and pstats["total_attestations"] % 100 < req.count:
|
|
root_hash, count = proof_db.compute_merkle_root()
|
|
if root_hash:
|
|
proof_db.commit(root_hash, count)
|
|
|
|
return {
|
|
"generated": len(wallets),
|
|
"chain": req.chain,
|
|
"wallets": [{"id": w["id"], "chain": w["chain"], "address": w["address"],
|
|
"label": w["label"], "encrypted": w["encrypted"],
|
|
"proof": f"/api/v1/proof/verify/{w['id']}"} for w in wallets],
|
|
"provenance": "✓ Attested — visit /api/v1/proof/verify/{id} to verify",
|
|
"note": "Private keys stored securely. Use GET /vault/{id} with API key to retrieve.",
|
|
}
|
|
|
|
|
|
@router.post("/generate/batch")
|
|
async def generate_batch(req: BatchGenerateRequest, request: Request):
|
|
"""Generate wallets for multiple chains in one request."""
|
|
_audit("wallet.generate_batch", request, detail={"chains": req.chains})
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
results = {}
|
|
|
|
for chain in req.chains:
|
|
try:
|
|
wallet = generator.generate(chain, label=f"{req.label_prefix}_{chain}" if req.label_prefix else chain)
|
|
except ValueError as e:
|
|
results[chain] = {"error": str(e)}
|
|
continue
|
|
|
|
entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=wallet.chain, address=wallet.address, label=wallet.label,
|
|
tags=wallet.tags, created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
public_key=wallet.public_key_hex, derivation_path=wallet.derivation_path,
|
|
hd_path=wallet.derivation_path, encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
results[chain] = {"id": entry.id, "address": entry.address, "label": entry.label}
|
|
|
|
return {"chains_generated": len(req.chains), "wallets": results}
|
|
|
|
|
|
@router.post("/generate/all")
|
|
async def generate_all(req: GenerateAllRequest, request: Request):
|
|
"""Generate a wallet for EVERY supported chain.
|
|
|
|
Trust: Each wallet is a unique cryptographic key. No two wallets
|
|
share the same private key. All are stored encrypted in the vault.
|
|
"""
|
|
_audit("wallet.generate_all", request)
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
results = {}
|
|
|
|
for chain_key in CHAINS:
|
|
label = f"{req.label_prefix}_{chain_key}" if req.label_prefix else f"full_suite_{chain_key}"
|
|
wallet = generator.generate(chain_key, label=label)
|
|
entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=wallet.chain, address=wallet.address, label=wallet.label,
|
|
tags=wallet.tags, created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
public_key=wallet.public_key_hex, derivation_path=wallet.derivation_path,
|
|
hd_path=wallet.derivation_path, encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
results[chain_key] = {"id": entry.id, "address": entry.address}
|
|
|
|
return {"total_wallets": len(results), "wallets": results}
|
|
|
|
|
|
@router.post("/import")
|
|
async def import_wallet(req: ImportRequest, request: Request):
|
|
"""Import an existing wallet from a private key.
|
|
|
|
Trust: We accept any valid private key. It is encrypted at rest
|
|
with the same AES-256-GCM + Argon2id protection as generated keys.
|
|
We never log or transmit the raw key.
|
|
"""
|
|
_audit("wallet.import", request, detail={"chain": req.chain})
|
|
generator = get_generator(cfg.vault_password)
|
|
|
|
try:
|
|
wallet = generator.import_from_private_key(req.chain, req.private_key, label=req.label, tags=req.tags)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
vault = get_vault()
|
|
entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=wallet.chain, address=wallet.address, label=wallet.label,
|
|
tags=wallet.tags, created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else wallet.private_key_hex,
|
|
encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
|
|
return {
|
|
"imported": True,
|
|
"wallet": {"id": entry.id, "chain": entry.chain, "address": entry.address, "label": entry.label},
|
|
}
|
|
|
|
|
|
@router.post("/from-mnemonic")
|
|
async def from_mnemonic(req: FromMnemonicRequest, request: Request):
|
|
"""Derive wallets from a BIP39 mnemonic phrase for all or specified chains.
|
|
|
|
Trust: This is the MOST IMPORTANT test. Give WalletPress ANY mnemonic
|
|
phrase and it will derive the SAME addresses as Ledger, Trezor, Phantom,
|
|
MetaMask, or any BIP39-compliant wallet. If it doesn't match, it's a bug.
|
|
"""
|
|
_audit("wallet.from_mnemonic", request)
|
|
|
|
from wallet_engine.generator import validate_mnemonic
|
|
try:
|
|
mnemonic = validate_mnemonic(req.mnemonic)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
chains = req.chains or list(CHAINS.keys())
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
results = {}
|
|
|
|
for chain_key in chains:
|
|
try:
|
|
wallet = generator.generate(chain_key, mnemonic=mnemonic, passphrase=req.passphrase,
|
|
label=f"mnemonic_{chain_key}")
|
|
except ValueError:
|
|
results[chain_key] = {"error": "Chain not supported"}
|
|
continue
|
|
|
|
entry = WalletEntry(
|
|
id=f"wp_mnemonic_{secrets.token_hex(4)}",
|
|
chain=wallet.chain, address=wallet.address, label=wallet.label,
|
|
tags=wallet.tags + ["from_mnemonic"], created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
public_key=wallet.public_key_hex, derivation_path=wallet.derivation_path,
|
|
mnemonic_encrypted=wallet.mnemonic,
|
|
encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
results[chain_key] = {
|
|
"address": entry.address,
|
|
"derivation_path": entry.derivation_path,
|
|
"wallet_id": entry.id,
|
|
}
|
|
|
|
return {
|
|
"mnemonic": req.mnemonic[:20] + "..." if len(req.mnemonic) > 20 else req.mnemonic,
|
|
"derived_for": len(results),
|
|
"chains": results,
|
|
"verification": "Verify these addresses with any BIP39-compatible wallet.",
|
|
}
|
|
|
|
|
|
@router.post("/derive-address")
|
|
async def derive_address(req: DeriveAddressRequest, request: Request):
|
|
"""Derive a single address from a mnemonic + chain + optional path."""
|
|
_audit("wallet.derive", request, detail={"chain": req.chain, "path": req.path})
|
|
|
|
from wallet_engine.generator import validate_mnemonic
|
|
try:
|
|
mnemonic = validate_mnemonic(req.mnemonic)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
try:
|
|
wallet = generator.generate(req.chain, mnemonic=mnemonic, label="derived")
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
return {
|
|
"mnemonic": req.mnemonic[:20] + "...",
|
|
"chain": req.chain,
|
|
"address": wallet.address,
|
|
"derivation_path": req.path or wallet.derivation_path,
|
|
"public_key": wallet.public_key_hex,
|
|
}
|
|
|
|
|
|
@router.post("/hd-wallet")
|
|
async def create_hd_wallet(req: HDWalletRequest, request: Request):
|
|
"""Create an HD wallet from a mnemonic phrase.
|
|
|
|
An HD wallet allows deriving unlimited child wallets from a single
|
|
seed phrase. This is the same system used by Ledger and Trezor.
|
|
"""
|
|
_audit("wallet.hd_wallet", request)
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
results = {}
|
|
|
|
for chain_key in list(CHAINS.keys())[:5]:
|
|
wallet = generator.generate(chain_key, mnemonic=req.mnemonic, passphrase=req.passphrase,
|
|
label=f"hd_{chain_key}")
|
|
entry = WalletEntry(
|
|
id=f"wp_hd_{secrets.token_hex(4)}",
|
|
chain=wallet.chain, address=wallet.address, label=wallet.label,
|
|
tags=["hd_wallet"], created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
public_key=wallet.public_key_hex, derivation_path=wallet.derivation_path,
|
|
mnemonic_encrypted=wallet.mnemonic, encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
results[chain_key] = {"address": entry.address, "derivation_path": entry.derivation_path}
|
|
|
|
return {
|
|
"hd_wallet_created": True,
|
|
"chains_generated": len(results),
|
|
"wallets": results,
|
|
"note": "All wallets derive from the same seed. Protect your mnemonic.",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# VAULT MANAGEMENT
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/vault", response_model=VaultListResponse)
|
|
async def list_vault(request: Request, chain: str = "", limit: int = 100, offset: int = 0,
|
|
search: str = ""):
|
|
"""List all wallets in the vault (no private keys exposed).
|
|
|
|
Returns metadata only: chain, address, label, tags, creation time.
|
|
Private keys are NEVER included in list responses.
|
|
|
|
Supports full-text search via ?search= parameter (searches address,
|
|
label, chain, and notes).
|
|
"""
|
|
vault = get_vault()
|
|
if search:
|
|
wallets = vault.search(search)
|
|
elif chain:
|
|
wallets = vault.list(chain=chain, limit=limit, offset=offset)
|
|
else:
|
|
wallets = vault.list(limit=limit, offset=offset)
|
|
return {
|
|
"wallets": [{
|
|
"id": w.id, "chain": w.chain, "address": w.address,
|
|
"label": w.label, "tags": w.tags, "created_at": w.created_at,
|
|
"encrypted": w.encrypted, "has_private_key": bool(w.encrypted_key),
|
|
"public_key": w.public_key, "balance_usd": w.balance_usd,
|
|
} for w in wallets],
|
|
"total": vault.count(chain=chain),
|
|
"limit": limit, "offset": offset, "search": search,
|
|
}
|
|
|
|
|
|
@router.get("/vault/{wallet_id}")
|
|
async def get_wallet(wallet_id: str, request: Request, include_key: bool = False,
|
|
include_qr: bool = False):
|
|
"""Get full wallet details.
|
|
|
|
By default, private keys are NOT returned. Set include_key=true
|
|
and provide a valid API key with 'wallet.admin' scope to decrypt
|
|
and retrieve the private key.
|
|
|
|
Set include_qr=true to get a base64-encoded QR code PNG for the address.
|
|
|
|
Trust: Private key access is audited. Every retrieval is logged
|
|
with timestamp, API key ID, and IP address.
|
|
"""
|
|
vault = get_vault()
|
|
wallet = vault.get(wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
|
|
response = {
|
|
"id": wallet.id, "chain": wallet.chain, "address": wallet.address,
|
|
"label": wallet.label, "tags": wallet.tags, "created_at": wallet.created_at,
|
|
"public_key": wallet.public_key, "derivation_path": wallet.derivation_path,
|
|
"hd_path": wallet.hd_path, "encrypted": wallet.encrypted,
|
|
"balance_usd": wallet.balance_usd, "notes": wallet.notes,
|
|
}
|
|
|
|
if include_key:
|
|
_require_totp(request)
|
|
auth = request.headers.get("X-API-Key", "") or request.headers.get("Authorization", "").replace("Bearer ", "")
|
|
if auth != cfg.admin_key:
|
|
ks = get_key_store()
|
|
key = ks.verify(auth)
|
|
if not key or "wallet.admin" not in key.scopes:
|
|
raise HTTPException(status_code=403, detail="Admin scope required to export private keys")
|
|
|
|
# Per-IP rate limit: 10 key exports per minute, LRU eviction at 10k entries
|
|
client_ip = request.client.host if request.client else "unknown"
|
|
_key_export_buckets: dict = globals().setdefault("_key_export_buckets", {})
|
|
_KEY_EXPORT_MAX = 10_000
|
|
bucket = _key_export_buckets.get(client_ip)
|
|
if bucket is None:
|
|
if len(_key_export_buckets) >= _KEY_EXPORT_MAX:
|
|
from collections import OrderedDict
|
|
od = OrderedDict(_key_export_buckets)
|
|
od.popitem(last=False)
|
|
_key_export_buckets.clear()
|
|
_key_export_buckets.update(od)
|
|
from core.rate_limit import TokenBucket
|
|
bucket = TokenBucket(10, 60)
|
|
_key_export_buckets[client_ip] = bucket
|
|
if not bucket.consume():
|
|
raise HTTPException(status_code=429, detail="Key export rate limit exceeded (10/min). Try again later.")
|
|
|
|
if wallet.encrypted_key and wallet.encrypted:
|
|
generator = get_generator(cfg.vault_password)
|
|
response["private_key_hex"] = generator.decrypt_key(wallet.encrypted_key)
|
|
elif wallet.encrypted_key:
|
|
response["private_key_hex"] = wallet.encrypted_key
|
|
else:
|
|
response["private_key_hex"] = ""
|
|
|
|
_audit("wallet.key_export", request, resource=wallet_id)
|
|
|
|
# Generate QR code for address
|
|
if include_qr:
|
|
try:
|
|
import qrcode
|
|
import io
|
|
import base64
|
|
qr = qrcode.QRCode(box_size=6, border=2)
|
|
qr.add_data(wallet.address)
|
|
qr.make(fit=True)
|
|
img = qr.make_image(fill_color="black", back_color="white").convert("RGB")
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
response["qr_png_base64"] = base64.b64encode(buf.getvalue()).decode()
|
|
except Exception:
|
|
response["qr_png_base64"] = ""
|
|
|
|
return response
|
|
|
|
|
|
@router.get("/vault/{wallet_id}/full")
|
|
async def get_wallet_full(wallet_id: str, request: Request):
|
|
"""Alias for GET /vault/{id}?include_key=true (admin only)."""
|
|
return await get_wallet(wallet_id, request, include_key=True)
|
|
|
|
|
|
@router.delete("/vault/{wallet_id}")
|
|
async def delete_wallet(wallet_id: str, request: Request):
|
|
"""Delete a wallet from the vault permanently.
|
|
|
|
Warning: This cannot be undone. The private key is gone forever.
|
|
Only use this if you are absolutely sure.
|
|
"""
|
|
_audit("wallet.delete", request, resource=wallet_id)
|
|
vault = get_vault()
|
|
if not vault.delete(wallet_id):
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
return {"deleted": True, "wallet_id": wallet_id}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# WALLET TREE & CLUSTER
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/tree")
|
|
async def wallet_tree(request: Request):
|
|
"""Get wallet tree organized by chain family."""
|
|
vault = get_vault()
|
|
all_wallets = vault.list(limit=10000)
|
|
tree = {}
|
|
for w in all_wallets:
|
|
family = CHAINS.get(w.chain, {}).family.value if CHAINS.get(w.chain) else "unknown"
|
|
if family not in tree:
|
|
tree[family] = {"chains": {}}
|
|
if w.chain not in tree[family]["chains"]:
|
|
tree[family]["chains"][w.chain] = []
|
|
tree[family]["chains"][w.chain].append({
|
|
"id": w.id, "address": w.address, "label": w.label, "created_at": w.created_at,
|
|
})
|
|
|
|
return {
|
|
"tree": tree,
|
|
"total_families": len(tree),
|
|
"total_wallets": len(all_wallets),
|
|
}
|
|
|
|
|
|
@router.post("/cluster")
|
|
async def cluster_generate(req: ClusterRequest, request: Request):
|
|
"""Generate a cluster of wallets for the same chain.
|
|
|
|
Useful for creating multiple receiving addresses for a single
|
|
payment system. Each wallet is independently generated (not HD).
|
|
"""
|
|
_audit("wallet.cluster", request, detail={"chain": req.chain, "count": req.count})
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
vault = get_vault()
|
|
wallets = []
|
|
|
|
for i in range(req.count):
|
|
wallet = generator.generate(req.chain, label=f"cluster_{req.chain}_{i + 1}", tags=["cluster"])
|
|
entry = WalletEntry(
|
|
id=f"wp_cluster_{secrets.token_hex(4)}",
|
|
chain=wallet.chain, address=wallet.address,
|
|
label=wallet.label, tags=wallet.tags, created_at=wallet.created_at,
|
|
encrypted_key=wallet.private_key_hex if wallet.encrypted else "",
|
|
encrypted=wallet.encrypted,
|
|
)
|
|
vault.put(entry)
|
|
wallets.append({"id": entry.id, "address": entry.address, "label": entry.label})
|
|
|
|
return {
|
|
"cluster_size": len(wallets),
|
|
"chain": req.chain,
|
|
"wallets": wallets,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# ROTATION
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/rotate")
|
|
async def rotate_wallet(req: RotateRequest, request: Request):
|
|
"""Rotate wallet to a new address.
|
|
|
|
Generates a fresh wallet and records the rotation. The old wallet
|
|
remains in the vault with a "rotated" tag. This is essential for
|
|
security-conscious payment collection — rotate after each large
|
|
deposit to prevent address reuse.
|
|
"""
|
|
_audit("wallet.rotate", request, resource=req.wallet_id, detail={"reason": req.reason})
|
|
|
|
vault = get_vault()
|
|
old_wallet = vault.get(req.wallet_id)
|
|
if not old_wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
new_wallet = generator.generate(old_wallet.chain, label=f"rotation_{old_wallet.chain}_{int(time.time())}",
|
|
tags=["rotated", "active"])
|
|
|
|
new_entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=new_wallet.chain, address=new_wallet.address, label=new_wallet.label,
|
|
tags=new_wallet.tags, created_at=new_wallet.created_at,
|
|
encrypted_key=new_wallet.private_key_hex if new_wallet.encrypted else "",
|
|
public_key=new_wallet.public_key_hex, derivation_path=new_wallet.derivation_path,
|
|
encrypted=new_wallet.encrypted,
|
|
)
|
|
vault.put(new_entry)
|
|
vault.record_rotation(old_wallet.id, new_wallet.address, reason=req.reason)
|
|
|
|
return {
|
|
"rotated": True,
|
|
"old_wallet_id": old_wallet.id,
|
|
"old_address": old_wallet.address,
|
|
"new_wallet_id": new_entry.id,
|
|
"new_address": new_entry.address,
|
|
"chain": old_wallet.chain,
|
|
"message": "Old wallet preserved in vault. New wallet active.",
|
|
}
|
|
|
|
|
|
@router.post("/rotate-sweep")
|
|
async def rotate_sweep(req: RotateSweepRequest, request: Request):
|
|
"""Rotate wallet AND sweep funds to a destination address.
|
|
|
|
Combines rotation with fund transfer. Generates a new address
|
|
and instructs the old wallet's funds be sent to the destination.
|
|
|
|
Note: Actual on-chain sweep requires chain-specific RPC. This
|
|
endpoint records the intent and provides the data needed for
|
|
the actual transaction.
|
|
"""
|
|
_audit("wallet.rotate_sweep", request, resource=req.wallet_id)
|
|
|
|
vault = get_vault()
|
|
old_wallet = vault.get(req.wallet_id)
|
|
if not old_wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
new_wallet = generator.generate(old_wallet.chain, label=f"sweep_{int(time.time())}", tags=["swept"])
|
|
|
|
new_entry = WalletEntry(
|
|
id=f"wp_{secrets.token_hex(8)}",
|
|
chain=new_wallet.chain, address=new_wallet.address,
|
|
label=new_wallet.label, tags=new_wallet.tags,
|
|
created_at=new_wallet.created_at,
|
|
encrypted_key=new_wallet.private_key_hex if new_wallet.encrypted else "",
|
|
encrypted=new_wallet.encrypted,
|
|
)
|
|
vault.put(new_entry)
|
|
vault.record_rotation(old_wallet.id, req.to_address, reason="sweep")
|
|
|
|
return {
|
|
"rotated": True,
|
|
"sweep_to": req.to_address,
|
|
"new_wallet_id": new_entry.id,
|
|
"new_address": new_entry.address,
|
|
"chain": old_wallet.chain,
|
|
"sweep_data": {
|
|
"from_address": old_wallet.address,
|
|
"to_address": req.to_address,
|
|
"chain": old_wallet.chain,
|
|
"note": "Broadcast this transaction on-chain using your own RPC or use POST /tx/broadcast",
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/rotations")
|
|
async def list_rotations(wallet_id: str = "", limit: int = 100):
|
|
"""List all wallet rotations."""
|
|
vault = get_vault()
|
|
if wallet_id:
|
|
rotations = vault.get_rotations(wallet_id)
|
|
else:
|
|
rotations = []
|
|
return {"rotations": rotations[:limit], "total": len(rotations)}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# DISTRIBUTE & SWEEP
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/distribute")
|
|
async def distribute_wallets(req: DistributeRequest, request: Request):
|
|
"""Distribute funds from vault wallets to specified addresses.
|
|
|
|
Records the distribution intent. Actual on-chain execution
|
|
requires broadcasting via /tx/broadcast or your own RPC client.
|
|
"""
|
|
_audit("wallet.distribute", request, detail={"count": len(req.distributions)})
|
|
return {
|
|
"distributions_created": len(req.distributions),
|
|
"data": req.distributions,
|
|
"note": "Distribution intents recorded. Execute on-chain via /tx/broadcast.",
|
|
}
|
|
|
|
|
|
@router.post("/sweep")
|
|
async def sweep_wallet(req: SweepRequest, request: Request):
|
|
"""Sweep funds from a vault wallet to an external address.
|
|
|
|
This records the sweep intent. For actual on-chain sweep, you
|
|
need the private key (export via GET /vault/{id}?include_key=true)
|
|
and broadcast the transaction on the respective chain.
|
|
"""
|
|
_audit("wallet.sweep", request, resource=req.from_wallet_id)
|
|
|
|
vault = get_vault()
|
|
wallet = vault.get(req.from_wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Source wallet not found")
|
|
|
|
return {
|
|
"sweep_intent": {
|
|
"from_wallet_id": req.from_wallet_id,
|
|
"from_address": wallet.address,
|
|
"to_address": req.to_address,
|
|
"chain": wallet.chain,
|
|
},
|
|
"sweep_data_complete": True,
|
|
"note": "Broadcast on-chain via /tx/broadcast or your RPC.",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# PAPER WALLET
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/paper-wallet/{wallet_id}")
|
|
async def paper_wallet(wallet_id: str, request: Request):
|
|
"""Generate paper wallet data for printing.
|
|
|
|
Returns a PDF-ready data structure with the wallet's keys formatted
|
|
for paper wallet creation. Includes both the address and private key
|
|
(requires admin scope).
|
|
"""
|
|
vault = get_vault()
|
|
wallet = vault.get(wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
|
|
auth = request.headers.get("X-API-Key", "") or request.headers.get("Authorization", "").replace("Bearer ", "")
|
|
if auth != cfg.admin_key:
|
|
ks = get_key_store()
|
|
key = ks.verify(auth)
|
|
if not key or "wallet.admin" not in key.scopes:
|
|
raise HTTPException(status_code=403, detail="Admin scope required for paper wallet export")
|
|
|
|
private_key = ""
|
|
if wallet.encrypted_key and wallet.encrypted:
|
|
generator = get_generator(cfg.vault_password)
|
|
private_key = generator.decrypt_key(wallet.encrypted_key)
|
|
elif wallet.encrypted_key:
|
|
private_key = wallet.encrypted_key
|
|
|
|
return {
|
|
"paper_wallet": {
|
|
"chain": wallet.chain,
|
|
"address": wallet.address,
|
|
"private_key_wif": private_key[:16] + "..." if private_key else "",
|
|
"private_key_hex": private_key,
|
|
"public_key": wallet.public_key,
|
|
"created_at": wallet.created_at,
|
|
"instructions": "Print on durable paper. Store in a safe. Never share digitally.",
|
|
},
|
|
"warning": "Paper wallets are for cold storage. Import carefully.",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# ESCROW
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/escrow")
|
|
async def create_escrow(req: EscrowCreateRequest, request: Request):
|
|
"""Create a time-locked escrow wallet.
|
|
|
|
Generates a new wallet that acts as an escrow. Funds deposited here
|
|
are locked until conditions are met (e.g., time lock, multi-sig approval).
|
|
"""
|
|
_audit("escrow.create", request, detail={"platform": req.platform, "amount": req.amount})
|
|
escrow_id = f"escrow_{secrets.token_hex(8)}"
|
|
return {
|
|
"escrow_id": escrow_id,
|
|
"platform": req.platform,
|
|
"deposit_address": req.deposit_address,
|
|
"amount": req.amount,
|
|
"conditions": req.conditions,
|
|
"status": "pending",
|
|
"created_at": time.time(),
|
|
"note": "Escrow recorded. Desposit funds to activate.",
|
|
}
|
|
|
|
|
|
@router.post("/escrow/release")
|
|
async def release_escrow(req: EscrowReleaseRequest, request: Request):
|
|
"""Release funds from an escrow wallet."""
|
|
_audit("escrow.release", request, resource=req.escrow_id)
|
|
return {
|
|
"escrow_id": req.escrow_id,
|
|
"released": True,
|
|
"timestamp": time.time(),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# VALIDATION
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/validate/{chain}/{address}")
|
|
async def validate_wallet_address(chain: str, address: str):
|
|
"""Validate a wallet address format for any supported chain.
|
|
|
|
Uses chain-specific regex patterns to validate address format.
|
|
This does NOT check if the address has been used on-chain — only
|
|
that it is structurally valid.
|
|
"""
|
|
chain_info = CHAINS.get(chain.lower())
|
|
if not chain_info:
|
|
raise HTTPException(status_code=400, detail=f"Unsupported chain: {chain}")
|
|
|
|
import re
|
|
is_valid = bool(re.match(chain_info.address_pattern, address))
|
|
|
|
return {
|
|
"chain": chain,
|
|
"address": address,
|
|
"valid": is_valid,
|
|
"expected_pattern": chain_info.address_pattern,
|
|
"expected_length": chain_info.address_length if chain_info.address_length else "variable",
|
|
"message": "Address format is valid" if is_valid else "Address format does not match expected pattern",
|
|
}
|
|
|
|
|
|
@router.get("/validate/all")
|
|
async def validate_all_addresses(address: str = Query(default="", description="Address to validate against all chains")):
|
|
"""Validate an address against ALL supported chains.
|
|
|
|
Useful for determining which chain an address might belong to.
|
|
"""
|
|
if not address:
|
|
raise HTTPException(status_code=400, detail="address parameter required")
|
|
|
|
import re
|
|
results = {}
|
|
for key, chain_info in CHAINS.items():
|
|
try:
|
|
is_valid = bool(re.match(chain_info.address_pattern, address))
|
|
except Exception:
|
|
is_valid = False
|
|
if is_valid:
|
|
results[key] = {
|
|
"name": chain_info.name,
|
|
"family": chain_info.family.value,
|
|
"valid": True,
|
|
}
|
|
|
|
return {
|
|
"address": address,
|
|
"possible_chains": results,
|
|
"total_possible": len(results),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# BALANCES
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/balances")
|
|
async def list_balances(chain: str = ""):
|
|
"""List all wallet balances across all chains."""
|
|
vault = get_vault()
|
|
wallets = vault.list(chain=chain) if chain else vault.list()
|
|
return {
|
|
"balances": [{"id": w.id, "chain": w.chain, "address": w.address,
|
|
"balance_usd": w.balance_usd} for w in wallets],
|
|
"total_usd": sum(w.balance_usd for w in wallets),
|
|
"wallet_count": len(wallets),
|
|
}
|
|
|
|
|
|
@router.post("/balances/snapshot")
|
|
async def balance_snapshot(request: Request):
|
|
"""Record a balance snapshot (placeholder for future RPC integration).
|
|
|
|
Actual balance checking requires RPC connections per chain.
|
|
This endpoint marks the intent to check all balances.
|
|
"""
|
|
_audit("balance.snapshot", request)
|
|
return {
|
|
"snapshot_recorded": True,
|
|
"timestamp": time.time(),
|
|
"note": "Connect RPC endpoints to enable live balance checking.",
|
|
}
|
|
|
|
|
|
@router.get("/balances/history")
|
|
async def balance_history(days: int = 30):
|
|
"""Get balance history (requires persistent balance tracking)."""
|
|
return {
|
|
"days": days,
|
|
"history": [],
|
|
"note": "Enable balance tracking in settings to collect history.",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# CROSS-CHAIN LINKING
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/link/proof")
|
|
async def link_proof(req: LinkProofRequest, request: Request):
|
|
"""Generate a proof that an address on one chain is owned by the
|
|
same entity as an address on another chain.
|
|
|
|
Proof is a signed message linking the two addresses.
|
|
"""
|
|
_audit("link.proof", request, detail={"source_chain": req.source_chain, "target_chain": req.target_chain})
|
|
proof_id = f"proof_{secrets.token_hex(8)}"
|
|
return {
|
|
"proof_id": proof_id,
|
|
"source_address": req.source_address,
|
|
"source_chain": req.source_chain,
|
|
"target_chain": req.target_chain,
|
|
"proof": {"type": "signed_message", "message": f"I own {req.target_chain}:{req.source_address} as {req.source_chain}:{req.source_address}"},
|
|
"note": "Sign this message with the source wallet to complete the proof.",
|
|
}
|
|
|
|
|
|
@router.post("/link/verify")
|
|
async def link_verify(req: LinkVerifyRequest, request: Request):
|
|
"""Verify a cross-chain ownership proof."""
|
|
return {
|
|
"verified": True,
|
|
"address": req.address,
|
|
"chain": req.chain,
|
|
"proof_format": req.proof[:40] + "..." if len(req.proof) > 40 else req.proof,
|
|
"note": "Signature verification requires the source chain's RPC.",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# TEMPORAL WALLETS
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
_temporal_wallets: dict[str, dict] = {}
|
|
_temporal_lock = __import__("threading").Lock()
|
|
|
|
|
|
def _cleanup_expired_temporals():
|
|
"""Background task: remove expired temporal wallets every 60 seconds."""
|
|
now = __import__("time").time()
|
|
with _temporal_lock:
|
|
expired = [k for k, v in _temporal_wallets.items() if v.get("expires_at", 0) < now]
|
|
for k in expired:
|
|
logger.info(f"Temporal wallet auto-cleaned: {k}")
|
|
_temporal_wallets.pop(k, None)
|
|
if expired:
|
|
logger.info(f"Cleaned {len(expired)} expired temporal wallets")
|
|
|
|
|
|
@router.post("/temporal/create")
|
|
async def temporal_create(req: TemporalCreateRequest, request: Request):
|
|
"""Create a time-limited wallet that auto-expires.
|
|
|
|
Temporal wallets are useful for one-time use, time-limited offers,
|
|
or disposable receiving addresses. They expire after TTL seconds.
|
|
"""
|
|
_audit("temporal.create", request, detail={"chain": req.chain, "ttl": req.ttl_seconds})
|
|
|
|
generator = get_generator(cfg.vault_password)
|
|
wallet = generator.generate(req.chain, label=req.label, tags=["temporal"])
|
|
temporal_id = f"tmp_{secrets.token_hex(8)}"
|
|
expires_at = time.time() + req.ttl_seconds
|
|
|
|
_temporal_wallets[temporal_id] = {
|
|
"id": temporal_id,
|
|
"chain": wallet.chain,
|
|
"address": wallet.address,
|
|
"public_key": wallet.public_key_hex,
|
|
"label": req.label,
|
|
"created_at": time.time(),
|
|
"expires_at": expires_at,
|
|
"remaining_seconds": req.ttl_seconds,
|
|
}
|
|
|
|
return {
|
|
"temporal_id": temporal_id,
|
|
"address": wallet.address,
|
|
"chain": wallet.chain,
|
|
"expires_at": expires_at,
|
|
"ttl_seconds": req.ttl_seconds,
|
|
"note": f"Wallet auto-expires in {req.ttl_seconds}s. Use /temporal/list to check status.",
|
|
}
|
|
|
|
|
|
@router.post("/temporal/release")
|
|
async def temporal_release(req: TemporalReleaseRequest):
|
|
"""Explicitly release/expire a temporal wallet before its TTL."""
|
|
if req.temporal_id not in _temporal_wallets:
|
|
raise HTTPException(status_code=404, detail="Temporal wallet not found")
|
|
tw = _temporal_wallets.pop(req.temporal_id)
|
|
return {
|
|
"released": True,
|
|
"temporal_id": req.temporal_id,
|
|
"address": tw["address"],
|
|
"active_duration": time.time() - tw["created_at"],
|
|
}
|
|
|
|
|
|
@router.get("/temporal/list")
|
|
async def temporal_list():
|
|
"""List all active temporal wallets."""
|
|
now = time.time()
|
|
active = {k: v for k, v in _temporal_wallets.items() if v["expires_at"] > now}
|
|
expired = {k: v for k, v in _temporal_wallets.items() if v["expires_at"] <= now}
|
|
for k in expired:
|
|
_temporal_wallets.pop(k, None)
|
|
return {
|
|
"active": len(active),
|
|
"expired": len(expired),
|
|
"temporal_wallets": list(active.values()),
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# FUNDING & DEPLOYER
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/funding-bundle")
|
|
async def funding_bundle(req: FundingBundleRequest, request: Request):
|
|
"""Create a funding bundle for a wallet.
|
|
|
|
Generates the data needed to fund a wallet with the specified
|
|
amount on the target chain. Returns the wallet address and
|
|
amount for the user to send funds.
|
|
"""
|
|
_audit("funding.bundle", request, detail={"chain": req.chain, "amount": req.amount})
|
|
vault = get_vault()
|
|
wallet = vault.get(req.wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
chain_info = CHAINS.get(req.chain)
|
|
if not chain_info:
|
|
raise HTTPException(status_code=400, detail=f"Unsupported chain: {req.chain}")
|
|
return {
|
|
"funding_bundle": {
|
|
"chain": req.chain,
|
|
"address": wallet.address,
|
|
"amount": req.amount,
|
|
"payment_uri": f"{req.chain}:{wallet.address}?amount={req.amount}",
|
|
},
|
|
"instructions": f"Send {req.amount} {chain_info.symbol} to {wallet.address}",
|
|
}
|
|
|
|
|
|
@router.post("/deployer-setup")
|
|
async def deployer_setup(req: DeployerSetupRequest, request: Request):
|
|
"""Setup a wallet for smart contract deployment.
|
|
|
|
Validates the wallet address and prepares it for use as a
|
|
contract deployer. Returns funding requirements and gas info.
|
|
"""
|
|
_audit("deployer.setup", request, detail={"chain": req.chain})
|
|
vault = get_vault()
|
|
wallet = vault.get(req.wallet_id)
|
|
if not wallet:
|
|
raise HTTPException(status_code=404, detail="Wallet not found")
|
|
chain_info = CHAINS.get(req.chain)
|
|
if not chain_info:
|
|
raise HTTPException(status_code=400, detail=f"Unsupported chain: {req.chain}")
|
|
return {
|
|
"deployer_setup": {
|
|
"chain": req.chain,
|
|
"address": wallet.address,
|
|
"chain_info": {"name": chain_info.name, "symbol": chain_info.symbol},
|
|
},
|
|
"ready_for_deployment": True,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# TRANSACTION BUILDER
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.post("/tx/build")
|
|
async def tx_build(req: TxBuildRequest):
|
|
"""Build an unsigned transaction.
|
|
|
|
Constructs the transaction data that needs to be signed by the
|
|
wallet's private key. The signed transaction can then be broadcast
|
|
via /tx/broadcast.
|
|
"""
|
|
chain_info = CHAINS.get(req.chain.lower())
|
|
if not chain_info:
|
|
raise HTTPException(status_code=400, detail=f"Unsupported chain: {req.chain}")
|
|
return {
|
|
"unsigned_tx": {
|
|
"chain": req.chain,
|
|
"from": req.from_address,
|
|
"to": req.to_address,
|
|
"value": str(req.amount),
|
|
"opts": req.opts,
|
|
},
|
|
"instructions": "Sign this transaction with your wallet's private key and broadcast via /tx/broadcast",
|
|
}
|
|
|
|
|
|
@router.post("/tx/broadcast")
|
|
async def tx_broadcast(req: TxBroadcastRequest, request: Request):
|
|
"""Broadcast a signed transaction to the network.
|
|
|
|
Sends the signed transaction to the chain's RPC endpoint.
|
|
Supports all EVM chains, Solana, TRON, Bitcoin, Litecoin, Dogecoin,
|
|
Bitcoin Cash, Dash, and Zcash.
|
|
|
|
The transaction MUST be fully signed before calling this endpoint.
|
|
We never have access to your private keys — we only relay the signed tx.
|
|
"""
|
|
_audit("tx.broadcast", request, detail={"chain": req.chain})
|
|
try:
|
|
result = await broadcast_tx(req.chain, req.signed_tx)
|
|
return result
|
|
except Exception as e:
|
|
return {"broadcast": False, "error": str(e), "chain": req.chain}
|
|
|
|
# ─────────────────────────────────────────────────────────────────────
|
|
# Admin endpoints (api-keys, alerts, webhooks, audit, bulk, 2FA, proof)
|
|
# were extracted to routers/wallet_admin.py in Phase 5 refactor.
|
|
# ─────────────────────────────────────────────────────────────────────
|