Kairo-Jinkendo/backend/auth.py

"""Authentication: bcrypt passwords, server-side sessions, FastAPI dependencies."""

from __future__ import annotations

import os
import secrets
from datetime import datetime, timedelta, timezone
from typing import Any, Optional

import bcrypt
from fastapi import Depends, Header, HTTPException
from psycopg2.extras import RealDictCursor

from db import get_connection
from services.audit import log_audit

AUTH_HEADER = "X-Auth-Token"
SESSION_DAYS = int(os.getenv("SESSION_DAYS", "30"))


def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def verify_password(password: str, stored_hash: str) -> bool:
    if not stored_hash:
        return False
    try:
        return bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8"))
    except ValueError:
        return False


def make_token() -> str:
    return secrets.token_urlsafe(32)


def _session_expiry() -> datetime:
    return datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS)


def get_user_by_email(email: str) -> Optional[dict[str, Any]]:
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT id, email, password_hash, display_name, portal_role, is_active
                FROM users
                WHERE LOWER(email) = LOWER(%s)
                """,
                (email.strip(),),
            )
            row = cur.fetchone()
            return dict(row) if row else None
    finally:
        conn.close()


def get_user_by_id(user_id: str) -> Optional[dict[str, Any]]:
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT id, email, display_name, portal_role, is_active, created_at
                FROM users
                WHERE id = %s
                """,
                (user_id,),
            )
            row = cur.fetchone()
            return dict(row) if row else None
    finally:
        conn.close()


def get_session(token: str) -> Optional[dict[str, Any]]:
    if not token:
        return None
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                SELECT
                    s.id AS session_id,
                    s.token,
                    s.user_id,
                    s.active_tenant_id,
                    s.expires_at,
                    u.email,
                    u.display_name,
                    u.portal_role,
                    u.is_active AS user_is_active
                FROM sessions s
                JOIN users u ON u.id = s.user_id
                WHERE s.token = %s AND s.expires_at > NOW()
                """,
                (token,),
            )
            row = cur.fetchone()
            if not row:
                return None
            session = dict(row)
            if not session.get("user_is_active"):
                return None
            return session
    finally:
        conn.close()


def _default_active_tenant(user_id: str) -> Optional[str]:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT tm.tenant_id
                FROM tenant_memberships tm
                JOIN tenants t ON t.id = tm.tenant_id
                WHERE tm.user_id = %s
                  AND tm.is_active = TRUE
                  AND t.is_active = TRUE
                ORDER BY tm.created_at ASC
                LIMIT 1
                """,
                (user_id,),
            )
            row = cur.fetchone()
            return str(row[0]) if row else None
    finally:
        conn.close()


def create_session(user_id: str, active_tenant_id: Optional[str] = None) -> dict[str, Any]:
    token = make_token()
    tenant_id = active_tenant_id or _default_active_tenant(user_id)
    expires_at = _session_expiry()
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                """
                INSERT INTO sessions (token, user_id, active_tenant_id, expires_at)
                VALUES (%s, %s, %s, %s)
                RETURNING token, expires_at, active_tenant_id
                """,
                (token, user_id, tenant_id, expires_at),
            )
            row = dict(cur.fetchone())
        conn.commit()
        row["user_id"] = user_id
        return row
    finally:
        conn.close()


def delete_session(token: str) -> None:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM sessions WHERE token = %s", (token,))
        conn.commit()
    finally:
        conn.close()


def set_session_active_tenant(token: str, tenant_id: str) -> bool:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE sessions s
                SET active_tenant_id = %s
                FROM users u
                WHERE s.token = %s
                  AND s.user_id = u.id
                  AND EXISTS (
                    SELECT 1 FROM tenant_memberships tm
                    JOIN tenants t ON t.id = tm.tenant_id
                    WHERE tm.user_id = s.user_id
                      AND tm.tenant_id = %s
                      AND tm.is_active = TRUE
                      AND t.is_active = TRUE
                  )
                """,
                (tenant_id, token, tenant_id),
            )
            updated = cur.rowcount > 0
        conn.commit()
        return updated
    finally:
        conn.close()


def login(email: str, password: str) -> dict[str, Any]:
    user = get_user_by_email(email)
    if not user or not user.get("is_active"):
        log_audit("auth.login_failed", details={"email": email.strip().lower()})
        raise HTTPException(status_code=401, detail="Ungültige Anmeldedaten")

    if not verify_password(password, user["password_hash"]):
        log_audit("auth.login_failed", user_id=str(user["id"]), details={"email": user["email"]})
        raise HTTPException(status_code=401, detail="Ungültige Anmeldedaten")

    session = create_session(str(user["id"]))
    log_audit("auth.login", user_id=str(user["id"]), details={"email": user["email"]})
    return {
        "token": session["token"],
        "expires_at": session["expires_at"].isoformat(),
        "user": {
            "id": str(user["id"]),
            "email": user["email"],
            "display_name": user["display_name"],
            "portal_role": user["portal_role"],
        },
    }


def logout(token: str) -> None:
    session = get_session(token)
    if session:
        log_audit(
            "auth.logout",
            user_id=str(session["user_id"]),
            tenant_id=str(session["active_tenant_id"]) if session.get("active_tenant_id") else None,
        )
    delete_session(token)


def require_auth(x_auth_token: Optional[str] = Header(default=None, alias=AUTH_HEADER)) -> dict[str, Any]:
    session = get_session(x_auth_token or "")
    if not session:
        raise HTTPException(status_code=401, detail="Nicht eingeloggt")
    return session


def require_portal_admin(session: dict[str, Any] = Depends(require_auth)) -> dict[str, Any]:
    if session.get("portal_role") != "admin":
        raise HTTPException(status_code=403, detail="Nur für Portal-Admins")
    return session