#!/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())