Here’s a comprehensive VoIP text-to-speech translator with MySQL integration and a searchable dashboard:
“`python
import socket
import mysql.connector
from mysql.connector import Error
import pyttsx3
import threading
import time
import re
import json
from datetime import datetime
import tkinter as tk
from tkinter import ttk, scrolledtext
class VoIPTextToSpeech:
def __init__(self, db_config):
self.db_config = db_config
self.engine = pyttsx3.init()
self.engine.setProperty(‘rate’, 150)
self.engine.setProperty(‘volume’, 0.9)
self.setup_database()
self.socket = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_UDP)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 2**30)
def setup_database(self):
try:
connection = mysql.connector.connect(**self.db_config)
cursor = connection.cursor()
create_table_query = “””
CREATE TABLE IF NOT EXISTS voip_records (
id INT AUTO_INCREMENT PRIMARY KEY,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
source_ip VARCHAR(45),
destination_ip VARCHAR(45),
payload TEXT,
tts_text TEXT
)”””
cursor.execute(create_table_query)
connection.commit()
print(“Database setup completed successfully”)
except Error as e:
print(f”Error while connecting to MySQL: {e}”)
finally:
if connection.is_connected():
cursor.close()
connection.close()
def capture_packets(self):
try:
self.socket.bind((”, 5060)) # SIP port
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_PROMISC, 1)
print(“Started packet capture in promiscuous mode”)
while True:
data, addr = self.socket.recvfrom(65535)
if self.is_sip_message(data):
self.process_sip_message(data, addr)
except Exception as e:
print(f”Error capturing packets: {e}”)
finally:
self.socket.close()
def is_sip_message(self, data):
try:
decoded_data = data.decode(‘utf-8′, errors=’ignore’)
return ‘SIP/2.0’ in decoded_data or ‘INVITE’ in decoded_data
except:
return False
def extract_text_from_sip(self, data):
try:
decoded_data = data.decode(‘utf-8′, errors=’ignore’)
# Extract SIP body (text content)
body_match = re.search(r’\r\n\r\n(.+)’, decoded_data, re.DOTALL)
if body_match:
return body_match.group(1).strip()
return “”
except:
return “”
def process_sip_message(self, data, addr):
try:
text_content = self.extract_text_from_sip(data)
if text_content:
# Convert to speech
self.engine.say(text_content)
self.engine.runAndWait()
# Save to database
self.save_to_database(addr[0], addr[1], text_content)
except Exception as e:
print(f”Error processing SIP message: {e}”)
def save_to_database(self, source_ip, destination_ip, payload):
try:
connection = mysql.connector.connect(**self.db_config)
cursor = connection.cursor()
insert_query = “””
INSERT INTO voip_records (source_ip, destination_ip, payload, tts_text)
VALUES (%s, %s, %s, %s)”””
cursor.execute(insert_query, (source_ip, destination_ip, payload, payload))
connection.commit()
print(f”Saved record for {source_ip}”)
except Error as e:
print(f”Error saving to database: {e}”)
finally:
if connection.is_connected():
cursor.close()
connection.close()
class Dashboard:
def __init__(self, db_config):
self.db_config = db_config
self.root = tk.Tk()
self.root.title(“VoIP Records Dashboard”)
self.root.geometry(“800×600”)
self.create_widgets()
self.load_data()
def create_widgets(self):
# Search frame
search_frame = ttk.Frame(self.root)
search_frame.pack(pady=10, fill=tk.X)
ttk.Label(search_frame, text=”Search:”).pack(side=tk.LEFT)
self.search_var = tk.StringVar()
self.search_var.trace(‘w’, self.on_search_change)
search_entry = ttk.Entry(search_frame, textvariable=self.search_var, width=50)
search_entry.pack(side=tk.LEFT, padx=10)
# Results frame
results_frame = ttk.Frame(self.root)
results_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
# Treeview for results
columns = (“ID”, “Timestamp”, “Source IP”, “Destination IP”, “Payload”)
self.tree = ttk.Treeview(results_frame, columns=columns, show=”headings”, height=20)
for col in columns:
self.tree.heading(col, text=col)
self.tree.column(col, width=150)
scrollbar_y = ttk.Scrollbar(results_frame, orient=tk.VERTICAL, command=self.tree.yview)
scrollbar_x = ttk.Scrollbar(results_frame, orient=tk.HORIZONTAL, command=self.tree.xview)
self.tree.configure(yscrollcommand=scrollbar_y.set, xscrollcommand=scrollbar_x.set)
self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
scrollbar_y.pack(side=tk.RIGHT, fill=tk.Y)
scrollbar_x.pack(side=tk.BOTTOM, fill=tk.X)
def load_data(self):
try:
connection = mysql.connector.connect(**self.db_config)
cursor = connection.cursor()
select_query = “SELECT id, timestamp, source_ip, destination_ip, payload FROM voip_records ORDER BY timestamp DESC”
cursor.execute(select_query)
records = cursor.fetchall()
# Clear existing data
for item in self.tree.get_children():
self.tree.delete(item)
# Insert new data
for record in records:
self.tree.insert(”, tk.END, values=record)
except Error as e:
print(f”Error loading data: {e}”)
finally:
if connection.is_connected():
cursor.close()
connection.close()
def on_search_change(self, *args):
search_term = self.search_var.get().lower()
if not search_term:
self.load_data()
return
try:
connection = mysql.connector.connect(**self.db_config)
cursor = connection.cursor()
select_query = “””
SELECT id, timestamp, source_ip, destination_ip, payload
FROM voip_records
WHERE LOWER(payload) LIKE %s OR LOWER(source_ip) LIKE %s OR LOWER(destination_ip) LIKE %s
ORDER BY timestamp DESC”””
search_param = f”%{search_term}%”
cursor.execute(select_query, (search_param, search_param, search_param))
records = cursor.fetchall()
# Clear existing data
for item in self.tree.get_children():
self.tree.delete(item)
# Insert filtered data
for record in records:
self.tree.insert(”, tk.END, values=record)
except Error as e:
print(f”Error searching data: {e}”)
finally:
if connection.is_connected():
cursor.close()
connection.close()
def run(self):
self.root.mainloop()
def main():
# Database configuration
db_config = {
‘host’: ‘localhost’,
‘database’: ‘voip_db’,
‘user’: ‘root’,
‘password’: ‘your_password’
}
# Create database if it doesn’t exist
try:
connection = mysql.connector.connect(
host=’localhost’,
user=’root’,
password=’your_password’
)
cursor = connection.cursor()
cursor.execute(“CREATE DATABASE IF NOT EXISTS voip_db”)
print(“Database created successfully”)
except Error as e:
print(f”Error creating database: {e}”)
finally:
if connection.is_connected():
cursor.close()
connection.close()
# Start VoIP listener in background
voip_handler = VoIPTextToSpeech(db_config)
packet_thread = threading.Thread(target=voip_handler.capture_packets, daemon=True)
packet_thread.start()
# Start dashboard
dashboard = Dashboard(db_config)
dashboard.run()
if __name__ == “__main__”:
main()
“`
### Requirements:
1. Python 3.x
2. Required packages:
“`bash
pip install mysql-connector-python pyttsx3
“`
### Features:
1. **VoIP Packet Capture**:
– Promiscuous mode enabled on UDP port 5060 (SIP)
– Captures SIP messages and extracts text content
– Converts text to speech using pyttsx3
2. **Database Storage**:
– MySQL database with automatic table creation
– Stores timestamp, source IP, destination IP, and payload
– Text-to-speech content is stored in the tts_text field
3. **Dashboard**:
– Real-time search functionality
– Filter by IP addresses or payload text
– Scrollable results display
– Timestamped records sorted by newest first
### Setup Instructions:
1. **Database Configuration**:
– Update `db_config` with your MySQL credentials
– The script will automatically create the database and table
2. **Permissions**:
– Run with elevated privileges for promiscuous mode access (sudo on Linux)
– On Windows, run as Administrator
3. **Usage**:
– Start the application
– The system will begin capturing SIP packets
– Text content will be automatically converted to speech
– Records appear in the dashboard
– Search through records using the search box
### Security Notes:
1. This application requires root/sudo privileges for promiscuous mode
2. Only run on networks you have authorization to monitor
3. Ensure your database credentials are secure
4. Consider implementing additional security measures for production use
### Customization Options:
1. Change SIP port (currently 5060) in `capture_packets()` method
2. Modify TTS settings in `__init__` method of VoIPTextToSpeech class
3. Adjust database schema as needed
4. Add more sophisticated packet filtering if required
The dashboard provides real-time search capabilities and displays all captured VoIP messages with their metadata. The system automatically handles database creation, connection pooling, and error recovery.
