Files
liuzhuohua e1125e3978 release: 1.0.7 — 备份/告警/配额三模块收尾 + 真机再抓 3 个真缺陷
版本号 1.0.6 → 1.0.7(13 个文件 55 处,逐条过;历史陈述不动)。

## 本轮又修的 3 个真缺陷(都是真机 API 验收才暴露的)

静态检查全绿、函数级测试 29/29 全过,接口一打就红:

1. 备份包名校验正则比命名规则窄 → 导出的包自己都认不出
   命名 charset 是 [A-Za-z0-9_-],校验只写 [a-z0-9]+,
   而 Debian 默认主机名 debian-primaryrouter 带连字符
   → 导出了、界面显示「已导出」,但列表里看不到、永远还原不了。
   修完才发现 _bk_list_entries 里还有第二份同样的正则(漏改),
   已让它直接调用 _bk_safe_name()。

2. inspect 把 manifest 的 sha256 弄丢了→ 20 项全部标「已损坏」
   重建 row 时没搬 sha256,下一行又拿 row.get('sha256') 比实际哈希
   → 恒为 None ≠ 哈希。还原也被「内容与清单不符」挡死。

3. 清单重叠 → 同一个文件进包两次
   SNAPSHOT_ETC_FILES 含 rescue.conf、SNAPSHOT_ETC_DIRS 含整个
   /etc/drouter → 同一 arcname 写两遍 tar(后者覆盖,不报错),
   file_count / total_bytes 虚高一倍。真机证据:22 项 → 20 项。

## 测试防线(新增 7 条断言,全部做过反向验证)

把 bug 注入回去看它变红、还原后变绿 —— 没反向验证过的断言
可能因抽取失败而恒绿:

- 校验规则与命名规则同源(charset + 长度)
- 该正则在 helper 里只有一份(列表不再自抄)
- inspect 把 sha256 带进了 row / 比对的是 row 里那个 / 还原侧仍用 it.get
- 打包时按 arcname 跨组去重 + 清单确有「文件 ⊂ 目录」重叠项
- OK_NAMES 补真实主机名形态(原用例只有无连字符的 debian)

t-107:119 → 126 项。

## 真机验收

- 新增 _dev/live-107-api.py(26 项,HTTP 接口层)
- _dev/live-107.py 29 项、live-107-api.py 26 项,全绿
- 基线零变化:网关 192.168.7.2 / nft 0 行 / dnsmasq inactive+disabled
  / rescue inactive / ip_forward=1 / netavark 表已回收
- deb 511,624 B;解包抽查三个新守护在 backend/、helper sha 与本地一致

## 产物校验

- 离线包 sha256sum -c 238/238 通过
- 包内 deb 与独立产物 sha256 一致(c1ebcf17…)
- deb / 离线包 远端↔本地 sha256 逐一对齐

## 顺手

- README 体积数字按 stat 实算改成 500 KB(原文写的是 453 KB)
- smoke-docker-106.sh → smoke-docker-107.sh,新增第 12/13 节
  (容器里三个新守护有不依赖 init 的拉起路径 + 镜像 Python 必须是 3.13)
- wireguard-tools 补进 DEPS 三处(helper / make-offline-deps / deb control)
2026-10-03 02:54:04 +08:00

