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

#!/usr/bin/env python3
"""A 项实测探针（在服务器上直接打测试后端 127.0.0.1:3000）

1. 登录拿 token（challenge 的正确 x 从测试 Redis db6 读取，模拟"人正确拖动滑块"）
2. POST /api/auth/avatar/upload 传 xss.svg（含 <script>）→ 期望 400 拒绝
3. POST /api/auth/avatar/upload 传 xss.png（SVG 内容、改名 .png）→ 期望 400（魔数兜底）
4. POST /api/auth/avatar/upload 传真 PNG → 期望 200，返回 url
5. POST /api/feedback/upload 同样三连（反馈图上传同源直出）
6. GET 上传后的真实图片 URL → 状态码 / Content-Type / X-Content-Type-Options
"""
import json
import os
import re
import socket
import sys
import urllib.error
import urllib.request
from urllib.parse import unquote, urlparse

BASE = "http://127.0.0.1:3000"
ENV_FILE = "/data/pms_dev/api-server/.env"
PHONE = "13900000009"
PASSWORD = "SecTest#2026"

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


def read_env(name, path=ENV_FILE):
    with open(path) as fh:
        for line in fh:
            line = line.strip()
            if line.startswith(name + "="):
                return line.split("=", 1)[1].strip()
    raise SystemExit(f"{name} not found in {path}")


class Redis:
    """极简 RESP 客户端（服务器没装 redis-cli）"""

    def __init__(self, url):
        u = urlparse(url)
        self.pwd = unquote(u.password or "")
        self.host, self.port, self.db = u.hostname, u.port or 6379, int((u.path or "/0").lstrip("/") or 0)
        self.sock = socket.create_connection((self.host, self.port), timeout=5)
        self.fh = self.sock.makefile("rwb")
        if self.pwd:
            self.cmd("AUTH", self.pwd)
        self.cmd("SELECT", str(self.db))

    def _reply(self):
        line = self.fh.readline()
        if line.startswith(b"+"):
            return line[1:].strip().decode()
        if line.startswith(b"-"):
            raise RuntimeError(line.decode().strip())
        if line.startswith(b":"):
            return int(line[1:])
        if line.startswith(b"$"):
            n = int(line[1:])
            if n == -1:
                return None
            data = self.fh.read(n + 2)
            return data[:-2].decode()
        raise RuntimeError(f"unexpected reply: {line!r}")

    def cmd(self, *args):
        out = b"*%d\r\n" % len(args)
        for a in args:
            if isinstance(a, str):
                a = a.encode()
            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, base=BASE):
    req = urllib.request.Request(base + 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 multipart(field, filename, content, ctype):
    boundary = "----secprobe" + str(os.getpid())
    head = (
        f"--{boundary}\r\n"
        f'Content-Disposition: form-data; name="{field}"; filename="{filename}"\r\n'
        f"Content-Type: {ctype}\r\n\r\n"
    ).encode()
    body = head + content + f"\r\n--{boundary}--\r\n".encode()
    return body, f"multipart/form-data; boundary={boundary}"


def login(redis):
    st, _, body = http("GET", "/api/auth/challenge")
    cid = json.loads(body)["data"]["challenge_id"]
    correct_x = redis.cmd("GET", f"challenge:{cid}")
    payload = json.dumps(
        {"phone": PHONE, "password": PASSWORD, "challenge_id": cid, "captcha_x": int(correct_x)}
    ).encode()
    st, _, body = http("POST", "/api/auth/login", payload, {"Content-Type": "application/json"})
    print(f"[login] status={st} body={body.decode()[:220]}")
    if st != 200:
        raise SystemExit("login failed, cannot continue")
    return json.loads(body)["data"]["token"]


def try_upload(token, path, field, filename, content, ctype, label):
    body, ct = multipart(field, filename, content, ctype)
    st, _, resp = http(
        "POST", path, body, {"Content-Type": ct, "Authorization": f"Bearer {token}"}
    )
    print(f"[{label}] {path} filename={filename} -> status={st} body={resp.decode()[:200]}")
    return st, resp


def main():
    redis = Redis(read_env("REDIS_URL"))
    token = login(redis)
    auth = {"Authorization": f"Bearer {token}"}

    print("\n===== 头像上传 /api/auth/avatar/upload =====")
    try_upload(token, "/api/auth/avatar/upload", "file", "xss.svg", SVG, "image/svg+xml", "A1 svg")
    try_upload(token, "/api/auth/avatar/upload", "file", "xss.png", SVG, "image/png", "A2 svg-as-png")
    st, resp = try_upload(token, "/api/auth/avatar/upload", "file", "ok.png", PNG, "image/png", "A3 real-png")
    ok_url = json.loads(resp)["data"]["avatar"] if st == 200 else None

    print("\n===== 反馈图上传 /api/feedback/upload =====")
    try_upload(token, "/api/feedback/upload", "file", "xss.svg", SVG, "image/svg+xml", "A4 svg")
    try_upload(token, "/api/feedback/upload", "file", "xss2.png", SVG, "image/png", "A5 svg-as-png")
    fb_st, fb_resp = try_upload(token, "/api/feedback/upload", "file", "ok.png", PNG, "image/png", "A6 real-png")

    print("\n===== 上传后静态访问（直连 :3000，验证 nosniff 由 nginx 加，这里看后端 Content-Type）=====")
    for url in filter(None, [ok_url, json.loads(fb_resp)["data"]["url"] if fb_st == 200 else None]):
        st, hdrs, body = http("GET", url)
        print(f"[A7] GET {url} -> status={st} content-type={hdrs.get('content-type')} bytes={len(body)}")

    print("\n===== 目录落盘检查 =====")
    for d in ("/data/pms_dev/api-server/uploads/avatars", "/data/pms_dev/api-server/uploads/feedback"):
        if os.path.isdir(d):
            print(f"{d}: {sorted(os.listdir(d))}")
    return 0


if __name__ == "__main__":
    sys.exit(main())