verify_encryption.py

"""
Script to verify wallet balance encryption in the database.
This demonstrates that balances are stored encrypted and cannot be read without the key.
"""
import asyncio
import asyncpg
from decimal import Decimal
from app.core.config import settings
from app.wallet.services.service import WalletEncryption


async def verify_encryption():
    """Verify that wallet balances are encrypted in the database"""
    
    print("\n" + "="*70)
    print("Wallet Balance Encryption Verification")
    print("="*70 + "\n")
    
    # Connect to database
    conn = await asyncpg.connect(
        host=settings.POSTGRES_HOST,
        port=settings.POSTGRES_PORT,
        database=settings.POSTGRES_DB,
        user=settings.POSTGRES_USER,
        password=settings.POSTGRES_PASSWORD
    )
    
    try:
        # Get a sample wallet
        row = await conn.fetchrow(
            "SELECT wallet_id, partner_id, encrypted_balance FROM trans.spa_wallets LIMIT 1"
        )
        
        if not row:
            print("❌ No wallets found in database. Create a wallet first.")
            return
        
        wallet_id = row['wallet_id']
        partner_id = row['partner_id']
        encrypted_balance = row['encrypted_balance']
        
        print(f"📊 Wallet Information:")
        print(f"   Wallet ID: {wallet_id}")
        print(f"   Partner ID: {partner_id}")
        print(f"\n🔒 Encrypted Balance (as stored in database):")
        print(f"   Type: {type(encrypted_balance)}")
        print(f"   Length: {len(encrypted_balance)} bytes")
        print(f"   Raw bytes (first 50): {encrypted_balance[:50]}")
        print(f"   Hex representation: {encrypted_balance.hex()[:100]}...")
        
        # Try to decrypt
        try:
            decrypted_balance = WalletEncryption.decrypt_balance(encrypted_balance)
            print(f"\n✅ Decrypted Balance (using encryption key):")
            print(f"   Balance: {decrypted_balance}")
            print(f"   Type: {type(decrypted_balance)}")
        except Exception as e:
            print(f"\n❌ Decryption failed: {e}")
        
        # Demonstrate encryption/decryption cycle
        print(f"\n🔄 Encryption/Decryption Cycle Test:")
        test_balance = Decimal("12345.67")
        print(f"   Original: {test_balance}")
        
        encrypted = WalletEncryption.encrypt_balance(test_balance)
        print(f"   Encrypted: {encrypted.hex()[:100]}...")
        
        decrypted = WalletEncryption.decrypt_balance(encrypted)
        print(f"   Decrypted: {decrypted}")
        
        if test_balance == decrypted:
            print(f"   ✅ Encryption/Decryption cycle successful!")
        else:
            print(f"   ❌ Encryption/Decryption cycle failed!")
        
        # Show that different values produce different encrypted data
        print(f"\n🔐 Encryption Uniqueness Test:")
        balance1 = Decimal("1000.00")
        balance2 = Decimal("2000.00")
        
        encrypted1 = WalletEncryption.encrypt_balance(balance1)
        encrypted2 = WalletEncryption.encrypt_balance(balance2)
        
        print(f"   Balance 1: {balance1}")
        print(f"   Encrypted 1: {encrypted1.hex()[:50]}...")
        print(f"   Balance 2: {balance2}")
        print(f"   Encrypted 2: {encrypted2.hex()[:50]}...")
        print(f"   Are they different? {encrypted1 != encrypted2} ✅")
        
        # Security note
        print(f"\n🛡️  Security Notes:")
        print(f"   • Balance is stored as binary data (BYTEA) in PostgreSQL")
        print(f"   • Encryption uses AES via Fernet (symmetric encryption)")
        print(f"   • Key is derived from SECRET_KEY using SHA256")
        print(f"   • Without the encryption key, the balance cannot be read")
        print(f"   • Even database administrators cannot see actual balances")
        print(f"   • Each encryption produces different ciphertext (includes IV)")
        
        # Show transaction amounts (not encrypted)
        print(f"\n📝 Transaction History (amounts are NOT encrypted):")
        transactions = await conn.fetch(
            """
            SELECT transaction_type, amount, balance_after, description, created_at
            FROM trans.spa_wallet_transactions
            WHERE partner_id = $1
            ORDER BY created_at DESC
            LIMIT 5
            """,
            partner_id
        )
        
        if transactions:
            for txn in transactions:
                print(f"   {txn['created_at']} | {txn['transaction_type']:6} | "
                      f"₹{txn['amount']:>10} | Balance: ₹{txn['balance_after']:>10} | "
                      f"{txn['description']}")
        else:
            print(f"   No transactions found")
        
        print(f"\n💡 Why transaction amounts are not encrypted:")
        print(f"   • Needed for reporting and analytics")
        print(f"   • Balance history is maintained in transactions")
        print(f"   • Only current balance in wallet table is encrypted")
        print(f"   • Transaction amounts are historical and less sensitive")
        
    finally:
        await conn.close()
    
    print("\n" + "="*70)
    print("Verification Complete")
    print("="*70 + "\n")


if __name__ == "__main__":
    print("""
    ╔══════════════════════════════════════════════════════════════╗
    ║     Wallet Balance Encryption Verification Script           ║
    ║                                                              ║
    ║  This script demonstrates that wallet balances are stored   ║
    ║  encrypted in the database and cannot be read without the   ║
    ║  encryption key.                                            ║
    ╚══════════════════════════════════════════════════════════════╝
    """)
    
    try:
        asyncio.run(verify_encryption())
    except Exception as e:
        print(f"\n❌ Error: {e}")
        print("\nMake sure:")
        print("  1. Database is running")
        print("  2. .env file is configured")
        print("  3. Wallet tables are created (run migration)")
        print("  4. At least one wallet exists in the database")