Compare commits
2 Commits
a6cd9632ee
...
96553b27bd
| Author | SHA1 | Date | |
|---|---|---|---|
|
96553b27bd | ||
|
|
4832ba751f |
@ -1,109 +0,0 @@
|
|||||||
import socket
|
|
||||||
import time
|
|
||||||
import select
|
|
||||||
from PyQt5.QtCore import QThread, pyqtSignal
|
|
||||||
from client_state import ClientState
|
|
||||||
|
|
||||||
class PhoneClient(QThread):
|
|
||||||
data_received = pyqtSignal(bytes, int)
|
|
||||||
state_changed = pyqtSignal(str, str, int)
|
|
||||||
|
|
||||||
def __init__(self, client_id):
|
|
||||||
super().__init__()
|
|
||||||
self.host = "localhost"
|
|
||||||
self.port = 12345
|
|
||||||
self.client_id = client_id
|
|
||||||
self.sock = None
|
|
||||||
self.running = True
|
|
||||||
self.state = ClientState(client_id)
|
|
||||||
|
|
||||||
def connect_socket(self):
|
|
||||||
retries = 3
|
|
||||||
for attempt in range(retries):
|
|
||||||
try:
|
|
||||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
||||||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
|
|
||||||
self.sock.settimeout(120)
|
|
||||||
self.sock.connect((self.host, self.port))
|
|
||||||
print(f"Client {self.client_id} connected to {self.host}:{self.port}")
|
|
||||||
return True
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Client {self.client_id} connection attempt {attempt + 1} failed: {e}")
|
|
||||||
if attempt < retries - 1:
|
|
||||||
time.sleep(1)
|
|
||||||
self.sock = None
|
|
||||||
return False
|
|
||||||
|
|
||||||
def run(self):
|
|
||||||
while self.running:
|
|
||||||
if not self.sock:
|
|
||||||
if not self.connect_socket():
|
|
||||||
print(f"Client {self.client_id} failed to connect after retries")
|
|
||||||
self.state_changed.emit("CALL_END", "", self.client_id)
|
|
||||||
break
|
|
||||||
try:
|
|
||||||
while self.running:
|
|
||||||
self.state.process_command(self)
|
|
||||||
self.state.check_handshake_timeout(self)
|
|
||||||
|
|
||||||
# Always check for incoming data, even during handshake
|
|
||||||
if self.sock is None:
|
|
||||||
print(f"Client {self.client_id} socket is None, exiting inner loop")
|
|
||||||
break
|
|
||||||
readable, _, _ = select.select([self.sock], [], [], 0.01)
|
|
||||||
if readable:
|
|
||||||
try:
|
|
||||||
if self.sock is None:
|
|
||||||
print(f"Client {self.client_id} socket is None before recv, exiting")
|
|
||||||
break
|
|
||||||
data = self.sock.recv(1024)
|
|
||||||
if not data:
|
|
||||||
print(f"Client {self.client_id} disconnected")
|
|
||||||
self.state_changed.emit("CALL_END", "", self.client_id)
|
|
||||||
break
|
|
||||||
self.state.handle_data(self, data)
|
|
||||||
except socket.error as e:
|
|
||||||
print(f"Client {self.client_id} socket error: {e}")
|
|
||||||
self.state_changed.emit("CALL_END", "", self.client_id)
|
|
||||||
break
|
|
||||||
|
|
||||||
self.msleep(1)
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Client {self.client_id} unexpected error in run loop: {e}")
|
|
||||||
self.state_changed.emit("CALL_END", "", self.client_id)
|
|
||||||
break
|
|
||||||
finally:
|
|
||||||
if self.sock:
|
|
||||||
try:
|
|
||||||
self.sock.close()
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Client {self.client_id} error closing socket: {e}")
|
|
||||||
self.sock = None
|
|
||||||
|
|
||||||
def send(self, message):
|
|
||||||
if self.sock and self.running:
|
|
||||||
try:
|
|
||||||
if isinstance(message, str):
|
|
||||||
data = message.encode('utf-8')
|
|
||||||
self.sock.send(data)
|
|
||||||
print(f"Client {self.client_id} sent: {message}, length={len(data)}")
|
|
||||||
else:
|
|
||||||
self.sock.send(message)
|
|
||||||
print(f"Client {self.client_id} sent binary data, length={len(message)}")
|
|
||||||
except socket.error as e:
|
|
||||||
print(f"Client {self.client_id} send error: {e}")
|
|
||||||
self.state_changed.emit("CALL_END", "", self.client_id)
|
|
||||||
|
|
||||||
def stop(self):
|
|
||||||
self.running = False
|
|
||||||
if self.sock:
|
|
||||||
try:
|
|
||||||
self.sock.close()
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Client {self.client_id} error closing socket in stop: {e}")
|
|
||||||
self.sock = None
|
|
||||||
self.quit()
|
|
||||||
self.wait(1000)
|
|
||||||
|
|
||||||
def start_handshake(self, initiator, keypair, peer_pubkey):
|
|
||||||
self.state.start_handshake(initiator, keypair, peer_pubkey)
|
|
||||||
@ -1,714 +0,0 @@
|
|||||||
"""
|
|
||||||
Voice codec integration for encrypted voice over GSM.
|
|
||||||
Implements Codec2 compression with FSK modulation for transmitting
|
|
||||||
encrypted voice data over standard GSM voice channels.
|
|
||||||
"""
|
|
||||||
|
|
||||||
import array
|
|
||||||
import math
|
|
||||||
import struct
|
|
||||||
from typing import Optional, Tuple, List
|
|
||||||
from dataclasses import dataclass
|
|
||||||
from enum import IntEnum
|
|
||||||
|
|
||||||
try:
|
|
||||||
import numpy as np
|
|
||||||
HAS_NUMPY = True
|
|
||||||
except ImportError:
|
|
||||||
HAS_NUMPY = False
|
|
||||||
|
|
||||||
# ANSI colors
|
|
||||||
RED = "\033[91m"
|
|
||||||
GREEN = "\033[92m"
|
|
||||||
YELLOW = "\033[93m"
|
|
||||||
BLUE = "\033[94m"
|
|
||||||
RESET = "\033[0m"
|
|
||||||
|
|
||||||
|
|
||||||
class Codec2Mode(IntEnum):
|
|
||||||
"""Codec2 bitrate modes."""
|
|
||||||
MODE_3200 = 0 # 3200 bps
|
|
||||||
MODE_2400 = 1 # 2400 bps
|
|
||||||
MODE_1600 = 2 # 1600 bps
|
|
||||||
MODE_1400 = 3 # 1400 bps
|
|
||||||
MODE_1300 = 4 # 1300 bps
|
|
||||||
MODE_1200 = 5 # 1200 bps (recommended for robustness)
|
|
||||||
MODE_700C = 6 # 700 bps
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class Codec2Frame:
|
|
||||||
"""Represents a single Codec2 compressed voice frame."""
|
|
||||||
mode: Codec2Mode
|
|
||||||
bits: bytes
|
|
||||||
timestamp: float
|
|
||||||
frame_number: int
|
|
||||||
|
|
||||||
|
|
||||||
class Codec2Wrapper:
|
|
||||||
"""
|
|
||||||
Wrapper for Codec2 voice codec.
|
|
||||||
In production, this would use py_codec2 or ctypes bindings to libcodec2.
|
|
||||||
This is a simulation interface for protocol development.
|
|
||||||
"""
|
|
||||||
|
|
||||||
# Frame sizes in bits for each mode
|
|
||||||
FRAME_BITS = {
|
|
||||||
Codec2Mode.MODE_3200: 64,
|
|
||||||
Codec2Mode.MODE_2400: 48,
|
|
||||||
Codec2Mode.MODE_1600: 64,
|
|
||||||
Codec2Mode.MODE_1400: 56,
|
|
||||||
Codec2Mode.MODE_1300: 52,
|
|
||||||
Codec2Mode.MODE_1200: 48,
|
|
||||||
Codec2Mode.MODE_700C: 28
|
|
||||||
}
|
|
||||||
|
|
||||||
# Frame duration in ms
|
|
||||||
FRAME_MS = {
|
|
||||||
Codec2Mode.MODE_3200: 20,
|
|
||||||
Codec2Mode.MODE_2400: 20,
|
|
||||||
Codec2Mode.MODE_1600: 40,
|
|
||||||
Codec2Mode.MODE_1400: 40,
|
|
||||||
Codec2Mode.MODE_1300: 40,
|
|
||||||
Codec2Mode.MODE_1200: 40,
|
|
||||||
Codec2Mode.MODE_700C: 40
|
|
||||||
}
|
|
||||||
|
|
||||||
def __init__(self, mode: Codec2Mode = Codec2Mode.MODE_1200):
|
|
||||||
"""
|
|
||||||
Initialize Codec2 wrapper.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
mode: Codec2 bitrate mode (default 1200 bps for robustness)
|
|
||||||
"""
|
|
||||||
self.mode = mode
|
|
||||||
self.frame_bits = self.FRAME_BITS[mode]
|
|
||||||
self.frame_bytes = (self.frame_bits + 7) // 8
|
|
||||||
self.frame_ms = self.FRAME_MS[mode]
|
|
||||||
self.frame_samples = int(8000 * self.frame_ms / 1000) # 8kHz sampling
|
|
||||||
self.frame_counter = 0
|
|
||||||
|
|
||||||
# Quiet initialization - no print
|
|
||||||
|
|
||||||
def encode(self, audio_samples) -> Optional[Codec2Frame]:
|
|
||||||
"""
|
|
||||||
Encode PCM audio samples to Codec2 frame.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
audio_samples: PCM samples (8kHz, 16-bit signed)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Codec2Frame or None if insufficient samples
|
|
||||||
"""
|
|
||||||
if len(audio_samples) < self.frame_samples:
|
|
||||||
return None
|
|
||||||
|
|
||||||
# In production: call codec2_encode(state, bits, samples)
|
|
||||||
# Simulation: create pseudo-compressed data
|
|
||||||
compressed = self._simulate_compression(audio_samples[:self.frame_samples])
|
|
||||||
|
|
||||||
frame = Codec2Frame(
|
|
||||||
mode=self.mode,
|
|
||||||
bits=compressed,
|
|
||||||
timestamp=self.frame_counter * self.frame_ms / 1000.0,
|
|
||||||
frame_number=self.frame_counter
|
|
||||||
)
|
|
||||||
|
|
||||||
self.frame_counter += 1
|
|
||||||
return frame
|
|
||||||
|
|
||||||
def decode(self, frame: Codec2Frame):
|
|
||||||
"""
|
|
||||||
Decode Codec2 frame to PCM audio samples.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
frame: Codec2 compressed frame
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
PCM samples (8kHz, 16-bit signed)
|
|
||||||
"""
|
|
||||||
if frame.mode != self.mode:
|
|
||||||
raise ValueError(f"Frame mode {frame.mode} doesn't match decoder mode {self.mode}")
|
|
||||||
|
|
||||||
# In production: call codec2_decode(state, samples, bits)
|
|
||||||
# Simulation: decompress to audio
|
|
||||||
return self._simulate_decompression(frame.bits)
|
|
||||||
|
|
||||||
def _simulate_compression(self, samples) -> bytes:
|
|
||||||
"""Simulate Codec2 compression (for testing)."""
|
|
||||||
# Convert to list if needed
|
|
||||||
if hasattr(samples, 'tolist'):
|
|
||||||
sample_list = samples.tolist()
|
|
||||||
elif hasattr(samples, '__iter__'):
|
|
||||||
sample_list = list(samples)
|
|
||||||
else:
|
|
||||||
sample_list = samples
|
|
||||||
|
|
||||||
# Extract basic features for simulation
|
|
||||||
if HAS_NUMPY and hasattr(samples, '__array__'):
|
|
||||||
# Convert to numpy array if needed
|
|
||||||
np_samples = np.asarray(samples, dtype=np.float32)
|
|
||||||
if len(np_samples) > 0:
|
|
||||||
mean_square = np.mean(np_samples ** 2)
|
|
||||||
energy = np.sqrt(mean_square) if not np.isnan(mean_square) else 0.0
|
|
||||||
zero_crossings = np.sum(np.diff(np.sign(np_samples)) != 0)
|
|
||||||
else:
|
|
||||||
energy = 0.0
|
|
||||||
zero_crossings = 0
|
|
||||||
else:
|
|
||||||
# Manual calculation without numpy
|
|
||||||
if sample_list and len(sample_list) > 0:
|
|
||||||
energy = math.sqrt(sum(s**2 for s in sample_list) / len(sample_list))
|
|
||||||
zero_crossings = sum(1 for i in range(1, len(sample_list))
|
|
||||||
if (sample_list[i-1] >= 0) != (sample_list[i] >= 0))
|
|
||||||
else:
|
|
||||||
energy = 0.0
|
|
||||||
zero_crossings = 0
|
|
||||||
|
|
||||||
# Pack into bytes (simplified)
|
|
||||||
# Ensure values are valid
|
|
||||||
energy_int = max(0, min(65535, int(energy)))
|
|
||||||
zc_int = max(0, min(65535, int(zero_crossings)))
|
|
||||||
data = struct.pack('<HH', energy_int, zc_int)
|
|
||||||
|
|
||||||
# Pad to expected frame size
|
|
||||||
data += b'\x00' * (self.frame_bytes - len(data))
|
|
||||||
|
|
||||||
return data[:self.frame_bytes]
|
|
||||||
|
|
||||||
def _simulate_decompression(self, compressed: bytes):
|
|
||||||
"""Simulate Codec2 decompression (for testing)."""
|
|
||||||
# Unpack features
|
|
||||||
if len(compressed) >= 4:
|
|
||||||
energy, zero_crossings = struct.unpack('<HH', compressed[:4])
|
|
||||||
else:
|
|
||||||
energy, zero_crossings = 1000, 100
|
|
||||||
|
|
||||||
# Generate synthetic speech-like signal
|
|
||||||
if HAS_NUMPY:
|
|
||||||
t = np.linspace(0, self.frame_ms/1000, self.frame_samples)
|
|
||||||
|
|
||||||
# Base frequency from zero crossings
|
|
||||||
freq = zero_crossings * 10 # Simplified mapping
|
|
||||||
|
|
||||||
# Generate harmonics
|
|
||||||
signal = np.zeros(self.frame_samples)
|
|
||||||
for harmonic in range(1, 4):
|
|
||||||
signal += np.sin(2 * np.pi * freq * harmonic * t) / harmonic
|
|
||||||
|
|
||||||
# Apply energy envelope
|
|
||||||
signal *= energy / 10000.0
|
|
||||||
|
|
||||||
# Convert to 16-bit PCM
|
|
||||||
return (signal * 32767).astype(np.int16)
|
|
||||||
else:
|
|
||||||
# Manual generation without numpy
|
|
||||||
samples = []
|
|
||||||
freq = zero_crossings * 10
|
|
||||||
|
|
||||||
for i in range(self.frame_samples):
|
|
||||||
t = i / 8000.0 # 8kHz sample rate
|
|
||||||
value = 0
|
|
||||||
for harmonic in range(1, 4):
|
|
||||||
value += math.sin(2 * math.pi * freq * harmonic * t) / harmonic
|
|
||||||
|
|
||||||
value *= energy / 10000.0
|
|
||||||
# Clamp to 16-bit range
|
|
||||||
sample = int(value * 32767)
|
|
||||||
sample = max(-32768, min(32767, sample))
|
|
||||||
samples.append(sample)
|
|
||||||
|
|
||||||
return array.array('h', samples)
|
|
||||||
|
|
||||||
|
|
||||||
class FSKModem:
|
|
||||||
"""
|
|
||||||
4-FSK modem for transmitting digital data over voice channels.
|
|
||||||
Designed to survive GSM/AMR/EVS vocoders.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, sample_rate: int = 8000, baud_rate: int = 600):
|
|
||||||
"""
|
|
||||||
Initialize FSK modem.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
sample_rate: Audio sample rate (Hz)
|
|
||||||
baud_rate: Symbol rate (baud)
|
|
||||||
"""
|
|
||||||
self.sample_rate = sample_rate
|
|
||||||
self.baud_rate = baud_rate
|
|
||||||
self.samples_per_symbol = int(sample_rate / baud_rate)
|
|
||||||
|
|
||||||
# 4-FSK frequencies (300-3400 Hz band)
|
|
||||||
self.frequencies = [
|
|
||||||
600, # 00
|
|
||||||
1200, # 01
|
|
||||||
1800, # 10
|
|
||||||
2400 # 11
|
|
||||||
]
|
|
||||||
|
|
||||||
# Preamble for synchronization (800 Hz, 100ms)
|
|
||||||
self.preamble_freq = 800
|
|
||||||
self.preamble_duration = 0.1 # seconds
|
|
||||||
|
|
||||||
# Quiet initialization - no print
|
|
||||||
|
|
||||||
def modulate(self, data: bytes, add_preamble: bool = True):
|
|
||||||
"""
|
|
||||||
Modulate binary data to FSK audio signal.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
data: Binary data to modulate
|
|
||||||
add_preamble: Whether to add synchronization preamble
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Audio signal (normalized float32 array or list)
|
|
||||||
"""
|
|
||||||
# Convert bytes to dibits (2-bit symbols)
|
|
||||||
symbols = []
|
|
||||||
for byte in data:
|
|
||||||
symbols.extend([
|
|
||||||
(byte >> 6) & 0x03,
|
|
||||||
(byte >> 4) & 0x03,
|
|
||||||
(byte >> 2) & 0x03,
|
|
||||||
byte & 0x03
|
|
||||||
])
|
|
||||||
|
|
||||||
# Generate audio signal
|
|
||||||
signal = []
|
|
||||||
|
|
||||||
# Add preamble
|
|
||||||
if add_preamble:
|
|
||||||
preamble_samples = int(self.preamble_duration * self.sample_rate)
|
|
||||||
if HAS_NUMPY:
|
|
||||||
t = np.arange(preamble_samples) / self.sample_rate
|
|
||||||
preamble = np.sin(2 * np.pi * self.preamble_freq * t)
|
|
||||||
signal.extend(preamble)
|
|
||||||
else:
|
|
||||||
for i in range(preamble_samples):
|
|
||||||
t = i / self.sample_rate
|
|
||||||
value = math.sin(2 * math.pi * self.preamble_freq * t)
|
|
||||||
signal.append(value)
|
|
||||||
|
|
||||||
# Modulate symbols
|
|
||||||
for symbol in symbols:
|
|
||||||
freq = self.frequencies[symbol]
|
|
||||||
if HAS_NUMPY:
|
|
||||||
t = np.arange(self.samples_per_symbol) / self.sample_rate
|
|
||||||
tone = np.sin(2 * np.pi * freq * t)
|
|
||||||
signal.extend(tone)
|
|
||||||
else:
|
|
||||||
for i in range(self.samples_per_symbol):
|
|
||||||
t = i / self.sample_rate
|
|
||||||
value = math.sin(2 * math.pi * freq * t)
|
|
||||||
signal.append(value)
|
|
||||||
|
|
||||||
# Apply smoothing to reduce clicks
|
|
||||||
if HAS_NUMPY:
|
|
||||||
audio = np.array(signal, dtype=np.float32)
|
|
||||||
else:
|
|
||||||
audio = array.array('f', signal)
|
|
||||||
audio = self._apply_envelope(audio)
|
|
||||||
|
|
||||||
return audio
|
|
||||||
|
|
||||||
def demodulate(self, audio) -> Tuple[bytes, float]:
|
|
||||||
"""
|
|
||||||
Demodulate FSK audio signal to binary data.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
audio: Audio signal
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Tuple of (demodulated data, confidence score)
|
|
||||||
"""
|
|
||||||
# Find preamble
|
|
||||||
preamble_start = self._find_preamble(audio)
|
|
||||||
if preamble_start < 0:
|
|
||||||
return b'', 0.0
|
|
||||||
|
|
||||||
# Skip preamble
|
|
||||||
data_start = preamble_start + int(self.preamble_duration * self.sample_rate)
|
|
||||||
|
|
||||||
# Demodulate symbols
|
|
||||||
symbols = []
|
|
||||||
confidence_scores = []
|
|
||||||
|
|
||||||
pos = data_start
|
|
||||||
while pos + self.samples_per_symbol <= len(audio):
|
|
||||||
symbol_audio = audio[pos:pos + self.samples_per_symbol]
|
|
||||||
symbol, confidence = self._demodulate_symbol(symbol_audio)
|
|
||||||
symbols.append(symbol)
|
|
||||||
confidence_scores.append(confidence)
|
|
||||||
pos += self.samples_per_symbol
|
|
||||||
|
|
||||||
# Convert symbols to bytes
|
|
||||||
data = bytearray()
|
|
||||||
for i in range(0, len(symbols), 4):
|
|
||||||
if i + 3 < len(symbols):
|
|
||||||
byte = (symbols[i] << 6) | (symbols[i+1] << 4) | (symbols[i+2] << 2) | symbols[i+3]
|
|
||||||
data.append(byte)
|
|
||||||
|
|
||||||
if HAS_NUMPY and confidence_scores:
|
|
||||||
avg_confidence = np.mean(confidence_scores)
|
|
||||||
else:
|
|
||||||
avg_confidence = sum(confidence_scores) / len(confidence_scores) if confidence_scores else 0.0
|
|
||||||
return bytes(data), avg_confidence
|
|
||||||
|
|
||||||
def _find_preamble(self, audio) -> int:
|
|
||||||
"""Find preamble in audio signal."""
|
|
||||||
# Simple energy-based detection
|
|
||||||
window_size = int(0.01 * self.sample_rate) # 10ms window
|
|
||||||
|
|
||||||
if HAS_NUMPY:
|
|
||||||
for i in range(0, len(audio) - window_size, window_size // 2):
|
|
||||||
window = audio[i:i + window_size]
|
|
||||||
|
|
||||||
# Check for preamble frequency
|
|
||||||
fft = np.fft.fft(window)
|
|
||||||
freqs = np.fft.fftfreq(len(window), 1/self.sample_rate)
|
|
||||||
|
|
||||||
# Find peak near preamble frequency
|
|
||||||
idx = np.argmax(np.abs(fft[:len(fft)//2]))
|
|
||||||
peak_freq = abs(freqs[idx])
|
|
||||||
|
|
||||||
if abs(peak_freq - self.preamble_freq) < 50: # 50 Hz tolerance
|
|
||||||
return i
|
|
||||||
else:
|
|
||||||
# Simple zero-crossing based detection without FFT
|
|
||||||
for i in range(0, len(audio) - window_size, window_size // 2):
|
|
||||||
window = list(audio[i:i + window_size])
|
|
||||||
|
|
||||||
# Count zero crossings
|
|
||||||
zero_crossings = 0
|
|
||||||
for j in range(1, len(window)):
|
|
||||||
if (window[j-1] >= 0) != (window[j] >= 0):
|
|
||||||
zero_crossings += 1
|
|
||||||
|
|
||||||
# Estimate frequency from zero crossings
|
|
||||||
estimated_freq = (zero_crossings * self.sample_rate) / (2 * len(window))
|
|
||||||
|
|
||||||
if abs(estimated_freq - self.preamble_freq) < 100: # 100 Hz tolerance
|
|
||||||
return i
|
|
||||||
|
|
||||||
return -1
|
|
||||||
|
|
||||||
def _demodulate_symbol(self, audio) -> Tuple[int, float]:
|
|
||||||
"""Demodulate a single FSK symbol."""
|
|
||||||
if HAS_NUMPY:
|
|
||||||
# FFT-based demodulation
|
|
||||||
fft = np.fft.fft(audio)
|
|
||||||
freqs = np.fft.fftfreq(len(audio), 1/self.sample_rate)
|
|
||||||
magnitude = np.abs(fft[:len(fft)//2])
|
|
||||||
|
|
||||||
# Find energy at each FSK frequency
|
|
||||||
energies = []
|
|
||||||
for freq in self.frequencies:
|
|
||||||
idx = np.argmin(np.abs(freqs[:len(freqs)//2] - freq))
|
|
||||||
energy = magnitude[idx]
|
|
||||||
energies.append(energy)
|
|
||||||
|
|
||||||
# Select symbol with highest energy
|
|
||||||
symbol = np.argmax(energies)
|
|
||||||
else:
|
|
||||||
# Goertzel algorithm for specific frequency detection
|
|
||||||
audio_list = list(audio) if hasattr(audio, '__iter__') else audio
|
|
||||||
energies = []
|
|
||||||
|
|
||||||
for freq in self.frequencies:
|
|
||||||
# Goertzel algorithm
|
|
||||||
omega = 2 * math.pi * freq / self.sample_rate
|
|
||||||
coeff = 2 * math.cos(omega)
|
|
||||||
|
|
||||||
s_prev = 0
|
|
||||||
s_prev2 = 0
|
|
||||||
|
|
||||||
for sample in audio_list:
|
|
||||||
s = sample + coeff * s_prev - s_prev2
|
|
||||||
s_prev2 = s_prev
|
|
||||||
s_prev = s
|
|
||||||
|
|
||||||
# Calculate magnitude
|
|
||||||
power = s_prev2 * s_prev2 + s_prev * s_prev - coeff * s_prev * s_prev2
|
|
||||||
energies.append(math.sqrt(abs(power)))
|
|
||||||
|
|
||||||
# Select symbol with highest energy
|
|
||||||
symbol = energies.index(max(energies))
|
|
||||||
|
|
||||||
# Confidence is ratio of strongest to second strongest
|
|
||||||
sorted_energies = sorted(energies, reverse=True)
|
|
||||||
confidence = sorted_energies[0] / (sorted_energies[1] + 1e-6)
|
|
||||||
|
|
||||||
return symbol, min(confidence, 10.0) / 10.0
|
|
||||||
|
|
||||||
def _apply_envelope(self, audio):
|
|
||||||
"""Apply smoothing envelope to reduce clicks."""
|
|
||||||
# Simple raised cosine envelope
|
|
||||||
ramp_samples = int(0.002 * self.sample_rate) # 2ms ramps
|
|
||||||
|
|
||||||
if len(audio) > 2 * ramp_samples:
|
|
||||||
if HAS_NUMPY:
|
|
||||||
# Fade in
|
|
||||||
t = np.linspace(0, np.pi/2, ramp_samples)
|
|
||||||
audio[:ramp_samples] *= np.sin(t) ** 2
|
|
||||||
|
|
||||||
# Fade out
|
|
||||||
audio[-ramp_samples:] *= np.sin(t[::-1]) ** 2
|
|
||||||
else:
|
|
||||||
# Manual fade in
|
|
||||||
for i in range(ramp_samples):
|
|
||||||
t = (i / ramp_samples) * (math.pi / 2)
|
|
||||||
factor = math.sin(t) ** 2
|
|
||||||
audio[i] *= factor
|
|
||||||
|
|
||||||
# Manual fade out
|
|
||||||
for i in range(ramp_samples):
|
|
||||||
t = ((ramp_samples - 1 - i) / ramp_samples) * (math.pi / 2)
|
|
||||||
factor = math.sin(t) ** 2
|
|
||||||
audio[-(i+1)] *= factor
|
|
||||||
|
|
||||||
return audio
|
|
||||||
|
|
||||||
|
|
||||||
class VoiceProtocol:
|
|
||||||
"""
|
|
||||||
Integrates voice codec and modem with the Icing protocol
|
|
||||||
for encrypted voice transmission over GSM.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, protocol_instance):
|
|
||||||
"""
|
|
||||||
Initialize voice protocol handler.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
protocol_instance: IcingProtocol instance
|
|
||||||
"""
|
|
||||||
self.protocol = protocol_instance
|
|
||||||
self.codec = Codec2Wrapper(Codec2Mode.MODE_1200)
|
|
||||||
self.modem = FSKModem(sample_rate=8000, baud_rate=600)
|
|
||||||
|
|
||||||
# Voice crypto state
|
|
||||||
self.voice_iv_counter = 0
|
|
||||||
self.voice_sequence = 0
|
|
||||||
|
|
||||||
# Buffers
|
|
||||||
if HAS_NUMPY:
|
|
||||||
self.audio_buffer = np.array([], dtype=np.int16)
|
|
||||||
else:
|
|
||||||
self.audio_buffer = array.array('h') # 16-bit signed integers
|
|
||||||
self.frame_buffer = []
|
|
||||||
|
|
||||||
print(f"{GREEN}[VOICE]{RESET} Voice protocol initialized")
|
|
||||||
|
|
||||||
def process_voice_input(self, audio_samples):
|
|
||||||
"""
|
|
||||||
Process voice input: compress, encrypt, and modulate.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
audio_samples: PCM audio samples (8kHz, 16-bit)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Modulated audio signal ready for transmission (numpy array or array.array)
|
|
||||||
"""
|
|
||||||
# Add to buffer
|
|
||||||
if HAS_NUMPY:
|
|
||||||
self.audio_buffer = np.concatenate([self.audio_buffer, audio_samples])
|
|
||||||
else:
|
|
||||||
self.audio_buffer.extend(audio_samples)
|
|
||||||
|
|
||||||
# Process complete frames
|
|
||||||
modulated_audio = []
|
|
||||||
|
|
||||||
while len(self.audio_buffer) >= self.codec.frame_samples:
|
|
||||||
# Extract frame
|
|
||||||
if HAS_NUMPY:
|
|
||||||
frame_audio = self.audio_buffer[:self.codec.frame_samples]
|
|
||||||
self.audio_buffer = self.audio_buffer[self.codec.frame_samples:]
|
|
||||||
else:
|
|
||||||
frame_audio = array.array('h', self.audio_buffer[:self.codec.frame_samples])
|
|
||||||
del self.audio_buffer[:self.codec.frame_samples]
|
|
||||||
|
|
||||||
# Compress with Codec2
|
|
||||||
compressed_frame = self.codec.encode(frame_audio)
|
|
||||||
if not compressed_frame:
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Encrypt frame
|
|
||||||
encrypted = self._encrypt_voice_frame(compressed_frame)
|
|
||||||
|
|
||||||
# Add FEC
|
|
||||||
protected = self._add_fec(encrypted)
|
|
||||||
|
|
||||||
# Modulate to audio
|
|
||||||
audio_signal = self.modem.modulate(protected, add_preamble=True)
|
|
||||||
modulated_audio.append(audio_signal)
|
|
||||||
|
|
||||||
if modulated_audio:
|
|
||||||
if HAS_NUMPY:
|
|
||||||
return np.concatenate(modulated_audio)
|
|
||||||
else:
|
|
||||||
# Concatenate array.array objects
|
|
||||||
result = array.array('f')
|
|
||||||
for audio in modulated_audio:
|
|
||||||
result.extend(audio)
|
|
||||||
return result
|
|
||||||
return None
|
|
||||||
|
|
||||||
def process_voice_output(self, modulated_audio):
|
|
||||||
"""
|
|
||||||
Process received audio: demodulate, decrypt, and decompress.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
modulated_audio: Received FSK-modulated audio
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Decoded PCM audio samples (numpy array or array.array)
|
|
||||||
"""
|
|
||||||
# Demodulate
|
|
||||||
data, confidence = self.modem.demodulate(modulated_audio)
|
|
||||||
|
|
||||||
if confidence < 0.5:
|
|
||||||
print(f"{YELLOW}[VOICE]{RESET} Low demodulation confidence: {confidence:.2f}")
|
|
||||||
return None
|
|
||||||
|
|
||||||
# Remove FEC
|
|
||||||
frame_data = self._remove_fec(data)
|
|
||||||
if not frame_data:
|
|
||||||
return None
|
|
||||||
|
|
||||||
# Decrypt
|
|
||||||
compressed_frame = self._decrypt_voice_frame(frame_data)
|
|
||||||
if not compressed_frame:
|
|
||||||
return None
|
|
||||||
|
|
||||||
# Decompress
|
|
||||||
audio_samples = self.codec.decode(compressed_frame)
|
|
||||||
|
|
||||||
return audio_samples
|
|
||||||
|
|
||||||
def _encrypt_voice_frame(self, frame: Codec2Frame) -> bytes:
|
|
||||||
"""Encrypt a voice frame using ChaCha20-CTR."""
|
|
||||||
if not self.protocol.hkdf_key:
|
|
||||||
raise ValueError("No encryption key available")
|
|
||||||
|
|
||||||
# Prepare frame data
|
|
||||||
frame_data = struct.pack('<BIH',
|
|
||||||
frame.mode,
|
|
||||||
frame.frame_number,
|
|
||||||
len(frame.bits)
|
|
||||||
) + frame.bits
|
|
||||||
|
|
||||||
# Generate IV for this frame (ChaCha20 needs 16 bytes)
|
|
||||||
iv = struct.pack('<Q', self.voice_iv_counter) + b'\x00' * 8 # 8 + 8 = 16 bytes
|
|
||||||
self.voice_iv_counter += 1
|
|
||||||
|
|
||||||
# Encrypt using ChaCha20
|
|
||||||
from encryption import chacha20_encrypt
|
|
||||||
key = bytes.fromhex(self.protocol.hkdf_key)
|
|
||||||
encrypted = chacha20_encrypt(frame_data, key, iv)
|
|
||||||
|
|
||||||
# Add sequence number and IV hint
|
|
||||||
return struct.pack('<HQ', self.voice_sequence, self.voice_iv_counter) + encrypted
|
|
||||||
|
|
||||||
def _decrypt_voice_frame(self, data: bytes) -> Optional[Codec2Frame]:
|
|
||||||
"""Decrypt a voice frame."""
|
|
||||||
if len(data) < 10:
|
|
||||||
return None
|
|
||||||
|
|
||||||
# Extract sequence and IV hint
|
|
||||||
sequence, iv_hint = struct.unpack('<HQ', data[:10])
|
|
||||||
encrypted = data[10:]
|
|
||||||
|
|
||||||
# Generate IV (16 bytes for ChaCha20)
|
|
||||||
iv = struct.pack('<Q', iv_hint) + b'\x00' * 8
|
|
||||||
|
|
||||||
# Decrypt
|
|
||||||
from encryption import chacha20_decrypt
|
|
||||||
key = bytes.fromhex(self.protocol.hkdf_key)
|
|
||||||
|
|
||||||
try:
|
|
||||||
decrypted = chacha20_decrypt(encrypted, key, iv)
|
|
||||||
|
|
||||||
# Parse frame
|
|
||||||
mode, frame_num, bits_len = struct.unpack('<BIH', decrypted[:7])
|
|
||||||
bits = decrypted[7:7+bits_len]
|
|
||||||
|
|
||||||
return Codec2Frame(
|
|
||||||
mode=Codec2Mode(mode),
|
|
||||||
bits=bits,
|
|
||||||
timestamp=0, # Will be set by caller
|
|
||||||
frame_number=frame_num
|
|
||||||
)
|
|
||||||
except Exception as e:
|
|
||||||
print(f"{RED}[VOICE]{RESET} Decryption failed: {e}")
|
|
||||||
return None
|
|
||||||
|
|
||||||
def _add_fec(self, data: bytes) -> bytes:
|
|
||||||
"""Add forward error correction."""
|
|
||||||
# Simple repetition code (3x) for testing
|
|
||||||
# In production: use convolutional code or LDPC
|
|
||||||
fec_data = bytearray()
|
|
||||||
|
|
||||||
for byte in data:
|
|
||||||
# Repeat each byte 3 times
|
|
||||||
fec_data.extend([byte, byte, byte])
|
|
||||||
|
|
||||||
return bytes(fec_data)
|
|
||||||
|
|
||||||
def _remove_fec(self, data: bytes) -> Optional[bytes]:
|
|
||||||
"""Remove FEC and correct errors."""
|
|
||||||
if len(data) % 3 != 0:
|
|
||||||
return None
|
|
||||||
|
|
||||||
corrected = bytearray()
|
|
||||||
|
|
||||||
for i in range(0, len(data), 3):
|
|
||||||
# Majority voting
|
|
||||||
votes = [data[i], data[i+1], data[i+2]]
|
|
||||||
byte_value = max(set(votes), key=votes.count)
|
|
||||||
corrected.append(byte_value)
|
|
||||||
|
|
||||||
return bytes(corrected)
|
|
||||||
|
|
||||||
|
|
||||||
# Example usage
|
|
||||||
if __name__ == "__main__":
|
|
||||||
# Test Codec2 wrapper
|
|
||||||
print(f"\n{BLUE}=== Testing Codec2 Wrapper ==={RESET}")
|
|
||||||
codec = Codec2Wrapper(Codec2Mode.MODE_1200)
|
|
||||||
|
|
||||||
# Generate test audio
|
|
||||||
if HAS_NUMPY:
|
|
||||||
t = np.linspace(0, 0.04, 320) # 40ms at 8kHz
|
|
||||||
test_audio = (np.sin(2 * np.pi * 440 * t) * 16384).astype(np.int16)
|
|
||||||
else:
|
|
||||||
test_audio = array.array('h')
|
|
||||||
for i in range(320):
|
|
||||||
t = i * 0.04 / 320
|
|
||||||
value = int(math.sin(2 * math.pi * 440 * t) * 16384)
|
|
||||||
test_audio.append(value)
|
|
||||||
|
|
||||||
# Encode
|
|
||||||
frame = codec.encode(test_audio)
|
|
||||||
print(f"Encoded frame: {len(frame.bits)} bytes")
|
|
||||||
|
|
||||||
# Decode
|
|
||||||
decoded = codec.decode(frame)
|
|
||||||
print(f"Decoded audio: {len(decoded)} samples")
|
|
||||||
|
|
||||||
# Test FSK modem
|
|
||||||
print(f"\n{BLUE}=== Testing FSK Modem ==={RESET}")
|
|
||||||
modem = FSKModem()
|
|
||||||
|
|
||||||
# Test data
|
|
||||||
test_data = b"Hello, secure voice!"
|
|
||||||
|
|
||||||
# Modulate
|
|
||||||
modulated = modem.modulate(test_data)
|
|
||||||
print(f"Modulated: {len(modulated)} samples ({len(modulated)/8000:.2f}s)")
|
|
||||||
|
|
||||||
# Demodulate
|
|
||||||
demodulated, confidence = modem.demodulate(modulated)
|
|
||||||
print(f"Demodulated: {demodulated}")
|
|
||||||
print(f"Confidence: {confidence:.2%}")
|
|
||||||
print(f"Match: {demodulated == test_data}")
|
|
||||||
@ -1,307 +0,0 @@
|
|||||||
import os
|
|
||||||
import struct
|
|
||||||
from typing import Optional, Tuple
|
|
||||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM, ChaCha20Poly1305
|
|
||||||
|
|
||||||
class MessageHeader:
|
|
||||||
"""
|
|
||||||
Header of an encrypted message (18 bytes total):
|
|
||||||
|
|
||||||
Clear Text Section (4 bytes):
|
|
||||||
- flag: 16 bits (0xBEEF by default)
|
|
||||||
- data_len: 16 bits (length of encrypted payload excluding tag)
|
|
||||||
|
|
||||||
Associated Data (14 bytes):
|
|
||||||
- retry: 8 bits (retry counter)
|
|
||||||
- connection_status: 4 bits (e.g., CRC required) + 4 bits padding
|
|
||||||
- iv/messageID: 96 bits (12 bytes)
|
|
||||||
"""
|
|
||||||
def __init__(self, flag: int, data_len: int, retry: int, connection_status: int, iv: bytes):
|
|
||||||
if not (0 <= flag < 65536):
|
|
||||||
raise ValueError("Flag must fit in 16 bits (0..65535)")
|
|
||||||
if not (0 <= data_len < 65536):
|
|
||||||
raise ValueError("Data length must fit in 16 bits (0..65535)")
|
|
||||||
if not (0 <= retry < 256):
|
|
||||||
raise ValueError("Retry must fit in 8 bits (0..255)")
|
|
||||||
if not (0 <= connection_status < 16):
|
|
||||||
raise ValueError("Connection status must fit in 4 bits (0..15)")
|
|
||||||
if len(iv) != 12:
|
|
||||||
raise ValueError("IV must be 12 bytes (96 bits)")
|
|
||||||
|
|
||||||
self.flag = flag # 16 bits
|
|
||||||
self.data_len = data_len # 16 bits
|
|
||||||
self.retry = retry # 8 bits
|
|
||||||
self.connection_status = connection_status # 4 bits
|
|
||||||
self.iv = iv # 96 bits (12 bytes)
|
|
||||||
|
|
||||||
def pack(self) -> bytes:
|
|
||||||
"""Pack header into 18 bytes."""
|
|
||||||
# Pack flag and data_len (4 bytes)
|
|
||||||
header = struct.pack('>H H', self.flag, self.data_len)
|
|
||||||
|
|
||||||
# Pack retry and connection_status (2 bytes)
|
|
||||||
# connection_status in high 4 bits of second byte, 4 bits padding as zero
|
|
||||||
ad_byte = (self.connection_status & 0x0F) << 4
|
|
||||||
ad_packed = struct.pack('>B B', self.retry, ad_byte)
|
|
||||||
|
|
||||||
# Append IV (12 bytes)
|
|
||||||
return header + ad_packed + self.iv
|
|
||||||
|
|
||||||
def get_associated_data(self) -> bytes:
|
|
||||||
"""Get the associated data for AEAD encryption (retry, conn_status, iv)."""
|
|
||||||
# Pack retry and connection_status
|
|
||||||
ad_byte = (self.connection_status & 0x0F) << 4
|
|
||||||
ad_packed = struct.pack('>B B', self.retry, ad_byte)
|
|
||||||
|
|
||||||
# Append IV
|
|
||||||
return ad_packed + self.iv
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def unpack(cls, data: bytes) -> 'MessageHeader':
|
|
||||||
"""Unpack 18 bytes into a MessageHeader object."""
|
|
||||||
if len(data) < 18:
|
|
||||||
raise ValueError(f"Header data too short: {len(data)} bytes, expected 18")
|
|
||||||
|
|
||||||
flag, data_len = struct.unpack('>H H', data[:4])
|
|
||||||
retry, ad_byte = struct.unpack('>B B', data[4:6])
|
|
||||||
connection_status = (ad_byte >> 4) & 0x0F
|
|
||||||
iv = data[6:18]
|
|
||||||
|
|
||||||
return cls(flag, data_len, retry, connection_status, iv)
|
|
||||||
|
|
||||||
class EncryptedMessage:
|
|
||||||
"""
|
|
||||||
Encrypted message packet format:
|
|
||||||
|
|
||||||
- Header (18 bytes):
|
|
||||||
* flag: 16 bits
|
|
||||||
* data_len: 16 bits
|
|
||||||
* retry: 8 bits
|
|
||||||
* connection_status: 4 bits (+ 4 bits padding)
|
|
||||||
* iv/messageID: 96 bits (12 bytes)
|
|
||||||
|
|
||||||
- Payload: variable length encrypted data
|
|
||||||
|
|
||||||
- Footer:
|
|
||||||
* Authentication tag: 128 bits (16 bytes)
|
|
||||||
* CRC32: 32 bits (4 bytes) - optional, based on connection_status
|
|
||||||
"""
|
|
||||||
def __init__(self, plaintext: bytes, key: bytes, flag: int = 0xBEEF,
|
|
||||||
retry: int = 0, connection_status: int = 0, iv: bytes = None,
|
|
||||||
cipher_type: int = 0):
|
|
||||||
self.plaintext = plaintext
|
|
||||||
self.key = key
|
|
||||||
self.flag = flag
|
|
||||||
self.retry = retry
|
|
||||||
self.connection_status = connection_status
|
|
||||||
self.iv = iv or generate_iv(initial=True)
|
|
||||||
self.cipher_type = cipher_type # 0 = AES-256-GCM, 1 = ChaCha20-Poly1305
|
|
||||||
|
|
||||||
# Will be set after encryption
|
|
||||||
self.ciphertext = None
|
|
||||||
self.tag = None
|
|
||||||
self.header = None
|
|
||||||
|
|
||||||
def encrypt(self) -> bytes:
|
|
||||||
"""Encrypt the plaintext and return the full encrypted message."""
|
|
||||||
# Create header with correct data_len (which will be set after encryption)
|
|
||||||
self.header = MessageHeader(
|
|
||||||
flag=self.flag,
|
|
||||||
data_len=0, # Will be updated after encryption
|
|
||||||
retry=self.retry,
|
|
||||||
connection_status=self.connection_status,
|
|
||||||
iv=self.iv
|
|
||||||
)
|
|
||||||
|
|
||||||
# Get associated data for AEAD
|
|
||||||
aad = self.header.get_associated_data()
|
|
||||||
|
|
||||||
# Encrypt using the appropriate cipher
|
|
||||||
if self.cipher_type == 0: # AES-256-GCM
|
|
||||||
cipher = AESGCM(self.key)
|
|
||||||
ciphertext_with_tag = cipher.encrypt(self.iv, self.plaintext, aad)
|
|
||||||
elif self.cipher_type == 1: # ChaCha20-Poly1305
|
|
||||||
cipher = ChaCha20Poly1305(self.key)
|
|
||||||
ciphertext_with_tag = cipher.encrypt(self.iv, self.plaintext, aad)
|
|
||||||
else:
|
|
||||||
raise ValueError(f"Unsupported cipher type: {self.cipher_type}")
|
|
||||||
|
|
||||||
# Extract ciphertext and tag
|
|
||||||
self.tag = ciphertext_with_tag[-16:]
|
|
||||||
self.ciphertext = ciphertext_with_tag[:-16]
|
|
||||||
|
|
||||||
# Update header with actual data length
|
|
||||||
self.header.data_len = len(self.ciphertext)
|
|
||||||
|
|
||||||
# Pack everything together
|
|
||||||
packed_header = self.header.pack()
|
|
||||||
|
|
||||||
# Check if CRC is required (based on connection_status)
|
|
||||||
if self.connection_status & 0x01: # Lowest bit indicates CRC required
|
|
||||||
import zlib
|
|
||||||
# Compute CRC32 of header + ciphertext + tag
|
|
||||||
crc = zlib.crc32(packed_header + self.ciphertext + self.tag) & 0xffffffff
|
|
||||||
crc_bytes = struct.pack('>I', crc)
|
|
||||||
return packed_header + self.ciphertext + self.tag + crc_bytes
|
|
||||||
else:
|
|
||||||
return packed_header + self.ciphertext + self.tag
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def decrypt(cls, data: bytes, key: bytes, cipher_type: int = 0) -> Tuple[bytes, MessageHeader]:
|
|
||||||
"""
|
|
||||||
Decrypt an encrypted message and return the plaintext and header.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
data: The full encrypted message
|
|
||||||
key: The encryption key
|
|
||||||
cipher_type: 0 for AES-256-GCM, 1 for ChaCha20-Poly1305
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Tuple of (plaintext, header)
|
|
||||||
"""
|
|
||||||
if len(data) < 18 + 16: # Header + minimum tag size
|
|
||||||
raise ValueError("Message too short")
|
|
||||||
|
|
||||||
# Extract header
|
|
||||||
header_bytes = data[:18]
|
|
||||||
header = MessageHeader.unpack(header_bytes)
|
|
||||||
|
|
||||||
# Get ciphertext and tag
|
|
||||||
data_len = header.data_len
|
|
||||||
ciphertext_start = 18
|
|
||||||
ciphertext_end = ciphertext_start + data_len
|
|
||||||
|
|
||||||
if ciphertext_end + 16 > len(data):
|
|
||||||
raise ValueError("Message length does not match header's data_len")
|
|
||||||
|
|
||||||
ciphertext = data[ciphertext_start:ciphertext_end]
|
|
||||||
tag = data[ciphertext_end:ciphertext_end + 16]
|
|
||||||
|
|
||||||
# Get associated data for AEAD
|
|
||||||
aad = header.get_associated_data()
|
|
||||||
|
|
||||||
# Combine ciphertext and tag for decryption
|
|
||||||
ciphertext_with_tag = ciphertext + tag
|
|
||||||
|
|
||||||
# Decrypt using the appropriate cipher
|
|
||||||
try:
|
|
||||||
if cipher_type == 0: # AES-256-GCM
|
|
||||||
cipher = AESGCM(key)
|
|
||||||
plaintext = cipher.decrypt(header.iv, ciphertext_with_tag, aad)
|
|
||||||
elif cipher_type == 1: # ChaCha20-Poly1305
|
|
||||||
cipher = ChaCha20Poly1305(key)
|
|
||||||
plaintext = cipher.decrypt(header.iv, ciphertext_with_tag, aad)
|
|
||||||
else:
|
|
||||||
raise ValueError(f"Unsupported cipher type: {cipher_type}")
|
|
||||||
|
|
||||||
return plaintext, header
|
|
||||||
except Exception as e:
|
|
||||||
raise ValueError(f"Decryption failed: {e}")
|
|
||||||
|
|
||||||
def generate_iv(initial: bool = False, previous_iv: bytes = None) -> bytes:
|
|
||||||
"""
|
|
||||||
Generate a 96-bit IV (12 bytes).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
initial: If True, return a random IV
|
|
||||||
previous_iv: The previous IV to increment
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
A new IV
|
|
||||||
"""
|
|
||||||
if initial or previous_iv is None:
|
|
||||||
return os.urandom(12) # 96 bits
|
|
||||||
else:
|
|
||||||
# Increment the previous IV by 1 modulo 2^96
|
|
||||||
iv_int = int.from_bytes(previous_iv, 'big')
|
|
||||||
iv_int = (iv_int + 1) % (1 << 96)
|
|
||||||
return iv_int.to_bytes(12, 'big')
|
|
||||||
|
|
||||||
# Convenience functions to match original API
|
|
||||||
def encrypt_message(plaintext: bytes, key: bytes, flag: int = 0xBEEF,
|
|
||||||
retry: int = 0, connection_status: int = 0,
|
|
||||||
iv: bytes = None, cipher_type: int = 0) -> bytes:
|
|
||||||
"""
|
|
||||||
Encrypt a message using the specified parameters.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
plaintext: The data to encrypt
|
|
||||||
key: The encryption key (32 bytes for AES-256-GCM, 32 bytes for ChaCha20-Poly1305)
|
|
||||||
flag: 16-bit flag value (default: 0xBEEF)
|
|
||||||
retry: 8-bit retry counter
|
|
||||||
connection_status: 4-bit connection status
|
|
||||||
iv: Optional 96-bit IV (if None, a random one will be generated)
|
|
||||||
cipher_type: 0 for AES-256-GCM, 1 for ChaCha20-Poly1305
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
The full encrypted message
|
|
||||||
"""
|
|
||||||
message = EncryptedMessage(
|
|
||||||
plaintext=plaintext,
|
|
||||||
key=key,
|
|
||||||
flag=flag,
|
|
||||||
retry=retry,
|
|
||||||
connection_status=connection_status,
|
|
||||||
iv=iv,
|
|
||||||
cipher_type=cipher_type
|
|
||||||
)
|
|
||||||
return message.encrypt()
|
|
||||||
|
|
||||||
def decrypt_message(message: bytes, key: bytes, cipher_type: int = 0) -> bytes:
|
|
||||||
"""
|
|
||||||
Decrypt a message.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
message: The full encrypted message
|
|
||||||
key: The encryption key
|
|
||||||
cipher_type: 0 for AES-256-GCM, 1 for ChaCha20-Poly1305
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
The decrypted plaintext
|
|
||||||
"""
|
|
||||||
plaintext, _ = EncryptedMessage.decrypt(message, key, cipher_type)
|
|
||||||
return plaintext
|
|
||||||
|
|
||||||
# ChaCha20-CTR functions for voice streaming (without authentication)
|
|
||||||
def chacha20_encrypt(plaintext: bytes, key: bytes, nonce: bytes) -> bytes:
|
|
||||||
"""
|
|
||||||
Encrypt plaintext using ChaCha20 in CTR mode (no authentication).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
plaintext: Data to encrypt
|
|
||||||
key: 32-byte key
|
|
||||||
nonce: 16-byte nonce (for ChaCha20 in cryptography library)
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Ciphertext
|
|
||||||
"""
|
|
||||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
||||||
from cryptography.hazmat.backends import default_backend
|
|
||||||
|
|
||||||
if len(key) != 32:
|
|
||||||
raise ValueError("ChaCha20 key must be 32 bytes")
|
|
||||||
if len(nonce) != 16:
|
|
||||||
raise ValueError("ChaCha20 nonce must be 16 bytes")
|
|
||||||
|
|
||||||
cipher = Cipher(
|
|
||||||
algorithms.ChaCha20(key, nonce),
|
|
||||||
mode=None,
|
|
||||||
backend=default_backend()
|
|
||||||
)
|
|
||||||
encryptor = cipher.encryptor()
|
|
||||||
return encryptor.update(plaintext) + encryptor.finalize()
|
|
||||||
|
|
||||||
def chacha20_decrypt(ciphertext: bytes, key: bytes, nonce: bytes) -> bytes:
|
|
||||||
"""
|
|
||||||
Decrypt ciphertext using ChaCha20 in CTR mode (no authentication).
|
|
||||||
|
|
||||||
Args:
|
|
||||||
ciphertext: Data to decrypt
|
|
||||||
key: 32-byte key
|
|
||||||
nonce: 12-byte nonce
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
Plaintext
|
|
||||||
"""
|
|
||||||
# ChaCha20 is symmetrical - encryption and decryption are the same
|
|
||||||
return chacha20_encrypt(ciphertext, key, nonce)
|
|
||||||
@ -1,32 +0,0 @@
|
|||||||
#!/usr/bin/env python3
|
|
||||||
"""Debug script to trace the UI behavior"""
|
|
||||||
|
|
||||||
import sys
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
# Monkey patch the integrated_protocol to see what's being called
|
|
||||||
orig_file = Path(__file__).parent / "DryBox" / "integrated_protocol.py"
|
|
||||||
backup_file = Path(__file__).parent / "DryBox" / "integrated_protocol_backup.py"
|
|
||||||
|
|
||||||
# Read the original file
|
|
||||||
with open(orig_file, 'r') as f:
|
|
||||||
content = f.read()
|
|
||||||
|
|
||||||
# Add debug prints
|
|
||||||
debug_content = content.replace(
|
|
||||||
'def initiate_key_exchange(self, cipher_type=1, is_initiator=True):',
|
|
||||||
'''def initiate_key_exchange(self, cipher_type=1, is_initiator=True):
|
|
||||||
import traceback
|
|
||||||
print(f"\\n[DEBUG] initiate_key_exchange called with is_initiator={is_initiator}")
|
|
||||||
print("[DEBUG] Call stack:")
|
|
||||||
for line in traceback.format_stack()[:-1]:
|
|
||||||
print(line.strip())
|
|
||||||
print()'''
|
|
||||||
)
|
|
||||||
|
|
||||||
# Write the debug version
|
|
||||||
with open(orig_file, 'w') as f:
|
|
||||||
f.write(debug_content)
|
|
||||||
|
|
||||||
print("Debug patch applied. Run the UI now to see the trace.")
|
|
||||||
print("To restore: cp DryBox/integrated_protocol_backup.py DryBox/integrated_protocol.py")
|
|
||||||
Loading…
Reference in New Issue
Block a user