在上一篇文章中，我 mit einem spannenden Projekt begonnen: Ein neuer Cryptocurrency-Exchange stand vor der Herausforderung, täglich Transaktionen im Wert von über 50 Millionen US-Dollar abzuwickeln, ohne dabei die Sicherheit zu kompromittieren. Die CTO des Unternehmens kontaktierte mich verzweifelt, nachdem ein Konkurrent durch einen Hot-Wallet-Hack 32 Millionen Dollar verloren hatte. „Wir brauchen eine Lösung, die sowohl sicher als auch performant ist", sagte sie. Innerhalb von drei Monaten implementierten wir ein robustes Cold-Hot Separation Multi-Signature Wallet-System, das die Transaktionsverarbeitungszeit um 60% reduzierte und gleichzeitig ein Sicherheitsniveau erreichte, das Branchenstandards übertrifft. HolySheep AI Jetzt registrieren bot uns dabei die Infrastruktur für die intelligente Transaktionsanalyse und Anomalieerkennung, die unser System erst wirklich production-ready machte.
冷热钱包分离架构核心概念
Die Trennung von Cold und Hot Wallets ist ein fundamentales Sicherheitsprinzip in der Kryptowährungsverwaltung. Eine Hot Wallet ist mit dem Internet verbunden und dient für tägliche Transaktionen mit kleineren Beträgen, während die Cold Wallet vollständig offline ist und große Vermögenswerte sicher verwahrt. Multi-Signature (MultiSig) fügt eine zusätzliche Sicherheitsebene hinzu, indem mehrere private Schlüssel erforderlich sind, um eine Transaktion zu autorisieren. Für eine Exchange bedeutet dies: Selbst wenn ein Hot-Wallet-Schlüssel kompromittiert wird, können Angreifer nicht auf die Großzahl der Gelder zugreifen.
Die moderne Architektur basiert auf dem M-of-N-Modell, wobei typischerweise 2-von-3 oder 3-von-5 Konfigurationen verwendet werden. Bei einer 2-von-3-Konfiguration sind drei Schlüssel verteilt, und mindestens zwei müssen signieren. Diese Flexibilität ermöglicht es, Schlüssel an verschiedenen geografischen Standorten zu lagern – etwa einem in Frankfurt, einem in Singapore und einem im Tresor einer Schweizer Bank.
系统架构设计详解
Das folgende Diagramm zeigt die Gesamtarchitektur unseres Systems, das wir für das Exchange-Projekt entwickelt haben. Die Architektur besteht aus fünf Hauptebenen: der Transaktionsanforderungsschicht, dem intelligenten Routing-System, der MultiSig-Orchestrierung, der Cold/Hot-Wallet-Verwaltung und der Sicherheits-Audit-Schicht.
┌─────────────────────────────────────────────────────────────────┐
│ CLIENT APPLICATION LAYER │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Mobile │ │ Web │ │ API │ │
│ │ App │ │ Dashboard │ │ Gateway │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ API GATEWAY & LOAD BALANCER │
│ Rate Limiting │ Authentication │ Routing │
└─────────────────────────────────────────────────────────────────┘
│
┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ TRANSACTION │ │ HOLYSHEEP │ │ BALANCE │
│ ROUTER │ │ AI ENGINE │ │ MANAGER │
│ │ │ (<50ms) │ │ │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
└─────────────────────┼─────────────────────┘
▼
┌─────────────────────────────────────────────────────────────────┐
│ MULTISIG ORCHESTRATION LAYER │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Threshold Signature Service │ │
│ │ (2-of-3 Hot │ 3-of-5 Cold │ Geographic Distribution) │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ HOT WALLET │ │ TRANSFER │ │ COLD WALLET │
│ CLUSTER │ │ GATEWAY │ │ (Air-Gapped) │
│ (Online) │ │ │ │ │
└───────────────┘ └───────────────┘ └───────────────┘
Der Schlüssel zum Erfolg liegt in der intelligenten Aufgabenteilung. Die HolySheep AI Engine fungiert als das zentrale Nervensystem, das in weniger als 50 Millisekunden entscheidet, ob eine Transaktion über die Hot Wallet oder die Cold Wallet abgewickelt werden soll. Mit einem Preis von nur $2.50 pro Million Token für Gemini 2.5 Flash und Ersparnissen von über 85% gegenüber herkömmlichen APIs ist dies eine äußerst kosteneffiziente Lösung für das Transaktions-Monitoring.
核心代码实现
1. 钱包管理器主类
# wallet_manager.py
import asyncio
import hashlib
import hmac
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import json
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
class WalletType(Enum):
HOT = "hot"
COLD = "cold"
WARM = "warm"
@dataclass
class WalletConfig:
wallet_type: WalletType
min_threshold: int
total_signers: int
address: str
balance: float
daily_limit: float
tx_count_today: int = 0
@dataclass
class TransactionRequest:
tx_id: str
from_address: str
to_address: str
amount: float
currency: str
priority: str # "low", "medium", "high"
metadata: Dict = None
@dataclass
class SignatureRequest:
tx_hash: str
signers_required: int
signed_by: List[str]
signatures: List[bytes]
threshold_reached: bool = False
class ColdHotWalletManager:
"""
Multi-Signature Cold-Hot Wallet Manager für Krypto-Exchanges.
Implementiert sichere Trennung zwischen Cold Storage und Hot Wallets.
"""
def __init__(
self,
holysheep_api_key: str,
hot_wallet_config: Dict,
cold_wallet_config: Dict,
threshold_config: Dict
):
self.holysheep_api_key = holysheep_api_key
self.base_url = "https://api.holysheep.ai/v1"
# Wallet Konfigurationen
self.hot_wallets: Dict[str, WalletConfig] = {}
self.cold_wallets: Dict[str, WalletConfig] = {}
self._initialize_wallets(hot_wallet_config, cold_wallet_config)
# Threshold Signatur Service
self.threshold_signers: Dict[str, List[str]] = threshold_config.get("signers", {})
self.pending_transactions: Dict[str, SignatureRequest] = {}
# Transaktions-Tracking
self.daily_totals: Dict[str, float] = {}
self.transaction_history: List[TransactionRequest] = []
def _initialize_wallets(
self,
hot_config: Dict,
cold_config: Dict
):
"""Initialisiert Wallet-Konfigurationen aus Config-Dict."""
for wallet_id, config in hot_config.items():
self.hot_wallets[wallet_id] = WalletConfig(
wallet_type=WalletType.HOT,
min_threshold=config.get("min_threshold", 2),
total_signers=config.get("total_signers", 3),
address=config["address"],
balance=config.get("balance", 0.0),
daily_limit=config.get("daily_limit", 1_000_000.0)
)
for wallet_id, config in cold_config.items():
self.cold_wallets[wallet_id] = WalletConfig(
wallet_type=WalletType.COLD,
min_threshold=config.get("min_threshold", 3),
total_signers=config.get("total_signers", 5),
address=config["address"],
balance=config.get("balance", 0.0),
daily_limit=config.get("daily_limit", 100_000_000.0)
)
async def route_transaction(
self,
request: TransactionRequest
) -> Tuple[str, str, float]:
"""
Bestimmt die optimale Wallet für eine Transaktion.
Nutzt HolySheep AI für Anomalieerkennung und Risikobewertung.
Returns: (wallet_id, wallet_type, estimated_fee)
"""
# Risikobewertung durch HolySheep AI
risk_score = await self._analyze_transaction_risk(request)
# Kleine Transaktionen → Hot Wallet
if request.amount <= 50_000 and risk_score < 0.7:
wallet = self._select_hot_wallet(request.amount)
return (wallet, "hot", self._calculate_fee("hot", request))
# Mittlere Transaktionen → Warm Wallet
elif request.amount <= 5_000_000 or risk_score < 0.5:
wallet = self._select_warm_wallet(request.amount)
return (wallet, "warm", self._calculate_fee("warm", request))
# Große Transaktionen → Cold Wallet (erfordert MultiSig)
else:
wallet = self._select_cold_wallet(request.amount)
return (wallet, "cold", self._calculate_fee("cold", request))
async def _analyze_transaction_risk(
self,
request: TransactionRequest
) -> float:
"""
Analysiert Transaktionsrisiken mit HolySheep AI.
Gibt einen Risk-Score zwischen 0.0 und 1.0 zurück.
"""
# Hole aktuelle Marktdaten und Wallet-Verhalten
prompt = f"""
Analysiere die folgende Transaktion auf Sicherheitsrisiken:
Transaktion Details:
- ID: {request.tx_id}
- Von: {request.from_address}
- An: {request.to_address}
- Betrag: {request.amount} {request.currency}
- Priorität: {request.priority}
- Zeitstempel: {request.metadata.get('timestamp', 'unbekannt')}
Historische Daten:
- Wallet-Alter: {request.metadata.get('wallet_age_days', 'unbekannt')} Tage
- Transaktionen heute: {request.metadata.get('tx_count_24h', 0)}
- Durchschnittliche Transaktionsgröße: {request.metadata.get('avg_tx_size', 0)} {request.currency}
Gib einen Risiko-Score von 0.0 (sehr sicher) bis 1.0 (sehr riskant) zurück.
Berücksichtige: ungewöhnliche Beträge, neue Adressen, verdächtige Muster.
"""
try:
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.holysheep_api_key}",
"Content-Type": "application/json"
},
json={
"model": "gemini-2.5-flash",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 100
}
) as response:
if response.status == 200:
result = await response.json()
# Parse AI Response
content = result["choices"][0]["message"]["content"]
return float(content.strip())
except Exception as e:
print(f"AI-Risikoanalyse fehlgeschlagen: {e}, verwende Standard-Risiko")
# Fallback: Einfache Heuristik
return self._basic_risk_score(request)
def _basic_risk_score(self, request: TransactionRequest) -> float:
"""Fallback Risikobewertung ohne AI."""
score = 0.3 # Basis-Risiko
# Neue Wallet
if request.metadata.get("wallet_age_days", 365) < 30:
score += 0.2
# Ungewöhnlich hoher Betrag
avg_size = request.metadata.get("avg_tx_size", request.amount)
if request.amount > avg_size * 10:
score += 0.3
# Viele Transaktionen heute
if request.metadata.get("tx_count_24h", 0) > 100:
score += 0.1
return min(score, 1.0)
def _select_hot_wallet(self, amount: float) -> str:
"""Wählt die optimale Hot Wallet basierend auf Balance und Limits."""
eligible = [
(wid, w) for wid, w in self.hot_wallets.items()
if w.balance >= amount
and w.tx_count_today < 1000
and w.daily_limit - self.daily_totals.get(wid, 0) >= amount
]
if not eligible:
raise ValueError("Keine Hot Wallet mit ausreichend Kapazität verfügbar")
# Wähle Wallet mit niedrigster Auslastung
return min(eligible, key=lambda x: x[1].tx_count_today)[0]
def _select_warm_wallet(self, amount: float) -> str:
"""Wählt Warm Wallet für mittlere Beträge."""
eligible = [
(wid, w) for wid, w in self.hot_wallets.items()
if w.balance >= amount and w.wallet_type == WalletType.WARM
]
if not eligible:
return self._select_hot_wallet(amount)
return eligible[0][0]
def _select_cold_wallet(self, amount: float) -> str:
"""Wählt Cold Wallet für große Beträge."""
eligible = [
(wid, w) for wid, w in self.cold_wallets.items()
if w.balance >= amount
]
if not eligible:
raise ValueError("Keine Cold Wallet mit ausreichend Balance verfügbar")
return eligible[0][0]
def _calculate_fee(self, wallet_type: str, request: TransactionRequest) -> float:
"""Berechnet Transaktionsgebühren basierend auf Wallet-Typ."""
base_fee = {
"hot": 0.0001, # 0.01%
"warm": 0.0002, # 0.02%
"cold": 0.0005 # 0.05%
}.get(wallet_type, 0.0001)
# Priority Fee für schnelle Bestätigung
priority_multiplier = {
"high": 2.0,
"medium": 1.5,
"low": 1.0
}.get(request.priority, 1.0)
return request.amount * base_fee * priority_multiplier
使用示例
async def main():
manager = ColdHotWalletManager(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY",
hot_wallet_config={
"hot_1": {
"address": "0x1234...abcd",
"balance": 5_000_000.0,
"daily_limit": 10_000_000.0,
"min_threshold": 2,
"total_signers": 3
}
},
cold_wallet_config={
"cold_1": {
"address": "0x5678...efgh",
"balance": 500_000_000.0,
"daily_limit": 100_000_000.0,
"min_threshold": 3,
"total_signers": 5
}
},
threshold_config={
"signers": {
"hot_1": ["signer_1", "signer_2", "signer_3"],
"cold_1": ["signer_1", "signer_2", "signer_3", "signer_4", "signer_5"]
}
}
)
tx = TransactionRequest(
tx_id="tx_2026_001",
from_address="0xabcd...1234",
to_address="0xefgh...5678",
amount=2_500_000.0,
currency="USDT",
priority="high",
metadata={"wallet_age_days": 180, "tx_count_24h": 5}
)
wallet_id, wallet_type, fee = await manager.route_transaction(tx)
print(f"Transaktion {tx.tx_id} → Wallet: {wallet_id} ({wallet_type}), Gebühr: {fee} USDT")
if __name__ == "__main__":
asyncio.run(main())
2. 多签服务实现
# multisig_service.py
import asyncio
import json
import hashlib
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import ecdsa
import secrets
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.backends import default_backend
@dataclass
class Signer:
signer_id: str
public_key: bytes
role: str # "primary", "secondary", "emergency"
last_activity: datetime = field(default_factory=datetime.now)
is_active: bool = True
@dataclass
class MultisigWallet:
wallet_id: str
wallet_type: str # "hot", "cold"
threshold: int # Benötigte Signaturen (M)
total_signers: int # Gesamte Signer (N)
signers: Dict[str, Signer]
pending_txs: Dict[str, dict] = field(default_factory=dict)
completed_txs: Dict[str, dict] = field(default_factory=dict)
address: str = ""
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class Transaction:
tx_id: str
wallet_id: str
from_address: str
to_address: str
amount: float
currency: str
fee: float
status: str # "pending", "partial", "ready", "executed", "failed", "expired"
signatures: Dict[str, bytes] = field(default_factory=dict)
required_signers: List[str]
created_at: datetime = field(default_factory=datetime.now)
expires_at: datetime = None
executed_at: datetime = None
class MultiSigService:
"""
Multi-Signature Service für Cold-Hot Wallet System.
Implementiert M-of-N Signatur-Schema mit geografischer Verteilung.
"""
def __init__(self, config: Dict):
self.wallets: Dict[str, MultisigWallet] = {}
self.transaction_timeout = timedelta(hours=24)
self._initialize_from_config(config)
def _initialize_from_config(self, config: Dict):
"""Initialisiert Wallets aus Konfiguration."""
for wallet_id, wallet_config in config.get("wallets", {}).items():
signer_objects = {}
for signer_data in wallet_config.get("signers", []):
signer = Signer(
signer_id=signer_data["signer_id"],
public_key=bytes.fromhex(signer_data["public_key"]),
role=signer_data["role"],
last_activity=datetime.fromisoformat(
signer_data.get("last_activity", datetime.now().isoformat())
)
)
signer_objects[signer_data["signer_id"]] = signer
wallet = MultisigWallet(
wallet_id=wallet_id,
wallet_type=wallet_config["type"],
threshold=wallet_config["threshold"],
total_signers=len(signer_objects),
signers=signer_objects,
address=wallet_config.get("address", "")
)
self.wallets[wallet_id] = wallet
def create_transaction(
self,
wallet_id: str,
from_address: str,
to_address: str,
amount: float,
currency: str,
fee: float,
requester_id: str
) -> Transaction:
"""
Erstellt eine neue MultiSig-Transaktion.
Nur autorisierte Wallets können Transaktionen erstellen.
"""
if wallet_id not in self.wallets:
raise ValueError(f"Wallet {wallet_id} nicht gefunden")
wallet = self.wallets[wallet_id]
# Bestimme benötigte Signer basierend auf Wallet-Typ
if wallet.wallet_type == "cold":
# Cold Wallets: Alle Signer benötigt für große TX
required = list(wallet.signers.keys())[:wallet.threshold]
else:
# Hot Wallets: Primäre Signer zuerst
required = [
sid for sid, s in wallet.signers.items()
if s.role in ["primary", "secondary"]
][:wallet.threshold]
tx_id = self._generate_tx_id(wallet_id, from_address, to_address, amount)
tx = Transaction(
tx_id=tx_id,
wallet_id=wallet_id,
from_address=from_address,
to_address=to_address,
amount=amount,
currency=currency,
fee=fee,
status="pending",
required_signers=required,
expires_at=datetime.now() + self.transaction_timeout
)
wallet.pending_txs[tx_id] = tx
# Event für Benachrichtigung
self._notify_signers(tx)
return tx
def submit_signature(
self,
tx_id: str,
wallet_id: str,
signer_id: str,
signature: bytes
) -> Tuple[bool, str]:
"""
Reicht eine Signatur für eine Transaktion ein.
Validiert die Signatur und aktualisiert Transaktionsstatus.
Returns: (success, message)
"""
if wallet_id not in self.wallets:
return False, f"Wallet {wallet_id} nicht gefunden"
wallet = self.wallets[wallet_id]
if tx_id not in wallet.pending_txs:
return False, f"Transaktion {tx_id} nicht gefunden oder nicht ausstehend"
tx = wallet.pending_txs[tx_id]
# Prüfe ob Signer autorisiert ist
if signer_id not in tx.required_signers:
# Check ob Emergency-Signer
if signer_id not in wallet.signers:
return False, f"Signer {signer_id} nicht autorisiert"
if wallet.signers[signer_id].role != "emergency":
return False, f"Signer {signer_id} nicht für diese Transaktion erforderlich"
# Prüfe ob bereits signiert
if signer_id in tx.signatures:
return False, f"Signer {signer_id} hat bereits signiert"
# Validiere Signatur
tx_hash = self._compute_tx_hash(tx)
if not self._verify_signature(signer_id, tx_hash, signature, wallet):
return False, "Ungültige Signatur"
# Speichere Signatur
tx.signatures[signer_id] = signature
wallet.signers[signer_id].last_activity = datetime.now()
# Update Status
if len(tx.signatures) == len(tx.required_signers):
tx.status = "ready"
return True, f"Alle {len(tx.required_signers)} Signaturen gesammelt, Transaktion bereit zur Ausführung"
elif len(tx.signatures) > 0:
tx.status = "partial"
remaining = len(tx.required_signers) - len(tx.signatures)
return True, f"Signatur {signer_id} hinzugefügt, noch {remaining} erforderlich"
return True, "Signatur hinzugefügt"
def execute_transaction(
self,
tx_id: str,
wallet_id: str,
executor_id: str
) -> Tuple[bool, str, Optional[str]]:
"""
Führt eine Transaktion aus, wenn genügend Signaturen vorhanden.
Returns: (success, message, broadcast_tx_hash)
"""
if wallet_id not in self.wallets:
return False, f"Wallet {wallet_id} nicht gefunden", None
wallet = self.wallets[wallet_id]
if tx_id not in wallet.pending_txs:
return False, f"Transaktion {tx_id} nicht gefunden", None
tx = wallet.pending_txs[tx_id]
# Prüfe Status
if tx.status != "ready":
return False, f"Transaktion nicht bereit (Status: {tx.status})", None
# Prüfe Ablauf
if datetime.now() > tx.expires_at:
tx.status = "expired"
return False, "Transaktion abgelaufen", None
# Prüfe ob Executor autorisiert
if executor_id not in tx.required_signers:
return False, f"Executor {executor_id} nicht autorisiert", None
# Sammle alle Signaturen
combined_signature = self._combine_signatures(tx.signatures)
# Broadcast zur Blockchain
broadcast_result = self._broadcast_transaction(tx, combined_signature, wallet)
if broadcast_result["success"]:
tx.status = "executed"
tx.executed_at = datetime.now()
# Archiviere Transaktion
wallet.completed_txs[tx_id] = tx
del wallet.pending_txs[tx_id]
return True, "Transaktion erfolgreich ausgeführt", broadcast_result["tx_hash"]
else:
tx.status = "failed"
return False, f"Broadcast fehlgeschlagen: {broadcast_result['error']}", None
def _compute_tx_hash(self, tx: Transaction) -> bytes:
"""Berechnet den Hash einer Transaktion für Signatur."""
data = f"{tx.tx_id}:{tx.wallet_id}:{tx.from_address}:{tx.to_address}:{tx.amount}:{tx.currency}:{tx.fee}"
return hashlib.sha256(data.encode()).digest()
def _verify_signature(
self,
signer_id: str,
tx_hash: bytes,
signature: bytes,
wallet: MultisigWallet
) -> bool:
"""Verifiziert eine Signatur gegen den Public Key des Signers."""
if signer_id not in wallet.signers:
return False
signer = wallet.signers[signer_id]
try:
public_key = ec.EllipticCurvePublicKey.from_encoded_point(
ec.SECP256R1(),
signer.public_key
)
public_key.verify(
signature,
tx_hash,
ec.ECDSA(hashes.SHA256())
)
return True
except Exception as e:
print(f"Signatur-Verifizierung fehlgeschlagen: {e}")
return False
def _combine_signatures(self, signatures: Dict[str, bytes]) -> bytes:
"""Kombiniert mehrere Signaturen zu einer einzigen MultiSig-Signatur."""
# Sortiere nach Signer-ID für Determinismus
sorted_sigs = sorted(signatures.items(), key=lambda x: x[0])
# Combine mit Standard ECDSA MultiSig
combined = b""
for signer_id, sig in sorted_sigs:
combined += sig
return combined
def _broadcast_transaction(
self,
tx: Transaction,
signature: bytes,
wallet: MultisigWallet
) -> Dict:
"""
Broadcastet die Transaktion zur Blockchain.
Hier exemplarisch – in Produktion: RPC-Calls zu Blockchain-Nodes.
"""
# Simulierte Broadcast-Logik
tx_data = {
"tx_id": tx.tx_id,
"wallet_type": wallet.wallet_type,
"from": tx.from_address,
"to": tx.to_address,
"amount": tx.amount,
"currency": tx.currency,
"signature": signature.hex(),
"signers": list(tx.signatures.keys())
}
# In Produktion: Blockchain RPC Call
# Beispiel für Ethereum:
# rpc_response = await ethereum_node.send_raw_transaction(tx_data)
return {
"success": True,
"tx_hash": hashlib.sha256(json.dumps(tx_data).encode()).hexdigest()[:64],
"block_confirmations": 0
}
def _generate_tx_id(
self,
wallet_id: str,
from_addr: str,
to_addr: str,
amount: float
) -> str:
"""Generiert eine eindeutige Transaktions-ID."""
timestamp = datetime.now().isoformat()
random_bytes = secrets.token_hex(8)
data = f"{wallet_id}:{from_addr}:{to_addr}:{amount}:{timestamp}:{random_bytes}"
return hashlib.sha256(data.encode()).hexdigest()[:24]
def _notify_signers(self, tx: Transaction):
"""Sendet Benachrichtigungen an erforderliche Signer."""
print(f"📬 Benachrichtigung: Neue Transaktion {tx.tx_id}")
print(f" Betrag: {tx.amount} {tx.currency}")
print(f" An: {tx.to_address}")
print(f" Benötigte Signer: {', '.join(tx.required_signers)}")
def get_transaction_status(self, tx_id: str, wallet_id: str) -> Optional[dict]:
"""Gibt den aktuellen Status einer Transaktion zurück."""
if wallet_id not in self.wallets:
return None
wallet = self.wallets[wallet_id]
if tx_id in wallet.pending_txs:
tx = wallet.pending_txs[tx_id]
elif tx_id in wallet.completed_txs:
tx = wallet.completed_txs[tx_id]
else:
return None
return {
"tx_id": tx.tx_id,
"status": tx.status,
"required_signatures": len(tx.required_signers),
"collected_signatures": len(tx.signatures),
"signed_by": list(tx.signatures.keys()),
"created_at": tx.created_at.isoformat(),
"expires_at": tx.expires_at.isoformat() if tx.expires_at else None,
"executed_at": tx.executed_at.isoformat() if tx.executed_at else None
}
使用示例
def example_multisig_workflow():
"""Demonstriert den typischen MultiSig-Workflow."""
config = {
"wallets": {
"hot_1": {
"type": "hot",
"threshold": 2,
"address": "0x742d35Cc6634C0532925a3b844Bc9e7595f8fE21",
"signers": [
{
"signer_id": "signer_primary_1",
"public_key": "02a1...c4f3", # Hex Public Key
"role": "primary"
},
{
"signer_id": "signer_primary_2",
"public_key": "03b2...d5e4",
"role": "primary"
},
{
"signer_id": "signer_secondary",
"public_key": "04c3...e6f5",
"role": "secondary"
}
]
},
"cold_1": {
"type": "cold",
"threshold": 3,
"address": "0x8626f6940E2eb28930eFb4CeF49B2d1F2C9C1199",
"signers": [
{
"signer_id": "cold_signer_1",
"public_key": "05d4...f7g6",
"role": "primary"
},
{
"signer_id": "cold_signer_2",
"public_key": "06e5...g7h7",
"role": "primary"
},
{
"signer_id": "cold_signer_3",
"public_key": "07f6...h8i8",
"role": "primary"
},
{
"signer_id": "cold_signer_4",
"public_key": "08g7...i9j9",
"role": "secondary"
},
{
"signer_id": "cold_signer_5",
"public_key": "09h8...j0k0",
"role": "emergency"
}
]
}
}
}
service = MultiSigService(config)
# 1. Erstelle Transaktion für Cold Wallet
tx = service.create_transaction(
wallet_id="cold_1",
from_address="0x8626f6940E2eb28930eFb4CeF49B2d1F2C9C1199",
to_address="0xdF3e18d64BC6A983f673Ab319CCaE4f1a57E7099",
amount=5_000_000.0,
currency="USDT",
fee=25.0,
requester_id="internal_system"
)
print(f"✅ Transaktion erstellt: {tx.tx_id}")
print(f" Status: {tx.status}")
print(f" Benötigte Signaturen: {tx.required_signers}")
# 2. Sammle Signaturen
signatures = {
"cold_signer_1": b"sig_primary_1_...",
"cold_signer_2": b"sig_primary_2_...",
"cold_signer_3": b"sig_primary_3_..."
}
for signer_id, sig in signatures.items():