216 lines
8.4 KiB
Python
216 lines
8.4 KiB
Python
"""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) »."
|
|
)
|