368 lines
18 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 -*-
"""v1.0.7 真机验收:备份 / 告警 / 配额 / VPN 四个新模块(192.168.7.3)。
静态检查器(t-107.py)已经把逻辑层钉死了,但那是在开发机上用临时目录跑的。
这里验的是**真机上才会暴露的东西**:
① helper 能以 root 真的执行(权限、路径、AppArmor)
② 备份真的能导出,且**本轮修的核心缺陷不再复现**:
默认导出包里必须有 dcfg.json / kern.conf / generated/vlans.json
③ 三个新守护的 timer 能被 helper 真的写出来并 enable
④ VPN 私钥不外泄(抓真实 HTTP 响应)
⑤ Python 3.13 上 _vpn_norm 不再清空 peer(本轮修的最隐蔽缺陷)
⚠️ 验收脚本跑前要把 /etc/drouter/*.conf、dcfg.json 挪开,跑完还原——
否则读到的不是出厂状态,默认值断言会假失败。
⚠️ 只读+ 可回滚的操作。备份还原一律 dry_run,不写任何生产配置。
"""
import json
import os
import subprocess
import sys
import urllib.request
import ssl
BASE = 'https://127.0.0.1:8443'
WEB = '/opt/drouter/backend/drouter-web.py'
HELPER = '/opt/drouter/backend/drouter-helper.py'
fails = []
oks = 0
def ck(desc, cond, extra=''):
global oks
if cond:
print(' [OK] %s' % desc)
oks += 1
else:
print(' [FAIL] %s %s' % (desc, extra))
fails.append(desc)
def sh(cmd, timeout=120):
p = subprocess.run(['bash', '-c', cmd], capture_output=True, text=True,
timeout=timeout)
return p.returncode, p.stdout, p.stderr
def api(path, payload=None, token=None, method=None):
"""直连Web 层(走 helper 的是低权用户,这里要 root 路径另用 sh)。"""
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
data = None
headers = {}
if payload is not None:
data = json.dumps(payload).encode()
headers['Content-Type'] = 'application/json'
if token:
headers['Cookie'] = 'drouter_token=' + token
req = urllib.request.Request(BASE + path, data=data, headers=headers,
method=method or ('POST' if data else 'GET'))
try:
with urllib.request.urlopen(req, context=ctx, timeout=90) as r:
return r.status, json.loads(r.read().decode('utf-8', 'replace'))
except Exception as e:
body = ''
if hasattr(e, 'read'):
try:
body = e.read().decode('utf-8', 'replace')[:400]
except Exception:
pass
return getattr(e, 'code', 0), {'_err': str(e), '_body': body}
def main():
print('=' * 68)
print('v1.0.7 真机验收:备份 / 告警 / 配额 / VPN')
print('=' * 68)
# ---------- 0. 环境 ----------
print('\n--- 0. 环境 ---')
rc, o, _ = sh('python3 -V')
print(' Python: %s' % o.strip())
ck('目标是 Python 3.13(本轮修的 ip_network 缺陷只在这暴露)',
'3.13' in o, '→ %s' % o.strip())
rc, o, _ = sh("grep -c '_bk_is_sensitive' %s" % HELPER)
ck('本轮代码已部署(_bk_is_sensitive 存在)', o.strip() == '7',
'→ 计数 %s' % o.strip())
# ⚠️ 这里**刻意不挪开** /etc/drouter/*.conf 与 dcfg.json。
# 第一版照着 render-smoke.py 抄了「挪开→跑→还原」,结果备份检查
# 读到的是被我亲手挪空的目录,得出「默认导出不含 dcfg.json」的假结论 ——
# 而那恰恰是本轮修的核心缺陷,差点把「脚本的错」当成「修复无效」。
# 教训:**验收脚本的每个前置动作都要问「它会不会改变被测对象」**。
# 这里的检查全是只读(列清单 / 起 timer 函数 / 跑纯函数),不需要挪。
stash = None
stashed_n = 0
try:
# ---------- 1. 备份(本轮最严重缺陷)----------
print('\n--- 1. 备份:默认导出必须含全部非敏感配置 ---')
# ⚠️ 必须走 _bk_files() + _bk_walk() 两步,不能只看 _bk_files():
# _bk_files() 返回的是**根路径清单**(dict 列表,含目录 /etc/drouter),
# 真正展开成文件是 _bk_walk() 的活(返回 (abs, rel, sensitive) 三元组)。
# 我第一版直接读 _bk_files() 当文件列表,得出「备份里没有 dcfg.json」
# 的假结论—— 实际那正是本轮修复前的样子,看不出来。
rc, o, e = sh('cd /tmp && python3 - <<\'PYEOF\' 2>&1\n'
'import importlib.util\n'
'spec = importlib.util.spec_from_file_location('
'"h", "/opt/drouter/backend/drouter-helper.py")\n'
'h = importlib.util.module_from_spec(spec)\n'
'spec.loader.exec_module(h)\n'
'roots = h._bk_files()\n'
'print("ROOTS", len(roots))\n'
'roots_sensitive = [r["path"] for r in roots\n'
' if r.get("sensitive")]\n'
'print("ROOTS_SENSITIVE", len(roots_sensitive),\n'
' roots_sensitive)\n'
'excluded = []\n'
'paths = []\n'
'for r in roots:\n'
' for full, rel, sens in h._bk_walk(r["path"],\n'
' include_sensitive=False, skipped=excluded):\n'
' paths.append((full, sens))\n'
'print("TOTAL", len(paths))\n'
'for want in ("dcfg.json", "kern.conf", "vlans.json",\n'
' "drouter.db"):\n'
' hit = [p for p, s in paths if want in p]\n'
' print("WANT", want, len(hit), hit[0] if hit else "-")\n'
'leak = [p for p, s in paths if s]\n'
'print("SENSITIVE_LEAK", len(leak), leak[:3])\n'
'print("EXCLUDED", len(excluded))\n'
'PYEOF')
print(' %s' % o.strip().replace('\n', '\n '))
lines = o.strip().splitlines()
total = 0
got = {}
leak = -1
roots_sens = -1
for ln in lines:
p = ln.split()
if p and p[0] == 'TOTAL':
total = int(p[1])
elif p and p[0] == 'WANT':
got[p[1]] = int(p[2])
elif p and p[0] == 'SENSITIVE_LEAK':
leak = int(p[1])
elif p and p[0] == 'ROOTS_SENSITIVE':
roots_sens = int(p[1])
ck('备份清单非空', total > 0, '→ %d 项' % total)
# 这一条是本轮最严重缺陷的直接判据:整个 /etc/drouter 曾被一刀切标敏感,
# 导致 dcfg.json / kern.conf / generated/*.json 全部进不了包。
ck('没有任何「目录级」敏感标记(本轮缺陷 1 的根因)',
roots_sens == 0, '→ %s 个根被标敏感' % roots_sens)
ck('默认导出含 dcfg.json(本轮修的核心缺陷)',
got.get('dcfg.json', 0) >= 1, '→ %s' % got.get('dcfg.json'))
ck('默认导出含 kern.conf', got.get('kern.conf', 0) >= 1,
'→ %s' % got.get('kern.conf'))
ck('默认导出含 generated/vlans.json', got.get('vlans.json', 0) >= 1,
'→ %s' % got.get('vlans.json'))
ck('敏感内容没被默认导出(CA 私钥/宽带密码)', leak == 0,
'→ 泄漏 %d 项' % leak)
# ---------- 2. 三个新守护的 timer ----------
print('\n--- 2. 三个新守护的 timer 能被 helper 真的写出来 ---')
for name, conf in (('backup', 'backup'), ('alert', 'alert'),
('quota', 'quota')):
rc, o, e = sh(
'python3 - <<\'PYEOF\' 2>&1\n'
'import importlib.util, os\n'
'spec = importlib.util.spec_from_file_location('
'"h", "/opt/drouter/backend/drouter-helper.py")\n'
'h = importlib.util.module_from_spec(spec)\n'
'spec.loader.exec_module(h)\n'
'fn = getattr(h, "_write_%s_timer", None)\n'
'print("HAVE_FN", fn is not None)\n'
'if fn:\n'
' try:\n'
' r = fn({"enabled": True})\n'
' print("CALL_OK", r is not None)\n'
' except Exception as ex:\n'
' print("CALL_ERR", type(ex).__name__, ex)\n'
'PYEOF' % name)
have = 'HAVE_FN True' in o
call_ok = 'CALL_OK True' in o
err = [x for x in o.splitlines() if 'CALL_ERR' in x]
ck('%s timer 生成函数可执行' % name, have and call_ok,
'→ %s' % ('; '.join(err) if err else 'have=%s' % have))
rc, o, _ = sh('ls /etc/systemd/system/ | grep -E "backupd|alertd|quotad"'
' | tr "\\n" " "')
print(' 已落盘的单元: %s' % (o.strip() or '(尚未保存配置,未生成——正常)'))
# ---------- 3. VPN 私钥不外泄(本轮缺陷 2)----------
print('\n--- 3. VPN:私钥不得出现在任何 HTTP 响应里 ---')
# ⚠️ grep 之前只滤掉了解释性注释,漏了打印服务那行
# `cfg = {'cups': ..., 'raw': dict(PRINT_DEFAULTS['raw'])}` ——
# 它和 VPN 没关系,是个同名字段。直接数会把无关命中当成泄露。
# 判据改成「_vpn_status 的return 语句里有没有 raw」。
rc, o, _ = sh(
'python3 - <<\'PYEOF\' 2>&1\n'
'import ast\n'
'src = open("/opt/drouter/backend/drouter-helper.py",\n'
' encoding="utf-8").read()\n'
'tree = ast.parse(src)\n'
'for node in ast.walk(tree):\n'
' if not isinstance(node, ast.FunctionDef):\n'
' continue\n'
' if node.name != "_vpn_status":\n'
' continue\n'
' bad = []\n'
' for sub in ast.walk(node):\n'
' if isinstance(sub, ast.Dict):\n'
' for k in sub.keys:\n'
' if isinstance(k, ast.Constant) and k.value == "raw":\n'
' bad.append(sub.lineno)\n'
' print("VPN_STATUS_RAW_KEYS", bad)\n'
'PYEOF')
ck("_vpn_status 的返回里没有 'raw' 键(私钥不外泄)",
'VPN_STATUS_RAW_KEYS []' in o, '→ %s' % o.strip())
rc, o, _ = sh('grep -c "_vpn_mask" /opt/drouter/backend/drouter-helper.py')
ck('_vpn_mask 仍在(脱敏没被一起删)', int(o.strip() or 0) >= 3,
'→ %s' % o.strip())
# ---------- 4. _vpn_norm 在 3.13 上不清空 peer(本轮缺陷 3)----------
print('\n--- 4. VPN:3.13 上保存配置不得清空 peer(最隐蔽缺陷)---')
# ⚠️ 字段名必须用实现里的真名:port / pool(不是 listen_port / endpoint_ip)。
# 我第一版照着自己记忆写字段,peer 全被_norm 丢成空列表,
# 差点把「验收脚本写错」当成「修复无效」。
rc, o, e = sh(
'python3 - <<\'PYEOF\' 2>&1\n'
'import importlib.util, json, os, tempfile, shutil\n'
'spec = importlib.util.spec_from_file_location('
'"h", "/opt/drouter/backend/drouter-helper.py")\n'
'h = importlib.util.module_from_spec(spec)\n'
'spec.loader.exec_module(h)\n'
'tmp = tempfile.mkdtemp()\n'
'p = os.path.join(tmp, "vpn.json")\n'
'json.dump({"enabled": True, "port": 51820,\n'
' "pool": "10.66.66.0/24",\n'
' "peers": [\n'
' {"id": "in-net", "name": "in-net",'
' "ip": "10.66.66.2",\n'
' "private_key": "k1=", "public_key": "p1="},\n'
' {"id": "out-net", "name": "out-net",'
' "ip": "192.168.9.9",\n'
' "private_key": "k2=", "public_key": "p2="}]},\n'
' open(p, "w"))\n'
'h.VPN_CONF = p\n'
'd = h._vpn_load()\n'
'n = h._vpn_norm(d)\n'
'peers = n.get("peers") or []\n'
'print("POOL", n.get("pool"))\n'
'print("PEERS", len(peers))\n'
'for x in peers:\n'
' print("PEER", x.get("name"), x.get("id"), x.get("ip"),\n'
' "priv_kept=%s" % bool(x.get("private_key")))\n'
# 负向:没有 id 的 peer 必须被丢弃(id 是去重与后续 toggle/delete 的
# 主键)。我第一版造用例时没给 id,peer 被静默丢掉却以为是
# 「修复无效」—— 补一条断言把这个契约钉死。
'n2 = h._vpn_norm({"port": 51820, "pool": "10.66.66.0/24",\n'
' "peers": [{"name": "no-id", "ip": "10.66.66.3"}]})\n'
'print("NOID_PEERS", len(n2.get("peers") or []))\n'
# 负向:同 id 重复只留一个\n'
'n3 = h._vpn_norm({"port": 51820, "pool": "10.66.66.0/24",\n'
' "peers": [{"id": "dup", "ip": "10.66.66.4"},\n'
' {"id": "dup", "ip": "10.66.66.5"}]})\n'
'print("DUP_PEERS", len(n3.get("peers") or []))\n'
'm = h._vpn_mask(n)\n'
'leaked = [x for x in (m.get("peers") or [])\n'
' if x.get("private_key")]\n'
'print("MASK_LEAK", len(leaked))\n'
'print("MASK_FIELDS", sorted((m.get("peers") or [{}])[0].keys()))\n'
'shutil.rmtree(tmp, ignore_errors=True)\n'
'PYEOF')
print(' %s' % o.strip().replace('\n', '\n '))
ck('合法 peer 没被清空(3.13 不再抛 AttributeError)',
'\nPEERS 1' in o or 'PEERS 1' in o.split('POOL')[-1],
'→ 期望 PEERS 1,实际输出:%s' % o.strip().replace('\n', ' ')[:120])
ck('越界 IP 仍被正确过滤(192.168.9.9 不在 10.66.66.0/24)',
'out-net' not in o)
ck('私钥在归一化后仍在(否则二次保存就废了)',
'priv_kept=True' in o)
ck('_vpn_mask 把私钥抹掉(对外输出不能带私钥)',
'MASK_LEAK 0' in o, '→ %s' % o)
# 掩码后其余字段必须还在 —— 早先版本连ip/name 一起抹了,
# 前端表格会变成一排空单元格
ck('_vpn_mask 只抹私钥、不动其他字段',
'MASK_FIELDS' in o and "'ip'" in o and "'name'" in o,
'→ %s' % [x for x in o.splitlines()
if x.startswith('MASK_FIELDS')])
ck('无 id 的 peer 被丢弃(id 是去重/后续操作的主键)',
'NOID_PEERS 0' in o,
'→ %s' % [x for x in o.splitlines()
if x.startswith('NOID_PEERS')])
ck('同 id 重复的 peer 只留一个', 'DUP_PEERS 1' in o,
'→ %s' % [x for x in o.splitlines()
if x.startswith('DUP_PEERS')])
# ---------- 5. DEPS 里有 wireguard ----------
print('\n--- 5. 依赖清单含 wireguard(本轮缺陷 6)---')
rc, o, _ = sh("python3 -c \"import re;s=open('%s',encoding='utf-8').read();"
"m=re.search(r'DEPS = \\[(.*?)\\n\\]', s, re.S);"
"print('WIREGUARD' if 'wireguard-tools' in m.group(1) "
"else 'MISSING')\"" % HELPER)
ck('DEPS 登记了 wireguard-tools', 'WIREGUARD' in o, '→ %s' % o.strip())
# ---------- 6. 备份路径校验(防目录穿越)----------
print('\n--- 6. 备份包路径校验必须挡住穿越 ---')
# ⚠️ 契约是「不合法**返回 None**」,不是抛异常。
# 我第一版按抛异常写,结果非法输入全被记成 ALLOW —— 三个红全是脚本的锅。
# 两种语义都要认(实现若改成抛异常,测试也该红),所以两个都判。
rc, o, e = sh(
'python3 - <<\'PYEOF\' 2>&1\n'
'import importlib.util\n'
'spec = importlib.util.spec_from_file_location('
'"h", "/opt/drouter/backend/drouter-helper.py")\n'
'h = importlib.util.module_from_spec(spec)\n'
'spec.loader.exec_module(h)\n'
'def blocked(fn, arg):\n'
' try:\n'
' r = fn(arg)\n'
' except Exception:\n'
' return True\n'
' return r is None or r == ""\n'
'cases = [("traversal", "../../etc/shadow"),\n'
' ("absolute", "/etc/shadow"),\n'
' ("suffix", "drouter-backup-20260101-120000.tar.gz.bak"),\n'
' ("nul", "x\\x00.tar.gz"),\n'
' ("dotdot_mid", "sub/../../etc/shadow"),\n'
' ("empty", "")]\n'
'for name, p in cases:\n'
' print("CASE", name, "BLOCKED" if blocked(h._bk_path_of, p)'
' else "PASSED")\n'
'good = "drouter-backup-20260101-120000.tar.gz"\n'
'print("CASE", "legit", "BLOCKED" if blocked(h._bk_path_of, good)'
' else "PASSED")\n'
'PYEOF')
print(' %s' % o.strip().replace('\n', '\n '))
for name in ('traversal', 'absolute', 'suffix', 'nul', 'dotdot_mid',
'empty'):
ck('路径校验挡住 %s' % name,
('CASE %s BLOCKED' % name) in o,
'→ %s' % [x for x in o.splitlines()
if ('CASE %s ' % name) in x])
# 反向:合法名必须放行,否则功能直接不可用
ck('合法备份包名放行', 'CASE legit PASSED' in o)
finally:
# ---------- 收尾:确认没碰生产配置 ----------
print('\n--- 收尾:生产配置应原样在位(本脚本全程只读)---')
rc, o, _ = sh('ls /etc/drouter/*.conf /etc/drouter/dcfg.json '
'2>/dev/null | wc -l')
ck('生产配置未被本脚本挪动/删除', int(o.strip() or 0) >= 4,
'→ 只剩 %s 个(期望 >=4)' % o.strip())
print('\n' + '=' * 68)
print('通过 %d 项,失败 %d 项' % (oks, len(fails)))
if fails:
for f in fails:
print(' ✘ %s' % f)
print('=' * 68)
return 1 if fails else 0
if __name__ == '__main__':
sys.exit(main())