Week 12: Private Communications II
🎯 Learning Goals
By the end of this week, you should understand:
- Advanced anonymous communication systems and their security properties
- Metadata-resistant messaging protocols and applications
- Anonymous file sharing and publishing systems
- Censorship resistance techniques and pluggable transports
📖 Theoretical Content
Beyond Basic Anonymity: Advanced Communication Privacy
Limitations of Basic Anonymous Networks:
- Traffic Analysis: Pattern-based attacks on Tor and mix networks
- Metadata Leakage: Timing, volume, and behavioral information
- Scalability Issues: Performance degrades with increased privacy
- Usability Barriers: Complex setup and configuration requirements
Advanced Privacy Requirements:
- Metadata Resistance: Hide communication patterns and relationships
- Forward Security: Past communications remain secure if keys are compromised
- Deniability: Participants can plausibly deny communication occurred
- Censorship Resistance: Communicate despite network blocking and filtering
Metadata-Resistant Messaging
The Metadata Problem: Even with encrypted content, communication metadata reveals:
- Who communicates with whom (social graph)
- When communication occurs (timing analysis)
- How much data is exchanged (volume analysis)
- Communication patterns and habits
Riffle: Verifiable Anonymous Messaging
Design Principles:
- Verifiable Shuffles: Cryptographically prove correct message mixing
- Private Information Retrieval: Fetch messages without revealing which ones
- Bandwidth Efficiency: Better performance than traditional mix networks
Protocol Overview:
class RiffleProtocol:
def __init__(self, num_servers, num_clients):
self.servers = num_servers
self.clients = num_clients
self.mixnet_servers = []
self.anon_server = None
def send_message(self, sender, recipient, message):
# Phase 1: Anonymous Broadcast
encrypted_msg = self.encrypt_for_mixnet(message, recipient)
self.broadcast_through_mixnet(encrypted_msg)
# Phase 2: Private Retrieval
recipient.retrieve_messages_privately()
def encrypt_for_mixnet(self, message, recipient):
# Onion encryption for mix network
layers = []
for server in reversed(self.mixnet_servers):
message = server.encrypt(message)
layers.append(message)
return layers
def broadcast_through_mixnet(self, encrypted_msg):
# Send through verifiable shuffle network
for server in self.mixnet_servers:
encrypted_msg = server.mix_and_verify(encrypted_msg)
# Final anonymous server receives all messages
self.anon_server.receive_anonymous_messages(encrypted_msg)
Karaoke: Distributed Private Messaging
Key Innovation: Clients collaboratively generate scheduling and routing decisions
Architecture:
class KaraokeProtocol:
def __init__(self):
self.clients = []
self.coordinators = []
self.current_round = 0
def coordinate_round(self):
# Phase 1: Coordination
schedule = self.generate_communication_schedule()
# Phase 2: Shuffling
for layer in self.shuffling_layers:
messages = layer.shuffle_messages(schedule)
# Phase 3: Delivery
self.deliver_messages_anonymously(messages)
def generate_communication_schedule(self):
# Clients jointly decide who sends when
# Uses cryptographic sortition for fairness
return self.cryptographic_scheduling()
Anonymous File Sharing and Publishing
Freenet: Distributed Anonymous Storage
Design Goals:
- Censorship Resistance: Content cannot be removed by authorities
- Anonymity: Publishers and readers remain anonymous
- Plausible Deniability: Node operators cannot know what content they store
- Self-Organization: Network routes and stores content automatically
Key Mechanisms:
class FreenetNode:
def __init__(self, node_id):
self.node_id = node_id
self.routing_table = {}
self.data_store = {}
self.location = random.uniform(0, 1) # Position in key space
def insert_content(self, key, content):
# Content is stored at nodes near key location
target_location = self.hash_to_location(key)
# Route to nodes closest to target location
path = self.route_to_location(target_location)
# Store content with probabilistic caching
for node in path:
if node.should_cache(key, content):
node.store_content(key, content)
def retrieve_content(self, key):
# Request routes toward key location
target_location = self.hash_to_location(key)
# Search probabilistically
return self.probabilistic_search(target_location)
def route_to_location(self, target):
# Route using small-world navigation
next_hop = self.find_closest_node(target)
return next_hop.route_to_location(target)
IPFS with Privacy Enhancements:
Onion Routing for IPFS:
class PrivateIPFS:
def __init__(self):
self.tor_proxy = TorProxy()
self.ipfs_node = IPFSNode()
def publish_content_anonymously(self, content):
# Step 1: Upload through Tor
content_hash = self.ipfs_node.add(content, proxy=self.tor_proxy)
# Step 2: Announce on DHT anonymously
self.announce_through_mixnet(content_hash)
return content_hash
def retrieve_content_anonymously(self, content_hash):
# Retrieve through multiple Tor circuits
circuits = self.tor_proxy.create_multiple_circuits(3)
# Parallel retrieval for plausible deniability
for circuit in circuits:
try:
content = self.ipfs_node.get(content_hash, proxy=circuit)
return content
except Exception:
continue
Censorship Resistance
Pluggable Transports: Modular protocols that disguise Tor traffic as other types of network traffic
obfs4 (Obfuscation Protocol 4):
class Obfs4Transport:
def __init__(self, bridge_info):
self.bridge_info = bridge_info
self.session_key = None
self.obfuscation_key = self.derive_key(bridge_info.secret)
def establish_connection(self):
# Step 1: Perform obfuscated handshake
handshake_data = self.create_handshake()
obfuscated_handshake = self.obfuscate_data(handshake_data)
# Step 2: Exchange keys hidden in random-looking traffic
self.session_key = self.exchange_keys_covertly()
# Step 3: All subsequent traffic looks random
return self.encrypted_channel(self.session_key)
def obfuscate_data(self, data):
# Make data indistinguishable from random bytes
return self.stream_cipher(data, self.obfuscation_key)
def create_handshake(self):
# Handshake disguised as other protocol (HTTP, TLS, etc.)
return self.mimic_protocol_handshake()
Snowflake: Domain Fronting for Tor
class SnowflakeProxy:
def __init__(self):
self.webrtc_connection = None
self.domain_front = "ajax.googleapis.com" # Trusted domain
def establish_proxy_connection(self):
# Step 1: Connect to broker through domain fronting
broker_url = f"https://{self.domain_front}/broker/api"
client_offer = self.get_client_from_broker(broker_url)
# Step 2: Establish WebRTC connection
self.webrtc_connection = self.create_webrtc_connection(client_offer)
# Step 3: Proxy Tor traffic through WebRTC
self.proxy_tor_traffic()
def proxy_tor_traffic(self):
# Relay data between Tor client and Tor bridge
while self.webrtc_connection.is_active():
data = self.webrtc_connection.receive()
self.forward_to_tor_bridge(data)
Meek: HTTP Domain Fronting
class MeekTransport:
def __init__(self, front_domain, real_domain):
self.front_domain = front_domain # CDN edge server
self.real_domain = real_domain # Actual Tor bridge
def send_data(self, tor_data):
# HTTP request appears to go to CDN
headers = {
'Host': self.front_domain,
'X-Real-Host': self.real_domain # Hidden in headers
}
# CDN forwards to real bridge based on headers
response = self.http_post(
url=f"https://{self.front_domain}/meek/",
headers=headers,
data=self.encode_tor_data(tor_data)
)
return self.decode_tor_response(response)
🔍 Detailed Explanations
Signal Protocol Integration with Anonymous Networks
Double Ratchet with Metadata Protection:
class AnonymousSignalProtocol:
def __init__(self, identity_key, tor_proxy):
self.identity_key = identity_key
self.tor_proxy = tor_proxy
self.signal_session = SignalSession(identity_key)
def send_message_anonymously(self, recipient, plaintext):
# Step 1: Encrypt with Signal's Double Ratchet
signal_ciphertext = self.signal_session.encrypt(plaintext)
# Step 2: Wrap in anonymous routing
anonymous_envelope = self.create_anonymous_envelope(
recipient, signal_ciphertext
)
# Step 3: Send through Tor
self.tor_proxy.send(anonymous_envelope)
def create_anonymous_envelope(self, recipient, ciphertext):
# Add multiple layers of encryption for anonymity
envelope = {
'recipient_pseudonym': self.hash_recipient_id(recipient),
'encrypted_content': ciphertext,
'timing_padding': self.generate_random_delay(),
'size_padding': self.pad_to_fixed_size(ciphertext)
}
return envelope
Pond: Forward-Secure Messaging
Key Features:
- Forward Security: Past messages secure even if current keys compromised
- Deniability: Cannot prove who sent a message
- Metadata Protection: Hides communication patterns
class PondMessaging:
def __init__(self, identity):
self.identity = identity
self.contacts = {}
self.message_queue = []
def create_contact(self, contact_identity):
# Generate shared secret through key exchange
shared_secret = self.ecdh_key_exchange(contact_identity)
self.contacts[contact_identity.id] = {
'shared_secret': shared_secret,
'ratchet_state': self.initialize_ratchet(shared_secret),
'message_keys': []
}
def send_message(self, contact_id, message):
contact = self.contacts[contact_id]
# Step 1: Advance ratchet for forward security
message_key = contact['ratchet_state'].advance()
# Step 2: Encrypt message
ciphertext = self.encrypt(message, message_key)
# Step 3: Add to outgoing queue with random delay
delivery_time = time.time() + random.uniform(0, 3600) # Up to 1 hour
self.message_queue.append({
'recipient': contact_id,
'ciphertext': ciphertext,
'delivery_time': delivery_time
})
def process_message_queue(self):
# Send messages at scheduled times for timing privacy
current_time = time.time()
ready_messages = [msg for msg in self.message_queue
if msg['delivery_time'] <= current_time]
for message in ready_messages:
self.deliver_message_anonymously(message)
self.message_queue.remove(message)
Vuvuzela: Scalable Private Messaging
Architecture for Large-Scale Deployment:
class VuvuzelaSystem:
def __init__(self, num_servers=3):
self.mixnet_servers = [MixServer(i) for i in range(num_servers)]
self.coordinator = CoordinatorServer()
self.users = {}
def process_messaging_round(self):
# Step 1: Collect user messages
user_messages = self.collect_user_messages()
# Step 2: Add noise for differential privacy
noisy_messages = self.add_cover_traffic(user_messages)
# Step 3: Mix through server chain
mixed_messages = noisy_messages
for server in self.mixnet_servers:
mixed_messages = server.mix_messages(mixed_messages)
# Step 4: Deliver to recipients
self.deliver_mixed_messages(mixed_messages)
def add_cover_traffic(self, real_messages):
# Add dummy messages for privacy
num_dummies = self.calculate_noise_level(len(real_messages))
dummy_messages = []
for _ in range(num_dummies):
dummy = self.create_dummy_message()
dummy_messages.append(dummy)
return real_messages + dummy_messages
def calculate_noise_level(self, real_count):
# Differential privacy noise calculation
epsilon = 1.0 # Privacy parameter
sensitivity = 1 # One user can change count by 1
noise = np.random.laplace(0, sensitivity / epsilon)
return max(0, int(noise))
💡 Practical Examples
Example 1: Anonymous Whistleblowing Platform
Scenario: Secure document submission system for investigative journalism
class SecureDropSystem:
def __init__(self):
self.tor_hidden_service = HiddenService()
self.document_store = EncryptedFileSystem()
self.journalist_keys = {}
def submit_document_anonymously(self, document, metadata):
# Step 1: Access through Tor hidden service
submission_id = self.generate_submission_id()
# Step 2: Encrypt for journalists
encrypted_copies = []
for journalist in self.journalist_keys:
encrypted_doc = self.encrypt_for_journalist(document, journalist)
encrypted_copies.append(encrypted_doc)
# Step 3: Store with anonymous access
self.document_store.store_anonymous(
submission_id, encrypted_copies, metadata
)
# Step 4: Provide secure communication channel
communication_key = self.generate_communication_key()
return {
'submission_id': submission_id,
'communication_key': communication_key,
'access_url': self.get_hidden_service_url()
}
def journalist_retrieve_documents(self, journalist_key):
# Journalists access through Tor with authenticated session
authenticated_session = self.authenticate_journalist(journalist_key)
# Retrieve and decrypt documents
encrypted_documents = self.document_store.list_for_journalist(journalist_key)
documents = []
for enc_doc in encrypted_documents:
decrypted = self.decrypt_document(enc_doc, journalist_key)
documents.append(decrypted)
return documents
def secure_communication_channel(self, submission_id, communication_key):
# Enable anonymous two-way communication
# Journalist can ask questions, source can respond
channel = SecureChannel(communication_key)
# All communication through Tor hidden service
return self.tor_hidden_service.create_channel(submission_id, channel)
# Example usage
secure_drop = SecureDropSystem()
# Anonymous source submits document
document = "sensitive_government_document.pdf"
metadata = {"topic": "corruption", "agency": "redacted"}
submission = secure_drop.submit_document_anonymously(document, metadata)
print(f"Submission ID: {submission['submission_id']}")
print(f"Access via: {submission['access_url']}")
# Journalist retrieves documents
journalist = JournalistKey("alice@newspaper.com")
documents = secure_drop.journalist_retrieve_documents(journalist)
Example 2: Anonymous Social Network
Scenario: Social media platform with strong privacy guarantees
class AnonymousSocialNetwork:
def __init__(self):
self.mixer_network = MixerNetwork()
self.content_store = DistributedHashTable()
self.pseudonym_manager = PseudonymManager()
def create_anonymous_account(self):
# Generate cryptographic pseudonym
pseudonym = self.pseudonym_manager.generate_pseudonym()
# Create unlinkable posting credentials
posting_credential = self.generate_posting_credential(pseudonym)
# Setup private communication channels
inbox_address = self.create_private_inbox(pseudonym)
return {
'pseudonym': pseudonym,
'posting_credential': posting_credential,
'inbox_address': inbox_address
}
def post_content_anonymously(self, account, content, visibility):
# Step 1: Create content with timestamp obfuscation
post = {
'content': content,
'timestamp': self.obfuscate_timestamp(),
'visibility': visibility,
'author_proof': self.create_authorship_proof(account, content)
}
# Step 2: Distribute through mix network
encrypted_post = self.encrypt_for_distribution(post)
# Step 3: Store in distributed hash table
content_hash = self.content_store.store_anonymously(encrypted_post)
# Step 4: Announce availability through privacy-preserving broadcast
self.announce_content(content_hash, visibility)
return content_hash
def follow_user_privately(self, follower_account, target_pseudonym):
# Create unlinkable subscription
subscription_token = self.create_subscription_token(
follower_account, target_pseudonym
)
# Subscribe through mix network to hide relationship
self.mixer_network.subscribe_anonymously(subscription_token)
def private_message(self, sender_account, recipient_pseudonym, message):
# Step 1: Establish anonymous communication channel
channel = self.establish_anonymous_channel(
sender_account, recipient_pseudonym
)
# Step 2: Send through mix network with padding
padded_message = self.add_timing_padding(message)
# Step 3: Deliver to recipient's private inbox
self.mixer_network.deliver_to_inbox(
recipient_pseudonym, padded_message
)
# Example usage
social_network = AnonymousSocialNetwork()
# Create anonymous accounts
alice_account = social_network.create_anonymous_account()
bob_account = social_network.create_anonymous_account()
# Post content anonymously
content_hash = social_network.post_content_anonymously(
alice_account,
"This is an anonymous post about privacy",
visibility="public"
)
# Follow someone privately
social_network.follow_user_privately(
bob_account,
alice_account['pseudonym']
)
# Send private message
social_network.private_message(
bob_account,
alice_account['pseudonym'],
"Hello, I appreciate your privacy advocacy!"
)
Example 3: Censorship-Resistant News Distribution
Scenario: News organization operating under authoritarian censorship
class CensorshipResistantNews:
def __init__(self):
self.domain_fronting = DomainFrontingService()
self.tor_bridges = TorBridgeNetwork()
self.distributed_mirrors = DistributedMirrorNetwork()
self.pluggable_transports = PluggableTransportManager()
def publish_article(self, article, urgency_level):
# Step 1: Encrypt article for distribution
encrypted_article = self.encrypt_article(article)
# Step 2: Distribute through multiple channels based on urgency
if urgency_level == "critical":
self.emergency_distribution(encrypted_article)
else:
self.standard_distribution(encrypted_article)
def emergency_distribution(self, article):
# Use all available channels simultaneously
channels = [
self.domain_fronting,
self.tor_bridges,
self.distributed_mirrors,
self.mesh_networks,
self.satellite_broadcast
]
for channel in channels:
channel.distribute_urgently(article)
def setup_reader_access(self, reader_location):
# Determine best access method based on censorship environment
censorship_level = self.assess_censorship(reader_location)
if censorship_level == "high":
# Use multiple layers of circumvention
access_methods = [
self.setup_pluggable_transport(reader_location),
self.provide_bridge_info(reader_location),
self.enable_mesh_access(reader_location)
]
elif censorship_level == "medium":
access_methods = [
self.setup_domain_fronting(reader_location),
self.provide_mirror_list(reader_location)
]
else:
access_methods = [self.direct_access()]
return access_methods
def create_mobile_app_distribution(self):
# Distribute news app through censorship-resistant channels
# Method 1: F-Droid with Tor
fdroid_package = self.create_fdroid_package()
self.distribute_via_tor_hidden_service(fdroid_package)
# Method 2: APK through IPFS
apk_file = self.build_android_apk()
ipfs_hash = self.upload_to_ipfs(apk_file)
self.announce_ipfs_hash_anonymously(ipfs_hash)
# Method 3: Progressive Web App
pwa = self.create_progressive_web_app()
self.deploy_pwa_on_distributed_hosting(pwa)
def reader_feedback_channel(self):
# Anonymous feedback system for readers
feedback_system = AnonymousFeedbackSystem()
# Multiple submission methods
feedback_channels = [
feedback_system.tor_hidden_service(),
feedback_system.signal_anonymous_group(),
feedback_system.encrypted_email_dropbox(),
feedback_system.decentralized_message_board()
]
return feedback_channels
# Example usage
news_service = CensorshipResistantNews()
# Publish breaking news with high urgency
breaking_news = {
"headline": "Government Surveillance Program Exposed",
"content": "Leaked documents reveal extensive monitoring...",
"sources": ["encrypted_documents.pdf", "witness_testimony.mp4"],
"verification": "cryptographic_signatures"
}
news_service.publish_article(breaking_news, urgency_level="critical")
# Setup access for readers in different regions
china_access = news_service.setup_reader_access("china")
iran_access = news_service.setup_reader_access("iran")
free_country_access = news_service.setup_reader_access("netherlands")
print("Access methods for highly censored regions:", china_access)
print("Access methods for moderately censored regions:", iran_access)
print("Direct access for free regions:", free_country_access)