auth via DB

app/auth.py

@@ -0,0 +1,93 @@
+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"},
+    )