#!/usr/bin/env python3 """ DeVpn Provider v4.0 All-in-one installer + agent + NAT for provider exit nodes. Windows NAT uses WinDivert (kernel packet interception) instead of broken ICS/NetNat. Linux uses iptables MASQUERADE as before. Usage: python dvpn-provider.py # auto-detect: setup or agent python dvpn-provider.py --setup # force setup wizard python dvpn-provider.py --agent # force agent mode python dvpn-provider.py --uninstall # remove everything https://devpn.org/provider Copyright (c) 2026 DVPN LLC - Open Source (MIT License) """ import os import sys import json import time import struct import socket import platform import subprocess import urllib.request import urllib.error import shutil import threading import hashlib from datetime import datetime VERSION = "4.3.8" IS_WINDOWS = platform.system() == "Windows" if IS_WINDOWS: import ctypes # ── Paths ──────────────────────────────────────────────────────────── if IS_WINDOWS: CONFIG_DIR = r"C:\dvpn" CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json") WG_EXE = r"C:\Program Files\WireGuard\wg.exe" WG_QUICK = r"C:\Program Files\WireGuard\wireguard.exe" WG_CONF = os.path.join(CONFIG_DIR, "wg0.conf") AGENT_DEST = os.path.join(CONFIG_DIR, "dvpn-provider.py") NAT_DIR = os.path.join(CONFIG_DIR, "windivert") else: CONFIG_DIR = "/opt/dvpn-node" CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json") WG_EXE = "wg" WG_QUICK = "wg-quick" WG_CONF = "/etc/wireguard/wg0.conf" AGENT_DEST = os.path.join(CONFIG_DIR, "dvpn-provider.py") NAT_DIR = None # Stats file for Node Pro dashboard if IS_WINDOWS: STATS_DIR = os.path.join(CONFIG_DIR, "stats") STATS_FILE = os.path.join(STATS_DIR, "stats.json") else: STATS_DIR = "/opt/devpn" STATS_FILE = "/opt/devpn/stats.json" STATS_PORT = 8081 # Dashboard auto-update DASHBOARD_FILE = "/opt/devpn/dashboard.html" if not IS_WINDOWS else os.path.join(CONFIG_DIR, "dashboard.html") DASHBOARD_SERVICE = "devpn-dashboard" API_URL = "https://api.devpn.org" WG_INTERFACE = "wg0" INTERNAL_SUBNET = "10.0.0." INTERNAL_GW = "10.0.0.1" HEARTBEAT_INTERVAL = 30 PEER_SYNC_INTERVAL = 5 SPEEDTEST_INTERVAL = 86400 # 24 hours - daily speed test builds 7-day avg DOWNLOAD_TEST_URL = "http://speedtest.newark.linode.com/100MB-newark.bin" UPLOAD_TEST_URL = "https://speed.cloudflare.com/__up" location_data = None speed_data = {"speed_mbps": None, "upload_mbps": None, "latency_ms": None} last_speedtest = 0 # ═══════════════════════════════════════════════════════════════════════ # COMMON UTILITIES # ═══════════════════════════════════════════════════════════════════════ def log(msg): ts = datetime.now().strftime("%H:%M:%S") print(f"[{ts}] {msg}", flush=True) def is_admin(): if IS_WINDOWS: try: return ctypes.windll.shell32.IsUserAnAdmin() != 0 except: return False else: return os.geteuid() == 0 def run_cmd(cmd, timeout=30): try: r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout) return r.stdout.strip() if r.returncode == 0 else None except: return None def run_cmd_full(cmd, timeout=30): try: r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout) return r.stdout.strip(), r.stderr.strip(), r.returncode except Exception as e: return "", str(e), -1 def wg_cmd(*args): cmd_list = ([WG_EXE] if IS_WINDOWS else ["wg"]) + list(args) try: r = subprocess.run(cmd_list, capture_output=True, text=True, timeout=10) return r.stdout.strip() if r.returncode == 0 else None except: return None def http_post(url, data): try: payload = json.dumps(data).encode("utf-8") req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json", "User-Agent": "DeVpnProvider/4.0"}, method="POST") with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except Exception as e: log(f"HTTP POST error ({url}): {e}") return None def http_get(url): try: req = urllib.request.Request(url, headers={"User-Agent": "DeVpnProvider/4.0"}) with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except Exception as e: log(f"HTTP GET error ({url}): {e}") return None def http_download(url, dest): try: req = urllib.request.Request(url, headers={"User-Agent": "DeVpnProvider/4.0"}) with urllib.request.urlopen(req, timeout=120) as resp: with open(dest, "wb") as f: f.write(resp.read()) return True except Exception as e: log(f"Download error ({url}): {e}") return False def config_exists(): return os.path.exists(CONFIG_FILE) def load_config(): if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE) as f: return json.load(f) log(f"ERROR: Config not found at {CONFIG_FILE}") sys.exit(1) def save_config(config): os.makedirs(CONFIG_DIR, exist_ok=True) with open(CONFIG_FILE, "w") as f: json.dump(config, f, indent=2) def detect_external_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except: return None # ═══════════════════════════════════════════════════════════════════════ # WINDIVERT NAT ENGINE (Windows only) # ═══════════════════════════════════════════════════════════════════════ class WinDivertNAT: """ Userspace NAT using WinDivert. Replaces iptables MASQUERADE on Windows. Intercepts packets from 10.0.0.0/24, rewrites source IP to external IP, tracks connections for return traffic. """ def __init__(self, external_ip): self.external_ip = external_ip self.running = False self.lock = threading.Lock() # (proto, src_ip, src_port, dst_ip, dst_port) -> nat_port self.outbound_map = {} # (proto, nat_port, remote_ip, remote_port) -> (orig_src_ip, orig_src_port) self.inbound_map = {} # ICMP: (icmp_id, dst_ip) -> orig_src_ip self.icmp_map = {} self.next_port = 10000 self.max_port = 60000 self.last_seen = {} self.SESSION_TIMEOUT = 120 def _get_proto(self, packet): """Get protocol number - pydivert may return tuple (proto, hdr_len).""" p = packet.protocol return p[0] if isinstance(p, tuple) else p def _get_proto(self, packet): """Get protocol number - pydivert may return tuple (proto, hdr_len).""" p = packet.protocol return p[0] if isinstance(p, tuple) else p log(f"NAT engine initialized: {self.external_ip}") def _alloc_port(self): port = self.next_port self.next_port += 1 if self.next_port > self.max_port: self.next_port = 10000 self._cleanup() return port def _cleanup(self): now = time.time() expired = [k for k, t in self.last_seen.items() if now - t > self.SESSION_TIMEOUT] for key in expired: nat_port = self.outbound_map.pop(key, None) if nat_port: to_del = [rk for rk in self.inbound_map if rk[1] == nat_port] for rk in to_del: self.inbound_map.pop(rk, None) self.last_seen.pop(key, None) if expired: log(f"NAT: cleaned {len(expired)} expired sessions") def run(self): try: import pydivert except ImportError: log("ERROR: pydivert not installed") return self.running = True ext = self.external_ip filt = ( f"(ip.SrcAddr >= 10.0.0.1 and ip.SrcAddr <= 10.0.0.254" f")" f" or " f"(inbound and ip.DstAddr == {ext}" f")" ) log(f"NAT: starting WinDivert capture") try: with pydivert.WinDivert(filt) as w: log("NAT: capture active - traffic flowing") while self.running: try: packet = w.recv() except: if not self.running: break continue try: self._process(w, packet) except: try: w.send(packet) except: pass except Exception as e: log(f"NAT error: {e}") self.running = False def _process(self, w, packet): src = packet.src_addr dst = packet.dst_addr # OUTBOUND: 10.0.0.x -> internet if src.startswith(INTERNAL_SUBNET) and not dst.startswith(INTERNAL_SUBNET): proto = self._get_proto(packet) if proto == 1: # ICMP self._icmp_out(w, packet) return src_port = getattr(packet, 'src_port', 0) or 0 dst_port = getattr(packet, 'dst_port', 0) or 0 if src_port == 0: w.send(packet) return key = (proto, src, src_port, dst, dst_port) with self.lock: if key in self.outbound_map: nat_port = self.outbound_map[key] else: nat_port = self._alloc_port() self.outbound_map[key] = nat_port self.inbound_map[(proto, nat_port, dst, dst_port)] = (src, src_port) self.last_seen[key] = time.time() packet.src_addr = self.external_ip packet.src_port = nat_port w.send(packet, recalculate_checksum=True) return # INBOUND: internet -> our external IP (NAT return?) if dst == self.external_ip: proto = self._get_proto(packet) if proto == 1: self._icmp_in(w, packet) return dst_port = getattr(packet, 'dst_port', 0) or 0 src_port = getattr(packet, 'src_port', 0) or 0 if dst_port == 0: w.send(packet) return rkey = (proto, dst_port, src, src_port) with self.lock: if rkey in self.inbound_map: orig_ip, orig_port = self.inbound_map[rkey] packet.dst_addr = orig_ip packet.dst_port = orig_port w.send(packet, recalculate_checksum=True) return w.send(packet) return w.send(packet) def _icmp_out(self, w, packet): raw = packet.raw if len(raw) < 28: w.send(packet); return ihl = (raw[0] & 0x0F) * 4 icmp_id = struct.unpack("!H", raw[ihl+4:ihl+6])[0] with self.lock: self.icmp_map[(icmp_id, packet.dst_addr)] = packet.src_addr packet.src_addr = self.external_ip w.send(packet, recalculate_checksum=True) def _icmp_in(self, w, packet): raw = packet.raw if len(raw) < 28: w.send(packet); return ihl = (raw[0] & 0x0F) * 4 icmp_id = struct.unpack("!H", raw[ihl+4:ihl+6])[0] with self.lock: orig = self.icmp_map.get((icmp_id, packet.src_addr)) if orig: packet.dst_addr = orig w.send(packet, recalculate_checksum=True) else: w.send(packet) def stop(self): self.running = False # ═══════════════════════════════════════════════════════════════════════ # INSTALLER / SETUP WIZARD # ═══════════════════════════════════════════════════════════════════════ def print_banner(): print() print("=" * 55) print(" ____ __ __ ") print(" | _ \\ \\ \\ / /_ __ _ __ ") print(" | | | | \\ \\ / /| '_ \\| '_ \\ ") print(" | |_| | \\ V / | |_) | | | |") print(" |____/ \\_/ | .__/|_| |_|") print(" |_| ") print(f" Provider Setup v{VERSION}") print(" https://devpn.org/provider") print("=" * 55) print() def check_wg(): return os.path.exists(WG_EXE) if IS_WINDOWS else shutil.which("wg") is not None def install_wireguard(): if check_wg(): print("[OK] WireGuard already installed") return True print("[...] Installing WireGuard...") if IS_WINDOWS: url = "https://download.wireguard.com/windows-client/wireguard-installer.exe" inst = os.path.join(os.environ.get("TEMP", r"C:\Temp"), "wg-install.exe") try: print(" Downloading...") urllib.request.urlretrieve(url, inst) print(" Installing (may take a minute)...") subprocess.run([inst, "/S"], timeout=120) time.sleep(3) if os.path.exists(WG_EXE): print("[OK] WireGuard installed") return True print("[ERROR] Install failed. Get it from https://wireguard.com/install/") return False except Exception as e: print(f"[ERROR] {e}. Install from https://wireguard.com/install/") return False else: distro = run_cmd("cat /etc/os-release | grep ^ID= | cut -d= -f2 | tr -d '\"'") if distro in ("ubuntu", "debian"): os.system("apt-get update -qq && apt-get install -y -qq wireguard-tools curl") else: os.system("yum install -y -q wireguard-tools curl 2>/dev/null || apt-get install -y -qq wireguard-tools curl 2>/dev/null") return shutil.which("wg") is not None def install_windivert(): if not IS_WINDOWS: return True print("[...] Setting up WinDivert NAT engine...") # Install pydivert print(" Installing pydivert...") _, stderr, rc = run_cmd_full(f'"{sys.executable}" -m pip install pydivert', timeout=60) if rc != 0: print(f"[ERROR] pip install pydivert failed: {stderr}") return False # Download WinDivert os.makedirs(NAT_DIR, exist_ok=True) dll = os.path.join(NAT_DIR, "WinDivert.dll") syf = os.path.join(NAT_DIR, "WinDivert64.sys") if os.path.exists(dll) and os.path.exists(syf): print("[OK] WinDivert binaries present") return True print(" Downloading WinDivert 2.2.2...") zip_path = os.path.join(os.environ.get("TEMP", r"C:\Temp"), "windivert.zip") if not http_download("https://github.com/basil00/WinDivert/releases/download/v2.2.2/WinDivert-2.2.2-A.zip", zip_path): print("[ERROR] Download failed") return False print(" Extracting...") import zipfile try: with zipfile.ZipFile(zip_path, "r") as zf: for name in zf.namelist(): bn = os.path.basename(name) # Get x64 DLL and driver if bn == "WinDivert.dll" and "x64" in name: with zf.open(name) as s, open(dll, "wb") as d: d.write(s.read()) elif bn == "WinDivert64.sys": with zf.open(name) as s, open(syf, "wb") as d: d.write(s.read()) except Exception as e: print(f"[ERROR] Extract failed: {e}") return False finally: try: os.remove(zip_path) except: pass if os.path.exists(dll) and os.path.exists(syf): print(f"[OK] WinDivert installed in {NAT_DIR}") return True print("[ERROR] WinDivert files not found after extraction") return False def generate_wg_keys(): print("[...] Generating WireGuard keys...") if IS_WINDOWS: privkey = run_cmd(f'"{WG_EXE}" genkey') if not privkey: print("[ERROR] Key generation failed"); return None, None tmp = os.path.join(os.environ.get("TEMP", r"C:\Temp"), "dvpn_pk.tmp") with open(tmp, "w") as f: f.write(privkey) pubkey = run_cmd(f'type "{tmp}" | "{WG_EXE}" pubkey') try: os.remove(tmp) except: pass else: privkey = run_cmd("wg genkey") if not privkey: print("[ERROR] Key generation failed"); return None, None pubkey = run_cmd(f"echo '{privkey}' | wg pubkey") if privkey and pubkey: print(f"[OK] Public key: {pubkey[:20]}...") return privkey, pubkey return None, None def detect_location(): print("[...] Detecting location...") import subprocess as _sp services = [ {"url": "http://ip-api.com/json/", "ip": "query", "city": "city", "country": "country", "lat": "lat", "lon": "lon", "isp": "isp"}, {"url": "https://ipapi.co/json/", "ip": "ip", "city": "city", "country": "country_name", "lat": "latitude", "lon": "longitude", "isp": "org"}, {"url": "https://ipwhois.app/json/", "ip": "ip", "city": "city", "country": "country", "lat": "latitude", "lon": "longitude", "isp": "isp"}, {"url": "https://freeipapi.com/api/json", "ip": "ipAddress", "city": "cityName", "country": "countryName", "lat": "latitude", "lon": "longitude", "isp": ""}, ] for svc in services: try: data = http_get(svc["url"]) if data and data.get(svc["ip"]): loc = { "public_ip": data.get(svc["ip"], ""), "country": data.get(svc["country"], ""), "city": data.get(svc["city"], ""), "latitude": data.get(svc["lat"]), "longitude": data.get(svc["lon"]), "isp": data.get(svc["isp"], "") if svc["isp"] else "", } # IPv6 check — retry with IPv4-only services if ":" in loc["public_ip"]: print(f"[WARN] IPv6 detected ({loc['public_ip']}) — retrying with IPv4...") for ipurl in ["https://api.ipify.org", "https://ipv4.icanhazip.com"]: try: ipv4 = _sp.check_output(["curl", "-4", "-s", "--max-time", "10", ipurl], timeout=15).decode().strip() if ipv4 and ":" not in ipv4: loc["public_ip"] = ipv4 print(f"[OK] IPv4 resolved: {ipv4}") break except: continue # Final IPv6 guard — skip this service if still IPv6 if ":" in loc.get("public_ip", ""): continue print(f"[OK] {loc['city']}, {loc['country']} ({loc.get('isp','')})") print(f" Public IP: {loc['public_ip']}") return loc except: continue # Last resort: just get IP without location for ipurl in ["https://api.ipify.org", "https://ipv4.icanhazip.com", "https://ifconfig.me/ip"]: try: ipv4 = _sp.check_output(["curl", "-4", "-s", "--max-time", "10", ipurl], timeout=15).decode().strip() if ipv4 and ":" not in ipv4: print(f"[WARN] Got IP only (no location): {ipv4}") return {"public_ip": ipv4, "country": "", "city": "", "latitude": None, "longitude": None, "isp": ""} except: continue print("[WARN] All location detection services failed") return {} def read_claim_code_from_boot(): """Read claim code from boot partition (written by provisioning script)""" for path in ["/boot/firmware/devpn_claim_code.txt", "/boot/devpn_claim_code.txt"]: try: with open(path, "r") as f: code = f.read().strip() if code and code.startswith("DVPN-"): print(f"[OK] Found claim code on boot partition: {code}") return code, path except: pass return None, None def register_with_api(pid, pubkey, loc, wallet=""): print("[...] Registering with DeVpn...") data = {"provider_id": pid, "public_key": pubkey, "public_ip": loc.get("public_ip",""), "city": loc.get("city",""), "country": loc.get("country",""), "max_concurrent_users": 10} # Include claim code from boot partition if available claim_code, claim_path = read_claim_code_from_boot() if claim_code: data["claim_code"] = claim_code result = http_post(f"{API_URL}/api/provider/register", data) if result and result.get("status") in ("registered","created","updated","exists"): print(f"[OK] Registered: {result.get('status')}") if result.get("hw_linked"): print("[OK] Node linked to hardware order via claim code") # Clean up claim code file from boot partition (security) if claim_code and claim_path: try: os.remove(claim_path) print(f"[OK] Removed claim code file: {claim_path}") except: pass return result print("[ERROR] Registration failed") return None def create_wg_server_config(privkey): if IS_WINDOWS: conf = f"[Interface]\nPrivateKey = {privkey}\nAddress = {INTERNAL_GW}/24\nListenPort = 51820\n" else: iface = run_cmd("ip route | grep default | awk '{print $5}' | head -1") or "eth0" conf = f"""[Interface] PrivateKey = {privkey} Address = {INTERNAL_GW}/24 ListenPort = 51820 PostUp = iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -o {iface} -j MASQUERADE PostUp = iptables -A FORWARD -i wg0 -j ACCEPT PostUp = iptables -A FORWARD -o wg0 -j ACCEPT PostUp = sysctl -w net.ipv4.ip_forward=1 PostDown = iptables -t nat -D POSTROUTING -s 10.0.0.0/24 -o {iface} -j MASQUERADE PostDown = iptables -D FORWARD -i wg0 -j ACCEPT PostDown = iptables -D FORWARD -o wg0 -j ACCEPT """ os.makedirs(CONFIG_DIR, exist_ok=True) if IS_WINDOWS: with open(WG_CONF, "w") as f: f.write(conf) else: os.makedirs("/etc/wireguard", exist_ok=True) with open(WG_CONF, "w") as f: f.write(conf) os.chmod(WG_CONF, 0o600) print(f"[OK] WireGuard server config: {WG_CONF}") def setup_windows_forwarding(): print("[...] Enabling IP forwarding...") run_cmd('reg add "HKLM\\SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters" /v IPEnableRouter /t REG_DWORD /d 1 /f') run_cmd('powershell -Command "Get-NetIPInterface -AddressFamily IPv4 | Set-NetIPInterface -Forwarding Enabled"') run_cmd('netsh advfirewall firewall add rule name="DeVpn WireGuard" dir=in action=allow protocol=udp localport=51820') run_cmd('netsh advfirewall firewall add rule name="DeVpn Forward" dir=in action=allow protocol=any remoteip=10.0.0.0/24') print("[OK] IP forwarding enabled + firewall rules added") def start_wireguard(): print("[...] Starting WireGuard...") if IS_WINDOWS: subprocess.run([WG_QUICK, "/installtunnelservice", WG_CONF], capture_output=True, text=True, timeout=15) time.sleep(3) if wg_cmd("show", WG_INTERFACE): print("[OK] WireGuard tunnel active") else: print("[WARN] Tunnel may not have started") return True else: os.system(f"wg-quick down {WG_INTERFACE} 2>/dev/null") if os.system(f"wg-quick up {WG_INTERFACE}") == 0: print("[OK] WireGuard tunnel active") os.system(f"systemctl enable wg-quick@{WG_INTERFACE} 2>/dev/null") return True print("[ERROR] WireGuard start failed") return False def setup_service(): print("[...] Setting up auto-start...") if IS_WINDOWS: shutil.copy2(os.path.abspath(sys.argv[0]), AGENT_DEST) py = sys.executable task_cmd = f'"{py}" "{AGENT_DEST}" --agent' if getattr(sys, 'frozen', False): dest = os.path.join(CONFIG_DIR, "DeVpnProvider.exe") shutil.copy2(sys.executable, dest) task_cmd = f'"{dest}" --agent' run_cmd('schtasks /delete /tn "DeVpnProvider" /f 2>NUL') run_cmd(f'schtasks /create /tn "DeVpnProvider" /tr "{task_cmd}" /sc onstart /ru SYSTEM /rl HIGHEST /f') print("[OK] Scheduled task created") else: shutil.copy2(os.path.abspath(sys.argv[0]), AGENT_DEST) os.chmod(AGENT_DEST, 0o755) svc = f"[Unit]\nDescription=DeVpn Provider Agent\nAfter=network.target wg-quick@wg0.service\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 {AGENT_DEST} --agent\nRestart=always\nRestartSec=10\nWorkingDirectory={CONFIG_DIR}\n\n[Install]\nWantedBy=multi-user.target\n" with open("/etc/systemd/system/dvpn-provider.service", "w") as f: f.write(svc) os.system("systemctl daemon-reload && systemctl enable dvpn-provider && systemctl start dvpn-provider") print("[OK] systemd service created") def run_setup(): print_banner() if not is_admin(): print("[ERROR] Must run as Administrator/root.") input("\nPress Enter to exit..."); sys.exit(1) # Step 1: Name print("STEP 1: Provider Identity") print("-" * 40) name = "" while not name: name = input(" Provider name: ").strip().replace(" ", "-") if not name: print(" Cannot be empty.") pid = f"{name}-{hashlib.md5(f'{name}{time.time()}'.encode()).hexdigest()[:8]}" print(f" Provider ID: {pid}\n") # Step 2: Wallet print("STEP 2: Wallet (optional)") print("-" * 40) wallet = input(" Solana wallet (Enter to skip): ").strip() print() # Step 3: WireGuard print("STEP 3: WireGuard") print("-" * 40) if not install_wireguard(): input("\nPress Enter to exit..."); sys.exit(1) print() # Step 4: WinDivert (Windows) if IS_WINDOWS: print("STEP 4: NAT Engine") print("-" * 40) if not install_windivert(): print("[WARN] WinDivert setup failed - NAT may not work") print() # Step 5: Keys + Register step = 5 if IS_WINDOWS else 4 print(f"STEP {step}: Keys & Registration") print("-" * 40) privkey, pubkey = generate_wg_keys() if not privkey: input("\nPress Enter to exit..."); sys.exit(1) loc = detect_location() print() reg = register_with_api(pid, pubkey, loc, wallet) if not reg: input("\nPress Enter to exit..."); sys.exit(1) print() # Step 6: Configure & Start step += 1 print(f"STEP {step}: Configure & Start") print("-" * 40) create_wg_server_config(privkey) save_config({"provider_id": pid, "provider_name": name, "api_url": API_URL, "wg_pubkey": pubkey, "wallet_address": wallet, "tunnel_ip": INTERNAL_GW, "tunnel_subnet": "10.0.0.0/24", "version": VERSION, "installed_at": datetime.now().isoformat()}) print(f"[OK] Config saved") if IS_WINDOWS: setup_windows_forwarding() start_wireguard() setup_service() print() print("=" * 55) print(" SETUP COMPLETE!") print("=" * 55) print(f" Provider ID: {pid}") print(f" Location: {loc.get('city','?')}, {loc.get('country','?')}") print(f" Public IP: {loc.get('public_ip','?')}") print(f" Tunnel: {INTERNAL_GW}/24 :51820") print(f" NAT: {'WinDivert' if IS_WINDOWS else 'iptables'}") if wallet: print(f" Wallet: {wallet[:8]}...{wallet[-4:]}") print() print(" Dashboard: https://devpn.org/provider") print("=" * 55) print() if input("Start agent now? [Y/n]: ").strip().lower() != "n": run_agent() def run_uninstall(): print_banner() print("UNINSTALLING DeVpn Provider") print("-" * 40) # Notify API to remove provider if config_exists(): try: cfg = load_config() pid = cfg.get("provider_id") if pid: print(f"[...] Unregistering {pid} from network...") result = http_post(f"{API_URL}/api/provider/unregister", {"provider_id": pid}) if result: print(f"[OK] Unregistered: {result.get('status', 'done')}") else: print("[WARN] Could not reach API (removed locally only)") except Exception as e: print(f"[WARN] API unregister: {e}") if IS_WINDOWS: run_cmd(f'"{WG_QUICK}" /uninstalltunnelservice {WG_INTERFACE} 2>NUL') run_cmd('schtasks /delete /tn "DeVpnProvider" /f 2>NUL') run_cmd('reg add "HKLM\\SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters" /v IPEnableRouter /t REG_DWORD /d 0 /f') run_cmd('netsh advfirewall firewall delete rule name="DeVpn WireGuard" 2>NUL') run_cmd('netsh advfirewall firewall delete rule name="DeVpn Forward" 2>NUL') for d in [CONFIG_DIR, r"C:\MVPN", r"C:\mvpn"]: if os.path.exists(d): shutil.rmtree(d, ignore_errors=True) run_cmd(f'"{sys.executable}" -m pip uninstall pydivert -y 2>NUL') else: os.system(f"wg-quick down {WG_INTERFACE} 2>/dev/null") os.system("systemctl stop dvpn-provider 2>/dev/null; systemctl disable dvpn-provider 2>/dev/null") os.system("rm -f /etc/systemd/system/dvpn-provider.service; systemctl daemon-reload 2>/dev/null") if os.path.exists(CONFIG_DIR): shutil.rmtree(CONFIG_DIR, ignore_errors=True) print("[OK] Removed") # ═══════════════════════════════════════════════════════════════════════ # AGENT MODE # ═══════════════════════════════════════════════════════════════════════ def get_location_agent(): global location_data if location_data: return location_data services = [ {"url": "http://ip-api.com/json/", "ip": "query", "city": "city", "country": "country", "lat": "lat", "lon": "lon", "isp": "isp"}, {"url": "https://ipapi.co/json/", "ip": "ip", "city": "city", "country": "country_name", "lat": "latitude", "lon": "longitude", "isp": "org"}, {"url": "https://ipwhois.app/json/", "ip": "ip", "city": "city", "country": "country", "lat": "latitude", "lon": "longitude", "isp": "isp"}, {"url": "https://freeipapi.com/api/json", "ip": "ipAddress", "city": "cityName", "country": "countryName", "lat": "latitude", "lon": "longitude", "isp": ""}, ] for svc in services: try: data = http_get(svc["url"]) if data and data.get(svc["ip"]): location_data = { "public_ip": data.get(svc["ip"], ""), "country": data.get(svc["country"], ""), "city": data.get(svc["city"], ""), "latitude": data.get(svc["lat"]), "longitude": data.get(svc["lon"]), "isp": data.get(svc["isp"], "") if svc["isp"] else "", } # Skip IPv6 results if ":" in location_data.get("public_ip", ""): location_data = None continue log(f"Location: {location_data['city']}, {location_data['country']}") return location_data except: continue log("WARN: All location services failed") return location_data or {} def run_speedtest_agent(): global speed_data, last_speedtest if time.time() - last_speedtest < SPEEDTEST_INTERVAL and speed_data.get("speed_mbps"): return speed_data log("Running speed test...") dn = "NUL" if IS_WINDOWS else "/dev/null" try: out = run_cmd(f'curl -o {dn} -w "%{{speed_download}}" {DOWNLOAD_TEST_URL} 2>{dn}', timeout=120) if out: speed_data["speed_mbps"] = round(float(out) * 8 / 1_000_000, 2) except: pass try: if IS_WINDOWS: tmp = os.path.join(os.environ.get("TEMP", r"C:\Temp"), "dvpn_up.bin") run_cmd(f'fsutil file createnew "{tmp}" 52428800', timeout=10) out = run_cmd(f'curl -X POST --data-binary @"{tmp}" -w "%{{speed_upload}}" -o NUL {UPLOAD_TEST_URL} 2>NUL', timeout=120) try: os.remove(tmp) except: pass else: out = run_cmd(f'dd if=/dev/zero bs=1M count=50 2>/dev/null | curl -X POST --data-binary @- -w "%{{speed_upload}}" -o /dev/null {UPLOAD_TEST_URL} 2>/dev/null', timeout=120) if out: speed_data["upload_mbps"] = round(float(out) * 8 / 1_000_000, 2) except: pass try: if IS_WINDOWS: out = run_cmd("ping -n 3 api.devpn.org", timeout=15) if out and "Average" in out: speed_data["latency_ms"] = round(float(out.split("Average = ")[-1].strip().replace("ms","")), 2) else: out = run_cmd("ping -c 3 -q api.devpn.org", timeout=15) if out and "/" in out: parts = out.split("/") if len(parts) >= 5: speed_data["latency_ms"] = round(float(parts[4]), 2) except: pass last_speedtest = time.time() log(f"Speed: {speed_data.get('speed_mbps','?')} dn / {speed_data.get('upload_mbps','?')} up Mbps, lat: {speed_data.get('latency_ms','?')} ms") return speed_data def get_wg_peers(): peers = {} try: out = wg_cmd("show", WG_INTERFACE, "transfer") if out: for line in out.strip().split("\n"): p = line.split("\t") if len(p) >= 3: peers[p[0]] = {"bytes_received": int(p[1]), "bytes_sent": int(p[2]), "active": False} out = wg_cmd("show", WG_INTERFACE, "latest-handshakes") if out: now = time.time() for line in out.strip().split("\n"): p = line.split("\t") if len(p) >= 2 and p[0] in peers: hs = int(p[1]) if p[1] != "0" else 0 if hs > 0 and (now - hs) < 180: peers[p[0]]["active"] = True except: pass return peers def add_peer(pk, ip): try: cmd = ([WG_EXE] if IS_WINDOWS else ["wg"]) + ["set", WG_INTERFACE, "peer", pk, "allowed-ips", ip] if subprocess.run(cmd, capture_output=True, text=True, timeout=5).returncode == 0: log(f" + Peer: {ip}"); return True except: pass return False def remove_peer(pk): try: cmd = ([WG_EXE] if IS_WINDOWS else ["wg"]) + ["set", WG_INTERFACE, "peer", pk, "remove"] if subprocess.run(cmd, capture_output=True, text=True, timeout=5).returncode == 0: log(f" - Peer: {pk[:20]}..."); return True except: pass return False def sync_peers(pid): try: data = http_get(f"{API_URL}/api/provider/{pid}/peers") if not data: return 0, 0 sessions = data.get("peers", []) expected = {s["public_key"] for s in sessions} current = set(get_wg_peers().keys()) added = removed = 0 for s in sessions: if s["public_key"] not in current and add_peer(s["public_key"], s["allowed_ip"]): added += 1 for pk in current: if pk not in expected and remove_peer(pk): removed += 1 return added, removed except Exception as e: log(f"Peer sync error: {e}"); return 0, 0 def send_heartbeat(pid, peers): try: rx = sum(p["bytes_received"] for p in peers.values()) tx = sum(p["bytes_sent"] for p in peers.values()) act = sum(1 for p in peers.values() if p["active"]) data = {"provider_id": pid, "online": True, "current_users": len(peers), "bytes_sent": tx, "bytes_received": rx, "version": VERSION} try: _cfg = load_config() or {} if _cfg.get("is_cgnat"): data["is_cgnat"] = _cfg["is_cgnat"] except Exception as _de: log(f"Heartbeat config read error: {_de}") data.update(get_location_agent()) for k in ("speed_mbps", "upload_mbps", "latency_ms"): v = run_speedtest_agent().get(k) if v is not None: data[k] = v r = http_post(f"{API_URL}/api/provider/heartbeat", data) if r: log(f"Heartbeat OK | Peers: {len(peers)} (active: {act}) | TX: {tx//1024}KB RX: {rx//1024}KB") # Check for update notification in response if isinstance(r, dict) and r.get("update_available"): log(f"UPDATE AVAILABLE: v{r.get('latest_version', '?')} (current: v{VERSION})") return {"heartbeat": r, "update": True, "url": r.get("update_url"), "version": r.get("latest_version"), "force": VERSION < r.get("min_version", "0.0.0") if r.get("min_version") else False} return {"heartbeat": r, "update": False} return None except Exception as e: log(f"Heartbeat error: {e}"); return None def report_bandwidth(pid, peers): try: s = [{"public_key": pk, "bytes_received": i["bytes_received"], "bytes_sent": i["bytes_sent"]} for pk, i in peers.items()] if s: http_post(f"{API_URL}/api/provider/report-bandwidth", {"provider_id": pid, "sessions": s}) except: pass # ── Dashboard stats file + local HTTP API ──────────────────────────── _agent_start_time = None _bandwidth_today_rx = 0 _bandwidth_today_tx = 0 _bandwidth_total_rx = 0 _bandwidth_total_tx = 0 _last_hb_result = None _last_hb_time = None def write_stats_json(pid, peers, config, hb_result=None): """Write /opt/devpn/stats.json for the Node Pro dashboard.""" global _last_hb_result, _last_hb_time global _bandwidth_today_rx, _bandwidth_today_tx global _bandwidth_total_rx, _bandwidth_total_tx if hb_result: _last_hb_result = hb_result _last_hb_time = time.time() # Peer bandwidth rx = sum(p["bytes_received"] for p in peers.values()) tx = sum(p["bytes_sent"] for p in peers.values()) _bandwidth_today_rx = rx _bandwidth_today_tx = tx _bandwidth_total_rx = rx # WireGuard counters reset on restart, so current = session total _bandwidth_total_tx = tx # Uptime uptime_sec = None if _agent_start_time: uptime_sec = int(time.time() - _agent_start_time) # Update info update_available = False update_version = None hb = _last_hb_result if hb and hb.get("update"): update_available = True update_version = hb.get("version") # Also check update_pending.json if not update_available: try: upf = os.path.join(os.path.dirname(os.path.abspath(__file__)), "update_pending.json") if os.path.exists(upf): with open(upf) as f: upd = json.load(f) if upd.get("available"): update_available = True update_version = upd.get("version") except: pass # Earnings from heartbeat response dvpn_today = None dvpn_total = None if hb and isinstance(hb.get("heartbeat"), dict): dvpn_today = hb["heartbeat"].get("dvpn_earned_today") dvpn_total = hb["heartbeat"].get("dvpn_earned_total") loc = location_data or {} stats = { "provider_id": pid, "online": True, "dvpn_earned_today": dvpn_today, "dvpn_earned_total": dvpn_total, "bandwidth_today": _bandwidth_today_rx + _bandwidth_today_tx, "bandwidth_total": _bandwidth_total_rx + _bandwidth_total_tx, "current_users": sum(1 for p in peers.values() if p.get("active")), "max_concurrent_users": config.get("max_concurrent_users", 10), "uptime_seconds": uptime_sec, "started_at": _agent_start_time, "version": VERSION, "public_ip": loc.get("public_ip") or loc.get("query"), "speed_mbps": speed_data.get("speed_mbps"), "upload_mbps": speed_data.get("upload_mbps"), "last_heartbeat": _last_hb_time, "update_available": update_available, "update_version": update_version, "written_at": time.time() } try: os.makedirs(STATS_DIR, exist_ok=True) tmp = STATS_FILE + ".tmp" with open(tmp, "w") as f: json.dump(stats, f, indent=2) # Atomic rename if IS_WINDOWS: if os.path.exists(STATS_FILE): os.remove(STATS_FILE) os.rename(tmp, STATS_FILE) except Exception as e: log(f"Stats write error: {e}") NODE_DATA_FILE = "/home/devpn/node_data.json" def write_node_data(pid, peers, hb_result): """Write /home/devpn/node_data.json for the e-ink display script.""" if not os.path.isdir("/home/devpn"): return hb = hb_result.get("heartbeat", {}) if hb_result and isinstance(hb_result, dict) else {} # Active sessions sessions = sum(1 for p in peers.values() if p.get("active")) # Bandwidth today in MB bw_rx = sum(p["bytes_received"] for p in peers.values()) bw_tx = sum(p["bytes_sent"] for p in peers.values()) bandwidth_mb = int((bw_rx + bw_tx) / (1024 * 1024)) # Uptime percentage uptime_pct = 0.0 if _agent_start_time: elapsed = time.time() - _agent_start_time if elapsed > 0: uptime_pct = min(100.0, round(elapsed / 864 * 100, 1) if elapsed < 86400 else 100.0) # Speed tier from local speed test data spd = speed_data.get("speed_mbps") or 0 if spd >= 501: speed_tier = "Diamond" elif spd >= 251: speed_tier = "Platinum" elif spd >= 101: speed_tier = "Gold" elif spd >= 51: speed_tier = "Silver" else: speed_tier = "Bronze" # Staking multiplier from config cfg = load_config() stk = cfg.get("staking_multiplier", 1.0) staking_mult = f"{float(stk):.1f}x" if stk else "1.0x" data = { "online": True, "yesterday": float(hb.get("dvpn_earned_today") or 0), "total_earned": float(hb.get("dvpn_earned_total") or 0), "pending": float(hb.get("dvpn_pending") or 0), "sessions": sessions, "bandwidth_mb": bandwidth_mb, "uptime_pct": uptime_pct, "speed_tier": speed_tier, "staking_mult": staking_mult, "version": VERSION, "node_id": pid, } try: tmp = NODE_DATA_FILE + ".tmp" with open(tmp, "w") as f: json.dump(data, f, indent=2) os.rename(tmp, NODE_DATA_FILE) except Exception as e: log(f"node_data.json write error: {e}") def _stats_http_handler(): """Simple HTTP server on port 8080 serving /api/stats and /api/update.""" from http.server import HTTPServer, BaseHTTPRequestHandler class StatsHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): pass # Suppress request logging def do_GET(self): if self.path == '/api/stats': try: with open(STATS_FILE, 'r') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(data.encode()) except FileNotFoundError: self.send_response(503) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"stats not yet available"}') except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode()) else: self.send_response(404) self.end_headers() def do_POST(self): if self.path in ('/api/update', '/api/update-trigger'): # Write a trigger file the main loop can pick up try: trigger = os.path.join(STATS_DIR, "update_trigger") with open(trigger, 'w') as f: f.write(str(time.time())) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(b'{"status":"update triggered"}') except Exception as e: self.send_response(500) self.end_headers() self.wfile.write(str(e).encode()) else: self.send_response(404) self.end_headers() def do_OPTIONS(self): self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() try: server = HTTPServer(('0.0.0.0', STATS_PORT), StatsHandler) log(f"Stats HTTP server on port {STATS_PORT}") server.serve_forever() except Exception as e: log(f"Stats server error: {e}") def start_stats_server(): """Start the local stats HTTP server as a daemon thread.""" t = threading.Thread(target=_stats_http_handler, daemon=True, name="StatsHTTP") t.start() return t def ensure_nat_linux(): """Check and fix iptables MASQUERADE + ip_forward on Linux. Called every heartbeat to auto-repair if rules got lost (e.g. WG restart without PostUp).""" if IS_WINDOWS: return # Check ip_forward try: with open("/proc/sys/net/ipv4/ip_forward") as f: fwd = f.read().strip() if fwd != "1": run_cmd("sysctl -w net.ipv4.ip_forward=1") log("NAT-FIX: Enabled ip_forward") except: pass # Check iptables MASQUERADE for 10.0.0.0/24 out = run_cmd("iptables -t nat -S POSTROUTING") if out is None or "10.0.0.0/24" not in out: iface = run_cmd("ip route | grep default | awk '{print $5}' | head -1") or "eth0" run_cmd(f"iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -o {iface} -j MASQUERADE") log(f"NAT-FIX: Added MASQUERADE rule on {iface}") # Check FORWARD rules for wg0 out2 = run_cmd("iptables -S FORWARD") if out2 is None or "wg0" not in out2: run_cmd("iptables -A FORWARD -i wg0 -j ACCEPT") run_cmd("iptables -A FORWARD -o wg0 -j ACCEPT") log("NAT-FIX: Added FORWARD rules for wg0") # UPnP renewal — runs inside agent loop, no cron dependency _last_upnp_renewal = 0 UPNP_RENEWAL_INTERVAL = 1800 # 30 minutes def renew_upnp(): """Renew UPnP port mapping for UDP 51820 every 30 minutes. Runs inside the agent loop so it works even if cron is broken.""" global _last_upnp_renewal if IS_WINDOWS: return now = time.time() if now - _last_upnp_renewal < UPNP_RENEWAL_INTERVAL: return _last_upnp_renewal = now # Only renew if config says UPnP is enabled try: if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE) as f: cfg = json.load(f) if not cfg.get("upnp_enabled", False): return except: return # Check if upnpc is available if not shutil.which("upnpc"): return try: local_ip = run_cmd("hostname -I | awk '{print $1}'") if not local_ip: return # Delete old mapping first (ignore errors) run_cmd("upnpc -d 51820 UDP") # Add fresh mapping out, err, rc = run_cmd_full(f"upnpc -e 'DeVpn Node' -a {local_ip} 51820 51820 UDP") if rc == 0: log(f"UPnP renewed: {local_ip}:51820") else: log(f"UPnP renewal failed (rc={rc}): {err[:100]}") # Renew relay ports if this node is a relay try: result = subprocess.run(["ip", "link", "show", "wg-relay"], capture_output=True) if result.returncode == 0: subprocess.run(["upnpc", "-a", local_ip, "51821", "51821", "UDP"], capture_output=True, timeout=10) socat_check = subprocess.run(["pgrep", "-f", "UDP4-LISTEN:51822"], capture_output=True) if socat_check.returncode == 0: subprocess.run(["upnpc", "-a", local_ip, "51822", "51822", "UDP"], capture_output=True, timeout=10) except Exception: pass except Exception as e: log(f"UPnP renewal error: {e}") def start_nat_thread(): if not IS_WINDOWS: return None ext_ip = detect_external_ip() if not ext_ip: log("WARN: No external IP for NAT"); return None if os.path.exists(NAT_DIR): os.environ["PATH"] = NAT_DIR + ";" + os.environ.get("PATH", "") try: os.add_dll_directory(NAT_DIR) except: pass try: import pydivert except ImportError: log("WARN: pydivert not available - NAT disabled"); return None nat = WinDivertNAT(ext_ip) t = threading.Thread(target=nat.run, daemon=True, name="NAT") t.start() log(f"NAT engine started ({ext_ip})") return nat def update_dashboard(server_version): """Download latest dashboard HTML from CP1 and restart dashboard service.""" config = load_config() local_version = config.get("dashboard_version", "") if local_version == server_version: return # Already up to date if IS_WINDOWS: return # No dashboard on Windows log(f"DASHBOARD UPDATE: {local_version or 'none'} -> {server_version}") try: url = f"{API_URL}/static/node-dashboard.html" req = urllib.request.Request(url, headers={"User-Agent": f"DeVpn-Provider/{VERSION}"}) with urllib.request.urlopen(req, timeout=30) as resp: html = resp.read() if len(html) < 500: log("DASHBOARD UPDATE FAILED: Downloaded file too small"); return # Write dashboard file with open(DASHBOARD_FILE, "wb") as f: f.write(html) log(f"DASHBOARD UPDATE: Wrote {len(html)} bytes to {DASHBOARD_FILE}") # Restart dashboard service try: subprocess.run(["systemctl", "restart", DASHBOARD_SERVICE], capture_output=True, timeout=10) log(f"DASHBOARD UPDATE: Restarted {DASHBOARD_SERVICE}") except Exception as e: log(f"DASHBOARD UPDATE: Could not restart service: {e}") # Save version to config config["dashboard_version"] = server_version save_config(config) log(f"DASHBOARD UPDATE: Complete (v{server_version})") except Exception as e: log(f"DASHBOARD UPDATE FAILED: {e}") def run_system_update(version, url): """Download and run system-update.sh if version is newer than local.""" version_file = "/opt/dvpn/.system-update-version" # Check current version local_ver = "" try: if os.path.exists(version_file): with open(version_file) as f: local_ver = f.read().strip() except: pass if local_ver == version: return # Already applied log(f"SYSTEM UPDATE: {local_ver or 'none'} -> {version}") try: # Download script req = urllib.request.Request(url, headers={"User-Agent": f"DeVpn-Provider/{VERSION}"}) with urllib.request.urlopen(req, timeout=30) as resp: script = resp.read() if len(script) < 100: log("SYSTEM UPDATE FAILED: Script too small"); return # Write to temp file and execute tmp = "/tmp/dvpn-system-update.sh" with open(tmp, "wb") as f: f.write(script) os.chmod(tmp, 0o755) result = subprocess.run( ["bash", tmp], capture_output=True, timeout=120 ) if result.returncode == 0: # Write version file so it only runs once os.makedirs(os.path.dirname(version_file), exist_ok=True) with open(version_file, "w") as f: f.write(version) log(f"SYSTEM UPDATE: v{version} applied successfully") else: stderr = result.stderr.decode()[:200] if result.stderr else "" log(f"SYSTEM UPDATE: Script exited {result.returncode}: {stderr}") # Clean up try: os.remove(tmp) except: pass except Exception as e: log(f"SYSTEM UPDATE FAILED: {e}") def self_update(update_url, new_version): """Download new version and restart agent""" log(f"SELF-UPDATE: Downloading v{new_version} from {update_url}") try: script_path = os.path.abspath(__file__) backup_path = script_path + ".old" tmp_path = script_path + ".new" # Download new version req = urllib.request.Request(update_url, headers={"User-Agent": f"DeVpn-Provider/{VERSION}"}) with urllib.request.urlopen(req, timeout=30) as resp: new_code = resp.read() if len(new_code) < 1000: log("UPDATE FAILED: Downloaded file too small, aborting"); return False # Verify it's valid Python if b"VERSION" not in new_code or b"def run_agent" not in new_code: log("UPDATE FAILED: Downloaded file doesn't look like valid provider script"); return False # Write to tmp, backup old, move new with open(tmp_path, "wb") as f: f.write(new_code) if os.path.exists(backup_path): os.remove(backup_path) shutil.copy2(script_path, backup_path) shutil.move(tmp_path, script_path) os.chmod(script_path, 0o755) log(f"UPDATE SUCCESS: v{VERSION} -> v{new_version}") log("Restarting agent...") # Restart via systemd if available, otherwise exec if not IS_WINDOWS and os.path.exists("/usr/bin/systemctl"): # Find our service name for svc in ["dvpn-provider", "mvpn-provider", "devpn-provider"]: ret = subprocess.run(["systemctl", "is-active", svc], capture_output=True, text=True) if ret.stdout.strip() == "active": log(f"Restarting service: {svc}") subprocess.Popen(["systemctl", "restart", svc]) time.sleep(2) sys.exit(0) # No known service, just re-exec log("No systemd service found, re-executing...") os.execv(sys.executable, [sys.executable, script_path, "--agent"]) else: # Windows or no systemd log("Re-executing script...") os.execv(sys.executable, [sys.executable, script_path, "--agent"]) return True except Exception as e: log(f"UPDATE FAILED: {e}") # Rollback if possible backup_path = os.path.abspath(__file__) + ".old" if os.path.exists(backup_path): try: shutil.copy2(backup_path, os.path.abspath(__file__)) log("Rolled back to previous version") except: pass return False def configure_relay_peers(relay_peers): """Configure wg-relay interface to accept CGNAT node tunnels.""" import subprocess # Ensure socat is installed if subprocess.run(["which", "socat"], capture_output=True).returncode != 0: subprocess.run(["apt-get", "install", "-y", "-qq", "socat"], capture_output=True) log("[RELAY] Installed socat") WG_RELAY_IF = "wg-relay" WG_RELAY_PORT = 51821 RELAY_KEY_DIR = "/opt/devpn/relay-keys" os.makedirs(RELAY_KEY_DIR, exist_ok=True) privkey_path = os.path.join(RELAY_KEY_DIR, "relay_private.key") pubkey_path = os.path.join(RELAY_KEY_DIR, "relay_public.key") if not os.path.exists(privkey_path): privkey = subprocess.check_output(["wg", "genkey"]).decode().strip() with open(privkey_path, "w") as f: f.write(privkey) os.chmod(privkey_path, 0o600) pubkey = subprocess.check_output(["wg", "pubkey"], input=privkey.encode()).decode().strip() with open(pubkey_path, "w") as f: f.write(pubkey) log("[RELAY] Generated relay keypair") with open(privkey_path) as f: relay_privkey = f.read().strip() with open(pubkey_path) as f: relay_pubkey = f.read().strip() try: result = subprocess.run(["ip", "link", "show", WG_RELAY_IF], capture_output=True) if result.returncode != 0: conf = f"[Interface]\nPrivateKey = {relay_privkey}\nListenPort = {WG_RELAY_PORT}\n\n" for peer in relay_peers: if peer.get("cgnat_relay_pubkey") and peer.get("tunnel_subnet"): parts = peer["tunnel_subnet"].split("/")[0].split(".") peer_ip = f"{parts[0]}.{parts[1]}.{parts[2]}.2/32" conf += f"[Peer]\n" conf += f"# {peer.get('cgnat_provider_id', 'unknown')}\n" conf += f"PublicKey = {peer['cgnat_relay_pubkey']}\n" conf += f"AllowedIPs = {peer_ip}\n\n" conf_path = "/etc/wireguard/wg-relay.conf" with open(conf_path, "w") as f: f.write(conf) os.chmod(conf_path, 0o600) subprocess.run(["wg-quick", "up", WG_RELAY_IF], capture_output=True) log(f"[RELAY] Started {WG_RELAY_IF} on port {WG_RELAY_PORT} with {len(relay_peers)} peers") # UPnP forward relay tunnel port try: _lip2 = subprocess.check_output(["hostname", "-I"]).decode().strip().split()[0] subprocess.run(["upnpc", "-a", _lip2, "51821", "51821", "UDP"], capture_output=True, timeout=10) log(f"[RELAY] UPnP mapped port 51821 UDP") except Exception: pass else: current_peers = subprocess.check_output(["wg", "show", WG_RELAY_IF, "peers"]).decode().strip().split("\n") current_peers = set(p.strip() for p in current_peers if p.strip()) for peer in relay_peers: pk = peer.get("cgnat_relay_pubkey") if pk and pk not in current_peers and peer.get("tunnel_subnet"): parts = peer["tunnel_subnet"].split("/")[0].split(".") peer_ip = f"{parts[0]}.{parts[1]}.{parts[2]}.2/32" subprocess.run([ "wg", "set", WG_RELAY_IF, "peer", pk, "allowed-ips", peer_ip ], capture_output=True) log(f"[RELAY] Added peer {peer.get('cgnat_provider_id')}: {pk[:16]}...") except Exception as e: log(f"[RELAY] Error configuring relay peers: {e}") # Start socat UDP proxy for each CGNAT peer for peer in relay_peers: if peer.get("tunnel_subnet") and peer.get("client_relay_port"): parts = peer["tunnel_subnet"].split("/")[0].split(".") cgnat_tunnel_ip = f"{parts[0]}.{parts[1]}.{parts[2]}.2" client_port = peer.get("client_relay_port", 51822) # Check if socat already running for this port check = subprocess.run(["pgrep", "-f", f"UDP4-LISTEN:{client_port}"], capture_output=True) if check.returncode != 0: subprocess.Popen([ "socat", f"UDP4-LISTEN:{client_port},fork,reuseaddr,bind=0.0.0.0", f"UDP4:{cgnat_tunnel_ip}:51820" ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) log(f"[RELAY] socat proxy: port {client_port} -> {cgnat_tunnel_ip}:51820") # UPnP forward the client relay port try: _lip = subprocess.check_output(["hostname", "-I"]).decode().strip().split()[0] subprocess.run(["upnpc", "-a", _lip, str(client_port), str(client_port), "UDP"], capture_output=True, timeout=10) log(f"[RELAY] UPnP mapped port {client_port} UDP") except Exception as _ue: log(f"[RELAY] UPnP mapping failed for port {client_port}: {_ue}") # Register our relay pubkey with CP1 if not done try: config = load_config() if not config.get("relay_pubkey_registered"): resp = http_post(f"{API_URL}/api/provider/register-relay-pubkey", { "provider_id": config.get("provider_id"), "relay_public_key": relay_pubkey }) if resp: config["relay_pubkey_registered"] = True save_config(config) log("[RELAY] Registered relay pubkey with CP1") except Exception as e: log(f"[RELAY] Failed to register relay pubkey: {e}") def setup_cgnat_tunnel(): """If this node is CGNAT, establish outbound tunnel to assigned relay.""" import subprocess as _sp config = load_config() if config.get("is_cgnat") not in ("true", "likely"): return WG_RELAY_IF = "wg-relay" RELAY_KEY_DIR = "/opt/devpn/relay-keys" os.makedirs(RELAY_KEY_DIR, exist_ok=True) pid = config.get("provider_id") # Query CP1 for relay assignment FIRST (before handshake check) try: relay_info = http_get(f"{API_URL}/api/provider/my-relay?provider_id={pid}") if not relay_info or not relay_info.get("assigned"): log("[CGNAT] No relay assigned yet") return except Exception as e: log(f"[CGNAT] Failed to query relay: {e}") return relay_ip = relay_info.get("relay_public_ip") relay_port = relay_info.get("relay_wg_port", 51821) relay_pubkey = relay_info.get("relay_pubkey") tunnel_subnet = relay_info.get("tunnel_subnet") # Detect relay change and reset if needed current_relay = relay_info.get("relay_provider_id") saved_relay = config.get("relay_provider_id") if current_relay and saved_relay != current_relay: log(f"[CGNAT] Relay assignment: {saved_relay} -> {current_relay}") config["cgnat_relay_key_registered"] = False config["relay_tunnel_active"] = False config["relay_provider_id"] = current_relay save_config(config) _sp.run(["wg-quick", "down", WG_RELAY_IF], capture_output=True) # Check if tunnel already up and healthy (AFTER relay change detection) result = _sp.run(["ip", "link", "show", WG_RELAY_IF], capture_output=True) if result.returncode == 0: # Check if current tunnel points to correct relay try: wg_dump = _sp.check_output(["wg", "show", WG_RELAY_IF, "endpoints"]).decode().strip() if wg_dump and relay_ip and relay_ip not in wg_dump: log(f"[CGNAT] Tunnel points to wrong relay, tearing down. Expected {relay_ip}, got: {wg_dump}") _sp.run(["wg-quick", "down", WG_RELAY_IF], capture_output=True) config["cgnat_relay_key_registered"] = False config["relay_tunnel_active"] = False save_config(config) # Fall through to rebuild below result = _sp.run(["ip", "link", "show", WG_RELAY_IF], capture_output=True) except Exception: pass if result.returncode == 0: try: hs = _sp.check_output(["wg", "show", WG_RELAY_IF, "latest-handshakes"]).decode().strip() if hs: parts = hs.split("\t") if len(parts) >= 2 and int(parts[1]) > 0: age = int(time.time()) - int(parts[1]) if age < 180: return # Healthy else: log(f"[CGNAT] Tunnel handshake stale ({age}s), reconnecting...") _sp.run(["wg-quick", "down", WG_RELAY_IF], capture_output=True) except Exception: pass if not relay_ip: log("[CGNAT] Relay info incomplete, waiting for key exchange") return # Generate our relay keypair if not exists privkey_path = os.path.join(RELAY_KEY_DIR, "cgnat_relay_private.key") pubkey_path = os.path.join(RELAY_KEY_DIR, "cgnat_relay_public.key") if not os.path.exists(privkey_path): privkey = _sp.check_output(["wg", "genkey"]).decode().strip() with open(privkey_path, "w") as f: f.write(privkey) os.chmod(privkey_path, 0o600) pubkey = _sp.check_output(["wg", "pubkey"], input=privkey.encode()).decode().strip() with open(pubkey_path, "w") as f: f.write(pubkey) log("[CGNAT] Generated relay keypair") with open(privkey_path) as f: cgnat_privkey = f.read().strip() with open(pubkey_path) as f: cgnat_pubkey = f.read().strip() # Server says no subnet = key not registered on server, reset local state if not relay_info.get("tunnel_subnet") and config.get("cgnat_relay_key_registered"): log("[CGNAT] Server has no tunnel_subnet — resetting local key registration") config["cgnat_relay_key_registered"] = False config["relay_tunnel_active"] = False save_config(config) _sp.run(["wg-quick", "down", WG_RELAY_IF], capture_output=True) # Register our pubkey with CP1 if not done if not config.get("cgnat_relay_key_registered"): try: resp = http_post(f"{API_URL}/api/provider/register-relay-key", { "provider_id": pid, "relay_public_key": cgnat_pubkey }) if resp: config["cgnat_relay_key_registered"] = True save_config(config) tunnel_subnet = resp.get("tunnel_subnet", tunnel_subnet) relay_pubkey = resp.get("relay_pubkey", relay_pubkey) log(f"[CGNAT] Registered relay key, subnet: {tunnel_subnet}") except Exception as e: log(f"[CGNAT] Failed to register relay key: {e}") return if not relay_pubkey: log("[CGNAT] Waiting for relay node to register its pubkey") return # Build wg-relay config — CGNAT node is .2, relay is .1 parts = tunnel_subnet.split("/")[0].split(".") my_ip = f"{parts[0]}.{parts[1]}.{parts[2]}.2/30" conf = f"[Interface]\nPrivateKey = {cgnat_privkey}\nAddress = {my_ip}\n\n" conf += f"[Peer]\nPublicKey = {relay_pubkey}\n" conf += f"Endpoint = {relay_ip}:{relay_port}\n" conf += f"AllowedIPs = {parts[0]}.{parts[1]}.{parts[2]}.1/32\nPersistentKeepalive = 25\n" conf_path = "/etc/wireguard/wg-relay.conf" with open(conf_path, "w") as f: f.write(conf) os.chmod(conf_path, 0o600) try: _sp.run(["wg-quick", "up", WG_RELAY_IF], capture_output=True, check=True) log(f"[CGNAT] Tunnel UP to relay {relay_ip}:{relay_port}") config["relay_tunnel_active"] = True save_config(config) except _sp.CalledProcessError as e: log(f"[CGNAT] Failed to start tunnel: {e}") def run_agent(): log("=" * 55) log(f" DeVpn Provider Agent v{VERSION}") log(f" {platform.system()} {platform.machine()}") log("=" * 55) config = load_config() pid = config.get("provider_id") if not pid: log("ERROR: No provider_id in config"); sys.exit(1) log(f"Provider: {pid}") log(f"API: {API_URL}") log("") nat = start_nat_thread() # Dashboard stats global _agent_start_time _agent_start_time = time.time() start_stats_server() get_location_agent() run_speedtest_agent() # Immediate UPnP renewal on agent start renew_upnp() last_sync = 0 while True: try: now = time.time() peers = get_wg_peers() hb_result = send_heartbeat(pid, peers) report_bandwidth(pid, peers) # Auto-repair NAT rules if lost ensure_nat_linux() # Renew UPnP port mapping (every 30 min, inside agent — no cron needed) renew_upnp() # Write dashboard stats write_stats_json(pid, peers, config, hb_result) # Write display data for e-ink screen if hb_result: write_node_data(pid, peers, hb_result) # Check for dashboard auto-update if hb_result and isinstance(hb_result.get("heartbeat"), dict): dash_ver = hb_result["heartbeat"].get("dashboard_version") if dash_ver: update_dashboard(dash_ver) # Check for system update sys_ver = hb_result["heartbeat"].get("system_update_version") sys_url = hb_result["heartbeat"].get("system_update_url") if sys_ver and sys_url: run_system_update(sys_ver, sys_url) # Sync is_cgnat from server to local config try: _srv_cgnat = hb_result["heartbeat"].get("is_cgnat") if _srv_cgnat in ("true", "likely"): _ccfg = load_config() if _ccfg.get("is_cgnat") != _srv_cgnat: _ccfg["is_cgnat"] = _srv_cgnat save_config(_ccfg) log(f"[CGNAT] Updated local config: is_cgnat = {_srv_cgnat}") # Detect relay change _srv_relay = hb_result["heartbeat"].get("relay_provider_id") if _srv_relay: _rcfg = load_config() _old_relay = _rcfg.get("relay_provider_id") if _old_relay and _old_relay != _srv_relay: log(f"[CGNAT] Relay changed: {_old_relay} -> {_srv_relay}") _rcfg["cgnat_relay_key_registered"] = False _rcfg["relay_tunnel_active"] = False _rcfg["relay_provider_id"] = _srv_relay save_config(_rcfg) import subprocess as _rsp _rsp.run(["wg-quick", "down", "wg-relay"], capture_output=True) log("[CGNAT] Old tunnel torn down, will reconnect to new relay") elif not _old_relay: log(f"[CGNAT] First relay assignment: {_srv_relay}") _rcfg["relay_provider_id"] = _srv_relay _rcfg["cgnat_relay_key_registered"] = False _rcfg["relay_tunnel_active"] = False save_config(_rcfg) import subprocess as _rsp2 _rsp2.run(["wg-quick", "down", "wg-relay"], capture_output=True) except Exception as _cse: log(f"CGNAT config sync error: {_cse}") # Handle relay peers (if this node is a relay for CGNAT nodes) try: _rp = hb_result["heartbeat"].get("relay_peers", []) if _rp: configure_relay_peers(_rp) except Exception as _re: log(f"Relay peer config error: {_re}") # If CGNAT node, maintain tunnel to relay try: setup_cgnat_tunnel() except Exception as _ce: log(f"CGNAT tunnel error: {_ce}") # Check for dashboard-triggered update try: trigger_file = os.path.join(STATS_DIR, "update_trigger") if os.path.exists(trigger_file): os.remove(trigger_file) if _last_hb_result and _last_hb_result.get("update"): log("Dashboard triggered update, applying...") self_update(_last_hb_result["url"], _last_hb_result["version"]) except: pass # Handle updates if hb_result and hb_result.get("update"): update_mode = config.get("update_mode", "auto") # auto, manual, off if hb_result.get("force"): log("FORCED UPDATE: Version below minimum, updating now...") self_update(hb_result["url"], hb_result["version"]) elif update_mode == "auto": log(f"Auto-updating to v{hb_result['version']}...") self_update(hb_result["url"], hb_result["version"]) elif update_mode == "manual": log(f"Update available: v{hb_result['version']} — manual mode, waiting for user action") # Write update info to file for display/dashboard to read try: update_info = {"available": True, "version": hb_result["version"], "url": hb_result["url"], "checked_at": time.time()} update_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "update_pending.json") with open(update_file, "w") as uf: json.dump(update_info, uf) except: pass if now - last_sync >= PEER_SYNC_INTERVAL: a, r = sync_peers(pid) if a or r: log(f"Peer sync: +{a} -{r}") last_sync = now except KeyboardInterrupt: log("Shutting down...") if nat: nat.stop() break except Exception as e: log(f"Loop error: {e}") time.sleep(HEARTBEAT_INTERVAL) # ═══════════════════════════════════════════════════════════════════════ # ENTRY POINT # ═══════════════════════════════════════════════════════════════════════ def main(): a = sys.argv[1:] if "--uninstall" in a: run_uninstall() elif "--setup" in a: run_setup() elif "--agent" in a: run_agent() elif "--update" in a: print(f"Checking for updates (current: v{VERSION})...") try: req = urllib.request.Request(f"{API_URL}/api/updates/check?version={VERSION}") with urllib.request.urlopen(req, timeout=10) as resp: info = json.loads(resp.read()) if info.get("update_available"): print(f"Update available: v{info['latest_version']}") if input("Install now? (y/n): ").strip().lower() == "y": self_update(info["download_url"], info["latest_version"]) else: print("Skipped.") else: print("Already on latest version.") except Exception as e: print(f"Update check failed: {e}") elif "--version" in a: print(f"DeVpn Provider v{VERSION}") else: run_agent() if config_exists() else run_setup() if __name__ == "__main__": main()