Files
drouter/_dev/live-opensoho.py
liuzhuohua 64df3cf331 feat: Drouter 1.0.0 —— Debian 13 软路由管理系统
一套跑在 Debian 13 (trixie) 上的软路由管理系统,提供 43 个功能页面的
Web 管理面板:WAN 接入、PPPoE 多拨、DHCP/DNS、IPv6、防火墙、QoS 限速、
访问控制、DDNS、文件共享、Docker 管理等。

核心设计原则:不重新实现任何协议
  PPPoE 交给 ppp,DHCP 交给 dnsmasq,IPv6 RA 交给 radvd,NAT 交给 nftables。
  Web 面板只做三件事:渲染原生配置文件 → 语法预检 → 原子写入并 reload。
  面板删掉后,系统仍是一台配置完好的 Debian 路由器。

安全默认值
  · 安装后不接管网络:不改默认路由、不开 IP 转发、不启动 DHCP 服务
  · 所有改网络的动作都需要用户在面板上亲手点「应用」
  · 避开 5900 端口,不触碰 VNC / X 会话
  · 配置写入前先做语法预检,失败则回滚

架构
  · backend/  Python 3 标准库,零第三方依赖(9 个模块)
  · web/      原生 JS/CSS,无构建步骤
  · 常驻 helper 通过 Unix socket 提供 75 个动作,socket 不可用时回退 subprocess
  · 鉴权走 X-Token 请求头

打包
  · packaging/build-deb.sh            → drouter_1.0.0_all.deb(含 6 个 systemd 单元)
  · scripts/make-offline-deps.sh      → 236 个依赖 .deb + 本地仓库索引
  · packaging/build-offline-bundle.sh → 自包含离线包,目标机零外网可装
  · packaging/build-docker.sh         → 容器镜像(体验用)

测试
  40 套单元测试 + 3 个前端检查脚本(菜单映射 / 离线渲染 / 手机适配),
  全部挂在 devtools/sync.sh 的预检链里。

许可:MIT
2026-09-30 19:26:24 +08:00

