Chapter 11: Virtual Currency Security and Privacy Protection

11.1 Cryptocurrency Security Mechanisms

11.1.1 Private Key Management

import hashlib
import hmac
import os
import base58
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend
import secrets

class WalletKeyManager:
    def __init__(self):
        self.private_key = None
        self.public_key = None
        self.address = None

    def generate_new_keypair(self):
        """Generate new key pair"""
        # Generate private key using ECDSA secp256k1 curve
        self.private_key = ec.generate_private_key(ec.SECP256K1(), default_backend())

        # Derive public key from private key
        self.public_key = self.private_key.public_key()

        # Generate address
        self.address = self._public_key_to_address(self.public_key)

        return {
            'private_key': self._export_private_key(),
            'public_key': self._export_public_key(),
            'address': self.address
        }

    def _export_private_key(self):
        """Export private key (hex format)"""
        if not self.private_key:
            return None

        private_bytes = self.private_key.private_numbers().private_value.to_bytes(32, 'big')
        return private_bytes.hex()

    def _export_public_key(self):
        """Export public key (hex format)"""
        if not self.public_key:
            return None

        public_bytes = self.public_key.public_bytes(
            encoding=serialization.Encoding.X962,
            format=serialization.PublicFormat.UncompressedPoint
        )
        return public_bytes.hex()

    def _public_key_to_address(self, public_key):
        """Convert public key to address"""
        # Get public key bytes
        public_bytes = public_key.public_bytes(
            encoding=serialization.Encoding.X962,
            format=serialization.PublicFormat.UncompressedPoint
        )

        # SHA256 hash
        sha256_hash = hashlib.sha256(public_bytes).digest()

        # RIPEMD160 hash
        ripemd160 = hashlib.new('ripemd160')
        ripemd160.update(sha256_hash)
        hash160 = ripemd160.digest()

        # Add version prefix (0x00 for Bitcoin mainnet)
        versioned_hash = b'\x00' + hash160

        # Double SHA256 for checksum
        checksum = hashlib.sha256(hashlib.sha256(versioned_hash).digest()).digest()[:4]

        # Combine and encode with Base58
        address_bytes = versioned_hash + checksum
        address = base58.b58encode(address_bytes).decode()

        return address

    def sign_transaction(self, transaction_data):
        """Sign transaction"""
        if not self.private_key:
            raise ValueError("No private key available")

        # Serialize transaction data
        message = json.dumps(transaction_data, sort_keys=True).encode()

        # Sign
        signature = self.private_key.sign(
            message,
            ec.ECDSA(hashes.SHA256())
        )

        return signature.hex()

    def verify_signature(self, transaction_data, signature_hex):
        """Verify signature"""
        if not self.public_key:
            raise ValueError("No public key available")

        message = json.dumps(transaction_data, sort_keys=True).encode()
        signature = bytes.fromhex(signature_hex)

        try:
            self.public_key.verify(
                signature,
                message,
                ec.ECDSA(hashes.SHA256())
            )
            return True
        except:
            return False

class HDWallet:
    """Hierarchical Deterministic Wallet (BIP32)"""

    def __init__(self, seed=None):
        if seed is None:
            # Generate random seed
            self.seed = secrets.token_bytes(64)
        else:
            self.seed = seed

        self.master_private_key = None
        self.master_chain_code = None
        self._generate_master_keys()

    def _generate_master_keys(self):
        """Generate master key and chain code"""
        # Use HMAC-SHA512
        hmac_result = hmac.new(
            b"Bitcoin seed",
            self.seed,
            hashlib.sha512
        ).digest()

        # First 32 bytes are master private key, last 32 bytes are chain code
        self.master_private_key = hmac_result[:32]
        self.master_chain_code = hmac_result[32:]

    def derive_child_key(self, index, hardened=False):
        """Derive child key"""
        if hardened:
            index = index + 0x80000000  # Set hardened flag

        # Serialize index
        index_bytes = index.to_bytes(4, 'big')

        if hardened:
            # Hardened derivation: use private key
            data = b'\x00' + self.master_private_key + index_bytes
        else:
            # Normal derivation: use public key
            # (simplified, actual implementation needs to derive public key from private key)
            data = self.master_private_key + index_bytes

        # Use HMAC-SHA512
        hmac_result = hmac.new(
            self.master_chain_code,
            data,
            hashlib.sha512
        ).digest()

        # Child private key
        child_private_key = hmac_result[:32]
        child_chain_code = hmac_result[32:]

        return {
            'private_key': child_private_key.hex(),
            'chain_code': child_chain_code.hex(),
            'index': index
        }

    def generate_mnemonic(self, word_count=12):
        """Generate mnemonic words (simplified version)"""
        # Actual implementation should follow BIP39 standard
        # This is a simplified demonstration
        with open('/usr/share/dict/words', 'r') as f:
            words = [word.strip() for word in f.readlines()]

        # Use seed to deterministically select words
        random.seed(int.from_bytes(self.seed[:4], 'big'))
        mnemonic = random.sample(words, word_count)

        return ' '.join(mnemonic)

