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

June Major Update

Browse source
This commit is contained in:
Zenaku 2026-06-16 08:11:28 -05:00
parent c1fb1bc7c6
commit d6b122b975
12 changed files with 274 additions and 150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
README.md
View file

@ -1,4 +1,68 @@
# hydra-veil — usage
# Hydra-Veil — June Major Update
## Expanded Killswitch
`hydraveil-killswitch.sh` was added to the installer: a wrapper around `nftables` that blocks all traffic not routed through the tunnel. It defaults to `tun0`, but supports a configurable interface IP that the operator will send when creating the profile. The IP is currently hardcoded; it will be removed in the next commit.
## One Connection at a Time
First in, first out. When a new connection is established, the previous one receives a signal (`SIGUSR1`) and closes cleanly before the new one takes over.
## Unexpected Shutdowns
If the terminal is closed or the machine is powered off, the process receives `SIGHUP` and performs a full disconnect: stops `sing-box`, clears state, and releases the network. No orphan processes.
## Fewer Network Requests
Unnecessary calls made when establishing and closing connections have been removed.
---
## Repositories
| Repo | Role |
|---|---|
| `core` | Business logic, controllers, models |
| `cli` | Command-line interface |
| `essentials` | Network modules (Tor, WireGuard, proxies) |
| `installer` | Installation scripts and killswitch |
---
---
# Hydra-Veil — Gran Actualización de Junio
## Killswitch ampliado
Se añadió `hydraveil-killswitch.sh` al installer: un wrapper sobre `nftables` que bloquea todo el tráfico que no pase por el túnel. Por defecto usa `tun0`, pero soporta una IP de interfaz configurable que el operador enviará al crear el perfil. Por ahora la IP está hardcodeada; se eliminará en el próximo commit.
## Una conexión a la vez
Primero en entrar, primero en salir. Cuando se establece una nueva conexión, la anterior recibe una señal (`SIGUSR1`) y se cierra limpiamente antes de que la nueva tome el control.
## Cierres inesperados
Si se cierra la terminal o se apaga el equipo, el proceso recibe `SIGHUP` y ejecuta la desconexión completa: para `sing-box`, limpia el estado y libera la red. Sin procesos huérfanos.
## Menos solicitudes de red
Se eliminaron llamadas innecesarias que se hacían al establecer y cerrar conexiones.
---
## Repositorios
| Repo | Rol |
|---|---|
| `core` | Lógica de negocio, controladores, modelos |
| `cli` | Interfaz de línea de comandos |
| `essentials` | Módulos de red (Tor, WireGuard, proxies) |
| `installer` | Scripts de instalación y killswitch |
---
---
## Uso rápido — Desde Core (innecesario, usar CLI)
## 1. Create billing code / Crear billing code

10
core/Constants.py
View file

@ -51,7 +51,7 @@ class Constants:
HV_SESSION_STATE_HOME: Final[str] = f'{HV_STATE_HOME}/sessions'
HV_TOR_STATE_HOME: Final[str] = f'{HV_STATE_HOME}/tor'
# ── sing-box ──────────────────────────────────────────────────────────────
# ── sing-box ──────────────────────────────────────────────────────────────
SINGBOX_WRAPPER: Final[str] = os.environ.get(
'SINGBOX_WRAPPER', '/usr/local/bin/hydraveil-singbox'
)
@ -61,6 +61,9 @@ class Constants:
SINGBOX_CONFIG_DIR: Final[str] = f'{HV_DATA_HOME}/configs'
SINGBOX_PID_FILE: Final[str] = f'{HV_RUNTIME_DATA_HOME}/singbox.pid'
SINGBOX_LOG_FILE: Final[str] = f'{HV_RUNTIME_DATA_HOME}/singbox.log'
SINGBOX_TUN_IF: Final[str] = os.environ.get('SINGBOX_TUN_IF', 'tun0')
SINGBOX_INTERNAL_SUBNET: Final[str] = os.environ.get('SINGBOX_INTERNAL_SUBNET', '172.19.0.0/30')
SINGBOX_INTERNAL_ADDR: Final[str] = os.environ.get('SINGBOX_INTERNAL_ADDR', '172.19.0.1/30')
# ── killswitch / dns wrappers ─────────────────────────────────────────────
KILLSWITCH_WRAPPER: Final[str] = os.environ.get(
@ -68,4 +71,7 @@ class Constants:
)
RESOLVECTL_WRAPPER: Final[str] = os.environ.get(
'RESOLVECTL_WRAPPER', '/usr/local/bin/hydraveil-resolvectl'
)
)
VLESS_DNS_ENABLED: Final[bool] = os.environ.get('VLESS_DNS_ENABLED', 'true').lower() == 'true'
HYSTERIA2_DNS_ENABLED: Final[bool] = os.environ.get('HYSTERIA2_DNS_ENABLED', 'true').lower() == 'true'

