姚章浩
a3a1319a11
一次性把本地与 CI 的门禁收成一条命令：make verify = python3 scripts/ci_checks.py。 两边跑的是同一份逻辑，不再各写一套步骤（两边各写一套的结局是迟早只改一边）。 14 项必需检查：identity（检出 == 事件完整 SHA，只在 CI 必需）、commits、deps-install、 format、static-go、static-web、build-go、build-web、tests-go、tests-web、contracts、 security-secrets、security-deps、manifests-locked。make list 打印清单。 三条判据写死在代码里： - 失败即停，但剩下的记 not_run 并如实写进报告；aggregate() 只认 success， 未运行/被跳过都不算通过（不适用要在清单里声明 required_in，不许运行时悄悄 skip）。 - 工具缺失、依赖没按锁装、输出解析不了 ⇒ 失败，不当成"检查通过"。 - 测试必须真跑：go test 加 -count=1 并盘点通过/跳过数；前端读 vitest 的 JSON 报告， 空套件不算通过。跳过原因会打进报告（当前 72 个集成用例因缺 TEST_DATABASE_URL 跳过）。 新增四个自带反向自检的脚本（--self-test 故意弄坏必须红）： scripts/check_format.py（Go 判输出而非退出码 + 前端只能减的基线）、 scripts/check_secrets.py（只打印位置与类别，绝不打印值）、 scripts/check_deps_vuln.py（govulncheck 锁 v1.1.4 + npm audit 分级 + 到期例外）、 scripts/ci_checks.py（门禁 + 汇总）。 验证（本机实跑，全部为真实退出码）： make verify → 总体 success（档案 local），identity=skipped 已在清单里声明为 CI 必需； tests-go 执行 31 个包、通过 549 / 失败 0 / 跳过 72；tests-web 106 个套件、191 个用例全过； security-deps：govulncheck v1.1.4 0 命中、npm audit 生产与全量均 0。
313 lines
16 KiB
Python
313 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""依赖漏洞检查（Go + 前端）—— 判据 → 退出码。
|
||
|
||
两个扫描器：
|
||
· Go ：`govulncheck`（**可达性分析**：只报真的会被调用到的漏洞），版本锁死。
|
||
· 前端 ：`npm audit`（按 severity 分级；生产依赖与仅开发依赖分开判）。
|
||
|
||
分级与例外（正本 §4.5）：
|
||
· 生产依赖出现 high/critical ⇒ **阻断**。
|
||
· 仅开发依赖出现 high/critical ⇒ 阻断，**除非**在 `security/dependency-exceptions.json`
|
||
里有一条**有效**例外（必须有理由 / 责任人 / 批准人 / 到期日，且未过期）。
|
||
· moderate 及以下 ⇒ 只记录，不阻断。
|
||
· govulncheck 的命中**一律阻断**（它已经做过可达性分析），除非有有效例外。
|
||
· 扫描器缺工具、连不上漏洞库、输出解析不了 ⇒ **失败**，绝不当成"没有漏洞"。
|
||
|
||
为什么例外要写全字段：没有到期日的例外会永久留在仓库里，而半年后没人记得为什么。
|
||
批准人字段是给人看的责任锚点 —— 本脚本**不会**替任何人填它。
|
||
|
||
用法：
|
||
python3 scripts/check_deps_vuln.py # 真跑两个扫描器
|
||
python3 scripts/check_deps_vuln.py --self-test # 只跑分级逻辑的反向自检（用合成夹具）
|
||
python3 scripts/check_deps_vuln.py --skip-go # 只查前端（调试用，不算全绿）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
WEB = ROOT / "web"
|
||
EXCEPTIONS = ROOT / "security" / "dependency-exceptions.json"
|
||
|
||
GOVULNCHECK = "golang.org/x/vuln/cmd/govulncheck@v1.1.4" # 锁版本：不静默升级工具
|
||
BLOCKING_SEV = {"critical", "high"}
|
||
EXCEPTION_FIELDS = ("id", "severity", "scope", "reason", "owner", "approved_by", "expires_on")
|
||
|
||
|
||
def parse_json_stream(text: str) -> list[dict]:
|
||
"""govulncheck 的 JSON 是**多个 pretty-printed 对象首尾相接**，不是行分隔。
|
||
按行解析会崩（实测：'str' object has no attribute 'keys'）。"""
|
||
dec = json.JSONDecoder()
|
||
out, i, n = [], 0, len(text)
|
||
while i < n:
|
||
while i < n and text[i] in " \t\r\n":
|
||
i += 1
|
||
if i >= n:
|
||
break
|
||
try:
|
||
obj, j = dec.raw_decode(text, i)
|
||
except json.JSONDecodeError:
|
||
break
|
||
if isinstance(obj, dict):
|
||
out.append(obj)
|
||
i = j
|
||
return out
|
||
|
||
|
||
def load_exceptions(path: Path = EXCEPTIONS) -> tuple[list[dict], list[str]]:
|
||
"""返回 (条目, 加载问题)。文件缺失或坏掉都算问题 —— 不能静默当成"没有例外"。"""
|
||
if not path.is_file():
|
||
return [], [f"{path.relative_to(ROOT)} 不存在（例外机制应当显式存在，哪怕里面是空的）"]
|
||
try:
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as e:
|
||
return [], [f"{path.relative_to(ROOT)} 解析失败：{e}"]
|
||
entries = raw.get("exceptions") if isinstance(raw, dict) else None
|
||
if not isinstance(entries, list):
|
||
return [], [f"{path.relative_to(ROOT)} 里没有 exceptions 列表"]
|
||
return [e for e in entries if isinstance(e, dict)], []
|
||
|
||
|
||
def exception_state(entry: dict, today: dt.date) -> tuple[bool, str]:
|
||
"""一条例外是否**有效**。无效的例外等于没有例外（不许拿它换绿灯）。"""
|
||
missing = [f for f in EXCEPTION_FIELDS if not str(entry.get(f) or "").strip()]
|
||
if missing:
|
||
return False, f"例外缺少字段 {missing}"
|
||
try:
|
||
expires = dt.date.fromisoformat(str(entry["expires_on"]))
|
||
except ValueError:
|
||
return False, f"例外 {entry.get('id')} 的 expires_on 不是 YYYY-MM-DD"
|
||
if expires < today:
|
||
return False, f"例外 {entry.get('id')} 已于 {expires} 过期（过期即失效，要重新批）"
|
||
if str(entry["approved_by"]).strip().lower() in {"", "todo", "tbd", "pending", "self"}:
|
||
return False, f"例外 {entry.get('id')} 的 approved_by 是占位值（必须由人批）"
|
||
return True, ""
|
||
|
||
|
||
def matches(entry: dict, finding: dict) -> bool:
|
||
if str(entry.get("tool") or "") != finding["tool"]:
|
||
return False
|
||
if entry.get("package") and entry["package"] != finding.get("package"):
|
||
return False
|
||
return not (entry.get("id") and entry["id"] != finding.get("id"))
|
||
|
||
|
||
def grade(findings: list[dict], entries: list[dict], today: dt.date) -> dict:
|
||
"""纯函数：给定命中与例外，判定阻断 / 豁免 / 仅记录。--self-test 直接打这个函数。"""
|
||
blocking, exempted, notes = [], [], []
|
||
for f in findings:
|
||
sev = str(f.get("severity") or "").lower()
|
||
if f["tool"] == "npm-audit" and sev not in BLOCKING_SEV:
|
||
notes.append(f"{f['tool']} {f.get('package')} severity={sev} —— 低于阻断线，仅记录")
|
||
continue
|
||
if f["tool"] == "npm-audit" and f.get("scope") == "dev" and sev in BLOCKING_SEV:
|
||
pass # 开发依赖也走同一条路：要么修，要么有有效例外
|
||
cands = [e for e in entries if matches(e, f)]
|
||
if not cands:
|
||
blocking.append({**f, "reason": f"{sev or '未标severity'} 命中，且没有对应例外"})
|
||
continue
|
||
ok, why = exception_state(cands[0], today)
|
||
if ok:
|
||
exempted.append({**f, "reason": f"例外 {cands[0]['id']}（到期 {cands[0]['expires_on']}）"})
|
||
else:
|
||
blocking.append({**f, "reason": f"例外无效：{why}"})
|
||
return {"blocking": blocking, "exempted": exempted, "notes": notes}
|
||
|
||
|
||
# ── 扫描器 ────────────────────────────────────────────────────────────────
|
||
def scan_go() -> tuple[list[dict], str | None, str]:
|
||
go = shutil.which("go") or "/opt/citools/go/bin/go"
|
||
if not (Path(go).exists() or shutil.which("go")):
|
||
return [], "找不到 go，govulncheck 无法运行（工具缺失 ⇒ 失败，不当成无漏洞）", ""
|
||
env = dict(os.environ)
|
||
env.setdefault("GOFLAGS", "-mod=readonly")
|
||
env.setdefault("GOPROXY", "https://goproxy.cn,direct")
|
||
env.setdefault("GOTOOLCHAIN", "local")
|
||
try:
|
||
p = subprocess.run([go, "run", GOVULNCHECK, "-format", "json", "./..."],
|
||
cwd=ROOT, capture_output=True, text=True, env=env, timeout=900)
|
||
except subprocess.TimeoutExpired:
|
||
return [], "govulncheck 超时（漏洞库不可达时会这样）—— 未完成 ⇒ 失败", ""
|
||
objs = parse_json_stream(p.stdout)
|
||
if not objs:
|
||
return [], (f"govulncheck 没有产出任何可解析的 JSON（exit={p.returncode}）："
|
||
f"{(p.stderr or p.stdout)[:300]}"), ""
|
||
cfg = next((o["config"] for o in objs if isinstance(o.get("config"), dict)), {})
|
||
findings = []
|
||
for o in objs:
|
||
f = o.get("finding")
|
||
if not isinstance(f, dict) or not isinstance(f.get("osv"), dict):
|
||
continue
|
||
osv = f["osv"]
|
||
pkg = ""
|
||
for fr in f.get("trace") or []:
|
||
if isinstance(fr, dict) and fr.get("module"):
|
||
pkg = fr["module"]
|
||
break
|
||
findings.append({"tool": "govulncheck", "id": osv.get("id", ""), "package": pkg,
|
||
"severity": "", "scope": "build",
|
||
"detail": (osv.get("summary") or "")[:200]})
|
||
info = f"govulncheck {cfg.get('scanner_version', '?')}（DB {cfg.get('db_last_modified', '?')}）"
|
||
return findings, None, info
|
||
|
||
|
||
def npm_audit(args: list[str]) -> tuple[dict, str | None]:
|
||
if not (WEB / "package-lock.json").is_file():
|
||
return {}, "web/package-lock.json 不存在 ⇒ npm audit 无法按锁文件判定（先补锁文件）"
|
||
try:
|
||
p = subprocess.run(["npm", "audit", "--json", *args], cwd=WEB,
|
||
capture_output=True, text=True, timeout=600)
|
||
except subprocess.TimeoutExpired:
|
||
return {}, "npm audit 超时（registry 不可达？）—— 未完成 ⇒ 失败"
|
||
try:
|
||
return json.loads(p.stdout or "{}"), None
|
||
except json.JSONDecodeError:
|
||
return {}, f"npm audit 输出解析不了（exit={p.returncode}）：{(p.stdout or p.stderr)[:300]}"
|
||
|
||
|
||
def audit_findings(data: dict, scope: str) -> list[dict]:
|
||
out = []
|
||
for name, adv in (data.get("vulnerabilities") or {}).items():
|
||
if not isinstance(adv, dict):
|
||
continue
|
||
sev = str(adv.get("severity") or "").lower()
|
||
ids = [v.get("url", "").rsplit("/", 1)[-1]
|
||
for v in (adv.get("via") or []) if isinstance(v, dict) and v.get("url")]
|
||
out.append({"tool": "npm-audit", "id": ids[0] if ids else "", "package": name,
|
||
"severity": sev, "scope": scope,
|
||
"detail": f"isDirect={adv.get('isDirect')} range={adv.get('range')}"})
|
||
return out
|
||
|
||
|
||
# ── 反向自检：分级逻辑必须真的会红、也必须真的会豁免 ──────────────────────
|
||
def self_test() -> int:
|
||
today = dt.date(2026, 10, 11)
|
||
good_exc = {"tool": "npm-audit", "id": "GHSA-aaaa", "package": "tinypool", "severity": "critical",
|
||
"scope": "dev", "reason": "仅测试工具链，不进产物", "owner": "yao",
|
||
"approved_by": "yao", "expires_on": "2026-12-31"}
|
||
cases: list[tuple[str, list[dict], list[dict], dict]] = [
|
||
("生产依赖 high", [{"tool": "npm-audit", "id": "G1", "package": "vite", "severity": "high",
|
||
"scope": "prod"}], [], {"blocking": 1, "exempted": 0}),
|
||
("开发依赖 critical 且无例外", [{"tool": "npm-audit", "id": "G2", "package": "tinypool",
|
||
"severity": "critical", "scope": "dev"}], [], {"blocking": 1, "exempted": 0}),
|
||
("开发依赖 critical + 有效例外", [{"tool": "npm-audit", "id": "G3", "package": "tinypool",
|
||
"severity": "critical", "scope": "dev"}],
|
||
[{**good_exc, "id": "G3", "package": "tinypool"}],
|
||
{"blocking": 0, "exempted": 1}),
|
||
("例外已过期", [{"tool": "npm-audit", "id": "G4", "package": "tinypool", "severity": "critical",
|
||
"scope": "dev"}], [{**good_exc, "id": "G4", "expires_on": "2026-01-01"}],
|
||
{"blocking": 1, "exempted": 0}),
|
||
("例外缺批准人", [{"tool": "npm-audit", "id": "G5", "package": "tinypool", "severity": "critical",
|
||
"scope": "dev"}], [{**good_exc, "id": "G5", "approved_by": "TBD"}],
|
||
{"blocking": 1, "exempted": 0}),
|
||
("例外包名不匹配", [{"tool": "npm-audit", "id": "G6", "package": "other-pkg", "severity": "high",
|
||
"scope": "dev"}], [good_exc], {"blocking": 1, "exempted": 0}),
|
||
("moderate 一律只记录", [{"tool": "npm-audit", "id": "G7", "package": "x", "severity": "moderate",
|
||
"scope": "dev"}], [], {"blocking": 0, "exempted": 0}),
|
||
("govulncheck 命中一律阻断", [{"tool": "govulncheck", "id": "GO-2026-0001", "package": "stdlib",
|
||
"severity": "", "scope": "build"}], [], {"blocking": 1, "exempted": 0}),
|
||
("govulncheck 命中 + 有效例外", [{"tool": "govulncheck", "id": "GO-2026-0002", "package": "stdlib",
|
||
"severity": "", "scope": "build"}],
|
||
[{**good_exc, "tool": "govulncheck", "id": "GO-2026-0002", "package": "stdlib"}],
|
||
{"blocking": 0, "exempted": 1}),
|
||
]
|
||
problems: list[str] = []
|
||
for name, findings, excs, want in cases:
|
||
got = grade(findings, excs, today)
|
||
if len(got["blocking"]) != want["blocking"] or len(got["exempted"]) != want["exempted"]:
|
||
problems.append(f"{name}：期望 阻断{want['blocking']}/豁免{want['exempted']}，"
|
||
f"实际 阻断{len(got['blocking'])}/豁免{len(got['exempted'])}")
|
||
# 解析器自检：govulncheck 的 JSON 是"多个 pretty 对象首尾相接"
|
||
stream = ('{"config": {"scanner_version": "v1.1.4"}}\n'
|
||
'{\n "finding": {"osv": {"id": "GO-1", "summary": "s"}, '
|
||
'"trace": [{"module": "example.com/m"}]}\n}\n')
|
||
objs = parse_json_stream(stream)
|
||
if len(objs) != 2:
|
||
problems.append(f"JSON 流解析器没认出两个对象（得到 {len(objs)} 个）")
|
||
if problems:
|
||
print("❌ 依赖漏洞分级自检失败：", file=sys.stderr)
|
||
for p in problems:
|
||
print(f" · {p}", file=sys.stderr)
|
||
return 1
|
||
print(f"✅ 依赖漏洞分级自检通过：{len(cases)} 个判据 + JSON 流解析器全部符合预期")
|
||
return 0
|
||
|
||
|
||
def main(argv: list[str]) -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--self-test", action="store_true")
|
||
ap.add_argument("--skip-go", action="store_true", help="跳过 govulncheck（调试用，不算全绿）")
|
||
args = ap.parse_args(argv)
|
||
if args.self_test:
|
||
return self_test()
|
||
|
||
entries, exc_problems = load_exceptions()
|
||
today = dt.date.today()
|
||
findings: list[dict] = []
|
||
info: list[str] = []
|
||
tool_errors: list[str] = []
|
||
|
||
if not args.skip_go:
|
||
go_findings, err, go_info = scan_go()
|
||
if err:
|
||
tool_errors.append(err)
|
||
findings += go_findings
|
||
if go_info:
|
||
info.append(go_info)
|
||
|
||
prod, err = npm_audit(["--omit=dev"])
|
||
if err:
|
||
tool_errors.append(f"生产依赖：{err}")
|
||
else:
|
||
info.append(f"npm audit 生产依赖：{json.dumps((prod.get('metadata') or {}).get('vulnerabilities'), ensure_ascii=False)}")
|
||
findings += audit_findings(prod, "prod")
|
||
|
||
full, err = npm_audit([])
|
||
if err:
|
||
tool_errors.append(f"全量依赖：{err}")
|
||
else:
|
||
info.append(f"npm audit 全量：{json.dumps((full.get('metadata') or {}).get('vulnerabilities'), ensure_ascii=False)}")
|
||
prod_names = {f["package"] for f in findings if f["tool"] == "npm-audit" and f["scope"] == "prod"}
|
||
findings += [f for f in audit_findings(full, "dev") if f["package"] not in prod_names]
|
||
|
||
print("── 依赖漏洞检查 ──")
|
||
for line in info:
|
||
print(f" 工具/数据源：{line}")
|
||
for line in exc_problems:
|
||
print(f" ⚠️ 例外加载：{line}", file=sys.stderr)
|
||
|
||
verdict = grade(findings, entries, today)
|
||
for f in verdict["exempted"]:
|
||
print(f" ⚪ 已豁免：{f['tool']} {f.get('id') or f.get('package')} —— {f['reason']}")
|
||
for f in verdict["notes"]:
|
||
print(f" · 仅记录：{f}")
|
||
|
||
fail = bool(tool_errors) or bool(exc_problems) or bool(verdict["blocking"])
|
||
if tool_errors:
|
||
print("❌ 扫描器没跑成 —— 这**不是**\"没有漏洞\"（正本 §4.5）：", file=sys.stderr)
|
||
for e in tool_errors:
|
||
print(f" · {e}", file=sys.stderr)
|
||
if exc_problems:
|
||
print("❌ 例外文件有问题，无法据它做分级判定：", file=sys.stderr)
|
||
if verdict["blocking"]:
|
||
print("❌ 阻断级依赖漏洞：", file=sys.stderr)
|
||
for f in verdict["blocking"]:
|
||
print(f" {f['tool']} {f.get('id') or '?'} {f.get('package') or '?'} "
|
||
f"severity={f.get('severity') or 'n/a'} scope={f.get('scope')} —— {f['reason']}", file=sys.stderr)
|
||
print(" 处理：升级到修复版本（改 package.json/go.mod + 同步锁文件，独立提交）；"
|
||
"确有理由不修的，写进 security/dependency-exceptions.json 并写清责任人、批准人、到期日。",
|
||
file=sys.stderr)
|
||
if not fail:
|
||
print(f"✅ 依赖漏洞检查通过（阻断级 0；豁免 {len(verdict['exempted'])}；仅记录 {len(verdict['notes'])}）")
|
||
return 1 if fail else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main(sys.argv[1:])) |