from datetime import datetime, timedelta from typing import Optional, Union from jose import JWTError, jwt import bcrypt from app.core.config import settings from app.core.token_blacklist import token_blacklist def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """创建访问令牌""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.JWT_EXPIRE_MINUTES) to_encode.update({"exp": expire, "type": "access"}) encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) return encoded_jwt def create_refresh_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: """创建刷新令牌""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(days=settings.JWT_REFRESH_EXPIRE_DAYS) to_encode.update({"exp": expire, "type": "refresh"}) encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM) return encoded_jwt def verify_token(token: str, token_type: str = "access") -> Optional[dict]: """验证令牌""" try: # 首先检查令牌是否在黑名单中 if token_blacklist.is_blacklisted(token): return None payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) # 检查令牌类型 if payload.get("type") != token_type: return None return payload except JWTError: return None def verify_password(plain_password: str, hashed_password: str) -> bool: """验证密码""" try: return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) except: return False def get_password_hash(password: str) -> str: """获取密码哈希""" # bcrypt 限制密码长度为72字节,如果超过则截断 if len(password.encode('utf-8')) > 72: password = password.encode('utf-8')[:72].decode('utf-8', errors='ignore') salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') def create_password_reset_token(email: str) -> str: """创建密码重置令牌""" delta = timedelta(hours=1) # 1小时有效期 now = datetime.utcnow() expires = now + delta exp = expires.timestamp() encoded_jwt = jwt.encode( {"exp": exp, "nbf": now, "sub": email, "type": "password_reset"}, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM, ) return encoded_jwt def verify_password_reset_token(token: str) -> Optional[str]: """验证密码重置令牌""" try: payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) # 检查令牌类型 if payload.get("type") != "password_reset": return None return payload["sub"] except JWTError: return None # 密码强度验证 def validate_password_strength(password: str) -> dict: """验证密码强度""" errors = [] if len(password) < 8: errors.append("密码长度至少8位") if len(password) > 128: errors.append("密码长度不能超过128位") if not any(c.islower() for c in password): errors.append("密码必须包含至少一个小写字母") if not any(c.isupper() for c in password): errors.append("密码必须包含至少一个大写字母") if not any(c.isdigit() for c in password): errors.append("密码必须包含至少一个数字") # 检查特殊字符 special_chars = "!@#$%^&*()_+-=[]{}|;:,.<>?" if not any(c in special_chars for c in password): errors.append("密码必须包含至少一个特殊字符") return { "is_valid": len(errors) == 0, "errors": errors, "strength": calculate_password_strength(password) } def calculate_password_strength(password: str) -> str: """计算密码强度""" score = 0 # 长度评分 if len(password) >= 8: score += 1 if len(password) >= 12: score += 1 if len(password) >= 16: score += 1 # 字符类型评分 if any(c.islower() for c in password): score += 1 if any(c.isupper() for c in password): score += 1 if any(c.isdigit() for c in password): score += 1 if any(c in "!@#$%^&*()_+-=[]{}|;:,.<>?" for c in password): score += 1 # 根据评分返回强度等级 if score <= 2: return "弱" elif score <= 4: return "中等" elif score <= 6: return "强" else: return "非常强"