363 lines
16 KiB
Python
Executable File
363 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""通过 staging 管理 API 创建或轮换内部狗粮审核账号。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import secrets
|
||
import socket
|
||
import stat
|
||
import string
|
||
import sys
|
||
from typing import Any, Callable
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import urlencode
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
BASE_URL = "http://127.0.0.1:48080/admin-api"
|
||
EXPECTED_HOSTNAME = "mini-desktop"
|
||
EXPECTED_ACK = "ROTATE_DOGFOOD_REVIEWERS"
|
||
DEFAULT_ADMIN_USERNAME = "admin"
|
||
DEFAULT_PUBLIC_ADMIN_PASSWORD = "admin123"
|
||
ADMIN_LOGIN_REJECTION_CODES = frozenset({1_002_000_000, 1_002_000_001})
|
||
REVIEWER_PERMISSIONS = frozenset({
|
||
"project:review:query",
|
||
"project:review:approve",
|
||
"telemetry:game-stat:query",
|
||
})
|
||
READONLY_PERMISSIONS = frozenset({
|
||
"project:review:query",
|
||
"telemetry:game-stat:query",
|
||
})
|
||
ACCOUNT_SPECS = (
|
||
("dogfood_reviewer", "狗粮审核员", "dogfood_reviewer"),
|
||
("dogfood_readonly", "狗粮只读员", "dogfood_readonly"),
|
||
)
|
||
|
||
|
||
class ReviewerProvisionError(RuntimeError):
|
||
"""受控审核账号操作不满足安全前提。"""
|
||
|
||
|
||
def generate_password() -> str:
|
||
"""生成符合管理端 4-16 位约束的高熵随机密码。"""
|
||
alphabet = string.ascii_letters + string.digits
|
||
while True:
|
||
password = "".join(secrets.choice(alphabet) for _ in range(16))
|
||
if (any(char.islower() for char in password)
|
||
and any(char.isupper() for char in password)
|
||
and any(char.isdigit() for char in password)):
|
||
return password
|
||
|
||
|
||
class ApiClient:
|
||
"""固定访问本机 staging 管理 API，Bearer token 只保存在进程内存。"""
|
||
|
||
def __init__(self, admin_token: str, timeout: float = 10.0):
|
||
if not admin_token:
|
||
raise ReviewerProvisionError("管理员 token 不能为空")
|
||
self.admin_token = admin_token
|
||
self.timeout = timeout
|
||
|
||
def _request(self, method: str, path: str, *, query: dict[str, Any] | None = None,
|
||
payload: dict[str, Any] | None = None) -> Any:
|
||
url = BASE_URL + path
|
||
if query:
|
||
url += "?" + urlencode(query)
|
||
body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
request = Request(url, data=body, method=method, headers={
|
||
"Authorization": f"Bearer {self.admin_token}",
|
||
"Content-Type": "application/json",
|
||
})
|
||
try:
|
||
with urlopen(request, timeout=self.timeout) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||
# 不回显响应体，避免认证中间件或代理错误页带出敏感请求信息。
|
||
raise ReviewerProvisionError(f"管理 API 调用失败 method={method} path={path}") from exc
|
||
if result.get("code") != 0:
|
||
raise ReviewerProvisionError(
|
||
f"管理 API 拒绝操作 method={method} path={path} code={result.get('code')}")
|
||
return result.get("data")
|
||
|
||
def verify_rbac(self) -> dict[str, dict[str, Any]]:
|
||
"""校验两个专用角色存在且权限集合精确，作为角色权限验收入口。"""
|
||
roles = self._request("GET", "/system/role/simple-list")
|
||
# simple-list 不返回 permission，必须走受 system:menu:query 保护的完整菜单接口。
|
||
menus = self._request("GET", "/system/menu/list")
|
||
menu_permissions = {item["id"]: item.get("permission") for item in menus}
|
||
expected = {
|
||
"dogfood_reviewer": REVIEWER_PERMISSIONS,
|
||
"dogfood_readonly": READONLY_PERMISSIONS,
|
||
}
|
||
verified: dict[str, dict[str, Any]] = {}
|
||
for role_code, expected_permissions in expected.items():
|
||
matches = [role for role in roles if role.get("code") == role_code]
|
||
if len(matches) != 1:
|
||
raise ReviewerProvisionError(f"角色必须唯一存在 role={role_code} actual={len(matches)}")
|
||
role_id = matches[0]["id"]
|
||
menu_ids = self._request("GET", "/system/permission/list-role-menus",
|
||
query={"roleId": role_id})
|
||
actual_permissions = {menu_permissions.get(menu_id) for menu_id in menu_ids}
|
||
if (len(menu_ids) != len(expected_permissions)
|
||
or None in actual_permissions
|
||
or actual_permissions != set(expected_permissions)):
|
||
raise ReviewerProvisionError(
|
||
f"角色权限不精确 role={role_code} actual={sorted(map(str, actual_permissions))}")
|
||
verified[role_code] = {"id": role_id, "permissions": actual_permissions}
|
||
return verified
|
||
|
||
def find_user(self, username: str) -> dict[str, Any] | None:
|
||
page = self._request("GET", "/system/user/page", query={
|
||
"pageNo": 1,
|
||
"pageSize": 10,
|
||
"username": username,
|
||
})
|
||
matches = [user for user in page["list"] if user.get("username") == username]
|
||
if len(matches) > 1:
|
||
raise ReviewerProvisionError(f"审核账号不唯一 username={username}")
|
||
return matches[0] if matches else None
|
||
|
||
def create_user(self, username: str, nickname: str, password: str) -> int:
|
||
return int(self._request("POST", "/system/user/create", payload={
|
||
"username": username,
|
||
"nickname": nickname,
|
||
"password": password,
|
||
"remark": "staging 内部狗粮专用；账号密码由受控工具轮换",
|
||
"postIds": [],
|
||
"email": "",
|
||
"mobile": "",
|
||
"sex": 0,
|
||
"status": 0,
|
||
}))
|
||
|
||
def update_password(self, user_id: int, password: str) -> None:
|
||
self._request("PUT", "/system/user/update-password",
|
||
payload={"id": user_id, "password": password})
|
||
|
||
def assign_user_roles(self, user_id: int, role_ids: list[int]) -> None:
|
||
self._request("POST", "/system/permission/assign-user-role",
|
||
payload={"userId": user_id, "roleIds": role_ids})
|
||
|
||
def update_user_status(self, user_id: int, status: int) -> None:
|
||
self._request("PUT", "/system/user/update-status",
|
||
payload={"id": user_id, "status": status})
|
||
|
||
def get_user_role_ids(self, user_id: int) -> set[int]:
|
||
role_ids = self._request("GET", "/system/permission/list-user-roles",
|
||
query={"userId": user_id})
|
||
return {int(role_id) for role_id in role_ids}
|
||
|
||
def verify_password_login_rejected(self, username: str, password: str) -> int:
|
||
"""匿名验证公开默认密码已失效，只接受密码错误或账号禁用两个确定结果。"""
|
||
body = json.dumps({"username": username, "password": password}).encode("utf-8")
|
||
request = Request(BASE_URL + "/system/auth/login", data=body, method="POST", headers={
|
||
"Content-Type": "application/json",
|
||
})
|
||
try:
|
||
with urlopen(request, timeout=self.timeout) as response:
|
||
result = json.loads(response.read().decode("utf-8"))
|
||
except HTTPError as exc:
|
||
try:
|
||
result = json.loads(exc.read().decode("utf-8"))
|
||
except json.JSONDecodeError as decode_error:
|
||
raise ReviewerProvisionError("默认超管负例登录返回非 JSON") from decode_error
|
||
except (URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||
raise ReviewerProvisionError("默认超管负例登录调用失败") from exc
|
||
code = result.get("code")
|
||
if code not in ADMIN_LOGIN_REJECTION_CODES:
|
||
raise ReviewerProvisionError(f"默认公开超管密码未被确定拒绝 code={code}")
|
||
return int(code)
|
||
|
||
|
||
def rotate_reviewers(api: ApiClient, password_factory: Callable[[], str],
|
||
on_password: Callable[[str, str], None] | None = None) -> dict[str, Any]:
|
||
"""轮换审核账号并封禁默认公开超管；任一步失败均不签发通过证据。"""
|
||
roles = api.verify_rbac()
|
||
accounts: dict[str, str] = {}
|
||
evidence_accounts: list[dict[str, Any]] = []
|
||
for username, nickname, role_code in ACCOUNT_SPECS:
|
||
password = password_factory()
|
||
accounts[username] = password
|
||
if on_password is not None:
|
||
# 先持久化待应用密码，防止进程中断后已轮换账号的凭据永久丢失。
|
||
on_password(username, password)
|
||
user = api.find_user(username)
|
||
user_id = api.create_user(username, nickname, password) if user is None else int(user["id"])
|
||
|
||
# 禁用动作由后端同步删除该管理员的 MySQL 与 Redis OAuth2 token；失败时保持禁用。
|
||
api.update_user_status(user_id, 1)
|
||
if user is not None:
|
||
api.update_password(user_id, password)
|
||
role_id = int(roles[role_code]["id"])
|
||
api.assign_user_roles(user_id, [role_id])
|
||
if api.get_user_role_ids(user_id) != {role_id}:
|
||
raise ReviewerProvisionError(f"账号角色未收敛 username={username}")
|
||
api.update_user_status(user_id, 0)
|
||
evidence_accounts.append({
|
||
"username": username,
|
||
"userId": user_id,
|
||
"roleCode": role_code,
|
||
"roleId": role_id,
|
||
"permissions": sorted(roles[role_code]["permissions"]),
|
||
"oldTokensInvalidatedByStatusCycle": True,
|
||
"status": "ENABLED",
|
||
})
|
||
|
||
# 默认公开超管必须最后处理：禁用会立即使当前及历史 token 失效，之后不能再依赖该 token 调特权接口。
|
||
admin_password = password_factory()
|
||
accounts[DEFAULT_ADMIN_USERNAME] = admin_password
|
||
if on_password is not None:
|
||
on_password(DEFAULT_ADMIN_USERNAME, admin_password)
|
||
admin_user = api.find_user(DEFAULT_ADMIN_USERNAME)
|
||
if admin_user is None:
|
||
raise ReviewerProvisionError("staging 默认 admin 账号不存在，无法完成公开凭据收口")
|
||
admin_user_id = int(admin_user["id"])
|
||
api.update_password(admin_user_id, admin_password)
|
||
api.update_user_status(admin_user_id, 1)
|
||
rejection_code = api.verify_password_login_rejected(
|
||
DEFAULT_ADMIN_USERNAME, DEFAULT_PUBLIC_ADMIN_PASSWORD)
|
||
evidence_accounts.append({
|
||
"username": DEFAULT_ADMIN_USERNAME,
|
||
"userId": admin_user_id,
|
||
"accountType": "DEFAULT_BOOTSTRAP_ADMIN",
|
||
"passwordRotated": True,
|
||
"oldTokensInvalidatedByDisable": True,
|
||
"defaultPasswordLoginCode": rejection_code,
|
||
"status": "DISABLED",
|
||
})
|
||
return {"accounts": accounts, "evidenceAccounts": evidence_accounts}
|
||
|
||
|
||
def build_rbac_evidence(roles: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""把权限验收结果转换为不含 token、密码的稳定证据结构。"""
|
||
return [{
|
||
"roleCode": role_code,
|
||
"roleId": int(role["id"]),
|
||
"permissions": sorted(role["permissions"]),
|
||
} for role_code, role in sorted(roles.items())]
|
||
|
||
|
||
def require_private_file(path: Path) -> str:
|
||
"""读取 0400/0600 普通文件，拒绝符号链接。"""
|
||
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
||
try:
|
||
descriptor = os.open(path, flags)
|
||
except OSError as exc:
|
||
raise ReviewerProvisionError(f"敏感输入无法安全打开 path={path}") from exc
|
||
with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
|
||
file_stat = os.fstat(handle.fileno())
|
||
if not stat.S_ISREG(file_stat.st_mode):
|
||
raise ReviewerProvisionError(f"敏感输入必须是普通文件 path={path}")
|
||
if stat.S_IMODE(file_stat.st_mode) not in (0o400, 0o600):
|
||
raise ReviewerProvisionError(f"敏感输入权限必须是 0400 或 0600 path={path}")
|
||
value = handle.read().strip()
|
||
if not value:
|
||
raise ReviewerProvisionError(f"敏感输入不能为空 path={path}")
|
||
return value
|
||
|
||
|
||
def reserve_private_file(path: Path) -> None:
|
||
"""以 O_EXCL 预留 0600 输出，拒绝覆盖旧证据或跟随符号链接。"""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
|
||
descriptor = os.open(path, flags, 0o600)
|
||
os.close(descriptor)
|
||
|
||
|
||
def write_private_json(path: Path, payload: dict[str, Any]) -> None:
|
||
"""只重写本次预留的 0600 普通文件。"""
|
||
flags = os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0)
|
||
try:
|
||
descriptor = os.open(path, flags)
|
||
except OSError as exc:
|
||
raise ReviewerProvisionError(f"输出无法安全打开 path={path}") from exc
|
||
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
|
||
file_stat = os.fstat(handle.fileno())
|
||
if not stat.S_ISREG(file_stat.st_mode):
|
||
raise ReviewerProvisionError(f"输出不再是普通文件 path={path}")
|
||
if stat.S_IMODE(file_stat.st_mode) != 0o600:
|
||
raise ReviewerProvisionError(f"输出权限必须保持 0600 path={path}")
|
||
# 通过文件描述符复核类型和权限后再截断，避免检查前破坏被替换的目标。
|
||
os.ftruncate(handle.fileno(), 0)
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
handle.write("\n")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="受控创建或轮换 staging 狗粮审核账号")
|
||
parser.add_argument("--ack", required=True, help=f"必须精确填写 {EXPECTED_ACK}")
|
||
parser.add_argument("--admin-token-file", type=Path, required=True)
|
||
parser.add_argument("--credentials-out", type=Path)
|
||
parser.add_argument("--evidence-out", type=Path, required=True)
|
||
parser.add_argument("--verify-only", action="store_true", help="只校验角色权限并输出脱敏证据")
|
||
args = parser.parse_args()
|
||
if args.ack != EXPECTED_ACK:
|
||
raise ReviewerProvisionError(f"ack 必须精确设为 {EXPECTED_ACK}")
|
||
hostname = socket.gethostname().split(".", 1)[0]
|
||
if hostname != EXPECTED_HOSTNAME:
|
||
raise ReviewerProvisionError(f"只允许在 {EXPECTED_HOSTNAME} 执行 actual={hostname}")
|
||
if not args.verify_only and args.credentials_out is None:
|
||
raise ReviewerProvisionError("轮换模式必须提供 --credentials-out")
|
||
if args.credentials_out is not None and args.credentials_out == args.evidence_out:
|
||
raise ReviewerProvisionError("凭据与证据必须写入不同文件")
|
||
|
||
admin_token = require_private_file(args.admin_token_file)
|
||
api = ApiClient(admin_token)
|
||
if args.verify_only:
|
||
reserve_private_file(args.evidence_out)
|
||
roles = api.verify_rbac()
|
||
write_private_json(args.evidence_out, {
|
||
"status": "PASS",
|
||
"baseUrl": BASE_URL,
|
||
"roles": build_rbac_evidence(roles),
|
||
})
|
||
print(f"DOGFOOD_REVIEWER_RBAC_PASS roles={len(roles)} evidence={args.evidence_out}")
|
||
return 0
|
||
|
||
credentials_out = args.credentials_out
|
||
if credentials_out is None:
|
||
raise ReviewerProvisionError("轮换模式缺少凭据输出路径")
|
||
reserve_private_file(credentials_out)
|
||
try:
|
||
reserve_private_file(args.evidence_out)
|
||
except Exception:
|
||
credentials_out.unlink(missing_ok=True)
|
||
raise
|
||
|
||
generated_accounts: dict[str, str] = {}
|
||
|
||
def persist_pending(username: str, password: str) -> None:
|
||
generated_accounts[username] = password
|
||
write_private_json(credentials_out, {
|
||
"status": "PENDING",
|
||
"baseUrl": BASE_URL,
|
||
"accounts": generated_accounts,
|
||
})
|
||
|
||
result = rotate_reviewers(api, generate_password, persist_pending)
|
||
write_private_json(credentials_out, {
|
||
"status": "COMPLETE",
|
||
"baseUrl": BASE_URL,
|
||
"accounts": result["accounts"],
|
||
})
|
||
write_private_json(args.evidence_out, {
|
||
"status": "PASS",
|
||
"baseUrl": BASE_URL,
|
||
"accounts": result["evidenceAccounts"],
|
||
})
|
||
print(f"DOGFOOD_REVIEWER_PASS accounts={len(result['evidenceAccounts'])} evidence={args.evidence_out}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
raise SystemExit(main())
|
||
except (ReviewerProvisionError, FileExistsError, FileNotFoundError, PermissionError) as exc:
|
||
print(f"DOGFOOD_REVIEWER_FAIL {exc}", file=sys.stderr)
|
||
raise SystemExit(1)
|