#!/usr/bin/env python3 """ talku.py — Linux port of TalkU v2.3 (FIXED) Routes Vivox VoIP traffic by UDP port instead of IP ranges. """ import json import os import signal import subprocess import sys import time import urllib.request import ssl import tempfile API_URL = "https://talku.ddns.net:8000/exchange_keys/" API_KEY = "z~WXkukTav2^dodr5#9" WG_IFACE = "talkuwg" WSTUNNEL_LOG = "/tmp/talku-wstunnel.log" WSTUNNEL_PID = "/tmp/talku-wstunnel.pid" ROUTING_TABLE = 100 FW_MARK = 0x64 # Vivox signaling ports (STUN/control) VIVOX_PORTS = [3478, 3479, 3386, 5060, 5061, 19302, 500] # Vivox media port range (RTP - negotiated dynamically) VIVOX_MEDIA_PORT_START = 24000 VIVOX_MEDIA_PORT_END = 30000 R = "\033[0;31m"; G = "\033[0;32m"; C = "\033[0;36m"; Y = "\033[1;33m"; N = "\033[0m" def die(msg): print(f"{R}[ERROR]{N} {msg}"); sys.exit(1) def info(msg): print(f"{C}[INFO]{N} {msg}") def ok(msg): print(f"{G}[OK]{N} {msg}") def warn(msg): print(f"{Y}[WARN]{N} {msg}") def run_cmd(cmd, check=True, quiet=False): if quiet: return subprocess.run(cmd, check=check, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) return subprocess.run(cmd, check=check) def generate_keypair(): private = subprocess.check_output(["wg", "genkey"]).decode().strip() public = subprocess.check_output(["wg", "pubkey"], input=private.encode()).decode().strip() return private, public def get_config_from_server(public_key): payload = json.dumps({"clientPubKey": public_key, "apiKey": API_KEY}).encode() req = urllib.request.Request(API_URL, data=payload, headers={"Content-Type": "application/json"}, method="POST") ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE with urllib.request.urlopen(req, context=ctx) as resp: return json.loads(resp.read()) def get_gateway(): out = subprocess.check_output(["ip", "route", "show", "default"]).decode() for line in out.splitlines(): parts = line.split() if "default" in parts: return parts[parts.index("via") + 1] die("Could not determine default gateway") def wait_for_handshake(timeout=15): deadline = time.time() + timeout while time.time() < deadline: try: out = subprocess.check_output( ["wg", "show", WG_IFACE, "latest-handshakes"], stderr=subprocess.DEVNULL).decode() for line in out.splitlines(): parts = line.split() if len(parts) >= 2 and parts[1] != "0": return True except subprocess.CalledProcessError: pass time.sleep(0.5) return False def disable_rp_filter(): """Disable reverse path filtering to allow tunnel responses""" info("Disabling reverse path filtering...") for iface in ["all", "default", WG_IFACE]: run_cmd(["sysctl", "-w", f"net.ipv4.conf.{iface}.rp_filter=0"], check=False, quiet=True) ok("Reverse path filtering disabled") def setup_routing(server_config, private_key, gateway): remote_ip = server_config["remoteIp"] wg_port = server_config["endpoint"].split(":")[1] wstunnel_remote_port = server_config["wstunnelRemotePort"] address = server_config["address"] server_key = server_config["serverKey"] keepalive = str(server_config["presKeepAlive"]) # Tunnel IP tunnel_ip = address.split("/")[0] # 1. Start wstunnel info("Starting wstunnel...") log_file = open(WSTUNNEL_LOG, "w") proc = subprocess.Popen( ["wstunnel", "client", "--tls-sni-override", "talku.ddns.net", "-L", f"udp://{wg_port}:localhost:{wg_port}?timeout_sec=60", f"wss://{remote_ip}:{wstunnel_remote_port}"], stdout=log_file, stderr=log_file ) with open(WSTUNNEL_PID, "w") as f: f.write(str(proc.pid)) time.sleep(2) # Give wstunnel time to start # 2. Create WireGuard interface run_cmd(["ip", "link", "add", "dev", WG_IFACE, "type", "wireguard"]) # 3. WireGuard config wg_conf = f"""[Interface] PrivateKey = {private_key} # ListenPort removed to allow WireGuard to pick a random available port # and avoid conflict with wstunnel on {wg_port} [Peer] PublicKey = {server_key} Endpoint = 127.0.0.1:{wg_port} AllowedIPs = 0.0.0.0/0 PersistentKeepalive = {keepalive} """ with tempfile.NamedTemporaryFile(mode="w", suffix=".conf", delete=False) as tmp: tmp.write(wg_conf) tmp_path = tmp.name run_cmd(["wg", "setconf", WG_IFACE, tmp_path]) os.unlink(tmp_path) # 4. Assign IP and bring up run_cmd(["ip", "addr", "add", address, "dev", WG_IFACE]) run_cmd(["ip", "link", "set", "dev", WG_IFACE, "up"]) # 5. Route VPN server via real gateway (prevent loop) # FIX: Use 'replace' instead of 'add' to prevent "File exists" crash run_cmd(["ip", "route", "replace", f"{remote_ip}/32", "via", gateway], quiet=True, check=False) # Also route DNS servers to prevent issues try: with open("/etc/resolv.conf") as f: for dns_line in f.readlines(): if dns_line.strip().startswith("nameserver"): dns_ip = dns_line.split()[1] if not dns_ip.startswith("127."): # Skip local resolvers like systemd-resolved run_cmd(["ip", "route", "replace", f"{dns_ip}/32", "via", gateway], quiet=True, check=False) except Exception: pass # 6. Disable rp_filter BEFORE adding routes disable_rp_filter() # 7. Custom routing table: default via tunnel # FIX: Use 'replace' here too just in case run_cmd(["ip", "route", "replace", "default", "dev", WG_IFACE, "table", str(ROUTING_TABLE)], quiet=True, check=False) # 8. ip rule: route marked packets via table 100 # FIX: Ignore error if rule already exists from a prior crash run_cmd(["ip", "rule", "add", "fwmark", str(FW_MARK), "table", str(ROUTING_TABLE)], quiet=True, check=False) # 9. NAT/Masquerade for tunnel traffic info("Setting up NAT masquerade...") run_cmd(["iptables", "-t", "nat", "-A", "POSTROUTING", "-o", WG_IFACE, "-j", "MASQUERADE"], check=False) ok("NAT masquerade configured") # 10. Mark Vivox signaling ports (OUTPUT only) info("Setting up port-based marking...") for port in VIVOX_PORTS: run_cmd(["iptables", "-t", "mangle", "-A", "OUTPUT", "-p", "udp", "--dport", str(port), "-j", "MARK", "--set-mark", str(FW_MARK)], check=False) run_cmd(["iptables", "-t", "mangle", "-A", "OUTPUT", "-p", "tcp", "--dport", str(port), "-j", "MARK", "--set-mark", str(FW_MARK)], check=False) # 11. Mark Vivox media port range warn(f"Marking media ports {VIVOX_MEDIA_PORT_START}-{VIVOX_MEDIA_PORT_END}") run_cmd(["iptables", "-t", "mangle", "-A", "OUTPUT", "-p", "udp", "--dport", f"{VIVOX_MEDIA_PORT_START}:{VIVOX_MEDIA_PORT_END}", "-j", "MARK", "--set-mark", str(FW_MARK)], check=False) # 12. Use conntrack to mark established connections info("Setting up conntrack marking...") run_cmd(["iptables", "-t", "mangle", "-A", "OUTPUT", "-m", "conntrack", "--ctstate", "ESTABLISHED,RELATED", "-j", "CONNMARK", "--restore-mark"], check=False) run_cmd(["iptables", "-t", "mangle", "-A", "OUTPUT", "-p", "udp", "-m", "conntrack", "--ctstate", "NEW", "-j", "CONNMARK", "--save-mark"], check=False) ok("Routing rules configured") warn("Vivox signaling ports: " + str(VIVOX_PORTS)) warn("Vivox media port range: " + f"{VIVOX_MEDIA_PORT_START}-{VIVOX_MEDIA_PORT_END}") def bring_down_interface(): # Kill wstunnel if os.path.exists(WSTUNNEL_PID): try: with open(WSTUNNEL_PID) as f: os.kill(int(f.read().strip()), signal.SIGTERM) except Exception: pass os.remove(WSTUNNEL_PID) subprocess.run(["pkill", "-f", "wstunnel client"], stderr=subprocess.DEVNULL) # Remove iptables mangle rules for port in VIVOX_PORTS: subprocess.run(["iptables", "-t", "mangle", "-D", "OUTPUT", "-p", "udp", "--dport", str(port), "-j", "MARK", "--set-mark", "0x64"], stderr=subprocess.DEVNULL) subprocess.run(["iptables", "-t", "mangle", "-D", "INPUT", "-p", "udp", "--sport", str(port), "-j", "MARK", "--set-mark", "0x64"], stderr=subprocess.DEVNULL) # Remove ip rule and route subprocess.run(["ip", "rule", "del", "fwmark", "0x64", "table", "100"], stderr=subprocess.DEVNULL) subprocess.run(["ip", "route", "flush", "table", "100"], stderr=subprocess.DEVNULL) # Remove interface subprocess.run(["ip", "link", "delete", "dev", WG_IFACE], stderr=subprocess.DEVNULL) def debug_show_rules(): """Show current marking rules for debugging""" print(f"\n{C}=== MANGLE TABLE ==={N}") subprocess.run(["iptables", "-t", "mangle", "-L", "OUTPUT", "-n", "-v"]) print(f"\n{C}=== NAT TABLE ==={N}") subprocess.run(["iptables", "-t", "nat", "-L", "POSTROUTING", "-n", "-v"]) print(f"\n{C}=== IP RULES ==={N}") subprocess.run(["ip", "rule", "list"]) print(f"\n{C}=== TABLE 100 ==={N}") subprocess.run(["ip", "route", "show", "table", "100"]) def connect(): if os.path.exists(f"/sys/class/net/{WG_IFACE}"): die(f"Already connected. Run: sudo python3 {sys.argv[0]} disconnect") info("Generating WireGuard keypair...") private_key, public_key = generate_keypair() ok(f"Public key: {public_key}") info("Fetching config from TalkU server...") server_config = get_config_from_server(public_key) ok(f"Assigned address : {server_config['address']}") ok(f"Remote IP : {server_config['remoteIp']}") ok(f"wstunnel : udp/{server_config['endpoint'].split(':')[1]} → wss://{server_config['remoteIp']}:{server_config['wstunnelRemotePort']}") gateway = get_gateway() info(f"Default gateway : {gateway}") setup_routing(server_config, private_key, gateway) info("Waiting for handshake (up to 15s)...") if wait_for_handshake(timeout=15): ok("Handshake successful! TalkU is connected.") subprocess.run(["wg", "show", WG_IFACE]) if "--debug" in sys.argv: debug_show_rules() else: print(f"{R}[ERROR]{N} No handshake after 15s.") print(f"{Y}Check wstunnel log: {WSTUNNEL_LOG}{N}") with open(WSTUNNEL_LOG) as f: print(f.read()[-500:] if len(f.read()) > 500 else f.read()) bring_down_interface() die("Failed to connect.") def disconnect(): bring_down_interface() ok("Disconnected.") def status(): if os.path.exists(f"/sys/class/net/{WG_IFACE}"): ok("TalkU is CONNECTED") subprocess.run(["wg", "show", WG_IFACE]) if "--debug" in sys.argv: debug_show_rules() else: info("TalkU is DISCONNECTED") def test_vivox_routing(): """Test if Vivox ports are being routed correctly""" info("Testing Vivox port routing...") # Get a Vivox test IP (common Vivox servers) test_hosts = ["prod.talkgadget.google.com", "voice-east.proximic.com"] for port in VIVOX_PORTS[:3]: result = subprocess.run( ["iptables", "-t", "mangle", "-C", "OUTPUT", "-p", "udp", "--dport", str(port), "-j", "MARK", "--set-mark", str(FW_MARK)], capture_output=True ) status = "✓" if result.returncode == 0 else "✗" print(f" Port {port}: {status}") if __name__ == "__main__": if os.geteuid() != 0: die("Must run as root: sudo python3 talku.py ") commands = { "connect": connect, "disconnect": disconnect, "status": status, "debug": test_vivox_routing, } if len(sys.argv) < 2 or sys.argv[1] not in commands: print(f"Usage: sudo python3 {sys.argv[0]} {{connect|disconnect|status|debug}} [--debug]") print("\nOptions:") print(" connect - Start tunnel") print(" disconnect - Stop tunnel") print(" status - Show connection status") print(" debug - Test routing rules") print(" --debug - Show detailed info with connect/status") sys.exit(1) commands[sys.argv[1]]()