#!/usr/bin/env python3 """ LiveTalker HTTPS Voice Server Proper SSL/TLS support for microphone access """ import asyncio import json import logging import time import base64 import numpy as np import ssl import os from pathlib import Path from typing import Dict, Any from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="LiveTalker HTTPS Voice Server") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) active_connections: Dict[str, Dict] = {} def simple_vad(audio_data: np.ndarray, threshold: float = 0.01) -> tuple: """Energy-based voice activity detection""" if len(audio_data) == 0: return False, 0.0 rms = np.sqrt(np.mean(audio_data ** 2)) is_speech = rms > threshold confidence = min(rms / threshold, 1.0) if threshold > 0 else 0.0 return is_speech, confidence def create_ssl_certificate(): """Create self-signed SSL certificate for HTTPS""" try: from cryptography import x509 from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa import datetime import ipaddress # Generate private key private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, ) # Create certificate subject subject = issuer = x509.Name([ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "CA"), x509.NameAttribute(NameOID.LOCALITY_NAME, "LiveTalker"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "LiveTalker Voice Assistant"), x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), ]) # Create certificate with proper SANs for our use case cert = x509.CertificateBuilder().subject_name( subject ).issuer_name( issuer ).public_key( private_key.public_key() ).serial_number( x509.random_serial_number() ).not_valid_before( datetime.datetime.now(datetime.timezone.utc) ).not_valid_after( datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=365) ).add_extension( x509.SubjectAlternativeName([ x509.DNSName("localhost"), x509.DNSName("*.localhost"), x509.DNSName("bitbots01.tail51b02f.ts.net"), x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")), x509.IPAddress(ipaddress.IPv4Address("100.118.75.128")), ]), critical=False, ).sign(private_key, hashes.SHA256()) # Write certificate and key files cert_path = "livetalker.crt" key_path = "livetalker.key" with open(cert_path, "wb") as f: f.write(cert.public_bytes(serialization.Encoding.PEM)) with open(key_path, "wb") as f: f.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() )) print(f"āœ… SSL certificate created: {cert_path}") print(f"āœ… SSL private key created: {key_path}") return cert_path, key_path except Exception as e: print(f"āŒ Failed to create SSL certificate: {e}") return None, None @app.get("/") async def root(): """HTTPS voice interface""" html_content = """ LiveTalker HTTPS Voice Assistant

šŸ”’ LiveTalker HTTPS

Secure Voice Assistant with Real-time Processing

šŸ”’ HTTPS Enabled

Microphone access is now available with SSL/TLS encryption

šŸ“” Connection Status

Ready for secure connection

šŸŽ¤ Microphone Status

HTTPS required for microphone access

šŸŽµ Voice Activity Detection

Secure microphone access available
0
Audio Chunks
0
Speech Events
0%
Avg Confidence
šŸ”’ Secure HTTPS connection ready! Enable microphone to start voice conversation.

šŸ“Š Activity Log

HTTPS Voice Server initialized...
šŸ”’

HTTPS Secure

SSL/TLS encryption for microphone access

šŸŽÆ

Real-time VAD

Energy-based voice activity detection

āš”

Low Latency

WebSocket streaming audio processing

šŸŒ

Cross-platform

Works on all modern browsers

