kitos
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Authentication service — credential validation and password management."""
|
|
# Enable future language features for compatibility
|
|
from __future__ import annotations
|
|
|
|
# Import Session from sqlalchemy.orm
|
|
from sqlalchemy.orm import Session
|
|
|
|
# Import hash_password, verify_password from app.auth
|
|
from app.auth import hash_password, verify_password
|
|
|
|
# Import BusinessRuleViolation, PermissionViolation from app.domain.errors
|
|
from app.domain.errors import BusinessRuleViolation, PermissionViolation
|
|
|
|
# Import User from app.models.user
|
|
from app.models.user import User
|
|
|
|
# Assign _DUMMY_HASH = "$2b$12$LJ3m4ys3Lg3dMO/NpNmOaeVwFpWJMxlB2FLmEAo9fZr.S8H1vC4Wy"
|
|
_DUMMY_HASH = "$2b$12$LJ3m4ys3Lg3dMO/NpNmOaeVwFpWJMxlB2FLmEAo9fZr.S8H1vC4Wy"
|
|
|
|
|
|
# Define function authenticate_user
|
|
def authenticate_user(db: Session, *, username: str, password: str) -> User:
|
|
"""Validate credentials and return the User.
|
|
|
|
Raises BusinessRuleViolation for invalid credentials.
|
|
Raises PermissionViolation for disabled account.
|
|
Uses constant-time comparison to prevent timing attacks.
|
|
"""
|
|
# Assign user = db.query(User).filter(User.username == username).first()
|
|
user = db.query(User).filter(User.username == username).first()
|
|
# Assign hashed = user.hashed_password if user else _DUMMY_HASH
|
|
hashed = user.hashed_password if user else _DUMMY_HASH
|
|
# Assign password_valid = verify_password(password, hashed)
|
|
password_valid = verify_password(password, hashed)
|
|
|
|
# Check: user is None or not password_valid
|
|
if user is None or not password_valid:
|
|
# Raise BusinessRuleViolation
|
|
raise BusinessRuleViolation("Incorrect username or password")
|
|
# Check: not user.is_active
|
|
if not user.is_active:
|
|
# Raise PermissionViolation
|
|
raise PermissionViolation("Account is disabled. Contact an administrator.")
|
|
# Return user
|
|
return user
|
|
|
|
|
|
# Define function change_password
|
|
def change_password(
|
|
# Entry: db
|
|
db: Session,
|
|
# Entry: user
|
|
user: User,
|
|
*,
|
|
# Entry: current_password
|
|
current_password: str,
|
|
# Entry: new_password
|
|
new_password: str,
|
|
) -> None:
|
|
"""Change a user's password. Does NOT commit.
|
|
|
|
Raises BusinessRuleViolation if current password is wrong.
|
|
"""
|
|
# Check: not verify_password(current_password, user.hashed_password)
|
|
if not verify_password(current_password, user.hashed_password):
|
|
# Raise BusinessRuleViolation
|
|
raise BusinessRuleViolation("Current password is incorrect")
|
|
if verify_password(new_password, user.hashed_password):
|
|
raise BusinessRuleViolation(
|
|
"New password must be different from the current password"
|
|
)
|
|
user.hashed_password = hash_password(new_password)
|
|
# Assign user.must_change_password = False
|
|
user.must_change_password = False
|