"""Réglages d'accès de l'instance — inscription locale (fusion #146) et porte centrale unique (phase 2 du programme « compte partout »). Patron des réglages existants (``llm_config``, ``sso_config``) : une table à ligne unique (``id = 1``) créée paresseusement, qui prime sur les valeurs de ``app.config.settings`` (variables d'environnement / ``.env``), elles-mêmes aux défauts du code. Réglage « inscription locale ouverte » : - **fermée par défaut** (décision de Bruno, programme « compte partout ») : les comptes naissent dans le Hub Dracodev et entrent dans Flowdeck par la voie ObsiGate (SSO) / le provisionnement JIT ; le parcours public d'inscription locale est refusé proprement. - réversible à tout moment par un administrateur de l'instance (``GET``/``PUT /api/admin/reglages-acces``) ou en posant ``INSCRIPTION_LOCALE_OUVERTE=true`` avant tout enregistrement en base. Le réglage d'inscription ne touche **jamais** la connexion des comptes existants, locaux comme draco : il ne garde que la création de comptes locaux. Réglage « porte centrale » (phase 2, **distinct** du précédent) : - **désactivé par défaut à la livraison** ; activé par un administrateur, l'écran de connexion ne présente que l'entrée centrale (bouton ObsiGate SSO) et les identifiants locaux ordinaires ne sont plus acceptés — y compris pour les comptes locaux historiques. Les sessions déjà établies ne sont pas touchées, et la voie draco (SSO/JIT) reste la porte normale. - **Porte de secours administrateur** : UN compte administrateur local désigné peut encore s'authentifier localement, uniquement par le chemin discret ``/auth/login?provider=local&secours=1`` (jamais proposé sur l'écran principal). Ses identifiants sont posés par les variables d'environnement ``FLOWDECK_SECOURS_ADMIN_LOGIN`` et ``FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE`` — aucun secret en dur ; sans ces variables, la porte de secours est fermée. """ from __future__ import annotations import logging import os import secrets from app.config import settings logger = logging.getLogger(__name__) __all__ = [ "get_reglages_acces", "set_reglages_acces", "inscription_locale_ouverte", "message_inscription_fermee", "porte_centrale_active", "message_porte_centrale", "authentifier_secours", ] def _assurer_table(conn) -> None: """Crée la table des réglages d'accès si elle n'existe pas encore et y ajoute les colonnes des phases ultérieures si besoin.""" conn.execute(""" CREATE TABLE IF NOT EXISTS reglages_acces ( id INTEGER PRIMARY KEY CHECK (id = 1), inscription_locale_ouverte INTEGER, porte_centrale_active INTEGER, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) colonnes = {r[1] for r in conn.execute("PRAGMA table_info(reglages_acces)")} if "porte_centrale_active" not in colonnes: conn.execute( "ALTER TABLE reglages_acces ADD COLUMN porte_centrale_active INTEGER") def get_reglages_acces() -> dict: """Réglages d'accès effectifs — la ligne en base prime sur la config.""" reglages = { # Défaut du code et de la config : inscription fermée. "inscription_locale_ouverte": bool( getattr(settings, "inscription_locale_ouverte", False)), # Défaut du code et de la config : porte centrale désactivée. "porte_centrale_active": bool( getattr(settings, "porte_centrale_active", False)), } try: from app.db import get_conn with get_conn() as conn: _assurer_table(conn) row = conn.execute( "SELECT inscription_locale_ouverte, porte_centrale_active" " FROM reglages_acces WHERE id=1" ).fetchone() except Exception: # noqa: BLE001 — base pas encore prête → config seule return reglages if row is not None: if row["inscription_locale_ouverte"] is not None: reglages["inscription_locale_ouverte"] = bool( row["inscription_locale_ouverte"]) if row["porte_centrale_active"] is not None: reglages["porte_centrale_active"] = bool( row["porte_centrale_active"]) return reglages def set_reglages_acces(*, inscription_locale_ouverte: bool | None = None, porte_centrale_active: bool | None = None) -> dict: """Enregistre les réglages d'accès (ligne unique) et les renvoie.""" from app.db import get_conn actuels = get_reglages_acces() if inscription_locale_ouverte is not None: actuels["inscription_locale_ouverte"] = bool(inscription_locale_ouverte) if porte_centrale_active is not None: actuels["porte_centrale_active"] = bool(porte_centrale_active) with get_conn() as conn: _assurer_table(conn) conn.execute( "INSERT INTO reglages_acces (id, inscription_locale_ouverte," " porte_centrale_active, updated_at) VALUES (1, ?, ?," " CURRENT_TIMESTAMP)" " ON CONFLICT(id) DO UPDATE SET" " inscription_locale_ouverte=excluded.inscription_locale_ouverte," " porte_centrale_active=excluded.porte_centrale_active," " updated_at=CURRENT_TIMESTAMP", (1 if actuels["inscription_locale_ouverte"] else 0, 1 if actuels["porte_centrale_active"] else 0), ) conn.commit() return actuels def inscription_locale_ouverte() -> bool: """Vrai si l'inscription publique locale est ouverte sur l'instance.""" return bool(get_reglages_acces()["inscription_locale_ouverte"]) def porte_centrale_active() -> bool: """Vrai si la porte centrale unique est active sur l'instance.""" return bool(get_reglages_acces()["porte_centrale_active"]) def message_porte_centrale() -> str: """Message affiché quand la porte centrale refuse le local.""" from app.services import draco_urls return ( "La porte centrale est active sur cette instance : " "connectez-vous avec le bouton « Login with ObsiGate (SSO) » " "— les comptes se créent dans le Hub Dracodev " f"({draco_urls.hub_url()})." ) def _secours_identifiants() -> tuple[str, str]: """Identifiants de la porte de secours — variables d'environnement en priorité, repli sur la configuration. Jamais de valeur par défaut : vide = porte de secours fermée.""" login = (os.environ.get("FLOWDECK_SECOURS_ADMIN_LOGIN", "") or getattr(settings, "secours_admin_login", "") or "").strip() mot_de_passe = (os.environ.get("FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE", "") or getattr(settings, "secours_admin_mot_de_passe", "") or "") return login, mot_de_passe def authentifier_secours(login: str, mot_de_passe: str) -> dict | None: """Authentifie le compte administrateur de secours désigné. Les identifiants fournis doivent correspondre **exactement** aux variables d'environnement de la porte de secours (comparaison à temps constant), et le compte local correspondant doit exister, être actif et être administrateur. Renvoie le compte (dict) ou ``None`` — sans jamais dire quelle condition a échoué. """ attendu_login, attendu_mdp = _secours_identifiants() if not attendu_login or not attendu_mdp: return None if not secrets.compare_digest((login or "").strip(), attendu_login): return None if not secrets.compare_digest(mot_de_passe or "", attendu_mdp): return None from app.db import get_conn with get_conn() as conn: row = conn.execute( "SELECT * FROM users WHERE login=? OR email=?", (attendu_login, attendu_login), ).fetchone() if row is None: return None compte = dict(row) if not compte.get("is_admin") or not compte.get("is_active"): return None return compte def message_inscription_fermee() -> str: """Message affiché quand l'inscription locale est fermée (#146). Nomme le Hub Dracodev — là où les comptes se créent — avec son URL dérivée de l'hôte comme les autres liens de la suite (``draco_urls.hub_url()``, jamais un nom figé). """ from app.services import draco_urls return ( "L'inscription locale est fermée sur cette instance : " "les comptes se créent dans le Hub Dracodev " f"({draco_urls.hub_url()}), puis se connectent ici avec le " "bouton « Login with ObsiGate (SSO) »." )