11

app/core/security.py

@@ -0,0 +1,58 @@
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import bcrypt
+from jose import JWTError, jwt
+
+from app.config import get_settings
+
+settings = get_settings()
+
+ALGORITHM = "HS256"
+TOKEN_TYPE = "access"
+SESSION_PREFIX = "session:"
+
+
+def hash_password(password: str) -> str:
+    return bcrypt.hashpw(
+        password.encode("utf-8"),
+        bcrypt.gensalt(),
+    ).decode("utf-8")
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    try:
+        return bcrypt.checkpw(
+            plain.encode("utf-8"),
+            hashed.encode("utf-8"),
+        )
+    except (ValueError, TypeError):
+        return False
+
+
+def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str:
+    expire = datetime.now(UTC) + (
+        expires_delta
+        if expires_delta is not None
+        else timedelta(minutes=settings.access_token_expire_minutes)
+    )
+    payload = {
+        "sub": subject,
+        "exp": expire,
+        "type": TOKEN_TYPE,
+    }
+    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)
+
+
+def decode_access_token(token: str) -> dict[str, Any] | None:
+    try:
+        payload = jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM])
+    except JWTError:
+        return None
+    if payload.get("type") != TOKEN_TYPE:
+        return None
+    return payload
+
+
+def session_redis_key(token: str) -> str:
+    return f"{SESSION_PREFIX}{token}"