""" return HTMLResponse(content=html_content) @app.get("/health") async def health_check(): """Health check with SSL status""" return { "status": "healthy", "timestamp": time.time(), "protocol": "HTTPS", "active_connections": len(active_connections), "ssl_enabled": True, "features": { "secure_microphone": True, "voice_activity_detection": True, "real_time_processing": True, "ssl_encryption": True } } @app.websocket("/media-stream") async def websocket_endpoint(websocket: WebSocket): """Secure WebSocket for voice processing""" await websocket.accept() session_id = f"session_{int(time.time() * 1000)}" session = { "id": session_id, "websocket": websocket, "audio_buffer": [], "is_listening": False, "last_speech": 0, "https_enabled": True } active_connections[session_id] = session logger.info(f"Secure voice session: {session_id}") try: await websocket.send_json({ "type": "config", "session_id": session_id, "message": "Secure HTTPS voice processing ready", "ssl_enabled": True }) async for message in websocket.iter_json(): await handle_secure_message(session, message) except WebSocketDisconnect: logger.info(f"Secure session ended: {session_id}") except Exception as e: logger.error(f"Secure session error: {e}") try: await websocket.send_json({ "type": "error", "error": str(e) }) except: pass finally: if session_id in active_connections: del active_connections[session_id] async def handle_secure_message(session: Dict, message: Dict[str, Any]): """Handle secure WebSocket messages""" msg_type = message.get("type") if msg_type == "start_conversation": session["is_listening"] = True await session["websocket"].send_json({ "type": "conversation_started", "message": "šŸ”’ Secure voice conversation started!", "https_enabled": True }) elif msg_type == "audio" and session["is_listening"]: try: # Decode and process audio audio_data = base64.b64decode(message["data"]) audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0 if len(audio_np) > 0: # Run VAD is_speech, confidence = simple_vad(audio_np, threshold=0.015) await session["websocket"].send_json({ "type": "vad_result", "is_speech": bool(is_speech), "confidence": float(confidence), "timestamp": time.time(), "secure": True }) # Accumulate speech if is_speech: session["audio_buffer"].extend(audio_np.tolist()) session["last_speech"] = time.time() if len(session["audio_buffer"]) > 16000: # ~1 second await process_secure_speech(session) elif (len(session["audio_buffer"]) > 8000 and time.time() - session["last_speech"] > 0.8): await process_secure_speech(session) except Exception as e: await session["websocket"].send_json({ "type": "error", "error": f"Secure audio processing error: {str(e)}" }) elif msg_type == "stop_listening": session["is_listening"] = False if session["audio_buffer"]: await process_secure_speech(session) async def process_secure_speech(session: Dict): """Process speech with HTTPS security context""" if len(session["audio_buffer"]) < 4000: session["audio_buffer"] = [] return try: duration = len(session["audio_buffer"]) / 16000 text = f"šŸ”’ Secure speech: {duration:.1f}s voice segment detected" await session["websocket"].send_json({ "type": "speech_to_text", "text": text, "confidence": 0.9, "duration": duration, "secure": True }) # Secure AI response response = f"āœ… Received {duration:.1f}s of secure voice input! HTTPS encryption is working perfectly for real-time voice processing." await session["websocket"].send_json({ "type": "ai_response", "text": response, "processing_info": { "duration": f"{duration:.1f}s", "encryption": "HTTPS/TLS", "security": "enabled" } }) session["audio_buffer"] = [] logger.info(f"Processed {duration:.1f}s secure speech in session {session['id']}") except Exception as e: logger.error(f"Secure speech processing error: {e}") session["audio_buffer"] = [] if __name__ == "__main__": print("šŸ”’ LiveTalker HTTPS Voice Server") print("=" * 50) # Create SSL certificate cert_path, key_path = create_ssl_certificate() if cert_path and key_path and os.path.exists(cert_path) and os.path.exists(key_path): # Configure SSL context ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) ssl_context.load_cert_chain(cert_path, key_path) print("āœ… SSL certificate ready") print("āœ… HTTPS encryption enabled") print("āœ… Microphone access available") print("") print("šŸŒ HTTPS URLs:") print(" Local: https://localhost:8000") print(" Network: https://100.118.75.128:8000") print(" Tailnet: https://bitbots01.tail51b02f.ts.net:8000") print("") print("šŸ”§ Features:") print(" āœ… Secure microphone access") print(" āœ… Real-time voice processing") print(" āœ… SSL/TLS encryption") print(" āœ… Cross-platform compatibility") print("") print("āš ļø You may see browser security warnings for the self-signed certificate.") print(" Click 'Advanced' ā†’ 'Continue to site' to proceed.") try: uvicorn.run( app, host="0.0.0.0", port=8000, ssl_keyfile=key_path, ssl_certfile=cert_path, log_level="info" ) except KeyboardInterrupt: print("\nšŸ›‘ Server stopped") finally: # Cleanup certificate files for file_path in [cert_path, key_path]: if os.path.exists(file_path): os.remove(file_path) print(f"šŸ—‘ļø Cleaned up: {file_path}") else: print("āŒ Failed to create SSL certificate") print("Falling back to HTTP mode...") uvicorn.run( app, host="0.0.0.0", port=8000, log_level="info" )