foodlinkk-command-center/cockpit/app/auth_users.py

"""Cockpit user accounts: password hashing, CRUD, permissions."""
from __future__ import annotations

import hashlib
import json
import secrets
from typing import Any, Optional

from app.db import execute, fetch_all, fetch_one

# Nav / feature keys shown in Settings → Gebruikers
PAGE_CATALOG: list[dict[str, str]] = [
    {"key": "dashboard", "label": "Overzicht (CEO)"},
    {"key": "revenue", "label": "Revenue Cockpit"},
    {"key": "retail", "label": "Retail NL"},
    {"key": "marketing", "label": "Marketing"},
    {"key": "packaging", "label": "Packaging"},
    {"key": "beurs", "label": "Beurs Live"},
    {"key": "documents", "label": "Documenten"},
    {"key": "projects", "label": "Projecten"},
    {"key": "hermes", "label": "Telegram / Hermes"},
    {"key": "analytics", "label": "Analytics"},
    {"key": "export", "label": "Wereldexport"},
    {"key": "clients", "label": "Klanten"},
    {"key": "deals", "label": "Deals"},
    {"key": "products", "label": "Producten"},
    {"key": "suppliers", "label": "Leveranciers"},
    {"key": "reports", "label": "Rapporten"},
    {"key": "ops", "label": "IT Ops"},
    {"key": "agents", "label": "Agents"},
    {"key": "browser", "label": "Browser"},
    {"key": "studio", "label": "Studio"},
    {"key": "voice", "label": "Voice"},
    {"key": "settings", "label": "Instellingen"},
]

# path prefix → permission key (order: longest prefix first)
PATH_RULES: list[tuple[str, str]] = [
    ("/revenue-cockpit", "revenue"),
    ("/export-intel", "export"),
    ("/packaging", "packaging"),
    ("/marketing", "marketing"),
    ("/documents", "documents"),
    ("/analytics", "analytics"),
    ("/suppliers", "suppliers"),
    ("/products", "products"),
    ("/projects", "projects"),
    ("/settings", "settings"),
    ("/clients", "clients"),
    ("/browser", "browser"),
    ("/monitor", "browser"),
    ("/hermes", "hermes"),
    ("/retail", "retail"),
    ("/reports", "reports"),
    ("/studio", "studio"),
    ("/agents", "agents"),
    ("/beurs", "beurs"),
    ("/deals", "deals"),
    ("/voice", "voice"),
    ("/ops", "ops"),
    ("/", "dashboard"),
]

ADMIN_SEED_USERNAME = "aissa"
ADMIN_SEED_PASSWORD = "Foodlinkk#2026"


def hash_password(password: str) -> str:
    salt = secrets.token_hex(16)
    rounds = 200_000
    dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), rounds)
    return f"pbkdf2_sha256${rounds}${salt}${dk.hex()}"


def verify_password(password: str, password_hash: str) -> bool:
    try:
        algo, rounds_s, salt, digest = password_hash.split("$", 3)
        if algo != "pbkdf2_sha256":
            return False
        rounds = int(rounds_s)
        dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), rounds)
        return secrets.compare_digest(dk.hex(), digest)
    except Exception:
        return False


def _normalize_perms(raw: Any) -> dict[str, Any]:
    if raw is None:
        return {"pages": [], "manage_users": False}
    if isinstance(raw, str):
        try:
            raw = json.loads(raw)
        except Exception:
            return {"pages": [], "manage_users": False}
    if not isinstance(raw, dict):
        return {"pages": [], "manage_users": False}
    pages = raw.get("pages") or []
    if not isinstance(pages, list):
        pages = []
    return {
        "pages": [str(p) for p in pages],
        "manage_users": bool(raw.get("manage_users")),
    }


def public_user(row: dict[str, Any] | None) -> dict[str, Any] | None:
    if not row:
        return None
    perms = _normalize_perms(row.get("permissions"))
    role = row.get("role") or "user"
    if role == "admin":
        perms = {
            "pages": [p["key"] for p in PAGE_CATALOG],
            "manage_users": True,
        }
    return {
        "id": row["id"],
        "username": row["username"],
        "display_name": row.get("display_name") or row["username"],
        "role": role,
        "is_active": bool(row.get("is_active", True)),
        "avatar_url": row.get("avatar_url") or None,
        "permissions": perms,
        "created_at": row["created_at"].isoformat() if row.get("created_at") else None,
        "last_login_at": row["last_login_at"].isoformat() if row.get("last_login_at") else None,
    }


def ensure_admin_seed() -> None:
    """Create default admin aissa if table empty / user missing."""
    row = fetch_one("SELECT id FROM cockpit_users WHERE username = %s", (ADMIN_SEED_USERNAME,))
    if row:
        return
    execute(
        """
        INSERT INTO cockpit_users (username, password_hash, display_name, role, permissions)
        VALUES (%s, %s, %s, 'admin', %s::jsonb)
        ON CONFLICT (username) DO NOTHING
        """,
        (
            ADMIN_SEED_USERNAME,
            hash_password(ADMIN_SEED_PASSWORD),
            "Aissa",
            json.dumps({"pages": [p["key"] for p in PAGE_CATALOG], "manage_users": True}),
        ),
    )


