chore(server): 加重置后台 admin 口令的脚本（含自检与端到端验证）

yeg

2026-09-29 eddb360f199e6930b4abede276c04dec3b111d5e

chore(server): 加重置后台 admin 口令的脚本（含自检与端到端验证）

算法与应用完全一致（PBKDF2WithHmacSHA256/210000 轮/16 字节随机盐）。明文只写 /root/admin-password.txt(600)、不打印到 stdout；输出只有字符类别摘要、哈希自检、以及真调一次登录接口的成败（不打印 token）。顺带清零失败计数与锁定、并吊销所有未过期会话 —— 那次的告警报文里带出过 accessToken。

 .scratch/server-testenv/reset-admin-password.py

New file
@@ -0,0 +1,131 @@
#!/usr/bin/env python3
# 重置后台平台管理员 admin 的口令。
#
# 算法与应用完全一致（见 MerchantPasswordHasher）：
#   pbkdf2$210000$<base64 salt(16B)>$<base64 PBKDF2WithHmacSHA256(password, salt, 210000, 256bit)>
#
# 明文只写到 /root/admin-password.txt（600），**不打印到 stdout**。
# 输出只有：长度/字符类别的摘要 + 自检结果 + 端到端验证结果。
#
# 顺带做两件卫生工作：
#   * login_failed_count 清零、locked_until 置 NULL（等价于应用里的 resetStaffPassword）
#   * 吊销所有未过期的 admin_session —— 服务商那次告警的报文里带出过 accessToken，
#     换口令时把在用的会话一并作废，避免旧 token 继续可用；端到端验证自己产生的那个也会被吊销。
#
# 用法：python3 /root/reset-admin-password.py

import base64
import hashlib
import json
import os
import secrets
import subprocess
import sys
import urllib.request

LENGTH = 20
# 不含单引号/双引号/反斜杠/空格/`$`：既能满足服务商要求的四类组合，
# 又不会破坏 SQL 字符串字面量与 shell 引用。
CHARS = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789@%^*+_.-"
SPECIALS = "@%^*+_.-"
OUT_PATH = "/root/admin-password.txt"
LOGIN_URL = "http://127.0.0.1:8080/api/admin/auth/login"


def generate():
    while True:
        value = "".join(secrets.choice(CHARS) for _ in range(LENGTH))
        if (any(c.isupper() for c in value)
                and any(c.islower() for c in value)
                and any(c.isdigit() for c in value)
                and any(c in SPECIALS for c in value)):
            return value


def derive(password, salt, iterations):
    return hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations, dklen=32)


def matches(password, encoded):
    parts = encoded.split("$")
    if len(parts) != 4 or parts[0] != "pbkdf2":
        return False
    iterations = int(parts[1])
    salt = base64.b64decode(parts[2])
    expected = base64.b64decode(parts[3])
    return derive(password, salt, iterations) == expected


def mysql(sql):
    result = subprocess.run(["mysql", "-N", "-B", "-D", "talent_card", "-e", sql],
                            capture_output=True, text=True)
    if result.returncode != 0:
        print("mysql 执行失败: " + result.stderr.strip())
        sys.exit(1)
    return result.stdout.strip()


def revoke_sessions():
    mysql("update admin_session set revoked_at = now() where revoked_at is null;")


def login_works(password):
    """拿新口令真调一次登录接口。只回布尔值，绝不打印 token。"""
    request = urllib.request.Request(
        LOGIN_URL,
        data=json.dumps({"account": "admin", "password": password}).encode(),
        headers={"Content-Type": "application/json"},
        method="POST")
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            payload = json.loads(response.read().decode())
            token = (payload.get("data") or {}).get("accessToken")
            return response.status == 200 and bool(token)
    except Exception as exception:
        print("    登录接口返回异常: " + type(exception).__name__ + " " + str(exception))
        return False


def main():
    password = generate()

    salt = os.urandom(16)
    encoded = ("pbkdf2$210000$"
               + base64.b64encode(salt).decode()
               + "$"
               + base64.b64encode(derive(password, salt, 210000)).decode())

    # 口令取自上面的字符集（无引号），所以内联进 SQL 字面量是安全的
    mysql("update admin_operator set password_hash = '" + encoded
          + "', login_failed_count = 0, locked_until = null, updated_at = now()"
          + " where account = 'admin' and deleted = 0;")
    revoke_sessions()

    stored = mysql("select password_hash from admin_operator where account = 'admin' and deleted = 0")
    if not stored:
        print("!!! 没读到 admin 的哈希，中止")
        sys.exit(1)

    fd = os.open(OUT_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as handle:
        handle.write("account: admin\npassword: " + password + "\n")
    os.chmod(OUT_PATH, 0o600)

    print("已重置 admin 的口令。")
    print("  明文写入 " + OUT_PATH + "（权限 600），未打印到 stdout。")
    print("  长度 " + str(len(password)) + "；含大写/小写/数字/符号四类。")
    print("  失败计数与锁定已清零；未过期的 admin_session 已全部吊销。")
    print("  自检（用库里存的哈希重新算一遍该口令）: "
          + ("匹配 ✓" if matches(password, stored) else "不匹配 ✗"))

    passed = login_works(password)
    print("  端到端（真调一次登录接口，只看成功与否）: " + ("成功 ✓" if passed else "失败 ✗"))
    revoke_sessions()  # 把这次验证产生的会话也作废

    remaining = mysql("select count(*) from admin_session where revoked_at is null")
    print("  吊销后仍有效的 admin_session 数: " + remaining)
    return 0 if passed else 1


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