soapbox/soapbox/db.py

"""Brukere i sqlite.

Kun brukere og passord-hasher lever i databasen – alt innhold ligger i git.
Det gjør migrering enkel: klon innholdsrepoet og opprett brukerne på nytt.

Passord hashes med ``hashlib.scrypt`` fra standardbiblioteket for å slippe
en ekstra avhengighet. Parametrene (n=2**15, r=8, p=1) tilsvarer OWASP-
anbefalingene. ``maxmem`` må settes eksplisitt fordi OpenSSL ellers avviser
kall som trenger mer enn 32 MB, som er akkurat det disse parametrene bruker.
"""

from __future__ import annotations

import hashlib
import hmac
import secrets
import sqlite3
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path

ROLE_OWNER = "owner"
ROLE_GUEST = "guest"

SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    username TEXT NOT NULL UNIQUE,
    display_name TEXT NOT NULL DEFAULT '',
    password_hash TEXT NOT NULL,
    role TEXT NOT NULL CHECK (role IN ('owner', 'guest')),
    created_at TEXT NOT NULL
);
"""


@dataclass
class User:
    id: int
    username: str
    display_name: str
    role: str
    email: str = ""
    bio: str = ""

    @property
    def is_owner(self) -> bool:
        return self.role == ROLE_OWNER

    @property
    def guest_prefix(self) -> str | None:
        """Gjester publiserer under /<brukernavn>/, hovedbruker under /."""
        return None if self.is_owner else self.username


def connect(path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.executescript(SCHEMA)
    # Enkel migrering: kolonner lagt til etter første versjon
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)")}
    if "email" not in cols:
        conn.execute("ALTER TABLE users ADD COLUMN email TEXT NOT NULL DEFAULT ''")
    if "bio" not in cols:
        conn.execute("ALTER TABLE users ADD COLUMN bio TEXT NOT NULL DEFAULT ''")
    conn.commit()
    return conn


def hash_password(password: str) -> str:
    salt = secrets.token_bytes(16)
    digest = hashlib.scrypt(
        password.encode(), salt=salt, n=2**15, r=8, p=1, dklen=32, maxmem=64 * 1024 * 1024
    )
    return f"scrypt${salt.hex()}${digest.hex()}"


def verify_password(password: str, stored: str) -> bool:
    try:
        _, salt_hex, digest_hex = stored.split("$")
    except ValueError:
        return False
    digest = hashlib.scrypt(
        password.encode(),
        salt=bytes.fromhex(salt_hex),
        n=2**15,
        r=8,
        p=1,
        dklen=32,
        maxmem=64 * 1024 * 1024,
    )
    return hmac.compare_digest(digest.hex(), digest_hex)


def _row_to_user(row: sqlite3.Row | None) -> User | None:
    if row is None:
        return None
    return User(
        row["id"], row["username"], row["display_name"], row["role"], row["email"], row["bio"]
    )


def create_user(
    conn: sqlite3.Connection, username: str, password: str, role: str, display_name: str = ""
) -> User:
    conn.execute(
        "INSERT INTO users (username, display_name, password_hash, role, created_at) VALUES (?,?,?,?,?)",
        (
            username,
            display_name or username,
            hash_password(password),
            role,
            datetime.now(UTC).isoformat(timespec="seconds"),
        ),
    )
    conn.commit()
    return get_user_by_name(conn, username)


def get_user(conn: sqlite3.Connection, user_id: int) -> User | None:
    return _row_to_user(conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone())


def get_user_by_name(conn: sqlite3.Connection, username: str) -> User | None:
    return _row_to_user(
        conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
    )


def authenticate(conn: sqlite3.Connection, username: str, password: str) -> User | None:
    row = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
    if row and verify_password(password, row["password_hash"]):
        return _row_to_user(row)
    return None


def update_profile(
    conn: sqlite3.Connection, user_id: int, display_name: str, email: str, bio: str = ""
) -> None:
    conn.execute(
        "UPDATE users SET display_name=?, email=?, bio=? WHERE id=?",
        (display_name.strip(), email.strip(), bio.strip(), user_id),
    )
    conn.commit()


def authors(conn: sqlite3.Connection) -> dict[str, User]:
    """Brukernavn → bruker, til oppslag av visningsnavn/e-post ved bygging."""
    return {u.username: u for u in list_users(conn)}


def password_stamp(conn: sqlite3.Connection, user_id: int) -> str | None:
    """Kort, ikke-hemmelig fingeravtrykk av passord-hashen, lagret i sesjonen
    slik at et passordbytte ugyldiggjør alle eksisterende sesjoner."""
    row = conn.execute("SELECT password_hash FROM users WHERE id=?", (user_id,)).fetchone()
    return row["password_hash"][-16:] if row else None


def set_password(conn: sqlite3.Connection, user_id: int, password: str) -> None:
    conn.execute("UPDATE users SET password_hash=? WHERE id=?", (hash_password(password), user_id))
    conn.commit()


def list_users(conn: sqlite3.Connection) -> list[User]:
    rows = conn.execute("SELECT * FROM users ORDER BY role, username").fetchall()
    return [_row_to_user(r) for r in rows]


def delete_user(conn: sqlite3.Connection, user_id: int) -> None:
    conn.execute("DELETE FROM users WHERE id=? AND role='guest'", (user_id,))
    conn.commit()


def has_owner(conn: sqlite3.Connection) -> bool:
    return conn.execute("SELECT 1 FROM users WHERE role='owner' LIMIT 1").fetchone() is not None