"""校验 serve.py 手写的 SigV4 与 aws-sdk-go-v2 的 signer 逐字节一致。

为什么需要这个：SigV4 错一个字节，服务端只会回一句 SignatureDoesNotMatch，
既不说哪里错、也不会在本地暴露出来 —— 上传功能就成了「点了没反应」。
这里用官方 SDK 当参照实现做对拍，改完签名相关代码跑一遍即可。

用法（项目根）：
    python dev/theme-builder/check_sigv4.py

⚠️ 用例只放 ASCII 键：aws-sdk-go-v2 裸 SignHTTP 默认会二次转义 URI，
而 S3 客户端内部把 DisableURIPathEscaping 设成了 true（单次编码）。
即同一个路径两种行为，把非 ASCII 键放进来只会比出「SDK 的两个模式之差」。
我们跟随 S3 客户端的单次编码（AWS 文档的 RFC3986 规则），并由生成器把
slug 限制在 [a-z0-9-_]，从根上避开这个争议区。
"""
import os
import subprocess
import sys
import urllib.parse

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(os.path.dirname(HERE))
sys.path.insert(0, HERE)

import serve  # noqa: E402

KEY_ID = "AKIAIOSFODNN7EXAMPLE"
SECRET = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
TS = "20130524T000000Z"
EMPTY = serve.EMPTY_SHA256
BODY_HASH = serve._sha256(b"TamaBox theme builder probe\n")

# (说明, method, endpoint, extra cfg, key, query)
CASES = [
    ("path-style 桶级 HEAD", "HEAD", "http://127.0.0.1:9000", dict(bucket="tb-assets"), "", None),
    ("path-style 带前缀对象", "PUT", "http://127.0.0.1:9000",
     dict(bucket="tb-assets", prefix="themes"), "shiroko/bg1.webp", None),
    ("virtual-host 对象", "GET", "https://s3.us-west-004.backblazeb2.com",
     dict(bucket="my-bucket", region="us-west-004", pathStyle=False), "themes/shiroko/bg1.webp", None),
    ("R2 region=auto", "PUT", "https://acc.r2.cloudflarestorage.com",
     dict(bucket="cdn", region="auto", prefix="themes"), "x/a.webp", None),
    ("R2 虚拟主机", "GET", "https://acc.r2.cloudflarestorage.com",
     dict(bucket="cdn", region="auto", pathStyle=False), "themes/s/t.webp", None),
    ("带端口 https", "DELETE", "https://minio.lan:9443",
     dict(bucket="static", region="us-east-1"), "themes/s/t.webp", None),
    # 带 query：ListObjectsV2 的 canonical query 要按键排序、逐项编码（/ 编成 %2F）
    ("ListObjectsV2 三参数", "GET", "http://127.0.0.1:9000", dict(bucket="tb-assets"),
     "", {"list-type": "2", "max-keys": "1000", "prefix": "themes/shiroko/"}),
    ("ListObjectsV2 分页（乱序传入）", "GET", "https://acc.r2.cloudflarestorage.com",
     dict(bucket="cdn", region="auto", pathStyle=False), "",
     {"prefix": "themes/", "continuation-token": "1/abc+d", "max-keys": "2", "list-type": "2"}),
]


def build(endpoint, extra, key):
    cfg = serve.S3Config(dict(extra, endpoint=endpoint, accessKey=KEY_ID, secretKey=SECRET))
    if key:
        path = cfg.object_path(key)
    else:
        path = ("/" + cfg.bucket) if cfg.path_style else "/"
    return cfg, path


def go_sign(payloads):
    env = dict(os.environ, AWS_KEY_ID=KEY_ID, AWS_SECRET=SECRET,
               GOPROXY="https://goproxy.cn,direct", GOSUMDB="off", GOFLAGS="-mod=mod")
    p = subprocess.run(["go", "run", "./dev/sigv4ref"], input="\n".join(payloads),
                       capture_output=True, text=True, env=env, cwd=ROOT)
    if p.returncode != 0:
        sys.stderr.write(p.stderr[-2000:])
        raise SystemExit("go run 失败（检查 GOPROXY / Go 是否可用）")
    return [x for x in p.stdout.strip().splitlines() if x.strip()]


def main():
    payloads = []
    for _, method, endpoint, extra, key, query in CASES:
        cfg, path = build(endpoint, extra, key)
        body = BODY_HASH if method == "PUT" else EMPTY
        url = cfg.url(path) + (("?" + serve.canonical_query(query)) if query else "")
        payloads.append("\t".join([method, url, cfg.region, body, TS]))

    go = go_sign(payloads)
    if len(go) != len(CASES):
        raise SystemExit("返回条数不对：go=%d cases=%d" % (len(go), len(CASES)))

    bad = 0
    for (desc, method, endpoint, extra, key, query), g in zip(CASES, go):
        cfg, path = build(endpoint, extra, key)
        body = BODY_HASH if method == "PUT" else EMPTY
        py = serve.sign_v4(cfg, method, path, body, TS, cfg.host, serve.canonical_query(query))
        if g.rsplit("Signature=", 1)[-1] == py.rsplit("Signature=", 1)[-1]:
            print("OK   %s" % desc)
        else:
            bad += 1
            print("DIFF %s\n  go  %s\n  py  %s\n  path %s" % (desc, g, py, path))

    print("\n%d/%d 一致" % (len(CASES) - bad, len(CASES)))
    return 1 if bad else 0


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