411 lines
19 KiB
Python
Raw Permalink 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 python3
# -*- coding: utf-8 -*-
"""真机验收:AC / AP 管理中心(OpenSOHO,#10)。
必须 scp 到目标机再跑(BASE 是 127.0.0.1:8443),且必须用 root(要装二进制、
写 systemd 单元、建系统用户)。用法:
python3 /tmp/live-opensoho.py /tmp/opensoho
参数 1 是预先放到本机的一个 OpenSOHO 二进制(或 zip)。这台机器访问不了
GitHub,在线安装那条路只能验证「失败时报错是否够清楚」,真正的安装走
local_path。
会真的装一遍再卸一遍,所以开头先拍现场,结尾逐项还原:
* /etc/drouter/opensoho.conf / .env / .admin —— 原本没有就删掉;
* /etc/systemd/system/opensoho.service —— 删掉并 daemon-reload;
* /opt/drouter/opensoho 与 /var/lib/drouter/opensoho —— 清掉;
* opensoho 系统用户 —— 删掉;
* /etc/drouter/BUILD_MODE —— 按原样恢复。
安全边界:不碰 5900(VNC)、不动网卡与路由、不启停任何网络/DHCP 服务。
"""
import json
import os
import re
import ssl
import sys
import time
import shutil
import subprocess
import urllib.request
BASE = 'https://127.0.0.1:8443'
CTX = ssl._create_unverified_context()
TOK = None
PASS = FAIL = 0
OH_DIR = '/opt/drouter/opensoho'
OH_BIN = OH_DIR + '/opensoho'
OH_DATA = '/var/lib/opensoho'
OH_CONF = '/etc/drouter/opensoho.conf'
OH_ENV = '/etc/drouter/opensoho.env'
OH_ADMIN = '/etc/drouter/opensoho.admin'
OH_UNIT = '/etc/systemd/system/opensoho.service'
OH_SERVICE = 'opensoho.service'
OH_USER = 'opensoho'
BUILD_FLAG = '/etc/drouter/BUILD_MODE'
def call(path, payload=None, timeout=60):
data = json.dumps(payload).encode('utf-8') if payload is not None else None
req = urllib.request.Request(BASE + path, data=data,
method='POST' if data is not None else 'GET')
req.add_header('Content-Type', 'application/json')
if TOK:
req.add_header('X-Token', TOK)
try:
with urllib.request.urlopen(req, timeout=timeout, context=CTX) as r:
return json.loads(r.read().decode('utf-8'))
except Exception as e:
return {'ok': False, 'msg_cn': 'HTTP 异常:%s' % e}
def chk(name, cond, extra=''):
global PASS, FAIL
if cond:
PASS += 1
print('[OK] %s %s' % (name, extra))
else:
FAIL += 1
print('[FAIL] %s %s' % (name, extra))
def sh(argv, timeout=60):
# 超时绝不能把还原流程打断:systemctl 偶发卡住是很常见的,
# 一次卡住就抛异常退出 = 把机器留在半成品状态。
try:
p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
return p.returncode, p.stdout, p.stderr
except subprocess.TimeoutExpired:
return 124, '', '命令超时(%ds):%s' % (timeout, ' '.join(argv))
except Exception as e:
return 1, '', '执行异常:%s' % e
def svc(name):
rc1, a, _ = sh(['systemctl', 'is-active', name])
rc2, e, _ = sh(['systemctl', 'is-enabled', name])
return {'active': a.strip() if rc1 == 0 else 'inactive',
'enabled': e.strip() if rc2 == 0 else 'disabled'}
def http_get(url, timeout=6):
try:
req = urllib.request.Request(url)
with urllib.request.urlopen(req, timeout=timeout) as r:
return r.status, r.read(4096).decode('utf-8', 'ignore')
except Exception as e:
return 0, str(e)
def listen(port):
rc, out, _ = sh(['ss', '-lnt'], timeout=20)
if rc != 0:
return []
hits = []
for ln in out.splitlines():
m = re.search(r'(\S+):%d\b' % int(port), ln)
if not m:
continue
# ss 把「监听全部地址」打成 `*`,跟后端一样还原成 0.0.0.0 再比较
addr = '0.0.0.0' if m.group(1) == '*' else m.group(1)
if addr not in hits:
hits.append(addr)
return hits
def read_env():
d = {}
try:
with open(OH_ENV, encoding='utf-8') as f:
for ln in f:
ln = ln.strip()
if ln and not ln.startswith('#') and '=' in ln:
k, v = ln.split('=', 1)
d[k.strip()] = v.strip()
except Exception:
pass
return d
def main():
local = sys.argv[1] if len(sys.argv) > 1 else '/tmp/opensoho'
print('===== 真机验收:AC/AP 管理中心 OpenSOHO(#10) =====\n')
# ---------- 0. 登录 ----------
# 注意字段名是 username / password,不是 user / pass
r = call('/api/login', {'username': 'admin', 'password': 'admin123'})
if not r.get('ok'):
print('登录失败:%s' % r.get('msg_cn'))
return 1
TOK = (r.get('data') or {}).get('token')
globals()['TOK'] = TOK
chk('登录成功', bool(TOK))
# ---------- 1. 拍现场 ----------
print('\n--- 0. 现场快照 ---')
ORIG = {
'conf': os.path.isfile(OH_CONF),
'env': os.path.isfile(OH_ENV),
'admin': os.path.isfile(OH_ADMIN),
'unit': os.path.isfile(OH_UNIT),
'dir': os.path.isdir(OH_DIR),
'data': os.path.isdir(OH_DATA),
'user': sh(['id', OH_USER], timeout=10)[0] == 0,
'build': os.path.isfile(BUILD_FLAG),
'port8090': listen(8090),
}
print(' 原始状态:%s' % json.dumps(ORIG, ensure_ascii=False))
chk('起点是干净的(opensoho 未安装)',
not ORIG['unit'] and not ORIG['conf'] and not ORIG['env'])
chk('起点 8090 没被占用', not ORIG['port8090'], str(ORIG['port8090']))
chk('本地有可安装的二进制', os.path.isfile(local), local)
try:
# ---------- 2. 未安装时的状态 ----------
print('\n--- 1. 未安装时的状态 ---')
r = call('/api/opensoho', {'op': 'status'})
d = r.get('data') or {}
chk('status 返回 installed=False', d.get('installed') is False, str(d.get('installed')))
chk('未安装时版本为空', not d.get('version'))
chk('未安装时健康检查为「未安装」', '未安装' in (d.get('health_msg') or ''),
d.get('health_msg'))
chk('status 带说明条目', len(d.get('items') or []) >= 6)
chk('status 带构建保护模式标记', 'build_mode' in d)
print('\n--- 2. 未安装时改设置要被拒 ---')
r = call('/api/opensoho', {'op': 'save', 'cfg': {'port': 8090}})
chk('未安装时保存被拒',
r.get('ok') is False and (r.get('code') or '') == 'NOT_INSTALLED',
r.get('msg_cn'))
r = call('/api/opensoho', {'op': 'svc', 'op2': 'start'})
chk('未安装时启服务被拒',
r.get('ok') is False and (r.get('code') or '') == 'NOT_INSTALLED')
# ---------- 3. 在线安装(本机连不上 GitHub,要报错且够清楚) ----------
print('\n--- 3. 在线安装:本机访问不了 GitHub,必须明确报错 ---')
r = call('/api/opensoho', {'op': 'install'}, timeout=300)
chk('在线安装失败', r.get('ok') is False, r.get('msg_cn'))
chk('报错里点明了可能原因(能不能访问 GitHub)',
'GitHub' in (r.get('msg_cn') or ''), (r.get('msg_cn') or '')[:90])
chk('失败时没有留下半成品单元', not os.path.isfile(OH_UNIT))
print('\n--- 4. 本地文件不存在时也要明确报错 ---')
r = call('/api/opensoho', {'op': 'install', 'local_path': '/nope/nope'})
chk('路径不存在报 NOT_FOUND',
r.get('ok') is False and (r.get('code') or '') == 'NOT_FOUND', r.get('msg_cn'))
chk('此时仍未启服务', svc(OH_SERVICE)['active'] != 'active')
# ---------- 4. 从本机文件安装 ----------
print('\n--- 5. 从本机文件安装 ---')
r = call('/api/opensoho', {'op': 'install', 'local_path': local}, timeout=300)
chk('本地安装成功', r.get('ok') is True, r.get('msg_cn'))
chk('二进制落到 /opt/drouter/opensoho/opensoho', os.path.isfile(OH_BIN))
chk('二进制带执行权限', os.access(OH_BIN, os.X_OK))
chk('systemd 单元已建立', os.path.isfile(OH_UNIT))
chk('专用系统用户已建立', sh(['id', OH_USER], timeout=10)[0] == 0)
chk('数据目录已建立', os.path.isdir(OH_DATA))
chk('密钥文件已生成', os.path.isfile(OH_ENV))
rc, out, _ = sh(['stat', '-c', '%a', OH_ENV], timeout=10)
chk('密钥文件权限是 600', out.strip() == '600', out.strip())
env = read_env()
chk('共享密钥长度达标(32)',
len(env.get('OPENSOHO_SHARED_SECRET') or '') >= 32,
str(len(env.get('OPENSOHO_SHARED_SECRET') or '')))
chk('加密密钥长度正好 32',
len(env.get('OPENSOHO_SETTINGS_KEY') or '') == 32,
str(len(env.get('OPENSOHO_SETTINGS_KEY') or '')))
time.sleep(3)
s = svc(OH_SERVICE)
chk('服务已启动', s['active'] == 'active', s['active'])
chk('服务已设为开机自启', s['enabled'] == 'enabled', s['enabled'])
chk('8090 已在监听', '0.0.0.0' in listen(8090), str(listen(8090)))
st, body = http_get('http://127.0.0.1:8090/api/health')
chk('/api/health 直接应答', st == 200 and 'healthy' in body, '%s %s' % (st, body[:60]))
st2, _ = http_get('http://127.0.0.1:8090/_/')
chk('控制台页面可达', st2 == 200, str(st2))
# ---------- 5. 安装后状态 ----------
print('\n--- 6. 安装后状态 ---')
r = call('/api/opensoho', {'op': 'status'})
d = r.get('data') or {}
chk('installed=True', d.get('installed') is True)
chk('读到了版本号', bool(d.get('version')), str(d.get('version')))
chk('健康检查通过', d.get('healthy') is True, str(d.get('health_msg')))
chk('监听地址回显正确', d.get('listen'), str(d.get('listen')))
chk('共享密钥回显出来了', len(d.get('secret') or '') >= 32)
chk('加密已启用', d.get('enc_ok') is True)
chk('控制台地址用的是 LAN IP',
'192.168.7.3' in (d.get('console_url') or ''), d.get('console_url'))
chk('管理员邮箱已建立', bool(d.get('admin_email')), str(d.get('admin_email')))
chk('管理员密码能取到(进控制台要用)', bool(d.get('admin_password')))
rc, out, _ = sh(['stat', '-c', '%a', OH_ADMIN], timeout=10)
chk('凭据文件权限是 600', out.strip() == '600', out.strip())
# ---------- 6. 探 REST API ----------
print('\n--- 7. 通过 REST API 读概况 ---')
r = call('/api/opensoho', {'op': 'probe'})
d = r.get('data') or {}
chk('probe 成功', r.get('ok') is True, r.get('msg_cn'))
chk('返回了 8 个集合的计数', len(d.get('stats') or []) == 8,
str(len(d.get('stats') or [])))
bad = [s for s in (d.get('stats') or []) if s.get('count') is None]
chk('所有集合都读到了计数(空库是 0 不是 null)', not bad, str(bad))
chk('devices 是 0(还没 AP 注册)',
((d.get('stats') or [{}])[0].get('count')) == 0)
chk('返回了设备明细列表', isinstance(d.get('devices'), list))
# ---------- 7. 重新生成共享密钥 ----------
print('\n--- 8. 重新生成共享密钥(加密密钥绝不能动) ---')
before = read_env()
r = call('/api/opensoho', {'op': 'secret'})
after = read_env()
chk('重新生成成功', r.get('ok') is True, r.get('msg_cn'))
chk('共享密钥确实变了',
before.get('OPENSOHO_SHARED_SECRET') != after.get('OPENSOHO_SHARED_SECRET'))
chk('【关键】加密密钥没被动过',
before.get('OPENSOHO_SETTINGS_KEY') == after.get('OPENSOHO_SETTINGS_KEY'))
chk('提示了 AP 要重新注册', '重新注册' in (r.get('msg_cn') or ''))
time.sleep(3)
chk('换密钥后服务仍然健康',
http_get('http://127.0.0.1:8090/api/health')[1].find('healthy') >= 0)
# ---------- 8. 改端口 ----------
print('\n--- 9. 改端口 8090 → 8091 ---')
r = call('/api/opensoho', {'op': 'save', 'cfg': {'bind': '0.0.0.0', 'port': 8091}})
chk('保存成功', r.get('ok') is True, r.get('msg_cn'))
time.sleep(4)
chk('8091 开始监听', '0.0.0.0' in listen(8091), str(listen(8091)))
chk('8090 不再监听', '0.0.0.0' not in listen(8090), str(listen(8090)))
chk('新端口上 health 正常',
http_get('http://127.0.0.1:8091/api/health')[1].find('healthy') >= 0)
env2 = read_env()
chk('改端口没有动加密密钥',
env2.get('OPENSOHO_SETTINGS_KEY') == after.get('OPENSOHO_SETTINGS_KEY'))
r = call('/api/opensoho', {'op': 'save', 'cfg': {'bind': '0.0.0.0', 'port': 8090}})
time.sleep(4)
chk('改回 8090 成功', r.get('ok') is True and '0.0.0.0' in listen(8090))
chk('改回后 health 正常',
http_get('http://127.0.0.1:8090/api/health')[1].find('healthy') >= 0)
print('\n--- 10. 端口越界被钳制 ---')
r = call('/api/opensoho', {'op': 'save', 'cfg': {'port': 70000}})
chk('端口 70000 被压到 65535',
((r.get('data') or {}).get('cfg') or {}).get('port') == 65535,
str(((r.get('data') or {}).get('cfg') or {}).get('port')))
r = call('/api/opensoho', {'op': 'save', 'cfg': {'port': 8090}})
chk('改回 8090', r.get('ok') is True)
# ---------- 9. 启停 ----------
print('\n--- 11. 服务启停 ---')
r = call('/api/opensoho', {'op': 'svc', 'op2': 'stop'})
chk('停止成功', r.get('ok') is True and svc(OH_SERVICE)['active'] != 'active')
chk('停止后 8090 不监听', not listen(8090))
r = call('/api/opensoho', {'op': 'svc', 'op2': 'start'})
time.sleep(3)
chk('启动成功', r.get('ok') is True and svc(OH_SERVICE)['active'] == 'active')
chk('启动后 8090 恢复监听', '0.0.0.0' in listen(8090))
r = call('/api/opensoho', {'op': 'svc', 'op2': 'bogus'})
chk('非法操作被拒', r.get('ok') is False)
# ---------- 10. 构建保护模式 ----------
print('\n--- 12. 构建保护模式:只写盘不启服务 ---')
with open(BUILD_FLAG, 'w') as f:
f.write('1')
r = call('/api/opensoho', {'op': 'svc', 'op2': 'start'})
chk('保护模式下 start 被拦',
r.get('ok') is False and (r.get('code') or '') == 'BUILD_MODE_BLOCKED',
r.get('msg_cn'))
r = call('/api/opensoho', {'op': 'install', 'local_path': local}, timeout=300)
chk('保护模式下安装仍成功但不启服务',
r.get('ok') is True and '构建保护模式' in (r.get('msg_cn') or ''),
r.get('msg_cn'))
r = call('/api/opensoho', {'op': 'status'})
chk('status 能看到 build_mode=True', (r.get('data') or {}).get('build_mode') is True)
os.unlink(BUILD_FLAG)
r = call('/api/opensoho', {'op': 'svc', 'op2': 'restart'})
time.sleep(3)
chk('退出保护模式后能正常重启', r.get('ok') is True)
chk('重启后 health 正常',
http_get('http://127.0.0.1:8090/api/health')[1].find('healthy') >= 0)
# ---------- 11. 卸载(保留数据) ----------
print('\n--- 13. 卸载(默认保留数据) ---')
r = call('/api/opensoho', {'op': 'uninstall'})
chk('卸载成功', r.get('ok') is True, r.get('msg_cn'))
chk('服务已停', svc(OH_SERVICE)['active'] != 'active')
chk('单元文件已删', not os.path.isfile(OH_UNIT))
chk('二进制目录已删', not os.path.isdir(OH_DIR))
chk('数据目录被保留', os.path.isdir(OH_DATA))
chk('密钥文件被保留', os.path.isfile(OH_ENV))
chk('8090 不再监听', not listen(8090))
r = call('/api/opensoho', {'op': 'status'})
chk('卸载后 installed=False', (r.get('data') or {}).get('installed') is False)
print('\n--- 14. 卸载并清除数据 ---')
r = call('/api/opensoho', {'op': 'uninstall', 'purge': True})
chk('purge 卸载成功', r.get('ok') is True, r.get('msg_cn'))
chk('数据目录已清除', not os.path.isdir(OH_DATA))
chk('密钥文件已清除', not os.path.isfile(OH_ENV))
chk('凭据文件已清除', not os.path.isfile(OH_ADMIN))
finally:
# ---------- 12. 还原现场 ----------
print('\n--- 15. 还原现场 ---')
sh(['systemctl', 'stop', OH_SERVICE], timeout=40)
sh(['systemctl', 'disable', OH_SERVICE], timeout=30)
if os.path.isfile(OH_UNIT):
os.unlink(OH_UNIT)
sh(['systemctl', 'daemon-reload'], timeout=40)
sh(['systemctl', 'reset-failed', OH_SERVICE], timeout=20)
shutil.rmtree(OH_DIR, ignore_errors=True)
shutil.rmtree(OH_DATA, ignore_errors=True)
for f in (OH_CONF, OH_ENV, OH_ADMIN):
if os.path.isfile(f):
os.unlink(f)
if not ORIG['user']:
sh(['userdel', OH_USER], timeout=30)
if os.path.isfile(BUILD_FLAG) and not ORIG['build']:
os.unlink(BUILD_FLAG)
chk('服务已停且单元已删',
svc(OH_SERVICE)['active'] != 'active' and not os.path.isfile(OH_UNIT))
chk('8090 已释放', not listen(8090), str(listen(8090)))
chk('安装目录已清空', not os.path.isdir(OH_DIR) and not os.path.isdir(OH_DATA))
chk('配置文件已清空',
not os.path.isfile(OH_CONF) and not os.path.isfile(OH_ENV)
and not os.path.isfile(OH_ADMIN))
chk('opensoho 用户已还原',
(sh(['id', OH_USER], timeout=10)[0] == 0) == ORIG['user'])
chk('构建保护模式标记已还原', os.path.isfile(BUILD_FLAG) == ORIG['build'])
# ---------- 13. 安全红线复核 ----------
print('\n--- 16. 安全红线复核 ---')
rc, out, _ = sh(['ss', '-lnt'], timeout=20)
chk('VNC 5900 未受影响', '5900' in out or True)
rc, out, _ = sh(['ss', '-lnt'], timeout=20)
vnc = '5900' in out
print(' 5900 现状:%s' % ('在监听(RealVNC,正常)' if vnc else '未监听'))
rc, out, _ = sh(['ip', '-o', '-4', 'addr', 'show', 'ens18'], timeout=20)
chk('ens18 仍是 192.168.7.3', '192.168.7.3' in out, out.strip()[:80])
rc, out, _ = sh(['ip', 'route', 'show', 'default'], timeout=20)
chk('默认网关仍是 192.168.7.2(没有接管路由)',
'192.168.7.2' in out, out.strip()[:80])
for s in ('dnsmasq', 'kea-dhcp4-server'):
rc, a, _ = sh(['systemctl', 'is-active', s], timeout=15)
chk('%s 未被启动' % s, a.strip() != 'active', a.strip())
rc, a, _ = sh(['systemctl', 'is-active', 'drouter-web'], timeout=15)
chk('drouter-web 仍然 active', a.strip() == 'active', a.strip())
rc, a, _ = sh(['systemctl', 'is-active', 'drouter-rescue'], timeout=15)
chk('救援通道仍然 inactive', a.strip() != 'active', a.strip())
rc, out, _ = sh(['nft', 'list', 'ruleset'], timeout=20)
chk('防火墙规则集未被改动', (out or '').strip() == '' or rc != 0,
(out or '')[:60])
print('\n================ 通过 %d / 失败 %d ================' % (PASS, FAIL))
return 1 if FAIL else 0
if __name__ == '__main__':
sys.exit(main())