67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.exceptions import ConflictError, UnauthorizedError
|
|
from app.core.security import create_access_token, get_password_hash, verify_password
|
|
from app.models.user import RoleEnum, User
|
|
from app.schemas.auth import LoginRequest, UserCreate
|
|
|
|
audit_logger = logging.getLogger("app.audit")
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def register_user(self, payload: UserCreate) -> User:
|
|
existing_user = self.db.scalar(select(User).where(User.username == payload.username))
|
|
if existing_user is not None:
|
|
raise ConflictError("Username already exists")
|
|
|
|
user = User(
|
|
username=payload.username,
|
|
password_hash=get_password_hash(payload.password),
|
|
role=RoleEnum.client,
|
|
is_active=True,
|
|
)
|
|
self.db.add(user)
|
|
|
|
try:
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
except IntegrityError as exc:
|
|
self.db.rollback()
|
|
raise ConflictError("Username already exists") from exc
|
|
|
|
audit_logger.info("user_registered user_id=%s role=%s", user.id, user.role.value)
|
|
return user
|
|
|
|
def authenticate_user(self, payload: LoginRequest) -> str:
|
|
user = self.db.scalar(select(User).where(User.username == payload.username))
|
|
subject_hash = self._hash_subject(payload.username)
|
|
|
|
if user is None or not verify_password(payload.password, user.password_hash):
|
|
audit_logger.warning("login_failed subject_hash=%s", subject_hash)
|
|
raise UnauthorizedError("Invalid username or password")
|
|
|
|
if not user.is_active:
|
|
audit_logger.warning("login_rejected_inactive user_id=%s", user.id)
|
|
raise UnauthorizedError("Invalid username or password")
|
|
|
|
access_token = create_access_token(subject=str(user.id))
|
|
audit_logger.info("login_success user_id=%s role=%s", user.id, user.role.value)
|
|
return access_token
|
|
|
|
def get_user_by_id(self, user_id: int) -> User | None:
|
|
return self.db.get(User, user_id)
|
|
|
|
@staticmethod
|
|
def _hash_subject(subject: str) -> str:
|
|
return hashlib.sha256(subject.encode("utf-8")).hexdigest()[:12]
|