129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""设置用户登录密码（bcrypt 哈希写入 `sys_user.password_hash`）。
|
||
|
||
## ⚠️ 仅限演示环境
|
||
|
||
本脚本把演示口令写在源码里、也允许明文命令行传入，目的是让演示与联调**当天可用**。
|
||
`123456` / `666666` / `88888888` 这类弱口令**在生产环境等于没有密码**：
|
||
上线前必须全部更换，并由运维走单独的改密流程（本脚本只服务演示）。
|
||
|
||
## 为什么需要它
|
||
|
||
`sys_user.password_hash` 此前**全是占位符** —— 种子写 `'x'`、worker 身份写
|
||
`!worker-only-no-password-login!`，即"这个字段从来没过真实密码"。登录接口上线后，
|
||
不设密码就没人能登进来；这个脚本补的正是这一步。
|
||
|
||
## 用法
|
||
|
||
python tools/set_user_password.py --list # 只列现状，不改任何数据
|
||
python tools/set_user_password.py # 按内置演示规则设置
|
||
python tools/set_user_password.py --user 9002 --password 'xxx'
|
||
|
||
注意：bcrypt 每次加盐不同，**重复执行等于重设密码**（不是"已存在就跳过"）。这是有意的
|
||
——改密本来就该覆盖，但要清楚它不是幂等操作。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
from sqlalchemy import text
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from app.infrastructure.db import SessionFactory # noqa: E402
|
||
from app.service.auth_service import hash_password # noqa: E402
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr]
|
||
|
||
#: 演示口令（用户指定）。键是 `sys_user.id`。
|
||
#: 9001 客户 / 9002 员工（风控专员）/ 9003 管理员。
|
||
DEMO_PASSWORDS: dict[str, str] = {
|
||
"9001": "123456",
|
||
"9002": "666666",
|
||
"9003": "88888888",
|
||
}
|
||
|
||
#: 与 `AuthService.verify_password` 保持一致的识别方式：只有 bcrypt 格式才算"已设真密码"。
|
||
BCRYPT_PREFIXES = ("$2a$", "$2b$", "$2y$")
|
||
|
||
|
||
def _is_real_hash(value: str | None) -> bool:
|
||
return bool(value) and str(value).startswith(BCRYPT_PREFIXES)
|
||
|
||
|
||
async def list_users() -> None:
|
||
async with SessionFactory() as session:
|
||
rows = (
|
||
await session.execute(
|
||
text(
|
||
"SELECT id, username, user_type, status, password_hash "
|
||
"FROM sys_user ORDER BY id"
|
||
)
|
||
)
|
||
).mappings().all()
|
||
print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}密码状态")
|
||
for row in rows:
|
||
state = "已设（bcrypt）" if _is_real_hash(row["password_hash"]) else "占位符，无法登录"
|
||
print(
|
||
f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}"
|
||
f"{str(row['status']):<8}{state}"
|
||
)
|
||
|
||
|
||
async def set_password(user_id: str, password: str) -> int:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
async with SessionFactory() as session, session.begin():
|
||
result = await session.execute(
|
||
text(
|
||
"UPDATE sys_user SET password_hash = :hash, updated_at = :now "
|
||
"WHERE id = :user_id"
|
||
),
|
||
{"hash": hash_password(password), "now": now, "user_id": int(user_id)},
|
||
)
|
||
if result.rowcount == 0:
|
||
print(f"[失败] sys_user 里没有 id={user_id} 的用户")
|
||
return 1
|
||
print(f"[OK] id={user_id} 密码已设置（bcrypt）")
|
||
return 0
|
||
|
||
|
||
async def main() -> int:
|
||
parser = argparse.ArgumentParser(description="设置用户登录密码（bcrypt）")
|
||
parser.add_argument("--list", action="store_true", help="只列现状，不改数据")
|
||
parser.add_argument("--user", help="单个用户 id（配合 --password 使用）")
|
||
parser.add_argument("--password", help="要设置的明文密码")
|
||
args = parser.parse_args()
|
||
|
||
if args.list:
|
||
await list_users()
|
||
return 0
|
||
|
||
if args.user or args.password:
|
||
if not (args.user and args.password):
|
||
print("[失败] --user 与 --password 必须成对给出")
|
||
return 1
|
||
return await set_password(args.user, args.password)
|
||
|
||
print("按内置演示规则设置密码（生产环境必须更换）：")
|
||
failures = 0
|
||
for user_id, password in DEMO_PASSWORDS.items():
|
||
print(f" · id={user_id} → {len(password)} 位口令")
|
||
failures += await set_password(user_id, password)
|
||
print("\n设置后的现状：")
|
||
await list_users()
|
||
if failures:
|
||
return 1
|
||
print("\n可以登录了。接口：POST /api/v1/auth/tokens {\"username\": \"<username>\", \"password\": \"...\"}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(asyncio.run(main()))
|