Kanbante 0.46.6 — porte centrale unique : réglage administrateur, écran réduit à l'entrée SSO et porte de secours discrète
CI / test (pull_request) Failing after 2m20s
CI / test (push) Failing after 2m54s

This commit is contained in:
Spark
2026-11-07 04:19:08 +00:00
parent f8f8af63e1
commit e4531488d0
9 changed files with 504 additions and 16 deletions
+133 -15
View File
@@ -249,7 +249,7 @@ registry = connections_service.ForgeRegistry()
SESSION_COOKIE = "kanbante_session"
SESSION_DAYS = 7
_PUBLIC_PATHS = ("/api/health", "/webhooks/gitea", "/api/capture", "/login",
"/auth/draco")
"/login/secours", "/auth/draco")
def _session_secret() -> bytes:
@@ -536,11 +536,30 @@ def _safe_next(raw: str) -> str:
return "/"
def _ouvrir_session(username: str, next: str) -> RedirectResponse:
"""Réponse de connexion native : session signée posée, redirection."""
response = RedirectResponse(_safe_next(next), status_code=303)
response.set_cookie(
SESSION_COOKIE,
_make_session(username),
max_age=SESSION_DAYS * 86400,
httponly=True,
samesite="lax",
)
return response
@app.get("/login")
async def login_page(request: Request, next: str = "/"):
if not get_settings().auth_enabled or _session_user(request) is not None:
return RedirectResponse(_safe_next(next), status_code=303)
return render(request, "login.html", next=_safe_next(next), error="")
# Porte centrale (0.46.6) : en mode actif, le gabarit ne
# présente que l'entrée centrale (lien SSO) — aucun champ
# local (voir app/templates/login.html et docs/porte-centrale.md).
return render(
request, "login.html", next=_safe_next(next), error="",
porte_centrale=bool(get_settings().porte_centrale),
)
@app.post("/login")
@@ -559,6 +578,25 @@ async def login_submit(
request, "login.html", status_code=403,
next=_safe_next(next), error="Session expirée — réessaie.",
)
if settings.porte_centrale:
# Porte centrale unique (0.46.6) : en mode actif, les
# identifiants locaux ordinaires ne sont plus acceptés
# sur ce formulaire — ni mot de passe de base, ni repli
# d'environnement. Seules restent : l'entrée centrale
# (SSO ObsiGate, ci-dessous sur l'écran) et, pour
# l'administrateur désigné, la porte de secours discrète
# (/login/secours, identifiants d'environnement — jamais
# proposée ici). Les sessions déjà établies ne passent
# pas par ce formulaire et ne sont pas touchées.
return render(
request, "login.html", status_code=403,
next=_safe_next(next), porte_centrale=True,
error=(
"La porte centrale est active sur cette instance : "
"la connexion locale est fermée. Utilise « Se "
"connecter avec ObsiGate (SSO) » ci-dessous."
),
)
from app import accounts as accounts_service
user = models.get_user_by_username(username)
@@ -622,15 +660,88 @@ async def login_submit(
sub = claims.get("sub")
if sub and models.get_user_by_draco(sub) is None:
models.update_user(user["id"], draco_sub=sub)
response = RedirectResponse(_safe_next(next), status_code=303)
response.set_cookie(
SESSION_COOKIE,
_make_session(username),
max_age=SESSION_DAYS * 86400,
httponly=True,
samesite="lax",
return _ouvrir_session(username, next)
# ── Porte de secours administrateur (0.46.6) ──────────────────
# Chemin discret /login/secours : jamais lié ni proposé sur
# l'écran principal (ni en mode porte centrale, ni sinon).
# Un seul compte, désigné par l'environnement :
# KANBANTE_PORTE_SECOURS_USER / KANBANTE_PORTE_SECOURS_PASSWORD
# (voir app/config.py). Le compte désigné doit être un
# administrateur local actif (ou l'opérateur d'environnement) :
# un membre désigné par erreur ne passe pas. Sans les deux
# variables, la porte n'existe pas (404 sec, sans explication).
def _porte_secours_configuree(settings) -> bool:
return bool(settings.porte_secours_user and settings.porte_secours_password)
@app.get("/login/secours")
async def login_secours_page(request: Request, next: str = "/"):
settings = get_settings()
if not settings.auth_enabled or _session_user(request) is not None:
return RedirectResponse(_safe_next(next), status_code=303)
if not _porte_secours_configuree(settings):
return render(
request, "error.html", status_code=404,
message="Page introuvable.",
)
return render(
request, "login.html", next=_safe_next(next), error="",
mode_secours=True,
)
return response
@app.post("/login/secours")
async def login_secours_submit(
request: Request,
username: str = Form(""),
password: str = Form(""),
next: str = Form("/"),
csrf_token: str = Form(""),
):
settings = get_settings()
if not settings.auth_enabled:
return RedirectResponse("/", status_code=303)
if not _porte_secours_configuree(settings):
return render(
request, "error.html", status_code=404,
message="Page introuvable.",
)
if not _csrf_valid(request, csrf_token):
return render(
request, "login.html", status_code=403, mode_secours=True,
next=_safe_next(next), error="Session expirée — réessaie.",
)
identifiants_ok = secrets.compare_digest(
username, settings.porte_secours_user
) and secrets.compare_digest(password, settings.porte_secours_password)
compte = (
models.get_user_by_username(settings.porte_secours_user)
if identifiants_ok
else None
)
if compte is not None:
admin_ok = bool(compte["active"]) and compte["role"] == "admin"
else:
# Pas de ligne en base : seul l'opérateur d'environnement
# (administrateur synthétique) peut être le compte désigné.
admin_ok = settings.porte_secours_user == settings.auth_user
if not identifiants_ok or not admin_ok:
return render(
request, "login.html", status_code=401, mode_secours=True,
next=_safe_next(next), error="Identifiants incorrects.",
)
if compte is not None:
models.update_user(
compte["id"],
last_login_at=time.strftime(
"%Y-%m-%d %H:%M:%S", time.gmtime()
),
)
models.log_admin(compte["id"], "login.porte_secours", username)
return _ouvrir_session(settings.porte_secours_user, next)
@app.get("/logout")
@@ -2090,11 +2201,18 @@ async def admin_settings_save(request: Request, csrf_token: str = Form("")):
if raw is None or str(raw).strip() == "":
continue # champ vide = retour à la valeur d'environnement
try:
overrides[key] = (
str(raw).strip()
if converter is str
else converter(str(raw).strip())
)
if converter is bool:
# Booléen lu EXPLICITEMENT (0.46.6) : bool("false")
# vaudrait True — un réglage booléen du panneau
# (porte_centrale, sync_*, briefing_*) doit pouvoir
# se désactiver en saisissant « false ».
overrides[key] = str(raw).strip().lower() in (
"1", "true", "on", "oui",
)
elif converter is str:
overrides[key] = str(raw).strip()
else:
overrides[key] = converter(str(raw).strip())
except (TypeError, ValueError):
continue
import json as _json