charging-cabinet/test/security-hardening/smoke_prod.py

#!/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())