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