QKDpy: Quantum Key Distribution Library

🏗️ Architecture Overview

Detailed architecture diagrams are available in docs/diagrams/. Each diagram below is a high-level summary — click through to the linked file for full detail.

High-Level Module Architecture

The system is organized into 9 modular layers. Arrows represent dependency direction.

Full diagram → — complete module breakdown with dependency graph and directory structure

Protocol Execution Flow

All QKD protocols follow a template-method pattern defined in BaseProtocol.execute():

Full diagram → — BB84, E91, and CV-QKD sequence diagrams with detailed state transitions

Core Quantum Stack

Full diagram → — class hierarchy, Bloch sphere, noise models, security analysis

Key Management Pipeline

Full diagram → — Cascade protocol, LDPC belief propagation, Toeplitz hashing details

Network & Satellite QKD

Full diagram → — pass simulation, multi-party network, routing

Framework Integrations

Full diagram → — Qiskit, PennyLane, Cirq, QpiAI conversion flows

Cryptographic & Enterprise Module

Full diagram → — QuantumHash, ZK proofs, HSM, compliance, audit

ML Optimization Pipeline

Full diagram → — Bayesian/genetic optimization, neural prediction, edge deployment

End-to-End Data Flow

Full diagram → — input→output trace, type conversions, key sizes, instrumentation

API Surface & Usage Patterns

Full diagram → — public API overview, sequence diagrams, import map, configuration system

🌟 Highlights

🛰️ Satellite QKD

QKDpy includes a comprehensive Satellite Quantum Key Distribution module for simulating space-ground quantum links:

from qkdpy.network import SatelliteQKD, AtmosphericProfile, OrbitType

# Create a LEO satellite QKD system
sat_qkd = SatelliteQKD(
    orbit_type=OrbitType.LEO,
    altitude_km=500,
    ground_station_lat=28.5,   # Cape Canaveral
    ground_station_lon=-80.6,
)

# Simulate a satellite pass with atmospheric effects
atmosphere = AtmosphericProfile(
    visibility_km=23.0,
    turbulence_cn2=1e-14,
    aerosol_optical_depth=0.1,
)

results = sat_qkd.simulate_pass(
    duration_seconds=300,
    atmosphere=atmosphere,
)

print(f"Total key bits: {results['total_key_bits']:,.0f}")
print(f"Peak elevation: {max(results['elevation_angles']):.1f}°")

Features:

• 🌍 Orbital Mechanics: LEO/MEO/GEO orbit simulation with realistic slant range
• 🌫️ Atmospheric Effects: Rayleigh/Mie scattering, turbulence (Fried parameter), clouds
• 📡 Free-Space Optical Channel: Geometric spreading, pointing loss, beam wandering
• 🧠 ML Channel Prediction: Train models to predict optimal transmission windows

🤖 ML Integration

Optimize QKD performance with built-in machine learning:

from qkdpy import QKDOptimizer, EfficientQKDPredictor

# Bayesian optimization for protocol parameters
optimizer = QKDOptimizer("BB84")
results = optimizer.optimize_channel_parameters(
    {"loss": (0.0, 0.5), "noise_level": (0.0, 0.1)},
    objective_function,
    method="bayesian",  # or "genetic", "neural"
)

# Resource-efficient predictor for edge deployment
predictor = EfficientQKDPredictor(
    input_dim=5,
    max_memory_mb=128,  # Constrained for embedded systems
    enable_quantization=True,
    enable_pruning=True,
)

🔍 Observability & Instrumentation

QKDpy includes built-in structured observability for debugging, performance analysis, and operations telemetry:

from qkdpy.utils import OperationSpan, instrument, record_protocol_execution

# Context manager for timing any block
with OperationSpan("protocol.execute", protocol="BB84") as span:
    result = protocol.run()
    span.set_metadata(qber=result.qber)

# Decorator for automatic instrumentation
@instrument("ml.train")
def train_model(self, data):
    ...

# One-shot event recording
record_protocol_execution(
    protocol_name="BB84",
    key_length=256,
    qber=0.025,
    final_key_size=192,
    is_secure=True,
    duration_ms=145.2,
)

Features:

• OperationSpan — Context manager with automatic duration tracking and structured start/complete/failure events
• @instrument decorator — One-line function instrumentation with argument metadata capture
• record_ helpers* — Domain-specific events for protocol execution, ML training, and QBER diagnostics
• structlog backend — JSON output for log aggregation (ELK, Datadog) or pretty-printed console

🏷️ Product Tiers

QKDpy uses a cumulative three-tier licensing model. Each tier includes everything in the tiers below it.

Tier Comparison

Enterprise Suite

from qkdpy.enterprise import (
    HSMInterface,
    AuditLogger,
    ComplianceChecker,
    ComplianceStandard,
)

