Niko/nikovpn_openwrt
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions
nikovpn_openwrt/nikovpn.py
Niko Marmeladkov c887b77784
nikovpn.py: фиксы выбора протокола и сквозной проверки 
- choose_candidates: кандидату назначается его собственный протокол из
  scheme, а не общий выбранный — иначе TUIC/Naive-ссылки собирались
  как hysteria2 и не работали
- verify_tunnel: SSH-форвард строится через rt.ssh_args() — теперь
  передаются --identity/sshpass (раньше проверка падала с нестандартным
  именем ключа)
- BatchMode=yes только без пароля (с sshpass он ломал авторизацию)
- stop_service: добивает sing-box/ipt2socks вне procd — иначе Text
  file busy при перезаливке бинарников
2026-08-26 11:51:54 +03:00

1009 lines
40 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.9"
# dependencies = []
# ///
# =============================================================================
# NikoVPN — контроллер для OpenWRT (замена install.sh, запуск с ПК)
#
# uv run nikovpn.py [--host 192.168.1.1] install
# uv run nikovpn.py update | status | uninstall
#
# Управляет роутером по SSH: определяет архитектуру, заливает бинарники
# (локальные bin/ или f.n1ko.dev), спрашивает API vpnd.n1ko.dev, выбирает
# протокол/сервер (hysteria2 > tuic > naive > vless > shadowsocks),
# генерирует конфиг sing-box и поднимает прозрачный прокси.
#
# Креды VPN (login:password):
# --creds U:P | --creds-file F | env NIKOVPN_CREDS | /root/creds на роутере
#
# На роутере требуется только busybox + wget/curl + nft/iptables.
# =============================================================================
import argparse
import base64
import json
import os
import re
import shlex
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlsplit
BASE_URL = "https://f.n1ko.dev/nikovpn_openwrt"
API_URL = "https://vpnd.n1ko.dev/api/v1"
WORK_DIR = "/etc/nikovpn"
CREDS_FILE = "/root/creds"
SB_PATH = "/usr/bin/sing-box"
IPT_PATH = "/usr/bin/ipt2socks"
SOCKS_PORT = "1080"
TP_PORT = "12345"
RULE_TABLE = "100"
FW_MARK = "0x1"
PRIVATE_CIDRS = ["0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", "127.0.0.0/8",
"169.254.0.0/16", "172.16.0.0/12", "192.168.0.0/16",
"224.0.0.0/4", "240.0.0.0/4"]
log = lambda m: print(f"[*] {m}", flush=True)
warn = lambda m: print(f"[!] {m}", flush=True)
def die(m):
print(f"[ERROR] {m}", file=sys.stderr)
sys.exit(1)
# ============================================================================
# SSH-транспорт
# ============================================================================
class Router:
def __init__(self, host, user="root", port=22, password=None, key=None):
self.host, self.user, self.port = host, user, port
opts = ["-o", "StrictHostKeyChecking=accept-new",
"-o", "ConnectTimeout=10"]
pre = []
if password is not None:
if not shutil_which("sshpass"):
die("для парольного входа нужен sshpass")
pre = ["sshpass", "-p", password]
opts += ["-o", "PreferredAuthentications=password",
"-o", "PubkeyAuthentication=no"]
else:
opts += ["-o", "BatchMode=yes"]
if key:
opts += ["-i", str(key)]
self.pre, self.opts = pre, opts
self.base = pre + ["ssh", "-p", str(port)] + opts + [f"{user}@{host}"]
def ssh_args(self, *extra):
return self.pre + ["ssh", "-p", str(self.port)] + self.opts + list(extra)
def run(self, cmd, timeout=180, input_bytes=None, check=False):
r = subprocess.run(self.base + [cmd], capture_output=True,
timeout=timeout,
input=input_bytes if input_bytes is not None else b"")
if check and r.returncode != 0:
raise RuntimeError(
f"{cmd}: rc={r.returncode} {r.stderr.decode(errors='replace')[:500]}")
return r.returncode, r.stdout, r.stderr
def have(self, tool):
rc, out, _ = self.run(f"command -v {shlex.quote(tool)} >/dev/null 2>&1 && echo y")
return rc == 0 and b"y" in out
def exists(self, path):
rc, out, _ = self.run(f"test -e {shlex.quote(path)} && echo y")
return rc == 0 and b"y" in out
def read_file(self, path):
rc, out, _ = self.run(f"cat {shlex.quote(path)} 2>/dev/null")
return out if rc == 0 else None
def write_file(self, path, data: bytes, mode=None):
tmp = f"/tmp/nikovpn.{os.getpid()}.{os.path.basename(path)}"
cmd = f"cat > {shlex.quote(tmp)}"
if mode:
cmd += f" && chmod {mode} {shlex.quote(tmp)}"
cmd += f" && mv {shlex.quote(tmp)} {shlex.quote(path)}"
self.run(cmd, input_bytes=data, check=True)
def mkdirs(self, *dirs):
self.run("mkdir -p " + " ".join(shlex.quote(d) for d in dirs), check=True)
def shutil_which(x):
from shutil import which
return which(x)
# ============================================================================
# Архитектуры
# ============================================================================
def map_uname(u):
m = {
"x86_64|amd64": ("amd64", "x86_64", 1), "aarch64|arm64": ("arm64", "aarch64", 1),
"armv7l|armv7|arm": ("armv7", "armv7hf", 1), "armv6l|armv6": ("armv6", "armv6", 0),
"armv5tel|armv5te|armv5l": ("armv5", "armv5", 0),
"mips": ("mips", "mips", 0), "mipsel": ("mipsle", "mipsel", 1),
"mips64": ("mips64", "mips64", 0), "mips64el": ("mips64le", "mips64el", 0),
"i686|i586|i386|x86": ("386", "i686", 1), "riscv64": ("riscv64", "", 1),
"loongarch64|loong64": ("loong64", "", 1), "ppc64le": ("ppc64le", "", 0),
"s390x": ("s390x", "", 0),
}
for pat, v in m.items():
if u in pat.split("|"):
return v
return ("", "", 0)
def map_openwrt(da, uname_m):
if da.startswith("x86_64"):
return map_uname("x86_64")
if da.startswith("i386_"):
return map_uname("i686")
if da.startswith("aarch64_"):
return map_uname("aarch64")
if re.match(r"arm_cortex-a(5|7|8|9|15)", da) or da.startswith("arm_"):
return map_uname("armv7l")
if da == "arm_arm1176jzf-s_vfp":
return map_uname("armv6l")
if da in ("arm_arm926ej-s", "arm_fa526", "arm_xscale"):
return map_uname("armv5l")
for pfx, u in [("mips64el_", "mips64el"), ("mips64_", "mips64"),
("mipsel_", "mipsel"), ("mips_", "mips"),
("riscv64_", "riscv64"), ("loongarch64_", "loongarch64"),
("powerpc_", "ppc64le")]:
if da.startswith(pfx):
return map_uname(u)
return map_uname(uname_m)
def detect_arch(rt: Router):
da = ""
rel = rt.read_file("/etc/openwrt_release") or b""
mm = re.search(rb"DISTRIB_ARCH='([^']+)'", rel)
if mm:
da = mm.group(1).decode()
if not da and rt.have("apk"):
rc, out, _ = rt.run("apk --print-arch 2>/dev/null")
da = out.decode().strip() if rc == 0 else ""
if not da and rt.have("opkg"):
rc, out, _ = rt.run("opkg print-architecture 2>/dev/null")
if rc == 0:
for line in out.decode().splitlines():
p = line.split()
if len(p) >= 2 and p[0] == "arch" and p[1] not in ("all", "noarch"):
da = p[1]
break
rc, out, _ = rt.run("uname -m")
uname_m = out.decode().strip()
sb, ipt, naive = map_openwrt(da, uname_m) if da else map_uname(uname_m)
if not sb:
die(f"архитектура не поддерживается ({da or uname_m})")
return sb, ipt, naive, da or uname_m
# ============================================================================
# API vpnd.n1ko.dev (запросы идут с ПК)
# ============================================================================
def api_get(creds: str, path: str, timeout=30):
req = urllib.request.Request(
API_URL + path,
headers={"Authorization": "Basic " +
base64.b64encode(creds.encode()).decode()})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, r.read()
except urllib.error.HTTPError as e:
return e.code, e.read()
except Exception as e:
warn(f"API недоступен: {e}")
return None, b""
def get_creds(rt: Router, args) -> str:
cred = None
if getattr(args, "creds", None):
cred = args.creds.strip()
elif getattr(args, "creds_file", None):
cred = Path(args.creds_file).read_text().strip().splitlines()[0].strip()
elif os.environ.get("NIKOVPN_CREDS"):
cred = os.environ["NIKOVPN_CREDS"].strip()
else:
data = rt.read_file(CREDS_FILE)
if data:
for line in data.decode(errors="replace").splitlines():
line = line.strip()
if line and not line.startswith("#"):
cred = line
break
if not cred or ":" not in cred:
die("нет кредов: --creds login:password | --creds-file F | "
"env NIKOVPN_CREDS | /root/creds на роутере")
return "".join(str(cred).split())
# ============================================================================
# Разбор ссылок и выбор сервера
# ============================================================================
def parse_link(uri: str) -> dict:
sp = urlsplit(uri)
q = parse_qs(sp.query, keep_blank_values=True)
user = unquote(sp.username or "") if sp.username else ""
pw = unquote(sp.password or "") if sp.password else ""
return dict(uri=uri, scheme=sp.scheme.lower(),
host=unquote(sp.hostname or ""), port=sp.port or "",
user=user, password=pw, query=q, frag=unquote(sp.fragment))
def qp(link, name, default=""):
v = link["query"].get(name)
return v[0] if v else default
def host_alive(link) -> bool:
"""проверка с ПК (у роутера busybox-nc всегда возвращает ошибку).
UDP-протоколы (hysteria2/tuic) так проверить нельзя — считаем живыми."""
import socket
if link["scheme"] in ("hysteria2", "tuic"):
return True
host, port = link["host"], int(link["port"] or 443)
try:
s = socket.create_connection((host, port), timeout=4)
s.close()
return True
except OSError:
return False
def choose_candidates(rt: Router, links_text: bytes, naive_ok: bool) -> list:
prefs = ["hysteria2", "tuic", "naive", "vless", "shadowsocks"]
prefer_proto, prefer_srv = read_prefs(rt)
links = [parse_link(l) for l in links_text.decode(errors="replace").splitlines()
if "://" in l]
proto = None
for p in prefs:
if p == "naive" and not naive_ok:
continue
if prefer_proto and prefer_proto != p:
continue
if any(l["scheme"] == p or (p == "naive" and l["scheme"].startswith("naive"))
for l in links):
proto = p
break
if not proto:
die("ни один протокол недоступен (или предпочтение не найдено)")
cand = [l for l in links
if l["scheme"] == proto or (proto == "naive" and l["scheme"].startswith("naive"))]
if prefer_srv:
filt = [l for l in cand if prefer_srv.lower() in l["uri"].lower()]
if filt:
cand = filt
# плоский список: внутри протокола — живые первыми, протоколы по приоритету
picked = []
for p in prefs:
group = [l for l in links
if l["scheme"] == p or (p == "naive" and l["scheme"].startswith("naive"))]
alive = [l for l in group if host_alive(l)]
rest = [l for l in group if not host_alive(l)]
picked += alive + rest
for l in picked:
l["protocol"] = l["scheme"].split("+")[0]
l["sni"] = qp(l, "sni") or l["host"]
return picked
def read_prefs(rt: Router):
proto, srv = "", ""
data = rt.read_file(f"{WORK_DIR}/config")
if data:
for line in data.decode(errors="replace").splitlines():
mm = re.match(r'\s*PREFER_PROTOCOL="([^"]+)"', line)
if mm:
proto = mm.group(1)
mm = re.match(r'\s*PREFER_SERVER="([^"]+)"', line)
if mm:
srv = mm.group(1)
return proto, srv
# ============================================================================
# Генерация config.json sing-box
# ============================================================================
def tls_block(link, sni=None, insecure=False, utls_fp=None, reality=False):
t = {"enabled": True, "server_name": sni or link["sni"]}
if insecure:
t["insecure"] = True
if utls_fp:
t["utls"] = {"enabled": True, "fingerprint": utls_fp}
if reality:
t["reality"] = {"enabled": True,
"public_key": qp(link, "pbk"),
"short_id": qp(link, "sid")}
return t
def build_outbound(link) -> dict:
p = link["protocol"]
ob = {"tag": "main", "server": link["host"], "server_port": int(link["port"])}
if p == "hysteria2":
ob.update(type="hysteria2", password=link["user"],
tls=tls_block(link, insecure=qp(link, "insecure") == "1"))
if qp(link, "obfs"):
ob["obfs"] = {"type": qp(link, "obfs"),
"password": qp(link, "obfs-password")}
elif p == "tuic":
ob.update(type="tuic", uuid=link["user"], password=link["password"],
congestion_control=qp(link, "congestion_control", "bbr"),
udp_relay_mode=qp(link, "udp_relay_mode", "native"),
tls=tls_block(link))
elif p.startswith("naive"):
ob.update(type="naive", username=link["user"], password=link["password"],
quic=(link["scheme"] == "naive+quic"),
tls=tls_block(link, sni=link["host"]))
elif p == "vless":
ob.update(type="vless", uuid=link["user"])
if qp(link, "flow"):
ob["flow"] = qp(link, "flow")
sec = qp(link, "security")
fp = qp(link, "fp", "chrome")
ob["tls"] = tls_block(link, utls_fp=fp, reality=(sec == "reality"))
elif p == "shadowsocks":
try:
pad = "=" * (-len(link["user"]) % 4)
dec = base64.b64decode(link["user"] + pad).decode()
except Exception:
dec = ""
if ":" not in dec:
die("не удалось разобрать shadowsocks-ссылку")
method, sspass = dec.split(":", 1)
ob.update(type="shadowsocks", method=method, password=sspass)
plugin = unquote(qp(link, "plugin"))
if plugin:
popts = ""
if ";" in plugin:
plugin, popts = plugin.split(";", 1)
ob["plugin"], ob["plugin_opts"] = plugin, popts
else:
die(f"неизвестный протокол {p}")
return ob
def gen_config(rt: Router, link, tunnel: str):
main_ob = build_outbound(link)
if tunnel == "singbox":
inbound = {"type": "tproxy", "tag": "tproxy-in",
"listen": "0.0.0.0", "listen_port": int(TP_PORT)}
else:
inbound = {"type": "mixed", "tag": "mixed-in",
"listen": "127.0.0.1", "listen_port": int(SOCKS_PORT)}
cfg = {
"log": {"level": "info", "timestamp": True},
"inbounds": [inbound],
"outbounds": [main_ob, {"type": "direct", "tag": "direct"}],
"route": {"final": "main",
"rules": [{"ip_cidr": PRIVATE_CIDRS, "outbound": "direct"}]},
}
data = json.dumps(cfg, indent=2).encode()
rt.write_file(f"{WORK_DIR}/config.json", data)
rc, _, err = rt.run(f"{SB_PATH} check -c {WORK_DIR}/config.json")
if rc != 0:
die("конфиг sing-box не проходит проверку: " +
err.decode(errors="replace")[:300])
# ============================================================================
# fw.sh / init.d (выполняются на роутере — остаются shell)
# ============================================================================
FW_SH = r"""#!/bin/sh
# NikoVPN firewall (генерируется nikovpn.py)
WORK_DIR=/etc/nikovpn
STATE="$WORK_DIR/state"
[ -f "$STATE" ] && . "$STATE"
MODE="${MODE:-tcp}"
FW="${FW:-nft}"
TUNNEL="${TUNNEL:-ipt2socks}"
TP_PORT="${TP_PORT:-12345}"
VPN_PORT="${PORT:-0}"
MARK="0x1"
TABLE="100"
NFT_FILE="/etc/nftables.d/10-nikovpn.nft"
PRIVATE="0.0.0.0/8 10.0.0.0/8 100.64.0.0/10 127.0.0.0/8 169.254.0.0/16 172.16.0.0/12 192.168.0.0/16 224.0.0.0/4 240.0.0.0/4"
NFT_PRIVATE="$(echo $PRIVATE | tr ' ' ',')"
have() { command -v "$1" >/dev/null 2>&1; }
ip_rules_add() {
ip rule add fwmark $MARK table $TABLE 2>/dev/null
ip route add local default dev lo table $TABLE 2>/dev/null
true
}
ip_rules_del() {
ip rule del fwmark $MARK table $TABLE 2>/dev/null
ip route del local default dev lo table $TABLE 2>/dev/null
true
}
nft_write() {
cat > "$NFT_FILE" <<EOF2
table inet nikovpn {
set bypass4 { type ipv4_addr; flags interval;
elements = { $NFT_PRIVATE } }
set vpn_ports { type inet_service; elements = { $VPN_PORT } }
EOF2
if [ "$TUNNEL" = "singbox" ]; then
cat >> "$NFT_FILE" <<EOF2
chain prerouting_tcp {
type filter hook prerouting priority -1; policy accept;
ip daddr @bypass4 return
tcp dport 53 return
meta l4proto tcp tproxy to :$TP_PORT meta mark set 1
}
chain prerouting_udp {
type filter hook prerouting priority -1; policy accept;
ip daddr @bypass4 return
udp dport 53 return
meta l4proto udp tproxy to :$TP_PORT meta mark set 1
}
}
EOF2
else
cat >> "$NFT_FILE" <<EOF2
chain prerouting_tcp {
type nat hook prerouting priority 0; policy accept;
ip daddr @bypass4 return
tcp dport 53 return
meta l4proto tcp redirect to :$TP_PORT
}
chain output_tcp {
type nat hook output priority 0; policy accept;
ip daddr @bypass4 return
fib daddr type local return
tcp dport 53 return
tcp dport @vpn_ports return
meta l4proto tcp redirect to :$TP_PORT
}
EOF2
if [ "$MODE" = "full" ]; then
cat >> "$NFT_FILE" <<EOF2
chain prerouting_udp {
type filter hook prerouting priority -1; policy accept;
ip daddr @bypass4 return
udp dport 53 return
meta l4proto udp tproxy to :$TP_PORT meta mark set 1
}
}
EOF2
else
echo '}' >> "$NFT_FILE"
fi
fi
}
fw_start() {
if [ "$FW" = "nft" ]; then
have nft || return 1
nft_write
nft delete table inet nikovpn 2>/dev/null
nft -f "$NFT_FILE" || return 1
if [ "$MODE" = full ] || [ "$TUNNEL" = singbox ]; then ip_rules_add; fi
else
have iptables || return 1
if [ "$TUNNEL" = singbox ]; then
iptables -t mangle -N NIKOVPN_TP 2>/dev/null
iptables -t mangle -F NIKOVPN_TP
for n in $PRIVATE; do iptables -t mangle -A NIKOVPN_TP -d $n -j RETURN; done
iptables -t mangle -A NIKOVPN_TP -m addrtype --dst-type LOCAL -j RETURN
iptables -t mangle -A NIKOVPN_TP -p tcp --dport 53 -j RETURN
iptables -t mangle -A NIKOVPN_TP -p tcp -j TPROXY --on-port $TP_PORT --tproxy-mark $MARK/$MARK
iptables -t mangle -C PREROUTING -p tcp -j NIKOVPN_TP 2>/dev/null || iptables -t mangle -A PREROUTING -p tcp -j NIKOVPN_TP
else
iptables -t nat -N NIKOVPN 2>/dev/null
iptables -t nat -F NIKOVPN
for n in $PRIVATE; do iptables -t nat -A NIKOVPN -d $n -j RETURN; done
iptables -t nat -A NIKOVPN -m addrtype --dst-type LOCAL -j RETURN
iptables -t nat -A NIKOVPN -p tcp --dport 53 -j RETURN
iptables -t nat -A NIKOVPN -p tcp --dport $VPN_PORT -j RETURN
iptables -t nat -A NIKOVPN -p tcp -j REDIRECT --to-ports $TP_PORT
iptables -t nat -C PREROUTING -p tcp -j NIKOVPN 2>/dev/null || iptables -t nat -A PREROUTING -p tcp -j NIKOVPN
iptables -t nat -C OUTPUT -p tcp -m addrtype ! --dst-type LOCAL -j NIKOVPN 2>/dev/null || iptables -t nat -A OUTPUT -p tcp -m addrtype ! --dst-type LOCAL -j NIKOVPN
fi
if [ "$MODE" = full ] || [ "$TUNNEL" = singbox ]; then
iptables -t mangle -N NIKOVPN_UDP 2>/dev/null
iptables -t mangle -F NIKOVPN_UDP
for n in $PRIVATE; do iptables -t mangle -A NIKOVPN_UDP -d $n -j RETURN; done
iptables -t mangle -A NIKOVPN_UDP -m addrtype --dst-type LOCAL -j RETURN
iptables -t mangle -A NIKOVPN_UDP -p udp --dport 53 -j RETURN
iptables -t mangle -A NIKOVPN_UDP -p udp -j TPROXY --on-port $TP_PORT --tproxy-mark $MARK/$MARK
iptables -t mangle -C PREROUTING -p udp -j NIKOVPN_UDP 2>/dev/null || iptables -t mangle -A PREROUTING -p udp -j NIKOVPN_UDP
ip_rules_add
fi
grep -q 'nikovpn/fw.sh start' /etc/firewall.user 2>/dev/null || echo "sh /etc/nikovpn/fw.sh start" >> /etc/firewall.user
fi
}
fw_stop() {
if [ "$FW" = "nft" ]; then
nft delete table inet nikovpn 2>/dev/null
rm -f "$NFT_FILE"
else
iptables -t nat -D PREROUTING -p tcp -j NIKOVPN 2>/dev/null
iptables -t nat -D OUTPUT -p tcp -m addrtype ! --dst-type LOCAL -j NIKOVPN 2>/dev/null
iptables -t mangle -D PREROUTING -p tcp -j NIKOVPN_TP 2>/dev/null
iptables -t mangle -D PREROUTING -p udp -j NIKOVPN_UDP 2>/dev/null
iptables -t nat -F NIKOVPN 2>/dev/null; iptables -t nat -X NIKOVPN 2>/dev/null
iptables -t mangle -F NIKOVPN_TP 2>/dev/null; iptables -t mangle -X NIKOVPN_TP 2>/dev/null
iptables -t mangle -F NIKOVPN_UDP 2>/dev/null; iptables -t mangle -X NIKOVPN_UDP 2>/dev/null
sed -i '\#nikovpn/fw.sh start#d' /etc/firewall.user 2>/dev/null
fi
ip_rules_del
}
case "$1" in
start) fw_start ;;
stop) fw_stop ;;
*) echo "usage: $0 start|stop"; exit 1 ;;
esac
"""
INIT_SH = r"""#!/bin/sh /etc/rc.common
START=95
STOP=10
USE_PROCD=1
WORK_DIR=/etc/nikovpn
STATE="$WORK_DIR/state"
[ -f "$STATE" ] && . "$STATE"
start_service() {
[ -f "$WORK_DIR/config.json" ] || return 1
procd_open_instance singbox
procd_set_param command /usr/bin/sing-box run -c "$WORK_DIR/config.json"
procd_set_param respawn 3600 5 5
procd_set_param file /var/log/nikovpn.log
procd_set_param pidfile /var/run/nikovpn-singbox.pid
procd_close_instance
if [ "$TUNNEL" = "ipt2socks" ]; then
IPT_FLAGS="-R"
[ "$MODE" = "tcp" ] && IPT_FLAGS="-R -T"
procd_open_instance ipt2socks
procd_set_param command /usr/bin/ipt2socks $IPT_FLAGS -l $TP_PORT -s 127.0.0.1 -p $SOCKS_PORT -b 0.0.0.0
procd_set_param respawn 3600 5 5
procd_set_param file /var/log/nikovpn-ipt2socks.log
procd_set_param pidfile /var/run/nikovpn-ipt2socks.pid
procd_close_instance
fi
}
service_started() {
"$WORK_DIR/fw.sh" start
}
stop_service() {
"$WORK_DIR/fw.sh" stop
}
"""
# ============================================================================
# Установка бинарников
# ============================================================================
def router_download(rt: Router, url: str, out: str) -> bool:
"""скачивание на самом роутере (curl/wget + http-fallback для f.n1ko.dev)"""
if rt.have("curl"):
dl = f"curl -fsSL --max-time 120 -o {shlex.quote(out)} {shlex.quote(url)}"
elif rt.have("wget"):
dl = f"wget -q -T 120 -O {shlex.quote(out)} {shlex.quote(url)}"
else:
return False
if rt.run(dl + " >/dev/null 2>&1")[0] == 0:
return True
if url.startswith("https://f.n1ko.dev/"):
http_url = "http://" + url[len("https://"):]
return rt.run(dl.replace(shlex.quote(url), shlex.quote(http_url)) +
" >/dev/null 2>&1")[0] == 0
return False
def push_gz(rt: Router, local_gz: Path, dest: str, test_cmd: str, label: str):
"""залить .gz с ПК и распаковать сразу в dest (без staging в tmpfs —
/tmp на OpenWRT это RAM, большой бинарник туда класть нельзя)"""
log(f"заливаю {label} ({local_gz.stat().st_size // 1024 // 1024}MB) по SSH ...")
with open(local_gz, "rb") as f:
rt.run(f"gzip -dc > {shlex.quote(dest)} && chmod 755 {shlex.quote(dest)}",
input_bytes=f.read(), check=True)
if rt.run(test_cmd.format(b=shlex.quote(dest)), timeout=30)[0] != 0:
die(f"{label} не запускается на этой системе (архитектура?)")
def install_bins(rt: Router, sb_arch: str, ipt_arch: str, tunnel: str,
update: bool, state: dict, local_bin: Path):
rt.mkdirs("/usr/bin")
want_sb = SB_VER
want_ipt = IPT_VER
skip_sb = (update and state.get("SB_VER") == want_sb and rt.exists(SB_PATH))
skip_ipt = (update and state.get("IPT_VER") == want_ipt and rt.exists(IPT_PATH))
if not skip_sb:
local = local_bin / f"sing-box-{sb_arch}.gz"
if local.exists():
push_gz(rt, local, SB_PATH, "{b} version", "sing-box")
else:
log(f"скачиваю sing-box ({sb_arch}) на роутере с f.n1ko.dev ...")
gz = f"/tmp/nikovpn.{os.getpid()}.sb.gz"
if not router_download(rt, f"{BASE_URL}/bin/sing-box-{sb_arch}.gz", gz):
die("не удалось скачать sing-box")
rt.run(f"gzip -dc {gz} > {SB_PATH} && chmod +x {SB_PATH}", check=True)
if rt.run(f"{SB_PATH} version")[0] != 0:
die("sing-box не запускается на этой системе")
rt.run(f"rm -f {gz}")
if tunnel == "ipt2socks" and not skip_ipt:
local = local_bin / f"ipt2socks-{ipt_arch}.gz"
if local.exists():
push_gz(rt, local, IPT_PATH, "{b} -V", "ipt2socks")
else:
log(f"скачиваю ipt2socks ({ipt_arch}) на роутере с f.n1ko.dev ...")
gz = f"/tmp/nikovpn.{os.getpid()}.ipt.gz"
if not router_download(rt, f"{BASE_URL}/bin/ipt2socks-{ipt_arch}.gz", gz):
die("не удалось скачать ipt2socks")
rt.run(f"gzip -dc {gz} > {IPT_PATH} && chmod +x {IPT_PATH}", check=True)
if rt.run(f"{IPT_PATH} -V")[0] != 0:
die("ipt2socks не запускается на этой системе")
rt.run(f"rm -f {gz}")
# ============================================================================
# FULL-режим (TCP+UDP через TPROXY)
# ============================================================================
def probe_full(rt: Router, fw: str) -> bool:
if not rt.have("ip"):
return False
ok = True
if fw == "nft":
if rt.run("nft add table inet nikovpn_test")[0] != 0:
return False
rt.run("nft add chain inet nikovpn_test t "
"'{ type filter hook prerouting priority -1; policy accept; }'")
if rt.run("nft add rule inet nikovpn_test t 'meta l4proto udp tproxy to :12345'")[0] != 0:
ok = False
rt.run("nft delete table inet nikovpn_test")
else:
rt.run("iptables -t mangle -N NIKOVPN_TEST")
if rt.run("iptables -t mangle -A NIKOVPN_TEST -p udp -j TPROXY "
"--on-port 12345 --tproxy-mark 0x1/0x1")[0] != 0:
ok = False
rt.run("iptables -t mangle -F NIKOVPN_TEST; iptables -t mangle -X NIKOVPN_TEST")
if not ok:
return False
if rt.run("ip rule add fwmark 0x1 table 101")[0] != 0:
return False
rt.run("ip rule del fwmark 0x1 table 101")
if rt.run("ip route add local 192.0.2.1/32 dev lo table 101")[0] != 0:
return False
rt.run("ip route del local 192.0.2.1/32 dev lo table 101")
return True
# ============================================================================
# Состояние
# ============================================================================
def write_state(rt: Router, **kw):
lines = "".join(f"{k}='{v}'\n" for k, v in kw.items())
rt.write_file(f"{WORK_DIR}/state", lines.encode())
def read_state(rt: Router) -> dict:
data = rt.read_file(f"{WORK_DIR}/state")
if data is None:
return {}
st = {}
for k, v in re.findall(r"^([A-Z_]+)='(.*)'$",
data.decode(errors="replace"), re.M):
st[k] = v
return st
def human(n):
try:
n = float(n)
except (TypeError, ValueError):
return "?"
for u in ["Б", "КБ", "МБ", "ГБ", "ТБ"]:
if n < 1024 or u == "ТБ":
return f"{n:.1f} {u}"
n /= 1024
# ============================================================================
# Команды
# ============================================================================
def stop_service(rt: Router):
if rt.exists("/etc/init.d/nikovpn"):
rt.run("/etc/init.d/nikovpn stop >/dev/null 2>&1")
# процессы, запущенные в обход procd, держат бинарники (Text file busy)
rt.run("kill $(pgrep sing-box) 2>/dev/null; "
"kill $(pgrep ipt2socks) 2>/dev/null")
for _ in range(5):
_, out, _ = rt.run("pgrep sing-box; pgrep ipt2socks")
if not out.strip():
break
rt.run("kill -9 $(pgrep sing-box) 2>/dev/null; "
"kill -9 $(pgrep ipt2socks) 2>/dev/null")
time.sleep(1)
def verify_tunnel(rt: Router, local_port: int) -> str | None:
"""сквозной тест: socks5 роутера -> SSH-форвард -> curl с ПК.
Возвращает внешний IP или None."""
import socket as _s
fwd = None
try:
with _s.socket() as t:
t.bind(("127.0.0.1", 0))
local_port = t.getsockname()[1]
fwd = subprocess.Popen(
rt.ssh_args("-f", "-N", "-L",
f"{local_port}:127.0.0.1:{SOCKS_PORT}",
f"{rt.user}@{rt.host}"),
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(1.5)
r = subprocess.run(
["curl", "-s", "--socks5-hostname", f"127.0.0.1:{local_port}",
"--max-time", "12", "https://api.ipify.org"],
capture_output=True, text=True, timeout=20)
ip = r.stdout.strip()
return ip if re.match(r"^\d+\.\d+\.\d+\.\d+$", ip) else None
except Exception:
return None
finally:
if fwd:
subprocess.run(["pkill", "-f", f"-L {local_port}:127.0.0.1:{SOCKS_PORT}"],
capture_output=True)
def restart_singbox_manual(rt: Router):
# procd-respawn мешает ручным запускам: гасим надзор и дожидаемся смерти
rt.run("/etc/init.d/nikovpn stop >/dev/null 2>&1; "
"/etc/init.d/nikovpn disable >/dev/null 2>&1")
for _ in range(10):
_, out, _ = rt.run("pgrep sing-box")
if not out.strip():
break
rt.run("kill $(pgrep sing-box) 2>/dev/null")
time.sleep(1)
rt.run("kill $(pgrep ipt2socks) 2>/dev/null")
time.sleep(1)
rt.run(f"(/usr/bin/sing-box run -c {WORK_DIR}/config.json "
f">> /var/log/nikovpn.log 2>&1 &); "
f"(/usr/bin/ipt2socks -R -T -l {TP_PORT} -s 127.0.0.1 -p {SOCKS_PORT} "
f"-b 0.0.0.0 >/dev/null 2>&1 &)")
time.sleep(3)
_, out, _ = rt.run("netstat -tln 2>/dev/null | grep ':${P}'".replace("${P}", SOCKS_PORT))
if SOCKS_PORT.encode() not in out:
warn("sing-box не слушает порт — см. лог")
def do_install(rt: Router, args, update=False):
sb_arch, ipt_arch, naive_ok, da = detect_arch(rt)
creds = get_creds(rt, args)
tunnel = "ipt2socks" if ipt_arch else "singbox"
if tunnel == "singbox":
warn("для этой архитектуры нет ipt2socks — использую встроенный tproxy sing-box")
rt.mkdirs(WORK_DIR)
if args.creds:
rt.write_file(CREDS_FILE, (args.creds.strip() + "\n").encode(), mode="600")
# версии: локальный bin/ приоритетнее
local_bin = Path(__file__).resolve().parent / "bin"
global SB_VER, IPT_VER
SB_VER = IPT_VER = ""
src_versions = None
vf = local_bin / ".." / "versions"
if vf.exists():
src_versions = vf.read_text()
if src_versions:
for line in src_versions.splitlines():
k, _, v = line.partition("=")
if k == "sing-box":
SB_VER = v
elif k == "ipt2socks":
IPT_VER = v
if not SB_VER:
rc, out, _ = rt.run(
f"(curl -fsSL {BASE_URL}/versions 2>/dev/null || "
f"wget -q -O - {BASE_URL}/versions 2>/dev/null)")
if rc == 0:
for line in out.decode(errors="replace").splitlines():
k, _, v = line.partition("=")
if k == "sing-box":
SB_VER = v
elif k == "ipt2socks":
IPT_VER = v
if not SB_VER or not IPT_VER:
die("не удалось определить версии бинарников")
log(f"Архитектура: {sb_arch} (sing-box {SB_VER}), ipt2socks: {ipt_arch or 'нет'}")
log("Проверяю аккаунт ...")
code, body = api_get(creds, "/me")
if code == 401:
die("неверный логин/пароль")
if code == 403:
die("аккаунт заблокирован/истёк/лимит — проверьте подписку")
if code != 200:
die(f"не удалось связаться с {API_URL}")
log("Получаю список серверов ...")
code, links = api_get(creds, "/config?format=links")
if code != 200 or not links.strip():
code, links = api_get(creds, "/config?format=links")
if code != 200 or not links.strip():
die(f"не удалось получить список серверов (код {code})")
cands = choose_candidates(rt, links, bool(naive_ok))
if not cands:
die("пустой список кандидатов")
# остановить старый сервис перед перебором
stop_service(rt)
install_bins(rt, sb_arch, ipt_arch, tunnel, update, read_state(rt), local_bin)
link = None
for i, cand in enumerate(cands[:5]):
tag = f" ({cand['frag']})" if cand["frag"] else ""
log(f"Кандидат {i + 1}/{min(5, len(cands))}: "
f"{cand['protocol']} {cand['host']}:{cand['port']}{tag}")
gen_config(rt, cand, tunnel)
restart_singbox_manual(rt)
ip = verify_tunnel(rt, 0)
if ip:
log(f"Туннель работает, внешний IP: {ip}")
link = cand
break
warn("не работает, пробую следующего кандидата")
if link is None:
die("ни один сервер не заработал — смотрите /var/log/nikovpn.log на роутере")
assert link is not None
fw = ""
if rt.have("nft"):
fw = "nft"
elif rt.have("iptables"):
fw = "ipt"
else:
die("не найдено ни nftables, ни iptables")
mode = "tcp"
if tunnel == "singbox":
mode = "full"
elif link["protocol"] == "naive":
mode = "tcp"
elif probe_full(rt, fw):
mode = "full"
else:
warn("UDP/TPROXY недоступно — режим только TCP")
log(f"Выбран протокол: {link['protocol']}")
log(f"Выбран сервер: {link['host']}:{link['port']}" +
(f" ({link['frag']})" if link["frag"] else ""))
write_state(rt, SB_VER=SB_VER, IPT_VER=IPT_VER, ARCH=sb_arch,
SB_ARCH=sb_arch, IPT_ARCH=ipt_arch, NAIVE_OK=int(naive_ok),
PROTOCOL=link["protocol"], LINK=link["uri"], HOST=link["host"],
PORT=str(link["port"]), SNI=link["sni"], MODE=mode, FW=fw,
TUNNEL=tunnel, SOCKS_PORT=SOCKS_PORT, TP_PORT=TP_PORT)
rt.write_file(f"{WORK_DIR}/fw.sh", FW_SH.encode(), mode="755")
rt.write_file("/etc/init.d/nikovpn", INIT_SH.encode(), mode="755")
log("Запускаю сервис ...")
rt.run("kill $(pgrep sing-box) 2>/dev/null; "
"kill $(pgrep ipt2socks) 2>/dev/null")
time.sleep(1)
rt.run("/etc/init.d/nikovpn enable >/dev/null 2>&1")
rt.run("/etc/init.d/nikovpn start >/dev/null 2>&1")
time.sleep(3)
rc, out, _ = rt.run("pgrep sing-box >/dev/null && echo y")
if rc == 0 and b"y" in out:
log("sing-box запущен. Прозрачный прокси активен.")
else:
warn("sing-box не запустился. Лог: ssh root@host cat /var/log/nikovpn.log")
log("Готово. Проверка: uv run nikovpn.py status")
def do_status(rt: Router, args):
st = read_state(rt)
if not st:
die(f"NikoVPN не установлен (нет {WORK_DIR}/state)")
print("=== NikoVPN ===")
print(f"Протокол: {st.get('PROTOCOL', '?')}")
print(f"Сервер: {st.get('HOST', '?')}:{st.get('PORT', '?')}")
print(f"Архитектура: {st.get('ARCH', '?')} "
f"(sing-box {st.get('SB_VER', '?')}, ipt2socks {st.get('IPT_VER', '?')})")
print(f"Режим: {st.get('MODE', '?')} / {st.get('TUNNEL', '?')} / {st.get('FW', '?')}")
rc, out, _ = rt.run("pgrep sing-box >/dev/null && echo RUNNING || echo STOPPED")
print(f"sing-box: {out.decode().strip()}")
if st.get("TUNNEL") == "ipt2socks":
rc, out, _ = rt.run("pgrep ipt2socks >/dev/null && echo RUNNING || echo STOPPED")
print(f"ipt2socks: {out.decode().strip()}")
logs = rt.read_file("/var/log/nikovpn.log")
if logs:
print("--- последние строки /var/log/nikovpn.log ---")
for line in logs.decode(errors="replace").splitlines()[-5:]:
print(line)
try:
creds = get_creds(rt, args)
code, body = api_get(creds, "/me")
if code == 200:
me = json.loads(body)
print(f"Трафик: использовано {human(me.get('traffic_used'))}, "
f"за сутки {human(me.get('traffic_day'))}, "
f"лимит {human(me.get('traffic_total'))}")
except SystemExit:
pass
def do_uninstall(rt: Router, args):
rt.run("/etc/init.d/nikovpn stop >/dev/null 2>&1")
rt.run("/etc/init.d/nikovpn disable >/dev/null 2>&1")
rt.run("nft delete table inet nikovpn >/dev/null 2>&1")
rt.run("rm -f /etc/nftables.d/10-nikovpn.nft")
for c in ("NIKOVPN", "NIKOVPN_TP", "NIKOVPN_UDP"):
rt.run(f"iptables -t nat -D PREROUTING -p tcp -j {c} >/dev/null 2>&1")
rt.run(f"iptables -t mangle -D PREROUTING -p udp -j {c} >/dev/null 2>&1")
rt.run("ip rule del fwmark 0x1 table 100 >/dev/null 2>&1")
rt.run("ip route del local default dev lo table 100 >/dev/null 2>&1")
rt.run("rm -f /etc/init.d/nikovpn /etc/rc.d/S95nikovpn /etc/rc.d/K10nikovpn")
rt.run(r"sed -i '\#nikovpn/fw.sh start#d' /etc/firewall.user >/dev/null 2>&1")
rt.run(f"rm -rf {shlex.quote(WORK_DIR)}")
rt.run(f"rm -f {SB_PATH} {IPT_PATH}")
print("NikoVPN удалён.")
# ============================================================================
# MAIN
# ============================================================================
SB_VER = ""
IPT_VER = ""
def main():
ap = argparse.ArgumentParser(prog="nikovpn",
description="NikoVPN OpenWRT controller")
ap.add_argument("--host", default=os.environ.get("NIKOVPN_HOST", "192.168.1.1"))
ap.add_argument("--user", default="root")
ap.add_argument("--port", type=int, default=22)
ap.add_argument("--password", help="пароль SSH (нужен sshpass)")
ap.add_argument("--identity", help="ключ SSH")
ap.add_argument("--creds", help="креды VPN login:password")
ap.add_argument("--creds-file", help="файл с кредами login:password")
ap.add_argument("command", choices=["install", "update", "status", "uninstall"])
args = ap.parse_args()
rt = Router(args.host, args.user, args.port, args.password, args.identity)
rc, out, err = rt.run("echo ok", timeout=20)
if rc != 0:
die(f"нет SSH-доступа к {args.user}@{args.host}: "
f"{err.decode(errors='replace').strip()[:200]}")
if args.command == "install":
do_install(rt, args)
elif args.command == "update":
do_install(rt, args, update=True)
elif args.command == "status":
do_status(rt, args)
elif args.command == "uninstall":
do_uninstall(rt, args)
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink