JiriUhlir
94 lines
2.2 KiB
Python
94 lines
2.2 KiB
Python
from typing import Any
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
from passlib.context import CryptContext
|
|
|
|
from app.db.database import get_connection
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain_password: str, password_hash: str) -> bool:
|
|
if not password_hash:
|
|
return False
|
|
|
|
return pwd_context.verify(plain_password, password_hash)
|
|
|
|
|
|
def get_user_by_username(username: str) -> dict[str, Any] | None:
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE username = ?
|
|
""",
|
|
(username,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def get_user_by_id(user_id: int) -> dict[str, Any] | None:
|
|
con = get_connection()
|
|
|
|
row = con.execute(
|
|
"""
|
|
SELECT id, username, display_name, email, password_hash, role, is_active, created_at, updated_at
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def authenticate_user(username: str, password: str) -> dict[str, Any] | None:
|
|
user = get_user_by_username(username.strip())
|
|
|
|
if not user or not user.get("is_active"):
|
|
return None
|
|
|
|
if not verify_password(password, user.get("password_hash") or ""):
|
|
return None
|
|
|
|
return user
|
|
|
|
|
|
def current_user(request: Request) -> dict[str, Any] | None:
|
|
user_id = request.session.get("user_id")
|
|
if not user_id:
|
|
return None
|
|
|
|
try:
|
|
user_id = int(user_id)
|
|
except (TypeError, ValueError):
|
|
request.session.pop("user_id", None)
|
|
return None
|
|
|
|
user = get_user_by_id(user_id)
|
|
if not user or not user.get("is_active"):
|
|
request.session.pop("user_id", None)
|
|
return None
|
|
|
|
return user
|
|
|
|
|
|
def require_user(request: Request) -> dict[str, Any]:
|
|
user = current_user(request)
|
|
if user:
|
|
return user
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_303_SEE_OTHER,
|
|
headers={"Location": "/portal/login"},
|
|
)
|