243 lines
7.3 KiB
Python
243 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Integration test for phone recorder encryption and upload.
|
|
|
|
This script simulates what the Android app does:
|
|
1. Creates a test audio file
|
|
2. Encrypts it using the same SREC format
|
|
3. Uploads to the server
|
|
4. Verifies the server decrypted it correctly
|
|
|
|
Usage:
|
|
# Start the server first:
|
|
python main.py
|
|
|
|
# In another terminal:
|
|
python test_integration.py [server_url]
|
|
|
|
# Default server URL is http://localhost:8000
|
|
"""
|
|
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from nacl.public import PublicKey, SealedBox
|
|
|
|
# Test configuration
|
|
DEFAULT_SERVER_URL = "http://localhost:8000"
|
|
CHUNK_SIZE = 64 * 1024 # 64KB, same as Android app
|
|
MAGIC_BYTES = b"SREC"
|
|
VERSION = 1
|
|
|
|
|
|
def create_test_audio_file(size_kb: int = 100) -> bytes:
|
|
"""Creates a fake audio file for testing (just random bytes)."""
|
|
return os.urandom(size_kb * 1024)
|
|
|
|
|
|
def encrypt_file_srec_format(data: bytes, public_key_bytes: bytes) -> bytes:
|
|
"""
|
|
Encrypts data using the SREC format (same as Android app).
|
|
|
|
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]
|
|
"""
|
|
public_key = PublicKey(public_key_bytes)
|
|
sealed_box = SealedBox(public_key)
|
|
|
|
output = io.BytesIO()
|
|
|
|
# Calculate chunks
|
|
chunks = [data[i:i + CHUNK_SIZE] for i in range(0, len(data), CHUNK_SIZE)]
|
|
chunk_count = len(chunks)
|
|
|
|
# Write header
|
|
output.write(MAGIC_BYTES)
|
|
output.write(struct.pack(">B", VERSION))
|
|
output.write(struct.pack(">I", chunk_count))
|
|
|
|
# Encrypt and write each chunk
|
|
for chunk in chunks:
|
|
encrypted = sealed_box.encrypt(chunk)
|
|
output.write(struct.pack(">I", len(encrypted)))
|
|
output.write(encrypted)
|
|
|
|
return output.getvalue()
|
|
|
|
|
|
def compute_sha256(data: bytes) -> str:
|
|
"""Computes SHA-256 hash of data."""
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def test_health_check(server_url: str) -> bool:
|
|
"""Tests the health endpoint."""
|
|
print("Testing health check...")
|
|
try:
|
|
response = requests.get(f"{server_url}/health", timeout=5)
|
|
if response.status_code == 200 and response.json().get("status") == "ok":
|
|
print(" ✓ Health check passed")
|
|
return True
|
|
else:
|
|
print(f" ✗ Health check failed: {response.text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ Health check error: {e}")
|
|
return False
|
|
|
|
|
|
def test_get_public_key(server_url: str) -> bytes | None:
|
|
"""Tests fetching the public key."""
|
|
print("Testing public key fetch...")
|
|
try:
|
|
response = requests.get(f"{server_url}/api/v1/public-key", timeout=5)
|
|
if response.status_code == 200:
|
|
import base64
|
|
key_b64 = response.json().get("public_key")
|
|
key_bytes = base64.b64decode(key_b64)
|
|
if len(key_bytes) == 32:
|
|
print(f" ✓ Got public key: {key_b64[:20]}...")
|
|
return key_bytes
|
|
else:
|
|
print(f" ✗ Invalid key length: {len(key_bytes)}")
|
|
return None
|
|
else:
|
|
print(f" ✗ Failed to get public key: {response.text}")
|
|
return None
|
|
except Exception as e:
|
|
print(f" ✗ Public key fetch error: {e}")
|
|
return None
|
|
|
|
|
|
def test_upload_and_verify(server_url: str, public_key: bytes) -> bool:
|
|
"""Tests file upload with encryption and hash verification."""
|
|
print("Testing encrypted upload...")
|
|
|
|
# Create test data
|
|
original_data = create_test_audio_file(size_kb=50)
|
|
print(f" Created test file: {len(original_data)} bytes")
|
|
|
|
# Encrypt using SREC format
|
|
encrypted_data = encrypt_file_srec_format(original_data, public_key)
|
|
print(f" Encrypted to: {len(encrypted_data)} bytes")
|
|
|
|
# Compute hash of encrypted data (what the server will verify)
|
|
local_hash = compute_sha256(encrypted_data)
|
|
print(f" Local hash: {local_hash[:16]}...")
|
|
|
|
# Generate filename
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
filename = f"recording_{timestamp}.ogg.enc"
|
|
|
|
# Upload
|
|
try:
|
|
files = {"file": (filename, encrypted_data, "application/octet-stream")}
|
|
response = requests.post(
|
|
f"{server_url}/api/v1/recordings/upload",
|
|
files=files,
|
|
timeout=30
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
server_hash = result.get("received_hash", "")
|
|
verified = result.get("verified", False)
|
|
|
|
if server_hash == local_hash and verified:
|
|
print(f" ✓ Upload successful and hash verified!")
|
|
print(f" Server path: {result.get('path')}")
|
|
return True
|
|
elif server_hash != local_hash:
|
|
print(f" ✗ Hash mismatch!")
|
|
print(f" Local: {local_hash}")
|
|
print(f" Server: {server_hash}")
|
|
return False
|
|
else:
|
|
print(f" ⚠ Upload successful but not verified")
|
|
return False
|
|
else:
|
|
print(f" ✗ Upload failed: {response.status_code} - {response.text}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f" ✗ Upload error: {e}")
|
|
return False
|
|
|
|
|
|
def test_list_recordings(server_url: str) -> bool:
|
|
"""Tests listing recordings."""
|
|
print("Testing list recordings...")
|
|
try:
|
|
response = requests.get(f"{server_url}/api/v1/recordings", timeout=5)
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
count = result.get("count", 0)
|
|
print(f" ✓ Listed {count} recordings")
|
|
if count > 0:
|
|
recordings = result.get("recordings", [])
|
|
for rec in recordings[:3]: # Show first 3
|
|
print(f" - {rec}")
|
|
if count > 3:
|
|
print(f" ... and {count - 3} more")
|
|
return True
|
|
else:
|
|
print(f" ✗ List failed: {response.text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" ✗ List error: {e}")
|
|
return False
|
|
|
|
|
|
def main():
|
|
server_url = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_SERVER_URL
|
|
print(f"Testing server at: {server_url}\n")
|
|
|
|
results = []
|
|
|
|
# Test 1: Health check
|
|
results.append(("Health check", test_health_check(server_url)))
|
|
|
|
# Test 2: Get public key
|
|
public_key = test_get_public_key(server_url)
|
|
results.append(("Get public key", public_key is not None))
|
|
|
|
if public_key:
|
|
# Test 3: Upload with encryption and verification
|
|
results.append(("Upload & verify", test_upload_and_verify(server_url, public_key)))
|
|
|
|
# Test 4: List recordings
|
|
results.append(("List recordings", test_list_recordings(server_url)))
|
|
|
|
# Summary
|
|
print("\n" + "=" * 50)
|
|
print("Test Results:")
|
|
all_passed = True
|
|
for name, passed in results:
|
|
status = "✓ PASS" if passed else "✗ FAIL"
|
|
print(f" {status}: {name}")
|
|
if not passed:
|
|
all_passed = False
|
|
|
|
print("=" * 50)
|
|
if all_passed:
|
|
print("All tests passed!")
|
|
return 0
|
|
else:
|
|
print("Some tests failed!")
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|