Forge-Engine/auth/ensure_schema.py

from .connection import connect


def ensure_users_schema() -> None:
    conn = connect()
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS _users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            email TEXT NOT NULL UNIQUE,
            password_hash TEXT NOT NULL,
            nom TEXT NOT NULL,
            prenom TEXT NOT NULL,
            role TEXT NOT NULL DEFAULT 'user',
            totp_secret TEXT NOT NULL,
            totp_confirmed INTEGER NOT NULL DEFAULT 0,
            project_slug TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        );
        """
    )
    # Anti-bruteforce (voir rate_limit.py) : ajoutées en ALTER TABLE pour
    # ne rien casser sur une base de comptes déjà créée avant cette
    # fonctionnalité (le compte admin déjà inscrit sur ce poste, par
    # exemple) — CREATE TABLE IF NOT EXISTS ci-dessus n'aurait sinon
    # aucun effet sur une table qui existe déjà sans ces colonnes.
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(_users)").fetchall()}
    if "failed_attempts" not in cols:
        conn.execute("ALTER TABLE _users ADD COLUMN failed_attempts INTEGER NOT NULL DEFAULT 0")
    if "locked_until" not in cols:
        conn.execute("ALTER TABLE _users ADD COLUMN locked_until TEXT")
    if "support_slug" not in cols:
        # Le SEUL support de formation que ce compte possède (voir
        # set_support_slug.py) — mirroir exact de project_slug ci-dessus,
        # mais pour l'autre type de projet (entité racine séparée, voir
        # docs/plan/PLAN.md). Un compte peut avoir project_slug ET
        # support_slug simultanément (un de chaque), jamais deux du même
        # type pour un compte "user" restreint.
        conn.execute("ALTER TABLE _users ADD COLUMN support_slug TEXT")
    # Codes de récupération 2FA (voir recovery_codes.py) : table séparée
    # plutôt qu'une colonne JSON sur _users, pour pouvoir marquer/consommer
    # chaque code UN PAR UN (used_at) sans avoir à relire/réécrire toute
    # la liste à chaque connexion.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS _recovery_codes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            code_hash TEXT NOT NULL,
            used_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        );
        """
    )
    # Jetons de réinitialisation de mot de passe (voir password_reset.py) :
    # seul leur hash (SHA-256, un jeton aléatoire à haute entropie n'a pas
    # besoin d'un hash de mot de passe volontairement lent) est stocké, un
    # jeton envoyé par email reste donc inutilisable même en cas de fuite
    # de cette base.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS _password_reset_tokens (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            token_hash TEXT NOT NULL UNIQUE,
            expires_at TEXT NOT NULL,
            used_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        );
        """
    )
    conn.commit()
    conn.close()