0733efd186
* Add umedbazarov/ruh-vpn: VPN/proxy manager for sing-box New community plugin: bar widget, panel, service and control-center shortcut for managing SSH, VLESS, VMess, Shadowsocks and SOCKS5 connections through sing-box, with routing presets, custom rules, system-proxy/TUN modes and a kill switch. The bundled Python backend serves a loopback control API protected by a per-launch bearer token. * Address review: sanitize kill-switch ruleset, scope TUN capability, fix mux error path, disclose DNS - kill switch: only pre-resolved, canonicalized literal IPs enter the nft ruleset; domains are resolved first and anything unparseable is dropped, so subscription-supplied addresses can no longer inject nft syntax - TUN: CAP_NET_ADMIN is granted to a plugin-private copy of sing-box in a 0700 directory instead of the shared system binary; the copy is refreshed (clearing the cap) when the system binary changes, and the legacy grant on the shared binary is removed in the same polkit prompt - fix NameError in the mux startup failure path (undefined mux_name) that hid the log tail and skipped teardown - README: disclose plain-UDP DNS endpoints (8.8.8.8 via tunnel, 223.5.5.5 direct in rules mode) alongside the TUN DoH endpoint --------- Co-authored-by: Umedjon Bazarov <170195993+UmedjonBA@users.noreply.github.com>
75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
|
|
import aiofiles
|
|
|
|
from backend.models.server import RoutingRule, Server, parse_server, server_to_dict
|
|
from backend.paths import DATA_DIR, ensure_private_dir, protect_file
|
|
|
|
SERVERS_FILE = DATA_DIR / "servers.json"
|
|
RULES_FILE = DATA_DIR / "rules.json"
|
|
|
|
|
|
def ensure_dirs() -> None:
|
|
ensure_private_dir(DATA_DIR)
|
|
|
|
|
|
async def load_servers() -> list[Server]:
|
|
ensure_dirs()
|
|
if not SERVERS_FILE.exists():
|
|
return []
|
|
try:
|
|
async with aiofiles.open(SERVERS_FILE, "r") as f:
|
|
raw = await f.read()
|
|
data = json.loads(raw or "[]")
|
|
except (json.JSONDecodeError, ValueError):
|
|
return []
|
|
servers: list[Server] = []
|
|
for entry in data:
|
|
try:
|
|
servers.append(parse_server(entry))
|
|
except (ValueError, KeyError):
|
|
continue
|
|
return servers
|
|
|
|
|
|
async def save_servers(servers: list) -> None:
|
|
ensure_dirs()
|
|
data = [server_to_dict(s) for s in servers]
|
|
tmp = SERVERS_FILE.with_suffix(".json.tmp")
|
|
async with aiofiles.open(tmp, "w") as f:
|
|
await f.write(json.dumps(data, indent=2))
|
|
protect_file(tmp)
|
|
os.replace(tmp, SERVERS_FILE)
|
|
|
|
|
|
async def load_rules() -> list[RoutingRule]:
|
|
ensure_dirs()
|
|
if not RULES_FILE.exists():
|
|
return []
|
|
try:
|
|
async with aiofiles.open(RULES_FILE, "r") as f:
|
|
raw = await f.read()
|
|
data = json.loads(raw or "[]")
|
|
except (json.JSONDecodeError, ValueError):
|
|
return []
|
|
rules: list[RoutingRule] = []
|
|
for entry in data:
|
|
try:
|
|
rules.append(RoutingRule.model_validate(entry))
|
|
except ValueError:
|
|
continue
|
|
return rules
|
|
|
|
|
|
async def save_rules(rules: list[RoutingRule]) -> None:
|
|
ensure_dirs()
|
|
data = [r.model_dump(exclude_none=True) for r in rules]
|
|
tmp = RULES_FILE.with_suffix(".json.tmp")
|
|
async with aiofiles.open(tmp, "w") as f:
|
|
await f.write(json.dumps(data, indent=2))
|
|
protect_file(tmp)
|
|
os.replace(tmp, RULES_FILE)
|