58
core/controllers/ConnectionController.py
View file

@ -19,6 +19,7 @@ import os
import random
import re
import shutil
import signal
import subprocess
import sys
import tempfile
@ -143,6 +144,7 @@ class ConnectionController:
raise ConnectionError('The connection could not be established.')
return None
@staticmethod
def establish_session_connection(profile: SessionProfile, ignore: tuple[type[Exception]] = (), connection_observer: Optional[ConnectionObserver] = None):
@ -208,8 +210,32 @@ class ConnectionController:
SessionStateController.update_or_create(session_state)
return proxy_port_number or port_number
@staticmethod
def _signal_previous_process() -> None:
system_state = SystemStateController.get()
if system_state is None:
return
pid = system_state.pid
if pid is None or pid == os.getpid():
return
try:
os.kill(pid, signal.SIGUSR1)
time.sleep(0.5)
except (ProcessLookupError, PermissionError):
pass
@staticmethod
def establish_system_connection(profile: SystemProfile, ignore: tuple[type[Exception]] = (), connection_observer: Optional[ConnectionObserver] = None):
ConnectionController._signal_previous_process()
system_state = SystemStateController.get()
if system_state is not None:
try:
ConnectionController.terminate_system_connection(connection_observer=connection_observer)
except ConnectionTerminationError:
pass
if profile.connection.needs_operator_proxy():
operator_proxy_session = profile.get_operator_proxy_session()
@ -217,17 +243,29 @@ class ConnectionController:
if protocol == 'vless':
from core.controllers.encrypted_proxy.VlessController import VlessController
from core.services.encrypted_proxy.vless_service import parse_vless_link
import socket
server_ip = operator_proxy_session.server_ip
if server_ip is None:
vless = parse_vless_link(operator_proxy_session.links[0])
server_ip = socket.gethostbyname(vless['host'])
ok = VlessController(1080).enable(
operator_proxy_session.links[0],
operator_proxy_session.username,
server_ip,
connection_observer
)
elif protocol == 'hysteria2':
from core.controllers.encrypted_proxy.HysteriaController import HysteriaController
import socket
server_ip = operator_proxy_session.server_ip
if server_ip is None:
server_ip = socket.gethostbyname(operator_proxy_session.operator_hysteria2_host)
ok = HysteriaController(1080).enable(
operator_proxy_session.username,
operator_proxy_session.password,
operator_proxy_session.operator_hysteria2_host,
server_ip,
connection_observer
)
else:
@ -236,8 +274,11 @@ class ConnectionController:
if not ok:
raise ConnectionError('The connection could not be established.')
SystemStateController.create(profile.id)
token = SystemStateController.create(profile.id)
if connection_observer is not None:
connection_observer.notify('connected_token', {'session_token': token})
return
if ConfigurationController.get_endpoint_verification_enabled():
ProfileController.verify_wireguard_endpoint(profile, ignore=ignore)
@ -267,6 +308,7 @@ class ConnectionController:
ConnectionController.terminate_tor_connection()
time.sleep(1.0)
@staticmethod
def establish_tor_connection(connection_observer: Optional[ConnectionObserver] = None):
@ -484,7 +526,9 @@ class ConnectionController:
except CalledProcessError:
raise ConnectionError('The connection could not be established.')
SystemStateController.create(profile.id)
token = SystemStateController.create(profile.id)
if connection_observer is not None:
connection_observer.notify('connected_token', {'session_token': token})
try:
ConnectionController.await_connection(connection_observer=connection_observer)
@ -557,22 +601,22 @@ class ConnectionController:
return True
return False
@staticmethod
def verify_singbox_installation():
import subprocess
import os
from core.Errors import SingboxNotInstalledException
singbox_binary = '/usr/bin/sing-box'
singbox_wrapper = '/usr/local/bin/hydraveil-singbox'
if not os.path.exists(singbox_binary):
raise SingboxNotInstalledException(f'sing-box binary not found at {singbox_binary}')
if not os.path.exists(singbox_wrapper):
raise SingboxNotInstalledException(f'sing-box wrapper not found at {singbox_wrapper}')
try:
subprocess.run([singbox_binary, 'version'], capture_output=True, timeout=5, check=True)
except subprocess.CalledProcessError:

