import hashlib import secrets import uuid from dataclasses import dataclass from datetime import datetime, timedelta, timezone import jwt from jwt.exceptions import InvalidTokenError from pwdlib import PasswordHash from app.config import settings password_hash = PasswordHash.recommended() DUMMY_PASSWORD_HASH = password_hash.hash("opencloud-dummy-password") @dataclass(frozen=True) class AccessTokenPayload: user_id: uuid.UUID session_id: uuid.UUID def hash_password(password: str) -> str: return password_hash.hash(password) def verify_password(password: str, encoded_hash: str) -> bool: return password_hash.verify(password, encoded_hash) def create_random_token() -> str: return secrets.token_urlsafe(48) def hash_token(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() def create_access_token(user_id: uuid.UUID, session_id: uuid.UUID) -> str: now = datetime.now(timezone.utc) payload = { "sub": str(user_id), "sid": str(session_id), "type": "access", "iat": now, "exp": now + timedelta(minutes=settings.access_token_expire_minutes), "jti": str(uuid.uuid4()), } return jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm) def decode_access_token(token: str) -> AccessTokenPayload: try: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.jwt_algorithm]) if payload.get("type") != "access": raise InvalidTokenError("wrong token type") return AccessTokenPayload( user_id=uuid.UUID(payload["sub"]), session_id=uuid.UUID(payload["sid"]), ) except (InvalidTokenError, KeyError, TypeError, ValueError) as exc: raise ValueError("无效或已过期的访问令牌") from exc def utc_now() -> datetime: return datetime.now(timezone.utc) def refresh_expires_at() -> datetime: return utc_now() + timedelta(days=settings.refresh_token_expire_days)