Support/sp-hydra-veil-core
Support/sp-hydra-veil-core
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

radical change

Browse source
This commit is contained in:
Zenaku 2026-06-05 10:31:35 -05:00
parent 399c66e3cb
commit eff49cf1f7
11 changed files with 347 additions and 152 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
README.md
View file

@ -96,3 +96,4 @@ controller = HysteriaController(1080)
controller.disable(observer)
"
```
sudo ln -sf /run/systemd/resolve/resolv.conf /etc/resolv.conf

31
core/controllers/ConnectionController.py
View file

@ -217,14 +217,27 @@ class ConnectionController:
if protocol == 'vless':
from core.controllers.encrypted_proxy.VlessController import VlessController
VlessController(1080).enable(operator_proxy_session.links[0], operator_proxy_session.username, connection_observer)
ok = VlessController(1080).enable(
operator_proxy_session.links[0],
operator_proxy_session.username,
connection_observer
)
elif protocol == 'hysteria2':
from core.controllers.encrypted_proxy.HysteriaController import HysteriaController
HysteriaController(1080).enable(operator_proxy_session.username, operator_proxy_session.password, operator_proxy_session.operator_hysteria2_host, connection_observer)
ok = HysteriaController(1080).enable(
operator_proxy_session.username,
operator_proxy_session.password,
operator_proxy_session.operator_hysteria2_host,
connection_observer
)
else:
ok = False
if not ok:
raise ConnectionError('The connection could not be established.')
SystemStateController.create(profile.id)
return
if ConfigurationController.get_endpoint_verification_enabled():
ProfileController.verify_wireguard_endpoint(profile, ignore=ignore)
@ -338,7 +351,7 @@ class ConnectionController:
return subprocess.Popen(('proxychains4', '-f', proxychains_configuration_file_path, 'microsocks', '-p', str(proxy_port_number)), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
@staticmethod
def terminate_system_connection():
def terminate_system_connection(connection_observer: Optional[ConnectionObserver] = None):
system_state = SystemStateController.get()
if system_state is not None:
profile = ProfileController.get(system_state.profile_id)
@ -346,10 +359,10 @@ class ConnectionController:
protocol = profile.connection.get_protocol()
if protocol == 'vless':
from core.controllers.encrypted_proxy.VlessController import VlessController
VlessController(1080).disable()
VlessController(1080).disable(connection_observer)
elif protocol == 'hysteria2':
from core.controllers.encrypted_proxy.HysteriaController import HysteriaController
HysteriaController(1080).disable()
HysteriaController(1080).disable(connection_observer)
SystemState.dissolve()
return
@ -357,18 +370,18 @@ class ConnectionController:
raise CommandNotFoundError('nmcli')
if SystemStateController.exists():
process = subprocess.Popen(('nmcli', 'connection', 'delete', 'wg'), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
if completed_successfully or not ConnectionController.system_uses_wireguard_interface():
subprocess.run(('nmcli', 'connection', 'delete', 'hv-ipv6-sink'), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
ConnectionController.terminate_tor_connection()
SystemState.dissolve()
if connection_observer is not None:
connection_observer.notify('disconnected', {})
else:
raise ConnectionTerminationError('The connection could not be terminated.')
@staticmethod
def get_proxies(port_number: int):

21
core/controllers/ProfileController.py
View file

@ -86,51 +86,36 @@ class ProfileController:
profile_observer.notify('enabled', profile)
@staticmethod
def disable(profile: Union[SessionProfile, SystemProfile], explicitly: bool = True, ignore: tuple[type[Exception]] = (), profile_observer: ProfileObserver = None):
def disable(profile: Union[SessionProfile, SystemProfile], explicitly: bool = True, ignore: tuple[type[Exception]] = (), profile_observer: ProfileObserver = None, connection_observer: ConnectionObserver = None):
from core.controllers.ConnectionController import ConnectionController
if profile.is_session_profile():
if SessionStateController.exists(profile.id):
session_state = SessionStateController.get(profile.id)
if session_state is not None:
for port_number in session_state.network_port_numbers.tor:
ConnectionController.terminate_tor_session_connection(port_number)
session_state.dissolve(session_state.id)
if profile.is_system_profile():
subjects = ProfileController.get_all().values()
for subject in subjects:
if subject.is_session_profile():
if subject.connection.is_unprotected() and ProfileController.is_enabled(subject) and not ConnectionUnprotectedError in ignore:
raise ConnectionUnprotectedError('Disabling this system connection would leave one or more sessions exposed.')
if SystemStateController.exists():
system_state = SystemStateController.get()
if profile.id != system_state.profile_id:
raise ProfileDeactivationError('The profile could not be disabled.')
try:
ConnectionController.terminate_system_connection()
ConnectionController.terminate_system_connection(connection_observer=connection_observer)
except ConnectionTerminationError:
raise ProfileDeactivationError('The profile could not be disabled.')
if profile_observer is not None:
profile_observer.notify('disabled', profile, dict(
explicitly=explicitly,
))
profile_observer.notify('disabled', profile, dict(explicitly=explicitly))
time.sleep(1.0)

8
core/observers/ConnectionObserver.py
View file

@ -1,5 +1,9 @@
from essentials.observers.ConnectionObserver import ConnectionObserver as BaseConnectionObserver
class ConnectionObserver(BaseConnectionObserver):
pass
def __init__(self):
super().__init__()
self.on_connected = []
self.on_disconnected = []
self.on_error = []

45
core/services/encrypted_proxy/disable_service.py
View file

@ -1,33 +1,26 @@
import ssl
import subprocess
import time
import urllib.request
from pathlib import Path
from core.utils.encrypted_proxy.singbox import SingboxRunner
from core.utils.encrypted_proxy.dns import revert_dns_on_tun
from core.utils.encrypted_proxy import killswitch
def get_public_ip(timeout: int = 8) -> str:
ctx = ssl.create_default_context()
try:
req = urllib.request.Request(
endpoints = [
"https://api.ipify.org",
"https://ifconfig.me/ip",
headers={"User-Agent": "curl/7.0"}
"https://icanhazip.com",
]
for url in endpoints:
try:
r = subprocess.run(
["curl", "-s", "--max-time", str(timeout), url],
capture_output=True,
timeout=timeout + 2,
)
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
return resp.read().decode().strip()
except Exception as e:
return f"unknown ({e})"
def disable_proxy(tmp_dir: Path, wrapper: str, unit: str, observer=None) -> bool:
runner = SingboxRunner(wrapper, unit)
runner.stop()
for f in tmp_dir.glob("*-sing-box.json"):
f.unlink(missing_ok=True)
time.sleep(1)
ip = get_public_ip()
if observer:
observer.notify("disconnected", {"ip": ip})
return True
ip = r.stdout.decode().strip()
if ip and "." in ip and not ip.startswith("unknown"):
return ip
except Exception:
continue
return "unknown"

81
core/services/encrypted_proxy/hysteria_service.py
View file

@ -1,24 +1,33 @@
import socket
import subprocess
import time
from pathlib import Path
from core.Constants import Constants
from core.utils.encrypted_proxy.net import get_real_ip, verify_ip
from core.utils.encrypted_proxy.singbox import SingboxRunner
from core.utils.encrypted_proxy.dns import wait_for_tun, set_dns_on_tun, revert_dns_on_tun
from core.utils.encrypted_proxy import killswitch
from core.services.encrypted_proxy.disable_service import get_public_ip
def _resolve(host: str) -> str:
return socket.gethostbyname(host)
def build_hysteria_config(username: str, password: str,
server_host: str, socks5_port: int) -> dict:
server_ip = _resolve(server_host)
return {
"dns": {
"servers": [{"tag": "local", "type": "udp", "server": "9.9.9.9"}],
"final": "local",
"servers": [
{
"tag": "tunnel-dns",
"type": "udp",
"server": "9.9.9.9",
},
],
"final": "tunnel-dns",
"strategy": "ipv4_only",
"independent_cache": True,
},
"inbounds": [
{
@ -28,7 +37,7 @@ def build_hysteria_config(username: str, password: str,
"address": ["172.19.0.1/30"],
"mtu": 9000,
"auto_route": True,
"stack": "system",
"stack": "gvisor",
},
{
"type": "socks",
@ -57,43 +66,88 @@ def build_hysteria_config(username: str, password: str,
"rules": [
{"protocol": "dns", "action": "hijack-dns"},
{"ip_cidr": ["172.19.0.0/30"], "action": "hijack-dns"},
# DNS y servidor salen directo — nunca por el TUN
{"ip_cidr": ["9.9.9.9/32"], "outbound": "direct"},
{"ip_cidr": [f"{server_ip}/32"], "outbound": "direct"},
{"ip_is_private": True, "outbound": "direct"},
{"ip_version": 6, "outbound": "block"},
{"inbound": ["tun-in", "socks-in"], "outbound": "proxy"},
],
"final": "proxy",
"default_domain_resolver": "local",
"default_domain_resolver": "tunnel-dns",
"auto_detect_interface": True,
},
}
def _cleanup_all() -> None:
try:
SingboxRunner().stop()
except Exception:
pass
try:
revert_dns_on_tun()
except Exception:
pass
try:
killswitch.disarm()
except Exception:
pass
time.sleep(1)
def enable_hysteria(username: str, password: str, server_host: str,
socks5_port: int, observer=None) -> bool:
_cleanup_all()
server_ip = _resolve(server_host)
real_ip = get_real_ip()
if not killswitch.arm(server_ip):
if observer:
observer.notify("error", "Failed to arm kill switch")
return False
runner = SingboxRunner()
runner.stop()
config_path = Path(Constants.SINGBOX_CONFIG_DIR) / f"{username}-sing-box.json"
config = build_hysteria_config(username, password, server_host, socks5_port)
try:
runner.write_config(config_path, config)
ok = runner.start(config_path)
except Exception as e:
killswitch.disarm()
if observer:
observer.notify("error", str(e))
return False
if not ok:
killswitch.disarm()
if observer:
observer.notify("error", "sing-box not active after start")
return False
# Esperar que el proxy esté listo para rutear tráfico
time.sleep(3)
if not wait_for_tun(timeout=15.0):
killswitch.disarm()
runner.stop()
if observer:
observer.notify("error", "tun0 did not appear after 15s")
return False
set_dns_on_tun()
proxy_ip = verify_ip(socks5_port, retries=5, delay=3.0)
if not proxy_ip or proxy_ip == "unknown" or proxy_ip == real_ip:
killswitch.disarm()
revert_dns_on_tun()
runner.stop()
if observer:
observer.notify("error", "IP did not change — possible leak")
return False
if observer:
observer.notify("connected", {
"real_ip": real_ip,
@ -102,9 +156,12 @@ def enable_hysteria(username: str, password: str, server_host: str,
})
return True
def disable_hysteria(observer=None) -> bool:
revert_dns_on_tun()
SingboxRunner().stop()
killswitch.disarm()
time.sleep(1)
ip = get_public_ip()
if observer:
observer.notify("disconnected", {})
observer.notify("disconnected", {"ip": ip})
return True

75
core/services/encrypted_proxy/vless_service.py
View file

@ -1,10 +1,14 @@
from urllib.parse import unquote
from pathlib import Path
import socket
import subprocess
import time
from core.services.encrypted_proxy.disable_service import get_public_ip
from core.Constants import Constants
from core.utils.encrypted_proxy.net import get_real_ip, verify_ip
from core.utils.encrypted_proxy.singbox import SingboxRunner
import socket
import time
from core.utils.encrypted_proxy.dns import wait_for_tun, set_dns_on_tun, revert_dns_on_tun
from core.utils.encrypted_proxy import killswitch
def parse_vless_link(link: str) -> dict:
@ -39,13 +43,14 @@ def build_vless_config(vless: dict, socks5_port: int) -> dict:
"dns": {
"servers": [
{
"tag": "local",
"tag": "tunnel-dns",
"type": "udp",
"server": "9.9.9.9",
},
],
"final": "local",
"final": "tunnel-dns",
"strategy": "ipv4_only",
"independent_cache": True,
},
"inbounds": [
{
@ -55,7 +60,7 @@ def build_vless_config(vless: dict, socks5_port: int) -> dict:
"address": ["172.19.0.1/30"],
"mtu": 9000,
"auto_route": True,
"stack": "system",
"stack": "gvisor",
},
{
"type": "socks",
@ -89,43 +94,89 @@ def build_vless_config(vless: dict, socks5_port: int) -> dict:
"rules": [
{"protocol": "dns", "action": "hijack-dns"},
{"ip_cidr": ["172.19.0.0/30"], "action": "hijack-dns"},
{"ip_cidr": ["9.9.9.9/32"], "outbound": "direct"},
{"ip_cidr": [f"{server_ip}/32"], "outbound": "direct"},
{"ip_is_private": True, "outbound": "direct"},
{"ip_version": 6, "outbound": "block"},
{"inbound": ["tun-in", "socks-in"], "outbound": "proxy"},
],
"final": "proxy",
"default_domain_resolver": "local",
"default_domain_resolver": "tunnel-dns",
"auto_detect_interface": True,
},
}
def _cleanup_all() -> None:
try:
SingboxRunner().stop()
except Exception:
pass
try:
revert_dns_on_tun()
except Exception:
pass
try:
killswitch.disarm()
except Exception:
pass
time.sleep(1)
def enable_vless(vless_link: str, username: str,
socks5_port: int, observer=None) -> bool:
_cleanup_all()
vless = parse_vless_link(vless_link)
server_ip = socket.gethostbyname(vless["host"])
real_ip = get_real_ip()
if not killswitch.arm(server_ip):
if observer:
observer.notify("error", "Failed to arm kill switch")
return False
runner = SingboxRunner()
runner.stop()
config_path = Path(Constants.SINGBOX_CONFIG_DIR) / f"{username}-sing-box.json"
config = build_vless_config(vless, socks5_port)
try:
runner.write_config(config_path, config)
ok = runner.start(config_path)
except Exception as e:
killswitch.disarm()
if observer:
observer.notify("error", str(e))
return False
if not ok:
killswitch.disarm()
if observer:
observer.notify("error", "sing-box not active after start")
return False
# Esperar que el proxy esté listo para rutear tráfico
time.sleep(3)
if not wait_for_tun(timeout=15.0):
killswitch.disarm()
runner.stop()
if observer:
observer.notify("error", "tun0 did not appear after 15s")
return False
set_dns_on_tun()
proxy_ip = verify_ip(socks5_port, retries=5, delay=3.0)
if not proxy_ip or proxy_ip == "unknown" or proxy_ip == real_ip:
killswitch.disarm()
revert_dns_on_tun()
runner.stop()
if observer:
observer.notify("error", "IP did not change — possible leak")
return False
if observer:
observer.notify("connected", {
"real_ip": real_ip,
@ -136,7 +187,11 @@ def enable_vless(vless_link: str, username: str,
def disable_vless(observer=None) -> bool:
revert_dns_on_tun()
SingboxRunner().stop()
killswitch.disarm()
time.sleep(1)
ip = get_public_ip()
if observer:
observer.notify("disconnected", {})
observer.notify("disconnected", {"ip": ip})
return True

51
core/utils/encrypted_proxy/dns.py Normal file
View file

@ -0,0 +1,51 @@
import os
import subprocess
import time
RESOLVECTL_WRAPPER = "/usr/local/bin/hydraveil-resolvectl"
TUN_IFACE = "tun0"
TUN_DNS = "9.9.9.9"
_SUDO_KW = dict(
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
stdin=subprocess.DEVNULL,
env={**os.environ, "SUDO_ASKPASS": "/bin/false"},
text=True,
)
def wait_for_tun(timeout: float = 15.0, interval: float = 0.5) -> bool:
elapsed = 0.0
while elapsed < timeout:
result = subprocess.run(
["ip", "link", "show", TUN_IFACE],
capture_output=True,
)
if result.returncode == 0:
return True
time.sleep(interval)
elapsed += interval
return False
def set_dns_on_tun() -> bool:
result = subprocess.run(
["sudo", RESOLVECTL_WRAPPER, "set", TUN_DNS],
**_SUDO_KW,
)
if result.returncode != 0:
print(f"[DNS] Failed to configure tun0: {result.stderr.strip()}")
return False
return True
def revert_dns_on_tun() -> bool:
result = subprocess.run(
["sudo", RESOLVECTL_WRAPPER, "revert"],
**_SUDO_KW,
)
if result.returncode != 0:
print(f"[DNS] Failed to revert tun0: {result.stderr.strip()}")
return False
return True

36
core/utils/encrypted_proxy/killswitch.py Normal file
View file

@ -0,0 +1,36 @@
import subprocess
KILLSWITCH_WRAPPER = "/usr/local/bin/hydraveil-killswitch"
def arm(server_ip: str) -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "arm", server_ip],
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"[killswitch] Failed to arm: {result.stderr.strip()}")
return False
return True
def disarm() -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "disarm"],
capture_output=True,
text=True
)
if result.returncode != 0:
print(f"[killswitch] Failed to disarm: {result.stderr.strip()}")
return False
return True
def status() -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "status"],
capture_output=True,
text=True
)
return result.returncode == 0 and result.stdout.strip() == "armed"

3
core/utils/encrypted_proxy/net.py
View file

@ -54,7 +54,6 @@ def get_proxied_ip(socks5_port: int, timeout: int = 15) -> str:
stderr = r.stderr.decode().strip()
if ip and not ip.startswith("<") and "." in ip:
return ip
# Log silencioso para debug
if stderr:
print(f"[net] curl stderr ({url}): {stderr[:120]}")
except subprocess.TimeoutExpired:
@ -66,7 +65,7 @@ def get_proxied_ip(socks5_port: int, timeout: int = 15) -> str:
def verify_ip(socks5_port: int, retries: int = 5, delay: float = 3.0) -> str:
for attempt in range(1, retries + 1):
print(f"[net] verify_ip intento {attempt}/{retries}...")
print(f"[net] verify_ip attempt {attempt}/{retries}...")
ip = get_proxied_ip(socks5_port)
if ip != "unknown":
return ip

129
core/utils/encrypted_proxy/singbox.py
View file

@ -8,33 +8,17 @@ from core.Constants import Constants
class SingboxRunner:
"""
Gestiona el ciclo de vida de sing-box via wrapper sudo.
Flujo:
write_config() → start() → is_running() → stop()
El wrapper es el único binario con NOPASSWD en sudoers.
Nunca se llama sing-box ni ip directamente con sudo desde Python.
"""
_WRAPPER = Constants.SINGBOX_WRAPPER
_PID_FILE = Path(Constants.SINGBOX_PID_FILE)
_LOG_FILE = Path(Constants.SINGBOX_LOG_FILE)
_CONFIG_DIR = Path(Constants.SINGBOX_CONFIG_DIR)
_START_TIMEOUT_S = 15.0 # tiempo total de espera
_START_POLL_S = 0.5 # intervalo de polling
_START_INITIAL_S = 2.0 # espera inicial antes de empezar a verificar
# (el wrapper necesita ~1s para escribir el PID file)
# ── Público ───────────────────────────────────────────────────────────────
_START_TIMEOUT_S = 15.0
_START_POLL_S = 0.5
_START_INITIAL_S = 2.0
def start(self, config_path: Path) -> bool:
"""
Para cualquier instancia previa y lanza sing-box via wrapper.
Retorna True solo cuando el proceso está confirmado vivo.
"""
self.stop()
self._assert_config_safe(config_path)
self._ensure_log_file()
@ -49,11 +33,10 @@ class SingboxRunner:
)
except FileNotFoundError:
raise RuntimeError(
f'Wrapper no encontrado: {self._WRAPPER}\n'
'Ejecutá el installer primero.'
f'Wrapper not found: {self._WRAPPER}\n'
'Run the installer first.'
)
# Esperar que el wrapper escriba el PID file antes de empezar polling
time.sleep(self._START_INITIAL_S)
deadline = time.monotonic() + self._START_TIMEOUT_S
@ -63,15 +46,13 @@ class SingboxRunner:
time.sleep(self._START_POLL_S)
raise RuntimeError(
f'sing-box no levantó en {self._START_TIMEOUT_S}s.\n'
f'Último error:\n{self.get_last_error()}'
f'sing-box did not start within {self._START_TIMEOUT_S}s.\n'
f'Last error:\n{self.get_last_error()}'
)
def stop(self) -> None:
"""
Para sing-box limpiamente via wrapper.
Nunca lanza excepción — stop debe ser siempre seguro de llamar.
"""
self._kill_orphans()
try:
subprocess.run(
['sudo', self._WRAPPER, '', 'stop'],
@ -85,16 +66,7 @@ class SingboxRunner:
self._emergency_kill()
def is_running(self) -> bool:
"""
Verifica por PID file + señal 0.
PermissionError significa que el proceso VIVE (corre como root).
Si el PID file no existe todavía, verifica por pgrep como fallback
para cubrir la ventana entre que el proceso arranca y el wrapper
escribe el PID file.
"""
if not self._PID_FILE.exists():
# Fallback: el proceso puede estar vivo pero el PID file
# aún no fue escrito por el wrapper
try:
result = subprocess.run(
['pgrep', '-x', 'sing-box'],
@ -115,14 +87,9 @@ class SingboxRunner:
self._PID_FILE.unlink(missing_ok=True)
return False
except PermissionError:
# Proceso vivo pero de root — PermissionError es señal de vida
return True
def get_last_error(self) -> str:
"""
Lee las últimas 80 líneas del log via sudo para evitar
problemas de permisos (el log lo escribe root via wrapper).
"""
try:
result = subprocess.run(
['sudo', self._WRAPPER, '', 'log'],
@ -141,10 +108,6 @@ class SingboxRunner:
return ''
def write_config(self, config_path: Path, config: dict) -> None:
"""
Escribe el JSON de forma atómica con permisos 0o600.
Write en .tmp + rename — nunca deja un config a medias.
"""
self._assert_config_safe(config_path)
config_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
@ -158,44 +121,82 @@ class SingboxRunner:
tmp.unlink(missing_ok=True)
raise
# ── Privado ───────────────────────────────────────────────────────────────
def _ensure_log_file(self) -> None:
"""
Crea el log file con permisos del usuario actual si no existe.
Debe llamarse ANTES de que el wrapper lo tome como root.
"""
self._LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
if not self._LOG_FILE.exists():
self._LOG_FILE.touch(mode=0o600)
def _assert_config_safe(self, config_path: Path) -> None:
"""
Verifica que el config está dentro del directorio permitido.
Lanza ValueError si no — nunca continúa con path inseguro.
"""
try:
config_path.resolve().relative_to(self._CONFIG_DIR.resolve())
except ValueError:
raise ValueError(
f'Config fuera del directorio permitido.\n'
f'Config outside the allowed directory.\n'
f' Config: {config_path}\n'
f' Permitido: {self._CONFIG_DIR}'
f' Allowed: {self._CONFIG_DIR}'
)
def _emergency_kill(self) -> None:
"""
Kill directo por PID file cuando el wrapper no responde.
Último recurso — no usa sudo.
"""
if not self._PID_FILE.exists():
return
try:
pid = int(self._PID_FILE.read_text().strip())
if pid > 0:
os.kill(pid, 9)
except (ValueError, ProcessLookupError, PermissionError, OSError):
subprocess.run(
['sudo', 'kill', '-9', str(pid)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
timeout=5,
)
except (ValueError, OSError, subprocess.TimeoutExpired):
pass
finally:
self._PID_FILE.unlink(missing_ok=True)
def _kill_orphans(self) -> None:
try:
result = subprocess.run(
['pgrep', '-x', 'sing-box'],
capture_output=True,
text=True,
)
if result.returncode != 0:
return
pids = [
int(p)
for p in result.stdout.strip().splitlines()
if p.strip().isdigit()
]
if not pids:
return
subprocess.run(
['sudo', self._WRAPPER, '', 'stop'],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
timeout=15,
)
deadline = time.monotonic() + 4.0
while time.monotonic() < deadline:
check = subprocess.run(
['pgrep', '-x', 'sing-box'],
capture_output=True,
)
if check.returncode != 0:
return
time.sleep(0.3)
except (OSError, ValueError, subprocess.TimeoutExpired):
pass