mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-09-30 13:44:54 +02:00
98 lines
2.3 KiB
Python
98 lines
2.3 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import VerifyMismatchError
|
|
from jose import JWTError, jwt
|
|
|
|
from app.config import settings
|
|
|
|
ph = PasswordHasher(
|
|
time_cost=2,
|
|
memory_cost=65536,
|
|
parallelism=4,
|
|
hash_len=32,
|
|
salt_len=16,
|
|
)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""
|
|
Hash a password using Argon2id.
|
|
|
|
Args:
|
|
password: Plain text password to hash
|
|
|
|
Returns:
|
|
Argon2 hash as a string
|
|
"""
|
|
return ph.hash(password)
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""
|
|
Verify a password against its hash.
|
|
|
|
Args:
|
|
plain_password: Plain text password to verify
|
|
hashed_password: Argon2 hash to verify against
|
|
|
|
Returns:
|
|
True if password matches, False otherwise
|
|
"""
|
|
try:
|
|
ph.verify(hashed_password, plain_password)
|
|
return True
|
|
except VerifyMismatchError:
|
|
return False
|
|
|
|
|
|
def create_token(
|
|
subject: str | Any,
|
|
token_type: str,
|
|
expires_delta: timedelta,
|
|
) -> str:
|
|
"""Create a JWT token with the given subject and expiration."""
|
|
expire = datetime.now(UTC) + expires_delta
|
|
to_encode = {
|
|
"exp": expire,
|
|
"sub": str(subject),
|
|
"type": token_type,
|
|
}
|
|
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
|
|
|
|
|
|
def create_access_token(subject: str | Any) -> str:
|
|
"""Create an access token for the given subject."""
|
|
return create_token(
|
|
subject=subject,
|
|
token_type="access", # noqa: S106
|
|
expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
)
|
|
|
|
|
|
def create_refresh_token(subject: str | Any) -> str:
|
|
"""Create a refresh token for the given subject."""
|
|
return create_token(
|
|
subject=subject,
|
|
token_type="refresh", # noqa: S106
|
|
expires_delta=timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
|
)
|
|
|
|
|
|
def decode_token(token: str) -> dict | None:
|
|
"""
|
|
Decode and validate a JWT token.
|
|
|
|
Returns:
|
|
Decoded payload dict if valid, None otherwise
|
|
"""
|
|
try:
|
|
return jwt.decode(
|
|
token,
|
|
settings.SECRET_KEY,
|
|
algorithms=[settings.ALGORITHM],
|
|
)
|
|
except JWTError:
|
|
return None
|