def authenticate(username: str, password: str) -> Optional[dict[str, Any]]:
    row = fetch_one(
        "SELECT * FROM cockpit_users WHERE lower(username) = lower(%s)",
        (username.strip(),),
    )
    if not row or not row.get("is_active"):
        return None
    if not verify_password(password, row["password_hash"]):
        return None
    execute(
        "UPDATE cockpit_users SET last_login_at = NOW(), updated_at = NOW() WHERE id = %s",
        (row["id"],),
    )
    return public_user(row)


def get_user_by_id(user_id: int) -> Optional[dict[str, Any]]:
    row = fetch_one("SELECT * FROM cockpit_users WHERE id = %s", (user_id,))
    return public_user(row)


def list_users() -> list[dict[str, Any]]:
    rows = fetch_all("SELECT * FROM cockpit_users ORDER BY username ASC")
    return [public_user(r) for r in rows if r]


def create_user(
    username: str,
    password: str,
    display_name: str | None,
    role: str,
    permissions: dict[str, Any],
) -> dict[str, Any]:
    username = username.strip()
    if not username or not password:
        raise ValueError("Gebruikersnaam en wachtwoord zijn verplicht")
    if role not in ("admin", "user"):
        role = "user"
    perms = _normalize_perms(permissions)
    if role == "admin":
        perms["manage_users"] = True
        perms["pages"] = [p["key"] for p in PAGE_CATALOG]
    row = fetch_one(
        """
        INSERT INTO cockpit_users (username, password_hash, display_name, role, permissions)
        VALUES (%s, %s, %s, %s, %s::jsonb)
        RETURNING *
        """,
        (username, hash_password(password), (display_name or username).strip(), role, json.dumps(perms)),
    )
    return public_user(row)  # type: ignore


def update_user(
    user_id: int,
    *,
    display_name: str | None = None,
    role: str | None = None,
    is_active: bool | None = None,
    permissions: dict[str, Any] | None = None,
    password: str | None = None,
) -> dict[str, Any]:
    row = fetch_one("SELECT * FROM cockpit_users WHERE id = %s", (user_id,))
    if not row:
        raise ValueError("Gebruiker niet gevonden")
    new_name = display_name if display_name is not None else row.get("display_name")
    new_role = role if role in ("admin", "user") else row["role"]
    new_active = row["is_active"] if is_active is None else bool(is_active)
    new_perms = _normalize_perms(permissions if permissions is not None else row.get("permissions"))
    if new_role == "admin":
        new_perms["manage_users"] = True
        new_perms["pages"] = [p["key"] for p in PAGE_CATALOG]
    pw_hash = row["password_hash"]
    if password:
        pw_hash = hash_password(password)
    updated = fetch_one(
        """
        UPDATE cockpit_users
        SET display_name = %s, role = %s, is_active = %s, permissions = %s::jsonb,
            password_hash = %s, updated_at = NOW()
        WHERE id = %s
        RETURNING *
        """,
        (new_name, new_role, new_active, json.dumps(new_perms), pw_hash, user_id),
    )
    return public_user(updated)  # type: ignore


def delete_user(user_id: int, *, actor_id: int) -> None:
    if user_id == actor_id:
        raise ValueError("Je kunt jezelf niet verwijderen")
    row = fetch_one("SELECT role FROM cockpit_users WHERE id = %s", (user_id,))
    if not row:
        raise ValueError("Gebruiker niet gevonden")
    if row["role"] == "admin":
        admins = fetch_one(
            "SELECT COUNT(*)::int AS n FROM cockpit_users WHERE role = 'admin' AND is_active = TRUE"
        )
        if admins and admins["n"] <= 1:
            raise ValueError("Laatste admin kan niet verwijderd worden")
    execute("DELETE FROM cockpit_users WHERE id = %s", (user_id,))


def path_allowed(user: dict[str, Any] | None, path: str) -> bool:
    if not user:
        return False
    if user.get("role") == "admin":
        return True
    pages = set((user.get("permissions") or {}).get("pages") or [])
    # normalize
    p = path.split("?")[0] or "/"
    if p != "/" and p.endswith("/"):
        p = p.rstrip("/")
    for prefix, key in PATH_RULES:
        if prefix == "/":
            if p == "/" or p == "":
                return "dashboard" in pages
            continue
        if p == prefix or p.startswith(prefix + "/"):
            return key in pages
    return "dashboard" in pages


def can_manage_users(user: dict[str, Any] | None) -> bool:
    if not user:
        return False
    if user.get("role") == "admin":
        return True
    return bool((user.get("permissions") or {}).get("manage_users"))


def set_avatar(user_id: int, avatar_url: str | None) -> dict[str, Any]:
    row = fetch_one(
        """
        UPDATE cockpit_users
        SET avatar_url = %s, updated_at = NOW()
        WHERE id = %s
        RETURNING *
        """,
        (avatar_url, user_id),
    )
    if not row:
        raise ValueError("Gebruiker niet gevonden")
    return public_user(row)  # type: ignore