import json
import random

# Wallet key management demonstration
print("Wallet Key Management Demonstration")
print("=" * 50)

# Generate new wallet
wallet = WalletKeyManager()
keys = wallet.generate_new_keypair()

print("New Wallet Generated:")
print(f"Address: {keys['address']}")
print(f"Public Key: {keys['public_key'][:64]}...")
print(f"Private Key: {keys['private_key'][:16]}... (Sensitive! Conceal properly)")

# Transaction signing and verification
transaction = {
    'from': keys['address'],
    'to': '1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa',
    'amount': 0.5,
    'timestamp': datetime.now().isoformat()
}

print(f"\nSigning transaction:")
print(f"From: {transaction['from'][:10]}...")
print(f"To: {transaction['to'][:10]}...")
print(f"Amount: {transaction['amount']} BTC")

signature = wallet.sign_transaction(transaction)
print(f"Signature: {signature[:32]}...")

# Verify signature
is_valid = wallet.verify_signature(transaction, signature)
print(f"Signature verification result: {'✅ Valid' if is_valid else '❌ Invalid'}")

# HD Wallet demonstration
print(f"\n\nHD Wallet Demonstration")
print("=" * 50)

hd_wallet = HDWallet()
print(f"Master seed generated (64 bytes)")

# Generate mnemonic (simplified)
# mnemonic = hd_wallet.generate_mnemonic()
# print(f"Mnemonic: {mnemonic}")

# Derive child keys
print(f"\nDeriving child keys:")
for i in range(3):
    child_key = hd_wallet.derive_child_key(i, hardened=False)
    print(f"Child key {i}: {child_key['private_key'][:16]}...")

# Derive hardened child keys
print(f"\nDeriving hardened child keys:")
for i in range(3):
    child_key = hd_wallet.derive_child_key(i, hardened=True)
    print(f"Hardened child key {i}: {child_key['private_key'][:16]}...")

11.1.2 Multi-Signature Wallets

