998211c483
- 从服务器拉取完整代码 - 按框架规范整理项目结构 - 配置 Drone CI 测试环境部署 - 包含后端(FastAPI)、前端(Vue3)、管理端 技术栈: Vue3 + TypeScript + FastAPI + MySQL
73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
"""
|
|
安全相关功能
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Dict, Optional, Union
|
|
|
|
import bcrypt
|
|
from jose import JWTError, jwt
|
|
|
|
from .config import settings
|
|
|
|
|
|
def create_access_token(
|
|
subject: Union[str, Any],
|
|
expires_delta: Optional[timedelta] = None,
|
|
) -> str:
|
|
"""创建访问令牌"""
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(
|
|
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
|
|
)
|
|
|
|
to_encode = {"exp": expire, "sub": str(subject), "type": "access"}
|
|
encoded_jwt = jwt.encode(
|
|
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
|
|
)
|
|
return encoded_jwt
|
|
|
|
|
|
def create_refresh_token(
|
|
subject: Union[str, Any],
|
|
expires_delta: Optional[timedelta] = None,
|
|
) -> str:
|
|
"""创建刷新令牌"""
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
|
|
|
to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"}
|
|
encoded_jwt = jwt.encode(
|
|
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
|
|
)
|
|
return encoded_jwt
|
|
|
|
|
|
def decode_token(token: str) -> Dict[str, Any]:
|
|
"""解码令牌"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
raise ValueError("Invalid token")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""验证密码"""
|
|
return bcrypt.checkpw(
|
|
plain_password.encode("utf-8"), hashed_password.encode("utf-8")
|
|
)
|
|
|
|
|
|
def get_password_hash(password: str) -> str:
|
|
"""生成密码哈希"""
|
|
salt = bcrypt.gensalt()
|
|
hashed_password = bcrypt.hashpw(password.encode("utf-8"), salt)
|
|
return hashed_password.decode("utf-8")
|