Quantum Key Distribution: Unbreakable Cryptography with QKD

import numpy as np
from enum import Enum

class Basis(Enum):
    RECTILINEAR = 0  # |0⟩, |1⟩ (vertical/horizontal)
    DIAGONAL = 1      # |+⟩, |−⟩ (diagonal)

class BB84Protocol:
    """Simulate BB84 quantum key distribution"""
    def __init__(self, key_length=256):
        self.key_length = key_length
        self.quantum_channel_error = 0.01  # 1% intrinsic error
    
    def alice_prepare(self):
        """Alice prepares random qubits"""
        # Random bit string
        self.alice_bits = np.random.randint(0, 2, size=self.key_length * 4)
        
        # Random basis choices
        self.alice_bases = np.random.randint(0, 2, size=self.key_length * 4)
        
        # Encode: basis determines measurement orientation
        qubits = []
        for bit, basis in zip(self.alice_bits, self.alice_bases):
            qubit = {'bit': bit, 'basis': Basis(basis)}
            qubits.append(qubit)
        
        return qubits
    
    def bob_measure(self, qubits, eve_present=False):
        """Bob measures qubits in random bases"""
        # Random measurement bases
        self.bob_bases = np.random.randint(0, 2, size=len(qubits))
        
        measured_bits = []
        for i, qubit in enumerate(qubits):
            bob_basis = Basis(self.bob_bases[i])
            
            # If bases match, measurement is reliable
            if qubit['basis'] == bob_basis:
                measured_bit = qubit['bit']
            else:
                # Bases don't match: 50% chance of error
                measured_bit = np.random.randint(0, 2)
            
            # Quantum channel noise
            if np.random.random() < self.quantum_channel_error:
                measured_bit = 1 - measured_bit
            
            # ⚠️ Eve's eavesdropping
            if eve_present and np.random.random() < 0.5:
                # Eve's intercept-resend attack introduces errors
                measured_bit = np.random.randint(0, 2)
            
            measured_bits.append(measured_bit)
        
        return np.array(measured_bits)
    
    def sift_key(self, alice_bits, alice_bases, bob_bits, bob_bases):
        """Discard bits where bases don't match"""
        matching_bases = (alice_bases == bob_bases)
        
        sifted_alice = alice_bits[matching_bases]
        sifted_bob = bob_bits[matching_bases]
        
        print(f"Sifted key length: {len(sifted_alice)} (from {len(alice_bits)} transmitted)")
        
        return sifted_alice, sifted_bob
    
    def estimate_error_rate(self, alice_key, bob_key, sample_size=50):
        """Use subset to estimate quantum bit error rate (QBER)"""
        # Randomly sample bits to check
        indices = np.random.choice(len(alice_key), size=sample_size, replace=False)
        
        alice_sample = alice_key[indices]
        bob_sample = bob_key[indices]
        
        errors = np.sum(alice_sample != bob_sample)
        qber = errors / sample_size
        
        print(f"QBER: {qber:.2%}")
        
        # Remove sampled bits from key
        keep_mask = np.ones(len(alice_key), dtype=bool)
        keep_mask[indices] = False
        
        final_alice = alice_key[keep_mask]
        final_bob = bob_key[keep_mask]
        
        # ⚠️ Security threshold: QBER > 11% indicates eavesdropping
        if qber > 0.11:
            print("⚠️ High QBER detected! Possible eavesdropper (Eve)")
            return None, None, qber
        
        return final_alice, final_bob, qber
    
    def privacy_amplification(self, key):
        """Hash key to remove Eve's partial information"""
        # Simple privacy amplification via universal hashing
        # In production, use proper extractors
        np.random.seed(42)  # Shared public seed
        hash_matrix = np.random.randint(0, 2, size=(128, len(key)))
        
        final_key = (hash_matrix @ key) % 2
        
        return final_key

# Run full BB84 protocol
def run_qkd(eve_present=False):
    """Execute complete QKD session"""
    protocol = BB84Protocol(key_length=256)
    
    print("=== BB84 Quantum Key Distribution ===\n")
    
    # Step 1: Alice prepares qubits
    print("1. Alice prepares random qubits...")
    qubits = protocol.alice_prepare()
    
    # Step 2: Bob measures
    print("2. Bob measures qubits in random bases...")
    bob_bits = protocol.bob_measure(qubits, eve_present=eve_present)
    
    # Step 3: Classical basis reconciliation
    print("3. Basis reconciliation (public channel)...")
    alice_sifted, bob_sifted = protocol.sift_key(
        protocol.alice_bits, protocol.alice_bases,
        bob_bits, protocol.bob_bases
    )
    
    # Step 4: Error estimation
    print("4. Error rate estimation...")
    alice_key, bob_key, qber = protocol.estimate_error_rate(alice_sifted, bob_sifted)
    
    if alice_key is None:
        print("\n❌ Key exchange aborted due to eavesdropping")
        return None
    
    # Step 5: Privacy amplification
    print("5. Privacy amplification...")
    final_key = protocol.privacy_amplification(alice_key)
    
    print(f"\n✅ Final shared secret key ({len(final_key)} bits):")
    print(''.join(map(str, final_key[:32])), "...")
    
    return final_key

# Test with and without Eve
print("Without eavesdropper:")
run_qkd(eve_present=False)

print("\n" + "="*50 + "\n")

print("With eavesdropper:")
run_qkd(eve_present=True)

import hashlib
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

class QKDIntegration:
    """Integrate QKD with classical cryptography"""
    def __init__(self):
        self.quantum_key_pool = []
    
    def establish_qkd_session(self):
        """Run BB84 to establish shared key"""
        protocol = BB84Protocol(key_length=256)
        qubits = protocol.alice_prepare()
        bob_bits = protocol.bob_measure(qubits)
        
        alice_sifted, bob_sifted = protocol.sift_key(
            protocol.alice_bits, protocol.alice_bases,
            bob_bits, protocol.bob_bases
        )
        
        alice_key, bob_key, qber = protocol.estimate_error_rate(alice_sifted, bob_sifted)
        
        if alice_key is None:
            raise SecurityError("QKD session compromised")
        
        final_key = protocol.privacy_amplification(alice_key)
        
        # Store for one-time pad usage
        self.quantum_key_pool.append(final_key)
        
        return final_key
    
    def encrypt_message(self, message: bytes):
        """Use quantum key for AES encryption"""
        if not self.quantum_key_pool:
            self.establish_qkd_session()
        
        # Use quantum key as AES key
        qkey = self.quantum_key_pool.pop(0)
        aes_key = hashlib.sha256(qkey.tobytes()).digest()  # 256-bit key
        
        # AES-GCM for authenticated encryption
        iv = os.urandom(12)
        cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv))
        encryptor = cipher.encryptor()
        
        ciphertext = encryptor.update(message) + encryptor.finalize()
        
        return {
            'iv': iv,
            'ciphertext': ciphertext,
            'tag': encryptor.tag
        }