9
core/controllers/SystemStateController.py
View file

@ -1,3 +1,5 @@
import os
import uuid
from core.models.system.SystemState import SystemState
@ -13,7 +15,10 @@ class SystemStateController:
@staticmethod
def create(profile_id):
return SystemState(profile_id).save()
token = str(uuid.uuid4())
pid = os.getpid()
SystemState(profile_id, token, pid).save()
return token
@staticmethod
def update_or_create(system_state):
@ -21,4 +26,4 @@ class SystemStateController:
@staticmethod
def dissolve():
return SystemState.dissolve()
return SystemState.dissolve()

7
core/controllers/encrypted_proxy/HysteriaController.py
View file

@ -5,7 +5,7 @@ class HysteriaController:
self.socks5_port = socks5_port
def enable(self, username: str, password: str,
server_host: str, observer=None) -> bool:
server_host: str, server_ip: str, observer=None) -> bool:
if not username or not isinstance(username, str):
if observer:
observer.notify("error", "Invalid username")
@ -14,10 +14,15 @@ class HysteriaController:
if observer:
observer.notify("error", "Missing password or server_host")
return False
if not server_ip:
if observer:
observer.notify("error", "Missing server_ip")
return False
return enable_hysteria(
username=username,
password=password,
server_host=server_host,
server_ip=server_ip,
socks5_port=self.socks5_port,
observer=observer,
)

7
core/controllers/encrypted_proxy/VlessController.py
View file

@ -4,7 +4,7 @@ class VlessController:
def __init__(self, socks5_port: int):
self.socks5_port = socks5_port
def enable(self, vless_link: str, username: str, observer=None) -> bool:
def enable(self, vless_link: str, username: str, server_ip: str, observer=None) -> bool:
if not username or not isinstance(username, str):
if observer:
observer.notify("error", "Invalid username")
@ -13,9 +13,14 @@ class VlessController:
if observer:
observer.notify("error", "Invalid vless link")
return False
if not server_ip:
if observer:
observer.notify("error", "Missing server_ip")
return False
return enable_vless(
vless_link=vless_link,
username=username,
server_ip=server_ip,
socks5_port=self.socks5_port,
observer=observer,
)

3
core/models/OperatorProxySession.py
View file

@ -16,4 +16,5 @@ class OperatorProxySession:
operator_name: str
operator_domain: Optional[str] = None
operator_hysteria2_host: Optional[str] = None
operator_vless_host: Optional[str] = None
operator_vless_host: Optional[str] = None
server_ip: Optional[str] = None # pre-resolved IP — avoids DNS leak at connect time