class MultiSigWallet:
    def __init__(self, m, n, participants):
        """
        Initialize multi-signature wallet
        m: Minimum required signatures
        n: Total number of participants
        participants: List of participant public keys
        """
        if m > n or m < 1:
            raise ValueError("Invalid m-of-n configuration")

        self.m = m
        self.n = n
        self.participants = participants
        self.pending_transactions = {}
        self.transaction_counter = 0

    def create_transaction(self, transaction_data, creator_id):
        """Create transaction"""
        self.transaction_counter += 1
        tx_id = f"TX_{self.transaction_counter:04d}"

        transaction = {
            'id': tx_id,
            'data': transaction_data,
            'creator': creator_id,
            'signatures': {},
            'status': 'pending',
            'created_time': datetime.now()
        }

        self.pending_transactions[tx_id] = transaction
        return tx_id

    def sign_transaction(self, tx_id, signer_id, signature):
        """Sign transaction"""
        if tx_id not in self.pending_transactions:
            return {'success': False, 'error': 'Transaction not found'}

        transaction = self.pending_transactions[tx_id]

        if transaction['status'] != 'pending':
            return {'success': False, 'error': 'Transaction not in pending status'}

        if signer_id not in self.participants:
            return {'success': False, 'error': 'Signer not a participant'}

        if signer_id in transaction['signatures']:
            return {'success': False, 'error': 'Already signed'}

        # Add signature
        transaction['signatures'][signer_id] = {
            'signature': signature,
            'timestamp': datetime.now()
        }

        # Check if enough signatures collected
        if len(transaction['signatures']) >= self.m:
            transaction['status'] = 'ready'
            print(f"Transaction {tx_id} collected enough signatures ({len(transaction['signatures'])}/{self.m})")

        return {
            'success': True,
            'signatures_collected': len(transaction['signatures']),
            'signatures_required': self.m,
            'status': transaction['status']
        }

    def execute_transaction(self, tx_id):
        """Execute transaction"""
        if tx_id not in self.pending_transactions:
            return {'success': False, 'error': 'Transaction not found'}

        transaction = self.pending_transactions[tx_id]

        if transaction['status'] != 'ready':
            return {
                'success': False,
                'error': f'Transaction not ready to execute (current status: {transaction["status"]})'
            }

        # Verify all signatures (simplified)
        if len(transaction['signatures']) < self.m:
            return {'success': False, 'error': 'Insufficient signatures'}

        # Execute transaction
        transaction['status'] = 'executed'
        transaction['executed_time'] = datetime.now()

        print(f"Transaction {tx_id} executed successfully")

        return {
            'success': True,
            'transaction_id': tx_id,
            'executed_time': transaction['executed_time']
        }

    def get_transaction_status(self, tx_id):
        """Get transaction status"""
        if tx_id not in self.pending_transactions:
            return None

        transaction = self.pending_transactions[tx_id]
        return {
            'id': transaction['id'],
            'status': transaction['status'],
            'signatures_collected': len(transaction['signatures']),
            'signatures_required': self.m,
            'signers': list(transaction['signatures'].keys())
        }

    def get_pending_transactions(self):
        """Get all pending transactions"""
        pending = []
        for tx_id, transaction in self.pending_transactions.items():
            if transaction['status'] == 'pending':
                pending.append({
                    'id': tx_id,
                    'creator': transaction['creator'],
                    'signatures_collected': len(transaction['signatures']),
                    'created_time': transaction['created_time']
                })
        return pending

# Multi-signature wallet demonstration
print("\nMulti-Signature Wallet Demonstration")
print("=" * 50)

# Create 2-of-3 multi-signature wallet
participants = ['Alice', 'Bob', 'Charlie']
multisig_wallet = MultiSigWallet(m=2, n=3, participants=participants)

print(f"Created 2-of-3 multi-signature wallet")
print(f"Participants: {', '.join(participants)}")

# Alice creates transaction
transaction_data = {
    'to': '1RecipientAddress123',
    'amount': 10.5,
    'description': 'Payment to supplier'
}

tx_id = multisig_wallet.create_transaction(transaction_data, 'Alice')
print(f"\nAlice created transaction: {tx_id}")
print(f"Amount: {transaction_data['amount']} BTC")

# Alice signs
alice_signature = "alice_sig_" + secrets.token_hex(32)
result = multisig_wallet.sign_transaction(tx_id, 'Alice', alice_signature)
print(f"\nAlice signed: {result['signatures_collected']}/{result['signatures_required']} signatures")

# Bob signs
bob_signature = "bob_sig_" + secrets.token_hex(32)
result = multisig_wallet.sign_transaction(tx_id, 'Bob', bob_signature)
print(f"Bob signed: {result['signatures_collected']}/{result['signatures_required']} signatures")
print(f"Transaction status: {result['status']}")

# Execute transaction
if result['status'] == 'ready':
    execution_result = multisig_wallet.execute_transaction(tx_id)
    if execution_result['success']:
        print(f"\n✅ Transaction executed successfully!")

# Display transaction status
status = multisig_wallet.get_transaction_status(tx_id)
print(f"\nFinal transaction status:")
print(f"ID: {status['id']}")
print(f"Status: {status['status']}")
print(f"Signers: {', '.join(status['signers'])}")