# Hardware Security Module integration
hsm = get_hsm(provider=HSMProvider.SOFTWARE)  # or PKCS11
key_handle = hsm.generate_key("session_key", key_length=256)

# Tamper-evident audit logging
audit = AuditLogger(storage_path="audit.log")
audit.log_key_event(AuditEventType.KEY_GENERATED, "session_key")

# Compliance checking (NIST, FIPS, ISO, ETSI)
checker = ComplianceChecker([ComplianceStandard.NIST_SP_800_57])
report = checker.check_compliance()
print(report.export_markdown())
print(report.export_html())  # Enterprise-gated feature

Set your product tier via environment or config:

import os
os.environ["QKDPY_PRODUCT_TIER"] = "enterprise"

from qkdpy import set_config
set_config(product_tier="enterprise")

Compliance Standards Supported

🔐 Quantum-Safe Migration Toolkit

PREMIUM-tier toolkit for assessing and planning migration to quantum-resistant cryptography:

from qkdpy.enterprise.quantum_safe import (
    classic_enterprise_profile,
    generate_roadmap,
    QuantumSafeAssessment,
)

# Generate a crypto inventory from a classic enterprise profile
inventory = classic_enterprise_profile()
print(f"Total assets: {inventory.total_assets}")
print(f"Risk score: {inventory.risk_score:.0%}")

# Generate a phased migration roadmap
roadmap = generate_roadmap(inventory)
summary = roadmap.get_summary()
print(f"Target completion: {summary['target_completion']}")
print(f"Total steps: {summary['total_steps']}")

# Full assessment with recommendations
assessment = QuantumSafeAssessment(
    inventory=inventory,
    roadmap=roadmap,
)
report = assessment.to_dict()
for rec in report["recommendations"]:
    print(f"- {rec}")

Migration phases: Assess → Plan → Pilot → Migrate → Verify

📦 Features

Protocols

• BB84 (Standard + Decoy-State)
• E91 (Entanglement-based)
• B92, SARG04
• CV-QKD (Continuous-Variable)
• Device-Independent QKD
• HD-QKD (High-Dimensional)

Enterprise

• Product Tier Licensing — FREE/ENTERPRISE/PREMIUM with cumulative features
• Compliance Checking — ETSI GS QKD 014/016, ISO/IEC 23837, NIST SP 800-57, FIPS 140-2, ISO 27001
• HSM Integration — PKCS#11 interface with software fallback
• Audit Logging — Tamper-evident, structured event logging
• Quantum-Safe Migration — Crypto inventory, risk assessment, phased migration roadmap
• Key Escrow — Secure key recovery for enterprise deployments

Observability

• OperationSpan — Context manager for timed, structured operation tracking
• @instrument decorator — One-line function instrumentation
• Domain-specific events — Protocol execution, ML training, QBER diagnostics
• structlog backend — JSON or console output
• Correlation IDs — Trace operations across components

Framework Integrations

• Qiskit — IBM quantum SDK integration with noise models and transpilation
• Cirq — Google quantum framework including E91 protocol
• PennyLane — Quantum ML integration with noisy mixed-state simulation
• QpiAI — QpiAI Quantum SDK integration with statevector sampling

Infrastructure

• Structured exception hierarchy (QKDException)
• Centralized configuration management
• Structured logging with structlog
• Input validation decorators
• Type-safe (strict mypy)

🚀 Quick Start

# Install with uv (recommended)
pip install uv
uv pip install qkdpy

# Or with optional features
uv pip install qkdpy[ml]           # ML optimization
uv pip install qkdpy[enterprise]   # Enterprise features
uv pip install qkdpy[cirq]         # Cirq framework integration
uv pip install qkdpy[pennylane]    # PennyLane ML integration
uv pip install qkdpy[qiskit]       # Qiskit integration (noise models, transpilation)
uv pip install qkdpy[qpiai]        # QpiAI Quantum SDK integration
uv pip install qkdpy[all]          # Everything

from qkdpy import BB84, QuantumChannel

# Create channel and protocol
channel = QuantumChannel(loss=0.1, noise_model='depolarizing', noise_level=0.02)
bb84 = BB84(channel, key_length=256)

# Execute protocol
results = bb84.execute()
print(f"Key: {results['final_key'][:32]}...")
print(f"QBER: {results['qber']:.2%}")
print(f"Secure: {results['is_secure']}")

📐 Architecture Decisions

Key technical decisions are captured as Architecture Decision Records (ADRs):

🎯 Career Relevance

This library demonstrates expertise at the intersection of:

Real-world applications:

• 🛰️ Quantum satellite missions (like China's Micius)
• 🏦 Enterprise quantum-safe communications
• 🔬 Research in quantum networks

📄 License

Apache License 2.0 - See LICENSE

👤 Author

Pranava Kumar - Quantum Computing & Space Technology Enthusiast