47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import bcrypt
|
|
import jwt
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger("app.services.auth")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
return bcrypt.checkpw(plain.encode(), hashed.encode())
|
|
|
|
|
|
def create_access_token(user_id: int, email: str, token_type: str = "admin") -> str:
|
|
expire = datetime.now(timezone.utc) + timedelta(hours=settings.JWT_EXPIRE_HOURS)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"email": email,
|
|
"type": token_type,
|
|
"exp": expire,
|
|
}
|
|
token = jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
|
|
logger.info("Created JWT (%s) for id=%d, email=%s, expires=%s", token_type, user_id, email, expire)
|
|
return token
|
|
|
|
|
|
def create_client_token(client_id: int, email: str) -> str:
|
|
return create_access_token(client_id, email, token_type="client")
|
|
|
|
|
|
def decode_access_token(token: str) -> dict | None:
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
logger.warning("JWT expired")
|
|
return None
|
|
except jwt.InvalidTokenError as e:
|
|
logger.warning("Invalid JWT: %s", e)
|
|
return None
|