11.2 Privacy Protection Technologies

11.2.1 Mixing Services

class CoinMixer:
    """Coin mixing service (simplified demonstration)"""

    def __init__(self, mixing_fee_percentage=1.0):
        self.mixing_pool = []
        self.mixing_fee_percentage = mixing_fee_percentage
        self.mixing_rounds = {}
        self.round_counter = 0

    def create_mixing_round(self, target_amount, participants_count):
        """Create mixing round"""
        self.round_counter += 1
        round_id = f"ROUND_{self.round_counter:04d}"

        mixing_round = {
            'id': round_id,
            'target_amount': target_amount,
            'participants_count': participants_count,
            'deposits': [],
            'status': 'open',
            'created_time': datetime.now()
        }

        self.mixing_rounds[round_id] = mixing_round
        return round_id

    def deposit_to_mix(self, round_id, sender_address, amount, receive_address):
        """Deposit funds for mixing"""
        if round_id not in self.mixing_rounds:
            return {'success': False, 'error': 'Mixing round not found'}

        mixing_round = self.mixing_rounds[round_id]

        if mixing_round['status'] != 'open':
            return {'success': False, 'error': 'Mixing round not open'}

        # Check amount
        fee = amount * (self.mixing_fee_percentage / 100)
        net_amount = amount - fee

        if abs(net_amount - mixing_round['target_amount']) > 0.001:
            return {
                'success': False,
                'error': f'Amount must be {mixing_round["target_amount"]} + fee'
            }

        # Add deposit
        deposit = {
            'sender': sender_address,
            'receive_address': receive_address,
            'amount': amount,
            'fee': fee,
            'net_amount': net_amount,
            'timestamp': datetime.now()
        }

        mixing_round['deposits'].append(deposit)

        # Check if enough participants
        if len(mixing_round['deposits']) >= mixing_round['participants_count']:
            mixing_round['status'] = 'ready'
            print(f"Mixing round {round_id} ready to execute")

        return {
            'success': True,
            'deposit_index': len(mixing_round['deposits']) - 1,
            'participants': len(mixing_round['deposits']),
            'required_participants': mixing_round['participants_count']
        }

    def execute_mixing(self, round_id):
        """Execute mixing"""
        if round_id not in self.mixing_rounds:
            return {'success': False, 'error': 'Mixing round not found'}

        mixing_round = self.mixing_rounds[round_id]

        if mixing_round['status'] != 'ready':
            return {'success': False, 'error': 'Mixing round not ready'}

        # Shuffle receive addresses
        receive_addresses = [d['receive_address'] for d in mixing_round['deposits']]
        random.shuffle(receive_addresses)

        # Create outputs (breaking sender-receiver link)
        outputs = []
        for i, deposit in enumerate(mixing_round['deposits']):
            output = {
                'receive_address': receive_addresses[i],  # Shuffled address
                'amount': deposit['net_amount'],
                'original_sender': deposit['sender']  # For verification only, won't be public
            }
            outputs.append(output)

        mixing_round['outputs'] = outputs
        mixing_round['status'] = 'completed'
        mixing_round['completed_time'] = datetime.now()

        return {
            'success': True,
            'round_id': round_id,
            'outputs': outputs,
            'total_fee_collected': sum(d['fee'] for d in mixing_round['deposits'])
        }

    def get_mixing_status(self, round_id):
        """Get mixing status"""
        if round_id not in self.mixing_rounds:
            return None

        mixing_round = self.mixing_rounds[round_id]
        return {
            'id': round_id,
            'status': mixing_round['status'],
            'participants': len(mixing_round['deposits']),
            'required_participants': mixing_round['participants_count'],
            'target_amount': mixing_round['target_amount']
        }

# Mixing service demonstration
print("\nCoin Mixing Service Demonstration")
print("=" * 50)

mixer = CoinMixer(mixing_fee_percentage=1.0)

# Create mixing round
target_amount = 1.0  # 1 BTC
participants_count = 5
round_id = mixer.create_mixing_round(target_amount, participants_count)

