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

#!/usr/bin/env python3
"""生产环境 A 项实测（用服务端 JWT_SECRET 为该用户签发一个临时 token，仅调用上传接口）

安全说明：
- token 只在本次脚本内存在，不落盘、不外传；
- 只打"应被拒绝"的路径（拒绝发生在写盘/写库之前，不改任何数据）；
- 反馈图"应通过"路径只写一个我自己的测试文件，脚本末尾删除；
- 头像"应通过"路径不在生产执行（会覆盖真实用户 avatar），已在测试环境用同一 md5 二进制验证。
"""
import base64
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request

BASE = "http://127.0.0.1:9000"          # 直连后端（应用层行为）
DOMAIN = "https://app.anzhizhichong.com"  # 走 nginx（看响应头/CORS）
ENV_FILE = "/data/pms/api-server/.env"
UID = 5
PHONE = "13800000003"

SVG = b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>'
PNG = bytes.fromhex(
    "89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489"
    "0000000a49444154789c63000100000500010d0a2db40000000049454e44ae426082"
)


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} not found")


def b64(raw):
    return base64.urlsafe_b64encode(raw).rstrip(b"=")


def mint_token(secret, uid, phone):
    now = int(time.time())
    head = b64(json.dumps({"typ": "JWT", "alg": "HS256"}, separators=(",", ":")).encode())
    body = b64(
        json.dumps(
            {"sub": uid, "phone": phone, "exp": now + 3600, "iat": now}, separators=(",", ":")
        ).encode()
    )
    sig = b64(hmac.new(secret.encode(), head + b"." + body, hashlib.sha256).digest())
    return (head + b"." + body + b"." + sig).decode()


def http(method, url, data=None, headers=None):
    req = urllib.request.Request(url, 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 multipart(field, filename, content, ctype):
    b = "----prodprobe" + str(os.getpid())
    head = (
        f"--{b}\r\n"
        f'Content-Disposition: form-data; name="{field}"; filename="{filename}"\r\n'
        f"Content-Type: {ctype}\r\n\r\n"
    ).encode()
    return head + content + f"\r\n--{b}--\r\n".encode(), f"multipart/form-data; boundary={b}"


def main():
    token = mint_token(env("JWT_SECRET"), UID, PHONE)
    auth = {"Authorization": f"Bearer {token}"}

    st, _, body = http("GET", BASE + "/api/auth/me", headers=auth)
    print(f"[prod-token] GET /api/auth/me -> {st} {body.decode()[:160]}")
    if st != 200:
        raise SystemExit("临时 token 未被接受，终止")

    print("\n===== 生产 头像上传（拒绝路径，不写盘不写库）=====")
    for fn, content, ctype in (
        ("xss.svg", SVG, "image/svg+xml"),
        ("xss.png", SVG, "image/png"),
    ):
        data, ct = multipart("file", fn, content, ctype)
        st, _, body = http(
            "POST", BASE + "/api/auth/avatar/upload", data, {**auth, "Content-Type": ct}
        )
        print(f"  avatar filename={fn} -> {st} {body.decode()[:140]}")

    print("\n===== 生产 反馈图上传（拒绝路径 + 一个真实 PNG 测试文件）=====")
    url = None
    for fn, content, ctype, expect in (
        ("xss.svg", SVG, "image/svg+xml", "拒"),
        ("xss2.png", SVG, "image/png", "拒"),
        ("secprobe.png", PNG, "image/png", "过"),
    ):
        data, ct = multipart("file", fn, content, ctype)
        st, _, body = http("POST", BASE + "/api/feedback/upload", data, {**auth, "Content-Type": ct})
        print(f"  feedback filename={fn}（应{expect}）-> {st} {body.decode()[:140]}")
        if st == 200:
            url = json.loads(body)["data"]["url"]

    if url:
        st, hdrs, blob = http("GET", DOMAIN + url)
        print(
            f"\n[prod-static] GET {DOMAIN}{url} -> {st} content-type={hdrs.get('content-type')} "
            f"nosniff={hdrs.get('x-content-type-options')} bytes={len(blob)}"
        )
        path = "/data/pms/api-server" + url
        if os.path.isfile(path):
            os.remove(path)
            print(f"[cleanup] 已删除我自己的测试文件 {path}")

    print("\n===== 生产 CORS 白名单 =====")
    for origin in (
        "https://app.anzhizhichong.com",
        "https://test.anzhizhichong.com",
        "http://localhost:5173",
        "https://evil.example.com",
        "null",
    ):
        st, hdrs, _ = http("GET", DOMAIN + "/api/pms/health", headers={"Origin": origin})
        print(
            f"  Origin={origin:<32} -> {st} allow-origin="
            f"{hdrs.get('access-control-allow-origin', '(无)')}"
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())