54 lines
No EOL
1.5 KiB
Python
54 lines
No EOL
1.5 KiB
Python
import time
|
|
import threading
|
|
from core.utils.encrypted_proxy import killswitch
|
|
from core.utils.encrypted_proxy.net import get_proxied_ip
|
|
|
|
|
|
class KillswitchMonitor:
|
|
|
|
POLL_INTERVAL = 10
|
|
|
|
def __init__(self):
|
|
self._on_disconnect = None
|
|
self._on_leak = None
|
|
self._socks5_port = None
|
|
self._real_ip = None
|
|
self._running = False
|
|
self._stop_event = threading.Event()
|
|
|
|
def set_on_disconnect(self, fn):
|
|
self._on_disconnect = fn
|
|
|
|
def set_on_leak(self, fn):
|
|
self._on_leak = fn
|
|
|
|
def set_connection_data(self, socks5_port, real_ip):
|
|
self._socks5_port = socks5_port
|
|
self._real_ip = real_ip
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
self._stop_event.set()
|
|
|
|
def run_blocking(self):
|
|
self._running = True
|
|
self._stop_event.clear()
|
|
|
|
while self._running:
|
|
interrupted = self._stop_event.wait(timeout=self.POLL_INTERVAL)
|
|
if interrupted or not self._running:
|
|
break
|
|
|
|
if not killswitch.status():
|
|
if self._on_disconnect:
|
|
self._on_disconnect()
|
|
self._running = False
|
|
return
|
|
|
|
if self._socks5_port and self._real_ip:
|
|
proxied = get_proxied_ip(self._socks5_port, timeout=10)
|
|
if proxied == "unknown" or proxied == self._real_ip:
|
|
if self._on_leak:
|
|
self._on_leak()
|
|
self._running = False
|
|
return |