from app.domain.entities import user
import hashlib
import secrets
from datetime import datetime, timedelta, timezone
from uuid import UUID
import jwt
from passlib.context import CryptContext
import re
from app.infrastructure.settings import settings
from app.infrastructure.postgres.user_postgres import PostgresUserRepository
from app.infrastructure.postgres.refresh_token import PostgresRefreshTokenRepository
_pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
class AuthService:
def __init__(
self,
user_repo: PostgresUserRepository,
refresh_repo: PostgresRefreshTokenRepository,
):
self._user_repo = user_repo
self._refresh_repo = refresh_repo
# Password
def hash_password(self, plain: str) -> str:
return _pwd_context.hash(plain)
def verify_password(self, plain:str, hashed:str) -> bool:
return _pwd_context.verify(plain, hashed)
# Access token
def create_access_token(self, user_id:UUID, role:str) -> str:
expire = datetime.now(timezone.utc) + timedelta(
minutes=settings.access_token_expire_minute
)
payload = {
"sub": str(user_id),
"role":role,
"exp":expire
}
return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
def decode_access_token(self, token:str) -> UUID:
payload = jwt.decode(
token,
settings.jwt_secret_key,
algorithms=[settings.jwt_algorithm],
)
return UUID(payload["sub"])
# Refresh Token
def create_refresh_token(self) -> str:
return secrets.token_urlsafe(32)
def _hash_token(self, raw:str) -> str:
return hashlib.sha256(raw.encode()).hexdigest()
async def register(self, username: str, email: str, password: str, confirm_password: str, contact: int) -> None:
if password != confirm_password:
raise ValueError("Passwords do not match")
existing = await self._user_repo.get_by_email(email)
if existing:
raise ValueError("Email already registered")
hashed = self.hash_password(password)
return await self._user_repo.create(username, email, hashed, contact)
async def login(self, identifier:str, password:str)-> None:
pattern = r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
if re.fullmatch(pattern, identifier):
user = await self._user_repo.get_by_email(identifier)
else:
user = await self._user_repo.get_by_username(identifier)
if not user or not self.verify_password(password, user.password_hash):
raise ValueError("Invalid email/username or password")
if not user.is_active:
raise ValueError("Account is Deactivated")
access_token = self.create_access_token(user.id, user.role)
raw_refresh = self.create_refresh_token()
expires_at = datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_day)
await self._refresh_repo.create(user.id, self._hash_token(raw_refresh), expires_at)
return access_token, raw_refresh
async def refresh(self, raw_refresh_token:str) -> str:
token_hash = self._hash_token(raw_refresh_token)
user_id = await self._refresh_repo.get_valid(token_hash)
if not user_id:
raise ValueError("Invalid or epired refresh token")
return self.create_access_token(user_id)
user = await self._user_repo.get_by_id(user_id)
if not user or not user.is_active:
raise ValueError("User not found or deactivated")
return self.create_access_token(user_id, user.role)
async def logout(self, raw_refresh_token:str) -> None:
token_hash = self._hash_token(raw_refresh_token)
await self._refresh_repo.revoke(token_hash)