print(f"Created mixing round: {round_id}")
print(f"Target amount: {target_amount} BTC")
print(f"Required participants: {participants_count}")
print(f"Mixing fee: {mixer.mixing_fee_percentage}%")

# Simulate 5 users depositing
participants_data = [
    ('1AliceAddress', '1AliceReceiveAddr'),
    ('1BobAddress', '1BobReceiveAddr'),
    ('1CharlieAddress', '1CharlieReceiveAddr'),
    ('1DavidAddress', '1DavidReceiveAddr'),
    ('1EveAddress', '1EveReceiveAddr')
]

print(f"\nParticipants depositing:")
for i, (sender, receiver) in enumerate(participants_data, 1):
    fee = target_amount * (mixer.mixing_fee_percentage / 100)
    total_amount = target_amount + fee

    result = mixer.deposit_to_mix(round_id, sender, total_amount, receiver)

    if result['success']:
        print(f"{i}. {sender[:15]}... deposited {total_amount:.4f} BTC "
              f"({result['participants']}/{result['required_participants']})")

# Execute mixing
print(f"\nExecuting mixing...")
execution_result = mixer.execute_mixing(round_id)

if execution_result['success']:
    print(f"✅ Mixing completed!")
    print(f"Total fee collected: {execution_result['total_fee_collected']:.4f} BTC")

    print(f"\nOutput addresses (shuffled):")
    for i, output in enumerate(execution_result['outputs'], 1):
        print(f"{i}. {output['receive_address'][:20]}... receives {output['amount']:.4f} BTC")

    print(f"\n🔒 Privacy achieved: Original sender-receiver link broken")

11.2.2 Zero-Knowledge Proofs

class ZeroKnowledgeProof:
    """Zero-Knowledge Proof System (Simplified Demonstration)"""

    def __init__(self):
        self.commitment_store = {}

    def create_commitment(self, secret_value, randomness=None):
        """Create commitment"""
        if randomness is None:
            randomness = secrets.token_bytes(32)

        # Simplified commitment scheme: Hash(secret || randomness)
        commitment_data = str(secret_value).encode() + randomness
        commitment = hashlib.sha256(commitment_data).hexdigest()

        return {
            'commitment': commitment,
            'randomness': randomness.hex()
        }

    def verify_commitment(self, commitment, secret_value, randomness_hex):
        """Verify commitment"""
        randomness = bytes.fromhex(randomness_hex)
        commitment_data = str(secret_value).encode() + randomness
        calculated_commitment = hashlib.sha256(commitment_data).hexdigest()

        return calculated_commitment == commitment

    def prove_range(self, value, min_value, max_value):
        """Prove value is within range (without revealing exact value)"""
        # Simplified range proof
        if not (min_value <= value <= max_value):
            return None

        # Generate commitment
        commitment_data = self.create_commitment(value)

        # Create proof (simplified: in reality would use Bulletproofs or similar)
        proof = {
            'commitment': commitment_data['commitment'],
            'range': {'min': min_value, 'max': max_value},
            'proof_type': 'range_proof',
            'timestamp': datetime.now()
        }

        # Store private data (normally would not be stored)
        self.commitment_store[commitment_data['commitment']] = {
            'value': value,
            'randomness': commitment_data['randomness']
        }

        return proof

    def verify_range_proof(self, proof):
        """Verify range proof"""
        # Simplified verification
        commitment = proof['commitment']

        if commitment not in self.commitment_store:
            return {'valid': False, 'error': 'Commitment not found'}

        stored_data = self.commitment_store[commitment]
        value = stored_data['value']
        range_min = proof['range']['min']
        range_max = proof['range']['max']

        is_valid = range_min <= value <= range_max

        return {
            'valid': is_valid,
            'range': proof['range'],
            'proof_type': proof['proof_type']
        }

    def prove_membership(self, element, set_members):
        """Prove element is in set (without revealing which element)"""
        if element not in set_members:
            return None

        # Create commitments for all set members
        commitments = []
        for member in set_members:
            commitment_data = self.create_commitment(member)
            commitments.append(commitment_data['commitment'])

            # Store for verification
            self.commitment_store[commitment_data['commitment']] = {
                'value': member,
                'randomness': commitment_data['randomness']
            }

        # Create proof
        element_commitment = self.create_commitment(element)

        proof = {
            'element_commitment': element_commitment['commitment'],
            'set_commitments': commitments,
            'proof_type': 'membership_proof',
            'timestamp': datetime.now()
        }

        return proof

    def verify_membership_proof(self, proof):
        """Verify membership proof"""
        element_commitment = proof['element_commitment']
        set_commitments = proof['set_commitments']

        # Check if element commitment exists in set
        is_member = element_commitment in set_commitments

        return {
            'valid': is_member,
            'set_size': len(set_commitments),
            'proof_type': proof['proof_type']
        }

