143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
import base64
|
|
import struct
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
|
|
from nacl.public import PrivateKey, SealedBox
|
|
|
|
import config
|
|
|
|
# File format constants
|
|
MAGIC_BYTES = b"SREC"
|
|
SUPPORTED_VERSION = 1
|
|
|
|
|
|
def generate_keypair() -> Tuple[bytes, bytes]:
|
|
"""
|
|
Generates a new X25519 keypair.
|
|
|
|
Returns:
|
|
Tuple of (private_key_bytes, public_key_bytes)
|
|
"""
|
|
private_key = PrivateKey.generate()
|
|
public_key = private_key.public_key
|
|
return bytes(private_key), bytes(public_key)
|
|
|
|
|
|
def save_keypair(private_key: bytes, public_key: bytes) -> None:
|
|
"""Saves the keypair to files."""
|
|
config.PRIVATE_KEY_FILE.write_bytes(private_key)
|
|
config.PUBLIC_KEY_FILE.write_bytes(public_key)
|
|
print(f"Keys saved to {config.KEYS_DIR}")
|
|
|
|
|
|
def load_or_create_keypair() -> Tuple[bytes, bytes]:
|
|
"""
|
|
Loads existing keypair or creates a new one.
|
|
|
|
Returns:
|
|
Tuple of (private_key_bytes, public_key_bytes)
|
|
"""
|
|
if config.PRIVATE_KEY_FILE.exists() and config.PUBLIC_KEY_FILE.exists():
|
|
private_key = config.PRIVATE_KEY_FILE.read_bytes()
|
|
public_key = config.PUBLIC_KEY_FILE.read_bytes()
|
|
print("Loaded existing keypair")
|
|
else:
|
|
private_key, public_key = generate_keypair()
|
|
save_keypair(private_key, public_key)
|
|
print("Generated new keypair")
|
|
|
|
return private_key, public_key
|
|
|
|
|
|
def get_public_key_base64() -> str:
|
|
"""Returns the public key as a Base64-encoded string."""
|
|
_, public_key = load_or_create_keypair()
|
|
return base64.b64encode(public_key).decode("ascii")
|
|
|
|
|
|
def decrypt_file(encrypted_path: Path, output_path: Path) -> bool:
|
|
"""
|
|
Decrypts an encrypted recording file.
|
|
|
|
File format:
|
|
[4 bytes: "SREC" magic]
|
|
[1 byte: version (1)]
|
|
[4 bytes: chunk count (big-endian)]
|
|
For each chunk:
|
|
[4 bytes: encrypted length (big-endian)]
|
|
[N bytes: sealed box encrypted data]
|
|
|
|
Args:
|
|
encrypted_path: Path to the encrypted file
|
|
output_path: Path for the decrypted output
|
|
|
|
Returns:
|
|
True if decryption succeeded, False otherwise
|
|
"""
|
|
private_key_bytes, _ = load_or_create_keypair()
|
|
private_key = PrivateKey(private_key_bytes)
|
|
sealed_box = SealedBox(private_key)
|
|
|
|
try:
|
|
with open(encrypted_path, "rb") as f:
|
|
# Read and verify header
|
|
magic = f.read(4)
|
|
if magic != MAGIC_BYTES:
|
|
print(f"Invalid magic bytes: {magic}")
|
|
return False
|
|
|
|
version = struct.unpack(">B", f.read(1))[0]
|
|
if version != SUPPORTED_VERSION:
|
|
print(f"Unsupported version: {version}")
|
|
return False
|
|
|
|
chunk_count = struct.unpack(">I", f.read(4))[0]
|
|
|
|
# Decrypt chunks
|
|
with open(output_path, "wb") as out:
|
|
for i in range(chunk_count):
|
|
chunk_len = struct.unpack(">I", f.read(4))[0]
|
|
encrypted_chunk = f.read(chunk_len)
|
|
|
|
if len(encrypted_chunk) != chunk_len:
|
|
print(f"Unexpected end of file at chunk {i}")
|
|
return False
|
|
|
|
decrypted = sealed_box.decrypt(encrypted_chunk)
|
|
out.write(decrypted)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Decryption error: {e}")
|
|
return False
|
|
|
|
|
|
def get_output_path_for_recording(filename: str) -> Path:
|
|
"""
|
|
Determines the output path for a decrypted recording.
|
|
|
|
Organizes files into: recordings/YYYY/MM/DD/filename.ogg
|
|
|
|
Args:
|
|
filename: Original filename (e.g., "recording_2024-01-15_09-30-00.ogg.enc")
|
|
|
|
Returns:
|
|
Path for the decrypted file
|
|
"""
|
|
# Extract date from filename (recording_YYYY-MM-DD_HH-mm-ss.ogg.enc)
|
|
try:
|
|
parts = filename.replace("recording_", "").split("_")[0].split("-")
|
|
year, month, day = parts[0], parts[1], parts[2]
|
|
except (IndexError, ValueError):
|
|
# Fallback to "unknown" directory
|
|
year, month, day = "unknown", "00", "00"
|
|
|
|
# Remove .enc extension for output
|
|
output_filename = filename.replace(".enc", "")
|
|
|
|
output_dir = config.RECORDINGS_DIR / year / month / day
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
return output_dir / output_filename
|