dekun
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
"""注册、登录，以及首次启动时创建管理员。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.clock import utcnow
|
|
from app.config import settings
|
|
from app.errors import AppError
|
|
from app.models import User, UserRole
|
|
from app.security import hash_password, verify_password
|
|
from app.services.registry import register_user
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def register_account(
|
|
session: Session,
|
|
wallet_address: str,
|
|
username: str | None,
|
|
password: str | None,
|
|
) -> User:
|
|
name = _clean_username(username)
|
|
secret = (password or "").strip()
|
|
if name and not secret:
|
|
raise AppError("设置用户名时必须填写密码")
|
|
if secret and not name:
|
|
raise AppError("设置密码时必须填写用户名")
|
|
if secret and len(secret) < 6:
|
|
raise AppError("密码至少 6 位")
|
|
return register_user(
|
|
session,
|
|
wallet_address,
|
|
username=name,
|
|
password_hash=hash_password(secret) if secret else None,
|
|
)
|
|
|
|
|
|
def authenticate(
|
|
session: Session,
|
|
wallet_address: str | None,
|
|
username: str | None,
|
|
password: str | None,
|
|
) -> User:
|
|
name = _clean_username(username)
|
|
wallet = (wallet_address or "").strip()
|
|
secret = (password or "").strip()
|
|
if name and secret:
|
|
user = session.scalar(select(User).where(User.username == name))
|
|
if user is None or not verify_password(secret, user.password_hash):
|
|
raise AppError("账号或密码错误", 401)
|
|
return user
|
|
if wallet and secret:
|
|
user = session.scalar(select(User).where(User.wallet_address == wallet))
|
|
if user is None or not verify_password(secret, user.password_hash):
|
|
raise AppError("账号或密码错误", 401)
|
|
return user
|
|
if wallet:
|
|
user = session.scalar(select(User).where(User.wallet_address == wallet))
|
|
if user is None:
|
|
raise AppError("钱包未注册", 401)
|
|
if user.role == UserRole.ADMIN:
|
|
raise AppError("管理员请使用账号密码登录", 403)
|
|
return user
|
|
raise AppError("请填写钱包地址，或账号和密码")
|
|
|
|
|
|
def ensure_admin(session: Session) -> None:
|
|
username = _clean_username(settings.admin_username) or "admin"
|
|
existing = session.scalar(select(User).where(User.username == username))
|
|
if existing is not None:
|
|
if existing.role != UserRole.ADMIN:
|
|
existing.role = UserRole.ADMIN
|
|
session.flush()
|
|
return
|
|
generated = not settings.admin_password
|
|
password = settings.admin_password or secrets.token_urlsafe(12)
|
|
wallet = settings.admin_wallet.strip() or "admin-root"
|
|
if session.scalar(select(User).where(User.wallet_address == wallet)) is not None:
|
|
wallet = f"{wallet}-{secrets.token_hex(3)}"
|
|
session.add(
|
|
User(
|
|
wallet_address=wallet,
|
|
username=username,
|
|
password_hash=hash_password(password),
|
|
role=UserRole.ADMIN,
|
|
created_at=utcnow(),
|
|
)
|
|
)
|
|
session.flush()
|
|
if generated:
|
|
logger.warning("已创建管理员 %s，初始密码：%s", username, password)
|
|
|
|
|
|
def _clean_username(username: str | None) -> str | None:
|
|
if username is None:
|
|
return None
|
|
cleaned = username.strip().lower()
|
|
return cleaned or None
|