471b61d277
1) probe_export.py：任务状态语义**搞反了**（导出成功也会被判失败）
- download_tasks.status 的权威语义是 0 排队 / 1 处理中 / **2 完成** / 3 失败
（db/migrate.rs:225 列注释；写入侧 charge_records.rs::mark_task_completed 写的就是 2）。
- 该脚本此前**等 status == 1（处理中）当成功、把 status == 2（完成）判成"导出失败"**
—— 典型假红。已按权威语义订正，并与另外两处**本来就正确**的口径对齐：
web/src/pages/charge-records/DownloadCenter.tsx::statusText、
test/schema_alignment_e2e.py（等 status in (2,3) 再断言 == 2）。
- 同批修正退出码分类：账号缺 charge_record:export（403）由 exit 1 改为
**exit 2「未验证」** —— 本项要验证的是"降权成 pms 后能否往 exports/ 落盘"，
没权限根本走不到那一步，报 1 是**把夹具缺陷说成产品缺陷**（误报与假绿同样有害）。
- 允许用 PMS_PROBE_PHONE / PMS_PROBE_PASSWORD 换成有权限的账号。
- **C 项结论（由落盘物证直接判定，不依赖该脚本）**：测试环境
exports/充电记录__-.xlsx 5559 B、生产同类产物 5457 B，属主均为 pms:pms；
两环境 exports/uploads 目录均 pms:pms，两 api-server 单元均 User=pms/Group=pms
⇒ 降权进程确实能落盘。已在 README 写明并给出取证依据。
2) deploy-prod.ps1 的退出码（此前失败的探针只打印 ❌ 就继续，脚本仍以 0 退出）
- 消费链路探测失败（exit 1）现在**记入失败清单**并以 exit 1 收尾；
"未验证"（隧道没起）默认仍不计失败（环境问题会误报），
但新增 **-StrictVerify** 开关供 CI/无人值守使用，打开即 exit 1。
- 好处：把这条从"打印了 ❌ 却 exit 0"（仓库第 8 条假绿）里摘出来。
3) 统一退出码 0/1/2 落地到本批脚本，并对每个失败路径实测取证：
- test/tcp_e2e_smoke.py：13/13 通过（含未登录业务报文被静默丢弃）；先复位夹具再跑。
- test/security-hardening/ 新增 verdict.py 判据容器（未跟踪 → 本次入库）；
4 个探针全部按失败→exit 1 改造，并各实测过失败路径。
- 实测结果（2026-09-20 发布后回归）：smoke_prod 4/4 exit 0、
probe_prod 5/5 exit 0、probe_upload 6/6 exit 0。
4) 工具与文档
- tools/db.py 默认库由**生产 pms 改为测试 pms_dev**（可用 PMS_MYSQL_DB 覆盖，
切生产时打印醒目警告）—— "以为在测测试库、实际打生产"的隐患。
- tools/db_schema_diff.py：连不上必须是 exit 2，绝不伪装成"无漂移"的 0。
- 新增 tools/dev-tunnel.ps1（入库，SSH 隧道；口令走 PMS_SSH_PASSWORD 环境变量），
配套 tools/dev-tunnel-run.ps1 自愈启动器**保持 gitignore**（内含口令）。
- deploy/nginx/nginx.conf 镜像随线上同步（移除 stream 转发；与线上逐字节一致）。
- AGENTS.md / README.md：补「踩过的坑」第 7/8 条（掐断上游管道会杀被测进程、
打印 ❌ 却 exit 0 是假绿）、退出码约定、本地开发改 SSH 隧道、
在线判据口径订正、order_timeout 守门补齐、小程序 vue-tsc 门禁与其抓到的两个真 BUG。
发布验收（本次统一发布，按 device→api 顺序）：
- 测试环境与生产环境四服务 active，端口 9000/9001/3000/3001 正确，ERROR 均为 0；
- 生产迁移按判据 grep '更新列注释' 实测对齐 cabinets.status 列注释；
- 结构漂移 python tools/db_schema_diff.py = 0 处漂移 / exit 0（--vs-ddl 两库同样 0）；
- schema_alignment_e2e --env dev 全部通过 exit 0；
- 生产域名 /api/pms/cabinets -> 401、/pms/ -> 200；消费链路探测通过。
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
||
"""生产最终冒烟：SPA 资源 + 登录链路（错误凭据）+ 清理本次产生的失败计数键"""
|
||
import json
|
||
import socket
|
||
import urllib.error
|
||
import urllib.request
|
||
from urllib.parse import unquote, urlparse
|
||
|
||
from verdict import Verdict # 同目录：判据 + 统一退出码（0 通过 / 1 有失败 / 2 环境问题）
|
||
|
||
DOMAIN = "https://app.anzhizhichong.com"
|
||
ENV_FILE = "/data/pms/api-server/.env"
|
||
FAKE_PHONE = "00000000000"
|
||
|
||
|
||
def env(name):
|
||
with open(ENV_FILE) as fh:
|
||
for line in fh:
|
||
if line.startswith(name + "="):
|
||
return line.split("=", 1)[1].strip()
|
||
raise SystemExit(f"{name} missing")
|
||
|
||
|
||
class Redis:
|
||
def __init__(self, url):
|
||
u = urlparse(url)
|
||
self.sock = socket.create_connection((u.hostname, u.port or 6379), timeout=5)
|
||
self.fh = self.sock.makefile("rwb")
|
||
if u.password:
|
||
self.cmd("AUTH", unquote(u.password))
|
||
self.cmd("SELECT", str(int((u.path or "/0").lstrip("/") or 0)))
|
||
|
||
def _reply(self):
|
||
line = self.fh.readline()
|
||
if line.startswith(b"$"):
|
||
n = int(line[1:])
|
||
return None if n == -1 else self.fh.read(n + 2)[:-2].decode()
|
||
return line.strip().decode()
|
||
|
||
def cmd(self, *args):
|
||
out = b"*%d\r\n" % len(args)
|
||
for a in args:
|
||
a = a.encode() if isinstance(a, str) else a
|
||
out += b"$%d\r\n%s\r\n" % (len(a), a)
|
||
self.fh.write(out)
|
||
self.fh.flush()
|
||
return self._reply()
|
||
|
||
|
||
def http(method, path, data=None, headers=None):
|
||
req = urllib.request.Request(DOMAIN + path, data=data, method=method)
|
||
for k, v in (headers or {}).items():
|
||
req.add_header(k, v)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as resp:
|
||
return resp.status, dict(resp.headers), resp.read()
|
||
except urllib.error.HTTPError as exc:
|
||
return exc.code, dict(exc.headers), exc.read()
|
||
|
||
|
||
def main():
|
||
# 判据容器：失败 → exit 1（此前写死 return 0，属假绿）
|
||
v = Verdict("smoke_prod（生产冒烟：SPA / challenge / 登录参数校验）")
|
||
st, _, body = http("GET", "/pms/")
|
||
print(f"[SPA] GET /pms/ -> {st} bytes={len(body)}")
|
||
v.check("生产 SPA /pms/ 应 200", st == 200, f"status={st}")
|
||
asset = None
|
||
for token in body.decode(errors="ignore").split('"'):
|
||
if token.startswith("/pms/assets/") and token.endswith(".js"):
|
||
asset = token
|
||
break
|
||
if asset:
|
||
st, _, blob = http("GET", asset)
|
||
print(f"[SPA] GET {asset} -> {st} bytes={len(blob)}")
|
||
v.check(f"生产 SPA 资源 {asset} 应 200", st == 200, f"status={st}")
|
||
else:
|
||
v.check("生产 SPA 页面应引用 /pms/assets/*.js", False, "未在 HTML 中找到资源引用")
|
||
|
||
st, _, body = http("GET", "/api/pms/auth/challenge")
|
||
print(f"[login] GET /api/pms/auth/challenge -> {st} (JSON 长度 {len(body)})")
|
||
v.check("challenge 接口应 200", st == 200, f"status={st}")
|
||
|
||
st, _, body = http(
|
||
"POST",
|
||
"/api/pms/auth/login",
|
||
json.dumps({"phone": FAKE_PHONE, "password": "x", "challenge_id": "", "captcha_x": 0}).encode(),
|
||
{"Content-Type": "application/json"},
|
||
)
|
||
print(f"[login] POST /api/pms/auth/login（无 challenge）-> {st} {body.decode()[:120]}")
|
||
v.check("无 challenge 的登录必须被拒（非 200）", st != 200, f"status={st}")
|
||
|
||
redis = Redis(env("REDIS_URL"))
|
||
for key in (f"auth:fail:phone:{FAKE_PHONE}", "auth:fail:ip:unknown"):
|
||
redis.cmd("DEL", key)
|
||
print("[cleanup] 清理测试产生的 Redis 计数键完成")
|
||
return v.exit_code() # 0=判据全过，1=有判据失败
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|