用户系统

art-agent/backend/app/auth.py

@@ -0,0 +1,89 @@
+"""
+认证工具：密码哈希 + JWT 令牌 + FastAPI 依赖注入。
+"""
+
+import os
+from datetime import datetime, timedelta, timezone
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from pwdlib import PasswordHash
+from pwdlib.hashers.argon2 import Argon2Hasher
+from pwdlib.hashers.bcrypt import BcryptHasher
+from sqlmodel import Session, select
+
+from app.db import User, get_session
+
+# Argon2 优先用于新密码，BcryptHasher 兼容旧版已有哈希
+pwd_hash = PasswordHash((Argon2Hasher(), BcryptHasher()))
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
+
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+REFRESH_TOKEN_EXPIRE_DAYS = 7
+
+
+def _get_secret() -> str:
+    secret = os.getenv("JWT_SECRET", "")
+    if not secret:
+        raise RuntimeError("JWT_SECRET 环境变量未设置")
+    return secret
+
+
+def hash_password(plain: str) -> str:
+    return pwd_hash.hash(plain)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    return pwd_hash.verify(plain, hashed)
+
+
+def create_access_token(user_id: str) -> str:
+    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    return jwt.encode(
+        {"sub": user_id, "exp": expire, "type": "access"},
+        _get_secret(),
+        algorithm=ALGORITHM,
+    )
+
+
+def create_refresh_token(user_id: str) -> str:
+    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+    return jwt.encode(
+        {"sub": user_id, "exp": expire, "type": "refresh"},
+        _get_secret(),
+        algorithm=ALGORITHM,
+    )
+
+
+def decode_token(token: str, expected_type: str = "access") -> str:
+    """解码 JWT，返回 user_id。无效时抛 HTTPException 401。"""
+    try:
+        payload = jwt.decode(token, _get_secret(), algorithms=[ALGORITHM])
+        user_id: str = payload.get("sub", "")
+        token_type: str = payload.get("type", "")
+        if not user_id or token_type != expected_type:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效令牌")
+        return user_id
+    except JWTError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="令牌已过期或无效")
+
+
+def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    session: Session = Depends(get_session),
+) -> User:
+    """FastAPI 依赖：从 Bearer token 解析当前用户。"""
+    user_id = decode_token(token, "access")
+    user = session.exec(select(User).where(User.id == user_id)).first()
+    if not user or not user.is_active:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在或已禁用")
+    return user
+
+
+def require_admin(user: User = Depends(get_current_user)) -> User:
+    """FastAPI 依赖：要求当前用户是管理员。"""
+    if not user.is_admin:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="需要管理员权限")
+    return user