9
core/models/system/SystemState.py
View file

@ -1,8 +1,8 @@
from core.Constants import Constants
from dataclasses import dataclass
from dataclasses import dataclass, field
from dataclasses_json import dataclass_json
from pathlib import Path
from typing import Self
from typing import Optional, Self
import json
import os
import pathlib
@ -11,6 +11,8 @@ import pathlib
@dataclass
class SystemState:
profile_id: int
session_token: Optional[str] = field(default=None)
pid: Optional[int] = field(default=None)
def save(self: Self):
@ -31,7 +33,6 @@ class SystemState:
system_state_file_contents = open(f'{SystemState.__get_state_path()}/system.json', 'r').read()
system_state_dict = json.loads(system_state_file_contents)
# noinspection PyUnresolvedReferences
return SystemState.from_dict(system_state_dict)
except (FileNotFoundError, ValueError, KeyError):
@ -53,4 +54,4 @@ class SystemState:
@staticmethod
def __get_state_path():
return Constants.HV_STATE_HOME
return Constants.HV_STATE_HOME

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

@ -1,23 +1,14 @@
import socket
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
def _resolve(host: str) -> str:
return socket.gethostbyname(host)
def build_hysteria_config(username: str, password: str,
server_host: str, socks5_port: int,
server_ip: str = None) -> dict:
if server_ip is None:
server_ip = _resolve(server_host)
server_ip: str) -> dict:
return {
"dns": {
"servers": [{"tag": "tunnel-dns", "type": "udp", "server": "9.9.9.9"}],
@ -29,8 +20,8 @@ def build_hysteria_config(username: str, password: str,
{
"type": "tun",
"tag": "tun-in",
"interface_name": "tun0",
"address": ["172.19.0.1/30"],
"interface_name": Constants.SINGBOX_TUN_IF,
"address": [Constants.SINGBOX_INTERNAL_ADDR],
"mtu": 9000,
"auto_route": True,
"stack": "gvisor",
@ -61,7 +52,7 @@ def build_hysteria_config(username: str, password: str,
"route": {
"rules": [
{"protocol": "dns", "action": "hijack-dns"},
{"ip_cidr": ["172.19.0.0/30"], "action": "hijack-dns"},
{"ip_cidr": [Constants.SINGBOX_INTERNAL_SUBNET], "action": "hijack-dns"},
{"ip_cidr": [f"{server_ip}/32"], "outbound": "direct"},
{"ip_is_private": True, "outbound": "direct"},
{"ip_version": 6, "outbound": "block"},
@ -91,64 +82,58 @@ def _cleanup_all() -> None:
def enable_hysteria(username: str, password: str, server_host: str,
socks5_port: int, observer=None) -> bool:
server_ip: 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()
runner = SingboxRunner()
config_path = Path(Constants.SINGBOX_CONFIG_DIR) / f"{username}-sing-box.json"
config = build_hysteria_config(username, password, server_host, socks5_port,
server_ip=server_ip)
config = build_hysteria_config(username, password, server_host,
socks5_port, server_ip)
_connected = False
try:
# Fase 1 — arrancar sing-box
runner.write_config(config_path, config)
ok = runner.start(config_path)
if not runner.start(config_path):
if observer:
observer.notify("error", "sing-box not active after start")
return False
# Fase 2 — esperar a que tun aparezca
if not wait_for_tun(timeout=15.0):
if observer:
observer.notify("error", f"{Constants.SINGBOX_TUN_IF} did not appear after 15s")
return False
# Fase 3 — armar kill switch ahora que la interfaz existe
if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF):
if observer:
observer.notify("error", "Failed to arm kill switch")
return False
# Fase 4 — DNS
_C = Constants()
dns_ok = set_dns_on_tun() if _C.HYSTERIA2_DNS_ENABLED else False
_connected = True
if observer:
observer.notify("connected", {
"tunnel_if": Constants.SINGBOX_TUN_IF,
"socks5_port": socks5_port,
"dns_enabled": _C.HYSTERIA2_DNS_ENABLED,
"dns_active": dns_ok,
})
return True
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
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,
"proxy_ip": proxy_ip,
"socks5_port": socks5_port,
})
return True
finally:
if not _connected:
_cleanup_all()
def disable_hysteria(observer=None) -> bool:
@ -156,5 +141,5 @@ def disable_hysteria(observer=None) -> bool:
SingboxRunner().stop()
killswitch.disarm()
if observer:
observer.notify("disconnected", {})
observer.notify("disconnected", {"tunnel_if": Constants.SINGBOX_TUN_IF})
return True

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

@ -3,7 +3,6 @@ from pathlib import Path
import socket
import time
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
@ -34,10 +33,7 @@ def parse_vless_link(link: str) -> dict:
}
def build_vless_config(vless: dict, socks5_port: int, server_ip: str = None) -> dict:
if server_ip is None:
server_ip = socket.gethostbyname(vless["host"])
def build_vless_config(vless: dict, socks5_port: int, server_ip: str) -> dict:
return {
"dns": {
"servers": [{"tag": "tunnel-dns", "type": "udp", "server": "9.9.9.9"}],
@ -49,8 +45,8 @@ def build_vless_config(vless: dict, socks5_port: int, server_ip: str = None) ->
{
"type": "tun",
"tag": "tun-in",
"interface_name": "tun0",
"address": ["172.19.0.1/30"],
"interface_name": Constants.SINGBOX_TUN_IF,
"address": [Constants.SINGBOX_INTERNAL_ADDR],
"mtu": 9000,
"auto_route": True,
"stack": "gvisor",
@ -86,7 +82,7 @@ def build_vless_config(vless: dict, socks5_port: int, server_ip: str = None) ->
"route": {
"rules": [
{"protocol": "dns", "action": "hijack-dns"},
{"ip_cidr": ["172.19.0.0/30"], "action": "hijack-dns"},
{"ip_cidr": [Constants.SINGBOX_INTERNAL_SUBNET], "action": "hijack-dns"},
{"ip_cidr": [f"{server_ip}/32"], "outbound": "direct"},
{"ip_is_private": True, "outbound": "direct"},
{"ip_version": 6, "outbound": "block"},
@ -115,65 +111,55 @@ def _cleanup_all() -> None:
time.sleep(1)
def enable_vless(vless_link: str, username: str,
def enable_vless(vless_link: str, username: str, server_ip: 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()
vless = parse_vless_link(vless_link)
runner = SingboxRunner()
config_path = Path(Constants.SINGBOX_CONFIG_DIR) / f"{username}-sing-box.json"
config = build_vless_config(vless, socks5_port, server_ip=server_ip)
config = build_vless_config(vless, socks5_port, server_ip)
_connected = False
try:
runner.write_config(config_path, config)
ok = runner.start(config_path)
if not runner.start(config_path):
if observer:
observer.notify("error", "sing-box not active after start")
return False
if not wait_for_tun(timeout=15.0):
if observer:
observer.notify("error", f"{Constants.SINGBOX_TUN_IF} did not appear after 15s")
return False
if not killswitch.arm(server_ip, Constants.SINGBOX_TUN_IF):
if observer:
observer.notify("error", "Failed to arm kill switch")
return False
_C = Constants()
dns_ok = set_dns_on_tun() if _C.VLESS_DNS_ENABLED else False
_connected = True
if observer:
observer.notify("connected", {
"tunnel_if": Constants.SINGBOX_TUN_IF,
"socks5_port": socks5_port,
"dns_enabled": _C.VLESS_DNS_ENABLED,
"dns_active": dns_ok,
})
return True
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
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,
"proxy_ip": proxy_ip,
"socks5_port": socks5_port,
})
return True
finally:
if not _connected:
_cleanup_all()
def disable_vless(observer=None) -> bool:
@ -181,5 +167,5 @@ def disable_vless(observer=None) -> bool:
SingboxRunner().stop()
killswitch.disarm()
if observer:
observer.notify("disconnected", {})
observer.notify("disconnected", {"tunnel_if": Constants.SINGBOX_TUN_IF})
return True

37
core/utils/encrypted_proxy/dns.py
View file

@ -1,10 +1,9 @@
import os
import subprocess
import time
from core.Constants import Constants
RESOLVECTL_WRAPPER = "/usr/local/bin/hydraveil-resolvectl"
TUN_IFACE = "tun0"
TUN_DNS = "9.9.9.9"
_C = Constants()
_SUDO_KW = dict(
stdout=subprocess.DEVNULL,
@ -15,11 +14,24 @@ _SUDO_KW = dict(
)
def is_resolved_active() -> bool:
try:
result = subprocess.run(
["systemctl", "is-active", "systemd-resolved"],
capture_output=True,
text=True,
timeout=5,
)
return result.stdout.strip() == "active"
except (subprocess.TimeoutExpired, FileNotFoundError):
return False
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],
["ip", "link", "show", _C.SINGBOX_TUN_IF],
capture_output=True,
)
if result.returncode == 0:
@ -30,22 +42,31 @@ def wait_for_tun(timeout: float = 15.0, interval: float = 0.5) -> bool:
def set_dns_on_tun() -> bool:
if not is_resolved_active():
print("[DNS] systemd-resolved is not active — DNS on tun skipped")
return False
result = subprocess.run(
["sudo", RESOLVECTL_WRAPPER, "set", TUN_DNS],
["sudo", _C.RESOLVECTL_WRAPPER, "set", "9.9.9.9"],
**_SUDO_KW,
)
if result.returncode != 0:
print(f"[DNS] Failed to configure tun0: {result.stderr.strip()}")
print(f"[DNS] Failed to configure {_C.SINGBOX_TUN_IF}: {result.stderr.strip()}")
return False
return True
def revert_dns_on_tun() -> bool:
if not is_resolved_active():
return True
result = subprocess.run(
["sudo", RESOLVECTL_WRAPPER, "revert"],
["sudo", _C.RESOLVECTL_WRAPPER, "revert"],
**_SUDO_KW,
)
if result.returncode != 0:
print(f"[DNS] Failed to revert tun0: {result.stderr.strip()}")
print(f"[DNS] Failed to revert {_C.SINGBOX_TUN_IF}: {result.stderr.strip()}")
return False
return True

17
core/utils/encrypted_proxy/killswitch.py
View file

@ -1,13 +1,14 @@
import subprocess
from core.Constants import Constants
KILLSWITCH_WRAPPER = "/usr/local/bin/hydraveil-killswitch"
_C = Constants()
def arm(server_ip: str) -> bool:
def arm(server_ip: str, tunnel_if: str) -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "arm", server_ip],
["sudo", _C.KILLSWITCH_WRAPPER, "arm", server_ip, tunnel_if],
capture_output=True,
text=True
text=True,
)
if result.returncode != 0:
print(f"[killswitch] Failed to arm: {result.stderr.strip()}")
@ -17,9 +18,9 @@ def arm(server_ip: str) -> bool:
def disarm() -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "disarm"],
["sudo", _C.KILLSWITCH_WRAPPER, "disarm"],
capture_output=True,
text=True
text=True,
)
if result.returncode != 0:
print(f"[killswitch] Failed to disarm: {result.stderr.strip()}")
@ -29,8 +30,8 @@ def disarm() -> bool:
def status() -> bool:
result = subprocess.run(
["sudo", KILLSWITCH_WRAPPER, "status"],
["sudo", _C.KILLSWITCH_WRAPPER, "status"],
capture_output=True,
text=True
text=True,
)
return result.returncode == 0 and result.stdout.strip() == "armed"