feat: first commit, auth

core/secrets.py

@@ -0,0 +1,51 @@
+import hashlib
+import logging
+import secrets
+from typing import Any
+
+import jwt
+from argon2 import PasswordHasher
+from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
+
+from config import cfg
+from schemas.dto import KeyPair
+from schemas.jwt import JWTPayload
+from schemas.providers import ProvidersType
+
+ctx = PasswordHasher()
+logger = logging.getLogger(__name__)
+
+
+def hash_password(plain_password: str) -> ...:
+    return ctx.hash(plain_password)
+
+
+def verify_password(hashed_password: str, plain_password: str) -> bool:
+    try:
+        ctx.verify(hashed_password, plain_password)
+        return True
+    except (VerifyMismatchError, VerificationError, InvalidHashError):
+        return False
+    except Exception:
+        logger.exception("unexpected error while comparing password and hash")
+        return False
+
+
+def generate_jwt(payload: dict[str, Any]) -> str:
+    return jwt.encode(payload, cfg.private_key, "RS256")
+
+
+def decode_jwt(token: str) -> dict[str, Any]:
+    return jwt.decode(token, cfg.private_key, "RS256")
+
+
+def generate_pair(user_id: int, iss: ProvidersType) -> KeyPair:
+    payload = JWTPayload(sub=user_id, iss=iss)
+    access_token = generate_jwt(payload.model_dump())
+    refresh_token = secrets.token_urlsafe(32)
+
+    return KeyPair(access_token=access_token, refresh_token=refresh_token)
+
+
+def hash_refresh_token(token: str):
+    return hashlib.sha256(token.encode()).hexdigest()