# Copyright 2026 zhaoxi826 # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import annotations import os from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING, Annotated, Optional import jwt from fastapi import Depends, HTTPException, Request, status from pydantic import BaseModel, ValidationError from pwdlib import PasswordHash if TYPE_CHECKING: from kilostar.core.postgres_database.model import UserAuthority from kilostar.core.postgres_database.model.user import User ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 2 REFRESH_TOKEN_EXPIRE_DAYS = 7 _INSECURE_SECRETS = {"secret", "114514", "changethiskey12345"} class TokenData(BaseModel): """JWT 解码后的用户身份载荷。""" user_id: str username: Optional[str] = None exp: Optional[int] = None def _get_secret_key() -> str: from kilostar.utils.settings import get_settings key = get_settings().security.secret_key if not key or key in _INSECURE_SECRETS: raise RuntimeError( "未提供有效的 SECRET_KEY 或使用了不安全的默认值，请设置一个高熵的随机字符串" ) return key password_hasher = PasswordHash.recommended() class Accessor: """封装认证与口令哈希相关的静态工具方法。""" @staticmethod def _decode_token(token: str) -> TokenData: """解码并校验 JWT，返回 TokenData；过期或无效时抛 401。""" try: payload = jwt.decode(token, _get_secret_key(), algorithms=[ALGORITHM]) return TokenData(**payload) except jwt.ExpiredSignatureError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Token 已过期", ) except (jwt.InvalidTokenError, ValidationError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭证", ) @staticmethod def _create_access_token(data: dict) -> str: """根据 payload 生成带过期时间的 JWT 访问令牌。""" to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta( minutes=ACCESS_TOKEN_EXPIRE_MINUTES ) to_encode.update({"exp": int(expire.timestamp()), "type": "access"}) return jwt.encode(to_encode, _get_secret_key(), algorithm=ALGORITHM) @staticmethod def _create_refresh_token(data: dict) -> str: """生成长效 refresh token（默认 7 天有效期）。""" to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta( days=REFRESH_TOKEN_EXPIRE_DAYS ) to_encode.update({"exp": int(expire.timestamp()), "type": "refresh"}) return jwt.encode(to_encode, _get_secret_key(), algorithm=ALGORITHM) @staticmethod def verify_refresh_token(token: str) -> TokenData: """校验 refresh token 有效性并返回用户身份；过期或类型不对抛 401。""" try: payload = jwt.decode(token, _get_secret_key(), algorithms=[ALGORITHM]) if payload.get("type") != "refresh": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的 refresh token", ) return TokenData(**{k: v for k, v in payload.items() if k != "type"}) except jwt.ExpiredSignatureError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Refresh token 已过期，请重新登录", ) except (jwt.InvalidTokenError, ValidationError): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的 refresh token", ) @staticmethod def refresh_access_token(refresh_token: str) -> dict: """用 refresh token 换取新的 access token + refresh token 对。""" token_data = Accessor.verify_refresh_token(refresh_token) payload = {"user_id": token_data.user_id, "username": token_data.username} return { "access_token": Accessor._create_access_token(payload), "refresh_token": Accessor._create_refresh_token(payload), } @staticmethod def verify_password(plain_password: str, hashed_password: str) -> bool: """校验明文口令是否匹配数据库中存储的哈希。""" return password_hasher.verify(plain_password, hashed_password) @staticmethod def get_current_user(request: Request) -> TokenData: """从 Authorization Bearer 头解析当前请求的用户身份。""" auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="未提供认证头部", ) token = auth_header.split(" ")[1] return Accessor._decode_token(token) @staticmethod def login_hashed_password(user: "User", password: str) -> dict: """完成登录核验：找不到用户或密码错误抛 401，否则签发 access + refresh 令牌对。""" if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在", ) if not Accessor.verify_password(password, user.hashed_password): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误", ) token_payload = {"user_id": str(user.user_id), "username": user.user_name} return { "access_token": Accessor._create_access_token(data=token_payload), "refresh_token": Accessor._create_refresh_token(data=token_payload), } @staticmethod def hash_password(password: str) -> str: """对明文口令做强哈希；空值或不满足复杂度要求会抛 ValueError。""" if not password: raise ValueError("密码不能为空") if len(password) < 6: raise ValueError("密码长度不能小于 6 位") has_alpha = any(c.isalpha() for c in password) has_digit = any(c.isdigit() for c in password) if not (has_alpha and has_digit): raise ValueError("密码必须同时包含字母和数字") return password_hasher.hash(password) # ─── Role Check ────────────────────────────────────────────────────────────── def _user_not_found_detail(request: Request | None = None) -> str: from kilostar.utils.i18n import t loc = request.headers.get("accept-language") if request else None return t("user_not_found", accept_language=loc) async def get_authority(user_id: str) -> "UserAuthority": """通过 PostgresDatabase Actor 查出指定用户的 ``UserAuthority``；用户不存在时抛 401。""" from kilostar.utils.error import UserNotExistError from kilostar.utils.i18n import t from kilostar.utils.ray_hook import ray_actor_hook postgres_database = ray_actor_hook("postgres_database").postgres_database try: user_authority = await postgres_database.get_user_authority.remote( user_id=user_id ) return user_authority except UserNotExistError: raise HTTPException(status_code=401, detail=t("user_not_found")) except Exception as e: if "UserNotExistError" in str(e): raise HTTPException( status_code=401, detail=t("user_not_found") ) raise class RoleChecker: """FastAPI 依赖：在路由级别按 ``UserAuthority`` 做最低权限校验。 例：``Depends(RoleChecker(allowed_roles=UserAuthority.ADMINISTRATOR))``。 """ def __init__(self, **kwargs): self.allowed_roles = kwargs.get( "allowed_roles", ) async def __call__( self, token_data: Annotated[TokenData, Depends(Accessor.get_current_user)] ): """对当前请求执行权限比较，权限不足抛 403，否则把 ``TokenData`` 透传给路由。""" user_authority = await get_authority(token_data.user_id) if user_authority < self.allowed_roles: raise HTTPException( status_code=403, detail={ "message": f"User {token_data.user_id} does not have allowed roles" }, ) return token_data