# Zero-knowledge proof demonstration
print("\nZero-Knowledge Proof Demonstration")
print("=" * 50)

zkp = ZeroKnowledgeProof()

# Example 1: Range proof
print("Example 1: Prove account balance is within range")
actual_balance = 1500
min_required = 1000
max_allowed = 10000

range_proof = zkp.prove_range(actual_balance, min_required, max_allowed)

if range_proof:
    print(f"✅ Range proof created")
    print(f"Commitment: {range_proof['commitment'][:16]}...")
    print(f"Proving balance is between {min_required} and {max_allowed}")
    print(f"(Actual value not disclosed)")

    # Verify
    verification = zkp.verify_range_proof(range_proof)
    print(f"\nVerification result: {'✅ Valid' if verification['valid'] else '❌ Invalid'}")

# Example 2: Membership proof
print(f"\n\nExample 2: Prove address is in whitelist")
whitelist = ['1Address1', '1Address2', '1Address3', '1Address4', '1Address5']
user_address = '1Address3'

membership_proof = zkp.prove_membership(user_address, whitelist)

if membership_proof:
    print(f"✅ Membership proof created")
    print(f"Element commitment: {membership_proof['element_commitment'][:16]}...")
    print(f"Proving address is in whitelist of {len(whitelist)} addresses")
    print(f"(Specific address not disclosed)")

    # Verify
    verification = zkp.verify_membership_proof(membership_proof)
    print(f"\nVerification result: {'✅ Valid' if verification['valid'] else '❌ Invalid'}")
    print(f"Whitelist size: {verification['set_size']}")

# Example 3: Commitment scheme
print(f"\n\nExample 3: Commitment and reveal")
secret_number = 42
commitment_data = zkp.create_commitment(secret_number)

print(f"Commitment created: {commitment_data['commitment'][:16]}...")
print(f"(Secret number: {secret_number} committed)")

# Later reveal and verify
print(f"\nRevealing secret...")
is_valid = zkp.verify_commitment(
    commitment_data['commitment'],
    secret_number,
    commitment_data['randomness']
)

print(f"Commitment verification: {'✅ Valid' if is_valid else '❌ Invalid'}")
print(f"Revealed number: {secret_number}")

(Continuing with sections on privacy coins, security best practices, etc…)

11.5 Course Summary

This chapter detailed security and privacy protection technologies in the virtual currency space:

Key Points

1. Private Key Security: Foundation of all cryptocurrency security
2. Multi-Signature Mechanism: Distributes risk, improves security
3. Privacy Protection: Mixing services, zero-knowledge proofs, and other technologies protect transaction privacy
4. Security Best Practices: Cold storage, 2FA, regular audits are essential

Security Recommendations

• Never share private keys
• Use hardware wallets to store large amounts of assets
• Enable multi-factor authentication
• Regularly backup wallet data
• Remain vigilant about phishing attacks

Privacy Protection Recommendations

• Use different addresses for different transactions
• Consider using privacy-enhancing cryptocurrencies
• Understand transaction traceability
• Use mixing services when necessary

Security and privacy are the cornerstones of cryptocurrency. Through this chapter, you should understand the core principles of cryptocurrency security and privacy protection technologies. In practice, always maintain security awareness and choose appropriate security measures based on actual needs.