first commit

app/security.py

@@ -0,0 +1,71 @@
+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)