Compare commits

...
5 Commits
Author SHA1 Message Date
bruno c1aa6697f0 Merge pull request 'Flowdeck 7.85.7 — cohérence d'identité au changement de compte' (#59) from coherence-identite-session-draco into main
FlowDeck CI / test (push) Failing after 13m13s
FlowDeck CI / lint (push) Failing after 2s
FlowDeck CI / docker (push) Skipped
2026-11-08 14:43:23 -05:00
Spark f8b228396d Flowdeck 7.85.8 — entrée SSO automatique sans session locale (chantier entrée SSO, empilé sur la 7.85.7)
FlowDeck CI / lint (push) Failing after 1s
FlowDeck CI / lint (pull_request) Failing after 2s
FlowDeck CI / test (pull_request) Failing after 10m30s
FlowDeck CI / docker (pull_request) Skipped
FlowDeck CI / test (push) Failing after 3h1m6s
FlowDeck CI / docker (push) Skipped
La racine, sans session locale et sur instance peuplée, renvoie
au pont /auth/draco quand un cookie draco valide est présenté,
au lieu de servir la page publique déconnectée. Instance vierge
et formulaire local (?provider=local, le chemin prévu) jamais
interceptés ; sans cookie ou cookie invalide, page publique
inchangée. Symétrique du contrôle de cohérence de la 7.85.7
(session locale d'un autre compte).

Tests : tests/test_entree_sso.py (5) ; suites entrée + cohérence
12/12 ; suite complète comparée à la base (voir PR).
2026-11-08 18:08:35 +00:00
Spark a4852d3889 Flowdeck 7.85.7 — cohérence d'identité au changement de compte : la session locale s'aligne sur le cookie draco (middleware pages, marqueurs de session draco_sub/secours)
FlowDeck CI / lint (push) Failing after 0s
FlowDeck CI / test (push) Failing after 1s
FlowDeck CI / docker (push) Skipped
FlowDeck CI / lint (pull_request) Failing after 1s
FlowDeck CI / test (pull_request) Failing after 1s
FlowDeck CI / docker (pull_request) Skipped
2026-11-08 04:34:10 +00:00
bruno bed7857ada Merge pull request 'Flowdeck 7.85.6 — porte centrale unique (phase 2) : écran réduit à l'entrée ObsiGate SSO + porte de secours administrateur, désactivée par défaut' (#58) from porte-unique-phase2-porte-centrale into main
FlowDeck CI / lint (push) Failing after 4s
FlowDeck CI / test (push) Failing after 4s
FlowDeck CI / docker (push) Skipped
2026-11-07 10:31:54 -05:00
Spark cf2e39e15f Flowdeck 7.85.6 — porte centrale unique (phase 2) : écran réduit à l'entrée ObsiGate SSO + porte de secours administrateur par variables d'environnement, désactivée par défaut
FlowDeck CI / lint (push) Failing after 1s
FlowDeck CI / test (push) Failing after 0s
FlowDeck CI / docker (push) Skipped
2026-11-07 04:52:51 +00:00
14 changed files with 1021 additions and 37 deletions
+26
View File
@@ -1,3 +1,29 @@
## v7.85.8 (2026-11-08) — Entrée SSO automatique sans session locale (chantier « entrée SSO de bout en bout », empilée sur la 7.85.7)
Écart constaté par Bruno en test réel le 2026-11-08 (compte « testeur » créé au Hub) : arrivé chez Flowdeck avec une session centrale valide mais aucune session locale, l'application servait sa page publique déconnectée (« Log in », « Créer un compte dans le Hub ») comme si le cookie draco n'existait pas. Chaîne prouvée en navigateur réel dans une pile locale aux noms publics.
- **Entrée automatique par le pont à la racine** (`app/routers/dashboard/workspace.py`) : sans session locale, si l'instance est peuplée et qu'un cookie `draco_session` **valide** (`app.services.draco.claims_valides`) est présenté, la racine `/` renvoie au pont `/auth/draco?next=/workspaces` (JIT si besoin) au lieu de servir la page publique. **Gardes** : instance vierge (aucun compte) → page publique d'installation inchangée ; cookie absent ou invalide → page publique inchangée ; le pont tranche ensuite (rôle « aucun » → sa page 403 ; échec de résolution → retour au formulaire local `/auth/login?provider=local` — le chemin prévu, celui de tous les liens d'entrée internes — qui n'est jamais intercepté : pas de boucle). Symétrique du contrôle de cohérence de la 7.85.7 (middleware), qui couvre le cas inverse (session locale d'un AUTRE compte).
- **Tests** : nouveau `tests/test_entree_sso.py` (5) — racine renvoyée au pont, chaîne complète jusqu'à la session ouverte, page publique inchangée sans cookie et avec cookie invalide, formulaire local jamais intercepté. Suites entrée + cohérence : 12/12.
## v7.85.7 (2026-11-07) — Cohérence d'identité au changement de compte
Cas réel (capture) : connecté « fred » chez ObsiGate, l'utilisateur ouvre Flowdeck et y trouve son propre compte (BC) — la session locale préexistante gouvernait sans reconsulter l'identité centrale, le pont `/auth/draco` ne s'exécutant qu'à l'entrée SSO sans session locale.
- **Nouveau middleware `CoherenceDracoMiddleware`** (`app/middleware/coherence_draco.py`, le plus externe) : sur les **pages** (GET hors `/api/`, `/auth/`, `/static/`, `/s/`, `/f/`, `/scim/`), quand la requête porte une session locale valide ET un cookie `draco_session` **valide** (nouveau helper public `app.services.draco.claims_valides` : signature, expiration, révocation, `sub` présent) dont le `sub` diffère de celui que la session sert, la session locale est **révoquée** (`user_sessions`), le cookie est supprimé de la réponse et le navigateur est renvoyé au pont `/auth/draco?next=…` (JIT si besoin). Base de comparaison : le `draco_sub` inscrit dans le payload signé à la création par le pont — sinon le lien actif du compte (`draco.current_link`). Le sub inscrit neutralise le repli mono-utilisateur de la résolution (en mono-utilisateur, un sub étranger résout vers l'unique compte et ne doit pas détruire sa session en boucle).
- **Gardes** : aucun cookie ou cookie invalide/expiré/révoqué → rien (une panne de l'émettrice ne déconnecte personne) ; cookie que le pont ne peut pas servir (rôle « aucun » — `RoleAucunRefuse` —, dissociation, compte inactif) ou qui résout vers le même compte → session gardée, pas d'exclusion en boucle ; **compte local sans aucun lien** → l'identité centrale présente gagne (choix documenté ; en mode porte centrale, ce cas n'existe pas hors secours). **API non traversée** : un appel en vol n'est ni interrompu ni rejoué sous une autre identité ; dès la navigation suivante, la page sert l'identité du cookie et la session alignée gouverne l'API.
- **Marqueurs de session** (`app/auth/session.py`) : `create_session(..., draco_sub=…, secours=…)` ; le pont pose `draco_sub`, la connexion de la porte centrale pose `secours=True` (sessions exemptées du contrôle) ; `refresh_session` préserve les marqueurs ; nouveau `decode_payload` (payload complet, sans touch) sur lequel `decode_session` s'appuie.
- **Tests** : nouveau `tests/test_coherence_identite.py` (7) — le cas de la capture (session révoquée en base, pont servant ensuite le compte du cookie, identité lue par `/auth/user`), même sub / sans cookie / cookie invalide intacts, secours exemptée, compte sans lien aligné, marqueur inscrit par le pont et alignement au changement de cookie.
## v7.85.6 (2026-11-06) — Porte centrale unique (phase 2 du programme « compte partout »), désactivée par défaut
> Branche basée sur la 7.85.5 (phase 1, #146/#144) — PR distincte, à fusionner après elle.
- **Nouveau réglage « porte centrale »** (`porte_centrale_active` dans `app/services/reglages_acces.py`, même patron que le réglage d'inscription de la phase 1, **distinct** de lui, **désactivé par défaut à la livraison** ; activable/désactivable par un administrateur via `GET`/`PUT /api/admin/reglages-acces` ou la variable `PORTE_CENTRALE_ACTIVE`).
- **Mode actif** : l'écran de connexion ne présente que l'entrée centrale — le bouton « Login with ObsiGate (SSO) », bien visible, et le lien vers le Hub Dracodev où les comptes se créent (URL dérivée de l'hôte). `POST /auth/local-login` refuse les identifiants locaux ordinaires (403, message nommant la voie SSO), **y compris pour les comptes locaux historiques** ; l'inscription est également fermée dans ce mode. 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 `GET /auth/login?provider=local&secours=1` (le formulaire y marque ses envois `secours` ; 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` (comparaison à temps constant ; le compte local correspondant doit exister, être actif et administrateur ; aucun secret en dur, aucune valeur par défaut — sans variables, la porte est fermée).
- **Mode inactif** : comportement actuel intégralement préservé (formulaire local, inscription selon le réglage de la phase 1).
- **Tests** : `tests/test_porte_centrale.py` (nouveau, 15 tests) — défaut désactivé vérifié dans le code, réglage distinct de celui d'inscription, refus du compte ordinaire et du compte historique en mode actif, écran réduit à l'entrée centrale (aucun formulaire, aucun « secours »), inscription fermée en mode actif, session établie préservée, secours : écran discret, connexion admin réussie, mauvais mot de passe refusé, absence de drapeau refusée, compte non-admin désigné refusé, absence de variables = porte fermée ; mode inactif = comportement actuel ; aller-retour de l'API d'administration.
## v7.85.5 (2026-11-06) — Inscription locale fermée par défaut + échec de connexion qui nomme la voie SSO (fusion #146, #144)
- **#146 — l'inscription publique locale n'est plus ouverte par défaut** : nouveau réglage d'administration « inscription locale ouverte » (`app/services/reglages_acces.py`, patron `llm_config`/`sso_config` : ligne unique en base qui prime sur la variable `INSCRIPTION_LOCALE_OUVERTE`, défaut **fermée** — décision de Bruno : les comptes naissent dans le Hub Dracodev). Réversible à tout moment par un administrateur via `GET`/`PUT /api/admin/reglages-acces`.
+1 -1
View File
@@ -1 +1 @@
7.85.5
7.85.8
+43 -5
View File
@@ -36,16 +36,28 @@ class SessionManager:
"""
@staticmethod
def create_session(user_data: dict, request=None) -> str:
def create_session(user_data: dict, request=None, draco_sub: str | None = None,
secours: bool = False) -> str:
"""Create a signed session cookie value.
``request`` is optional — when provided the session is recorded in the
``user_sessions`` table (ip + user agent) and becomes revocable.
Cohérence d'identité (7.85.7) : ``draco_sub`` inscrit dans le
payload signé le sub draco que la session sert (posé par le
pont) — c'est la base de comparaison du contrôle de
cohérence, insensible au repli mono-utilisateur de la
résolution ; ``secours=True`` marque une session de la porte
de secours, exemptée de ce contrôle.
"""
payload = {
"user": public_user(user_data),
"created_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
}
if draco_sub:
payload["draco_sub"] = str(draco_sub)
if secours:
payload["secours"] = True
user_id = user_data.get("id")
if user_id:
sid = str(uuid4())
@@ -54,17 +66,31 @@ class SessionManager:
return _serializer.dumps(payload)
@staticmethod
def decode_session(cookie: str) -> dict | None:
"""Decode and validate a session cookie. Returns user data or None."""
def decode_payload(cookie: str) -> dict | None:
"""Decode the full session payload (user, sid, markers), or None.
Same validity checks as ``decode_session`` (signature,
expiration, server-side revocation) but no ``last_seen``
touch and no unwrapping — the coherence middleware reads
the identity markers from here.
"""
try:
payload = _serializer.loads(cookie, max_age=86400 * 7) # 7 days
except (BadSignature, SignatureExpired):
return None
sid = payload.get("sid") or ""
if sid and not _session_active(sid):
# Revoked or deleted session → treat as logged out.
return None
return payload
@staticmethod
def decode_session(cookie: str) -> dict | None:
"""Decode and validate a session cookie. Returns user data or None."""
payload = SessionManager.decode_payload(cookie)
if payload is None:
return None
sid = payload.get("sid") or ""
if sid:
_touch_session(sid)
return payload.get("user")
@@ -103,12 +129,24 @@ class SessionManager:
@staticmethod
def refresh_session(cookie: str, user_data: dict, request=None) -> str:
"""Re-sign a cookie keeping its session id (used after profile edits)."""
"""Re-sign a cookie keeping its session id (used after profile edits).
Les marqueurs de cohérence (``draco_sub``, ``secours`` —
7.85.7) de l'ancien payload sont préservés : un rafraîchi
de profil ne doit ni perdre l'identité servie ni l'exemption
de secours.
"""
sid = SessionManager.session_id(cookie) if cookie else None
payload = {
"user": public_user(user_data),
"created_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
}
if cookie:
ancien = SessionManager.decode_payload(cookie)
if ancien:
for marqueur in ("draco_sub", "secours"):
if ancien.get(marqueur):
payload[marqueur] = ancien[marqueur]
user_id = user_data.get("id")
if user_id:
if sid is None:
+12
View File
@@ -52,6 +52,18 @@ class Settings(BaseSettings):
# Inscription publique locale : fermée par défaut (décision de
# Bruno — les comptes naissent dans le Hub Dracodev).
inscription_locale_ouverte: bool = False
# Porte centrale unique (phase 2 du programme « compte partout ») :
# désactivée par défaut à la livraison ; même patron de réglage
# (la ligne en base prime). Activée, seuls la voie ObsiGate (SSO)
# et la porte de secours administrateur ouvrent une session.
porte_centrale_active: bool = False
# Porte de secours administrateur (porte centrale) — repli des
# variables d'environnement FLOWDECK_SECOURS_ADMIN_LOGIN /
# FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE, lues en priorité par
# `app/services/reglages_acces.py`. Aucune valeur par défaut :
# sans variables posées, la porte de secours est fermée.
secours_admin_login: str = ""
secours_admin_mot_de_passe: str = ""
# Rate limiting
rate_limit_enabled: bool = True
+8
View File
@@ -280,6 +280,14 @@ app.add_middleware(
from app.middleware.recherche_cors import RechercheCorsMiddleware
app.add_middleware(RechercheCorsMiddleware)
# Cohérence d'identité (7.85.7) : ajoutée en dernier, donc la plus
# externe — une page dont la session locale contredit le cookie
# draco valide présenté est renvoyée au pont avant tout service.
# Le middleware ne traverse que les GET de pages (voir son module).
from app.middleware.coherence_draco import CoherenceDracoMiddleware
app.add_middleware(CoherenceDracoMiddleware)
app.include_router(auth.router)
app.include_router(sso_router)
app.include_router(dashboard.router)
+110
View File
@@ -0,0 +1,110 @@
"""FlowDeck — cohérence d'identité au changement de compte (7.85.7).
Cas réel : connecté « fred » chez ObsiGate, l'utilisateur ouvre
Flowdeck et y trouve son propre compte (BC) — la session locale
préexistante gouvernait sans jamais reconsulter l'identité
centrale, le pont ``/auth/draco`` ne s'exécutant qu'à l'entrée
SSO sans session locale.
Ce middleware aligne la session sur l'identité centrale
**présentée** : quand une requête de page (GET hors ``/api/``,
``/auth/``, ``/static/``, ``/s/``, ``/f/``, ``/scim/``) porte à la
fois une session locale valide et un cookie ``draco_session``
**valide** (``app.services.draco.claims_valides`` : signature,
expiration, révocation) dont le ``sub`` diffère de celui que la
session sert, la session locale est révoquée (ligne
``user_sessions``), le cookie est supprimé de la réponse, et le
navigateur est renvoyé au pont ``/auth/draco?next=…`` (JIT si
besoin). Base de comparaison : le ``draco_sub`` inscrit dans le
payload signé de la session à sa création par le pont — sinon le
lien actif du compte (``draco.current_link``). Le sub inscrit
neutralise le repli mono-utilisateur de la résolution : en
mono-utilisateur, un sub étranger résout vers l'unique compte et
ne doit pas détruire sa session en boucle.
Gardes :
- aucun cookie draco, ou cookie invalide/expiré/révoqué → rien :
une panne ou une absence de l'émettrice ne déconnecte personne ;
- session de la **porte de secours** (marqueur ``secours`` du
payload) → exemptée : elle vit hors circuit central ;
- cookie que le pont ne peut pas servir (rôle « aucun » au
registre — ``user_for_draco`` lève ``RoleAucunRefuse`` —,
dissociation, compte inactif) ou qui résout vers le **même**
compte → session gardée : la détruire exclurait son titulaire
en boucle, le cookie refusé restant présent ;
- compte local **sans aucun lien** draco → l'identité centrale
présente gagne (le pont s'applique ; choix documenté). En mode
porte centrale, ce cas n'existe pas hors secours.
Périmètre : les **pages**. L'API n'est pas traversée par ce
middleware : un appel en vol n'est ni interrompu ni rejoué sous
une autre identité ; dès la navigation suivante, la page sert
l'identité du cookie et la session alignée gouverne l'API.
Après alignement, le ``draco_sub`` de la nouvelle session égale
celui du cookie : pas de boucle.
"""
from __future__ import annotations
import logging
from urllib.parse import quote
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import RedirectResponse
from app.auth.session import SessionManager
logger = logging.getLogger(__name__)
_EXEMPT_PREFIXES = ("/api/", "/auth/", "/static/", "/s/", "/f/", "/scim/")
class CoherenceDracoMiddleware(BaseHTTPMiddleware):
"""Aligne la session locale sur le cookie draco valide présenté."""
async def dispatch(self, request: Request, call_next):
if request.method != "GET":
return await call_next(request)
path = request.url.path
if any(path.startswith(p) for p in _EXEMPT_PREFIXES):
return await call_next(request)
cookie = request.cookies.get("flowdeck_session", "")
if not cookie:
return await call_next(request)
payload = SessionManager.decode_payload(cookie)
if payload is None:
return await call_next(request)
if payload.get("secours"):
return await call_next(request)
user = payload.get("user") or {}
user_id = user.get("id")
if not user_id:
return await call_next(request)
from app.services import draco
claims = draco.claims_valides(request)
if claims is None:
return await call_next(request)
sub_cookie = str(claims.get("sub", ""))
base = payload.get("draco_sub") or draco.current_link(user_id)
if base == sub_cookie:
return await call_next(request)
# Servabilité : le pont doit pouvoir servir un AUTRE compte,
# sinon la session est gardée (pas d'exclusion en boucle).
try:
candidat = draco.user_for_draco(request)
except Exception:
candidat = None
if candidat is None or candidat.get("id") == user_id:
return await call_next(request)
sid = payload.get("sid")
if sid:
SessionManager.revoke_session(sid)
cible = path
if request.url.query:
cible = f"{path}?{request.url.query}"
response = RedirectResponse(
url=f"/auth/draco?next={quote(cible, safe='')}", status_code=302)
response.delete_cookie("flowdeck_session", path="/")
return response
+14 -6
View File
@@ -158,15 +158,23 @@ def get_reglages_acces(_admin=Depends(admin_required)):
def put_reglages_acces(_admin=Depends(admin_required), body: dict = Body(default={})):
"""Modifie les réglages d'accès — réversible à tout moment.
Corps accepté : ``{"inscription_locale_ouverte": true|false}``.
Fermée (défaut) : aucun compte local ne se crée sur l'instance,
les comptes naissent dans le Hub Dracodev. La connexion des
comptes existants n'est jamais affectée.
Corps accepté : ``{"inscription_locale_ouverte": true|false,
"porte_centrale_active": true|false}`` (chaque clé optionnelle).
Inscription fermée (défaut) : aucun compte local ne se crée sur
l'instance, les comptes naissent dans le Hub Dracodev.
Porte centrale (défaut : inactive) : activée, seuls la voie
ObsiGate (SSO) et la porte de secours administrateur ouvrent
une session.
"""
from app.services import reglages_acces
kwargs = {}
if "inscription_locale_ouverte" in body:
return reglages_acces.set_reglages_acces(
inscription_locale_ouverte=bool(body["inscription_locale_ouverte"]))
kwargs["inscription_locale_ouverte"] = bool(
body["inscription_locale_ouverte"])
if "porte_centrale_active" in body:
kwargs["porte_centrale_active"] = bool(body["porte_centrale_active"])
if kwargs:
return reglages_acces.set_reglages_acces(**kwargs)
return reglages_acces.get_reglages_acces()
+111 -7
View File
@@ -94,15 +94,64 @@ body{{background:#191919;color:#fff;font-family:-apple-system,BlinkMacSystemFont
</html>"""
def _rendre_page_connexion(html: str) -> str:
"""Rend la page de connexion en tenant compte du réglage #146.
def _page_porte_centrale() -> str:
"""Écran de connexion en mode porte centrale (phase 2).
Inscription fermée : l'onglet Register disparaît et un avis nomme
le Hub Dracodev (lien dérivé de l'hôte, comme les autres liens de
la suite). Inscription ouverte : page intégralement inchangée.
Ne présente **que** l'entrée centrale : le bouton ObsiGate (SSO),
bien visible. Aucun formulaire local, aucun onglet Register —
la porte de secours administrateur passe par le chemin discret
``?secours=1``, jamais proposé ici.
"""
from app.services import draco_urls
hub = draco_urls.hub_url()
return """<!DOCTYPE html>
<html lang="fr">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>FlowDeck — Connexion</title>
<style>
body{background:#191919;color:#fff;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0;}
.box{background:#222;border:1px solid rgba(255,255,255,.08);border-radius:12px;padding:40px;width:100%;max-width:400px;}
.box h1{font-size:24px;margin:0 0 8px;}
.box p{color:rgba(255,255,255,.55);font-size:14px;line-height:1.6;margin:0 0 20px;}
.box a{color:#2383E2;}
.sso{display:flex;align-items:center;justify-content:center;gap:8px;width:100%;padding:14px;border-radius:8px;font-size:15px;font-weight:500;cursor:pointer;border:1px solid rgba(35,131,226,.6);background:#2383E2;color:#fff;text-decoration:none;box-sizing:border-box;}
.sso:hover{background:#2C8CEB;}
</style>
</head>
<body>
<div class="box">
<h1>FlowDeck</h1>
<p>Cette instance n'accepte que la connexion centrale de la suite
Dracodev. Les comptes se créent dans le
<a href="__HUB_URL__">Hub Dracodev</a>.</p>
<a class="sso" href="__SSO_LOGIN_URL__">🧭 Login with ObsiGate (SSO)</a>
</div>
</body>
</html>""".replace("__HUB_URL__", hub)
def _rendre_page_connexion(html: str, secours: bool = False) -> str:
"""Rend la page de connexion en tenant compte des réglages d'accès.
Porte centrale active (phase 2) : seul l'écran central est rendu
— sauf par le chemin discret de secours (``secours=True``), qui
rend le formulaire local en marquant ses envois ``secours``.
Inscription fermée (#146) : l'onglet Register disparaît et un
avis nomme le Hub Dracodev (lien dérivé de l'hôte, comme les
autres liens de la suite). Sinon : page intégralement inchangée.
"""
from app.services import draco_urls, reglages_acces
if reglages_acces.porte_centrale_active():
if not secours:
return _with_nonce(_page_porte_centrale())
html = html.replace(
"const body={email,password};",
"const body={email,password,secours:true};")
return _with_nonce(html)
if reglages_acces.inscription_locale_ouverte():
return _with_nonce(html)
html = html.replace(
@@ -257,6 +306,9 @@ def register_page(request: Request):
"""
from app.services import reglages_acces
if reglages_acces.porte_centrale_active():
return HTMLResponse(_with_nonce(_page_porte_centrale()),
status_code=403)
if not reglages_acces.inscription_locale_ouverte():
return HTMLResponse(_page_inscription_fermee(), status_code=403)
return HTMLResponse(_with_nonce(LOCAL_LOGIN_HTML.replace(
@@ -283,7 +335,13 @@ def login(request: Request, provider: str = Query("gitea")):
# Local login page (POST handled by /auth/local-login)
from fastapi.responses import HTMLResponse
if provider == "local":
return HTMLResponse(_rendre_page_connexion(LOCAL_LOGIN_HTML), status_code=200)
# Chemin discret de la porte de secours (phase 2) :
# ?secours=1 rend le formulaire local même en mode porte
# centrale — jamais lié depuis l'écran principal.
secours = request.query_params.get("secours") == "1"
return HTMLResponse(
_rendre_page_connexion(LOCAL_LOGIN_HTML, secours=secours),
status_code=200)
# OAuth flow — check if provider is configured
from app.auth.providers import get_provider
@@ -341,6 +399,13 @@ def register(request: Request, body: dict = Body(default={})):
# naissent dans le Hub Dracodev — aucun compte local ne se crée
# ici. La connexion des comptes existants n'est pas touchée.
from app.services import draco_urls, reglages_acces
if reglages_acces.porte_centrale_active():
from fastapi.responses import JSONResponse
return JSONResponse(
{"error": reglages_acces.message_porte_centrale(),
"hub_url": draco_urls.hub_url()},
status_code=403,
)
if not reglages_acces.inscription_locale_ouverte():
from fastapi.responses import JSONResponse
return JSONResponse(
@@ -400,6 +465,38 @@ def local_login(request: Request, body: dict = Body(default={})):
if not email or not password:
return JSONResponse({"error": "Email and password required"}, status_code=400)
# Porte centrale unique (phase 2) : les identifiants locaux
# ordinaires ne sont plus acceptés — comptes historiques inclus.
# Seule exception : la porte de secours administrateur, par le
# drapeau explicite `secours` (posé par le chemin discret) et les
# identifiants des variables d'environnement. Les sessions déjà
# établies ne passent pas par ici et ne sont pas touchées.
from app.services import reglages_acces as _reglages_acces
if _reglages_acces.porte_centrale_active():
if not body.get("secours"):
return JSONResponse(
{"error": _reglages_acces.message_porte_centrale()},
status_code=403,
)
compte_secours = _reglages_acces.authentifier_secours(email, password)
if compte_secours is None:
return JSONResponse(
{"error": "Porte de secours : identifiants invalides."},
status_code=401,
)
# Session marquée « secours » (7.85.7) : exemptée du
# contrôle de cohérence d'identité — elle vit hors circuit
# central par construction.
session = SessionManager.create_session(
compte_secours, request, secours=True)
_log_login(compte_secours["id"], request)
response = JSONResponse({"status": "ok", "user": {
"login": compte_secours["login"],
"name": compte_secours["full_name"]}})
response.set_cookie("flowdeck_session", session, httponly=True,
max_age=86400 * 7, samesite="lax", path="/")
return response
with get_conn() as conn:
user = conn.execute("SELECT * FROM users WHERE login=?", (email,)).fetchone()
if not user:
@@ -608,7 +705,14 @@ de l'instance reste possible par la page de connexion.</p>
)
if user is None:
return RedirectResponse(url="/auth/login?provider=local", status_code=302)
session = SessionManager.create_session(user, request)
# Cohérence d'identité (7.85.7) : la session créée par le pont
# inscrit le sub qu'elle sert dans son payload signé — base de
# comparaison du contrôle de cohérence, insensible au repli
# mono-utilisateur de la résolution.
_claims = draco.claims_valides(request)
session = SessionManager.create_session(
user, request,
draco_sub=str(_claims.get("sub", "")) if _claims else None)
_log_login(user["id"], request)
cible = next if next.startswith("/") and not next.startswith("//") else "/workspaces"
response = RedirectResponse(url=cible, status_code=302)
+28 -6
View File
@@ -55,16 +55,38 @@ async def dashboard(
# ── Not authenticated → show landing page ──
if not user:
# Allow through if DB is empty (fresh install)
_peuplee = False
try:
with get_conn() as conn:
count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
if count == 0:
from app.templating import ENV
env = ENV
template = env.get_template("landing.html")
return template.render(**_contexte_accueil())
_peuplee = conn.execute(
"SELECT COUNT(*) FROM users").fetchone()[0] > 0
except Exception:
logger.exception("dashboard")
if _peuplee:
# Entrée automatique par le pont (7.85.8, chantier
# « entrée SSO de bout en bout ») : un cookie draco
# VALIDE présenté sans session locale ne doit pas
# aboutir à la page publique comme si la session de
# la suite n'existait pas — le navigateur est renvoyé
# au pont /auth/draco (JIT si besoin), qui revient à
# /workspaces. Gardes : instance vierge (aucun compte)
# → page publique d'installation, inchangée ; cookie
# absent ou invalide → page publique, inchangée ; le
# pont tranche ensuite (rôle « aucun » → sa page 403 ;
# échec de résolution → retour au formulaire local
# /auth/login?provider=local, le chemin prévu, qui
# n'est jamais intercepté : pas de boucle). Symétrique
# du contrôle de cohérence de la 7.85.7 (middleware),
# qui couvre le cas inverse (session locale d'un
# AUTRE compte).
try:
from app.services import draco as _draco_entree
if _draco_entree.claims_valides(request) is not None:
return RedirectResponse(
url="/auth/draco?next=/workspaces", status_code=302)
except Exception:
logger.exception("dashboard — entrée automatique par le pont")
from app.templating import ENV
env = ENV
template = env.get_template("landing.html")
+24
View File
@@ -149,6 +149,30 @@ def is_revoked(jti: str) -> bool:
return jti in _revoked_cache
def claims_valides(request) -> dict | None:
"""Claims du cookie draco de la requête, validité complète, ou None.
Signature et expiration (``verify_token``), révocation du
``jti`` selon le cache alimenté par l'émettrice, ``sub``
présent — la même validité que celle qu'exige
``user_for_draco`` avant de résoudre. Consommé par le contrôle
de cohérence d'identité (7.85.7) : un cookie absent, invalide,
expiré ou révoqué vaut « aucune identité centrale présentée »
et ne doit rien déclencher.
"""
if not enabled():
return None
claims = verify_token(request.cookies.get(COOKIE_NAME, ""))
if not claims:
return None
jti = str(claims.get("jti", ""))
if jti and is_revoked(jti):
return None
if not str(claims.get("sub", "")):
return None
return claims
# ── Résolution sub → utilisateur local ──────────────────────────────────
def _linked_user(conn, sub: str) -> dict | None:
+114 -12
View File
@@ -1,4 +1,5 @@
"""Réglages d'accès de l'instance — inscription locale (fusion #146).
"""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
@@ -16,12 +17,32 @@ Réglage « inscription locale ouverte » :
``INSCRIPTION_LOCALE_OUVERTE=true`` avant tout enregistrement en
base.
Ce réglage ne touche **jamais** la connexion des comptes existants,
locaux comme draco : il ne garde que la création de comptes locaux.
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
@@ -32,18 +53,27 @@ __all__ = [
"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."""
"""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:
@@ -52,6 +82,9 @@ def get_reglages_acces() -> dict:
# 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
@@ -59,33 +92,43 @@ def get_reglages_acces() -> dict:
with get_conn() as conn:
_assurer_table(conn)
row = conn.execute(
"SELECT inscription_locale_ouverte FROM reglages_acces"
" WHERE id=1"
"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 and row["inscription_locale_ouverte"] is not None:
reglages["inscription_locale_ouverte"] = bool(
row["inscription_locale_ouverte"])
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) -> dict:
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,"
" updated_at) VALUES (1, ?, CURRENT_TIMESTAMP)"
" 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["inscription_locale_ouverte"] else 0,
1 if actuels["porte_centrale_active"] else 0),
)
conn.commit()
return actuels
@@ -96,6 +139,65 @@ def inscription_locale_ouverte() -> bool:
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).
+190
View File
@@ -0,0 +1,190 @@
"""Cohérence d'identité au changement de compte (7.85.7).
Cas réel (capture de Bruno) : connecté « fred » chez ObsiGate, il
ouvre Flowdeck et y trouve son propre compte — la session locale
préexistante gouvernait sans reconsulter l'identité centrale, le
pont ne s'exécutant qu'à l'entrée SSO sans session locale.
Le middleware ``CoherenceDracoMiddleware`` aligne la session sur
le cookie draco valide présenté : révocation de la session locale
+ renvoi des pages au pont. Gardes couvertes ici : même sub
(inchangée), aucun cookie / cookie invalide (intacte), session de
secours marquée (exemptée), compte local sans aucun lien
(l'identité centrale présente gagne), marqueur ``draco_sub``
inscrit par le pont dans le payload de session.
"""
import base64
import hashlib
import hmac
import json
import time
import pytest
from app.auth.session import SessionManager
from app.db import get_conn
from app.password_utils import hash_password
from app.services import draco
from tests.conftest import anon
SECRET = "coherence-secret"
SUB_PROPRIO = "coh-sub-proprio"
SUB_FRED = "coh-sub-fred"
MDP = "mot-de-passe-coh"
def _jeton(sub, secret=SECRET, exp=None):
def b64(d: bytes) -> str:
return base64.urlsafe_b64encode(d).rstrip(b"=").decode()
header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
payload = b64(json.dumps({
"sub": sub, "name": "Testeur", "iat": int(time.time()),
"exp": exp if exp is not None else int(time.time()) + 3600,
"jti": f"jti-{sub}",
}).encode())
sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(),
hashlib.sha256).digest()
return f"{header}.{payload}.{b64(sig)}"
@pytest.fixture(autouse=True)
def _env(monkeypatch):
monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
draco._revoked_cache.clear()
yield
draco._revoked_cache.clear()
def _creer(login, sub=None):
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO users (login, email, password_hash, full_name, is_active) "
"VALUES (?, ?, ?, ?, 1)",
(login, f"{login}@ex.ca", hash_password(MDP), login.title()))
uid = cur.lastrowid
if sub:
conn.execute(
"INSERT INTO draco_links (sub, user_id) VALUES (?, ?)",
(sub, uid))
conn.commit()
return uid
def _connecter(client, login):
r = client.post("/auth/local-login",
json={"email": login, "password": MDP})
assert r.status_code == 200, r.text
@pytest.fixture()
def deux_comptes(client):
proprio_id = _creer("cohproprio", SUB_PROPRIO)
fred_id = _creer("cohfred", SUB_FRED)
return proprio_id, fred_id
# ── (iii) le cas de Bruno : session d'un autre compte ────────────
def test_session_autre_compte_alignee_sur_le_cookie(client, deux_comptes):
proprio_id, _ = deux_comptes
anon(client)
_connecter(client, "cohproprio")
client.cookies.set("draco_session", _jeton(SUB_FRED))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"].startswith("/auth/draco")
assert "Max-Age=0" in r.headers.get("set-cookie", "")
# La session locale du proprio est révoquée côté serveur.
with get_conn() as conn:
rows = conn.execute(
"SELECT revoked FROM user_sessions WHERE user_id = ?",
(proprio_id,)).fetchall()
assert rows and all(row["revoked"] == 1 for row in rows)
# Le pont sert ensuite le compte du cookie.
r = client.get("/auth/draco", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "/workspaces"
me = client.get("/auth/user")
assert me.json()["authenticated"] is True
assert me.json()["user"]["login"] == "cohfred"
# ── (ii) même sub, (iv) pas de cookie, cookie invalide ───────────
def test_meme_sub_session_inchangee(client, deux_comptes):
anon(client)
_connecter(client, "cohproprio")
client.cookies.set("draco_session", _jeton(SUB_PROPRIO))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 200
me = client.get("/auth/user")
assert me.json()["user"]["login"] == "cohproprio"
def test_sans_cookie_session_intacte(client, deux_comptes):
anon(client)
_connecter(client, "cohproprio")
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 200
def test_cookie_invalide_session_intacte(client, deux_comptes):
anon(client)
_connecter(client, "cohproprio")
client.cookies.set(
"draco_session", _jeton(SUB_FRED, secret="mauvais-secret"))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 200
me = client.get("/auth/user")
assert me.json()["user"]["login"] == "cohproprio"
# ── (v) session de secours exemptée ──────────────────────────────
def test_session_secours_exemptee(client, deux_comptes):
proprio_id, _ = deux_comptes
anon(client)
session = SessionManager.create_session(
{"id": proprio_id, "login": "cohproprio",
"full_name": "Cohproprio", "is_admin": 0},
None, secours=True)
client.cookies.set("flowdeck_session", session)
client.cookies.set("draco_session", _jeton(SUB_FRED))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 200
# ── Garde 4 : compte sans lien, l'identité centrale gagne ────────
def test_compte_sans_lien_aligne(client):
_creer("cohproprio") # aucun lien draco
_creer("cohfred", SUB_FRED)
anon(client)
_connecter(client, "cohproprio")
client.cookies.set("draco_session", _jeton(SUB_FRED))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"].startswith("/auth/draco")
# ── Marqueur du pont ─────────────────────────────────────────────
def test_pont_inscrit_le_sub_servi_dans_la_session(client, deux_comptes):
anon(client)
client.cookies.set("draco_session", _jeton(SUB_FRED))
r = client.get("/auth/draco", follow_redirects=False)
assert r.status_code == 302
brut = client.cookies.get("flowdeck_session")
payload = SessionManager.decode_payload(brut)
assert payload is not None
assert payload.get("draco_sub") == SUB_FRED
# Changement de compte central : le cookie du proprio aligne
# la session (base = sub inscrit, pas le lien courant).
client.cookies.set("draco_session", _jeton(SUB_PROPRIO))
r = client.get("/workspaces", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"].startswith("/auth/draco")
+118
View File
@@ -0,0 +1,118 @@
"""Entrée SSO automatique sans session locale (7.85.8).
Chantier « entrée SSO de bout en bout » — écart constaté par
Bruno en test réel le 2026-11-08 : arrivé chez Flowdeck avec une
session centrale valide (cookie ``draco_session``) mais aucune
session locale, l'application servait sa page publique
déconnectée (« Log in », « Notion-style ») comme
si le cookie n'existait pas. Attendu : à l'arrivée sur la racine,
passage automatique par le pont ``/auth/draco`` et service de
l'application connectée au compte désigné par le cookie.
Couvert : redirection de la racine vers le pont, chaîne complète
jusqu'à la session ouverte, page publique inchangée sans cookie
ou avec un cookie invalide, formulaire local jamais intercepté
(le chemin prévu ``?provider=local``), pont qui tranche en cas de
rôle « aucun » (403, pas de boucle).
Montage repris de tests/test_coherence_identite.py.
"""
import base64
import hashlib
import hmac
import json
import time
import pytest
from app.db import get_conn
from app.password_utils import hash_password
from app.services import draco
from tests.conftest import anon
SECRET = "entree-secret"
SUB_TESTEUR = "ent-sub-testeur"
MDP = "mot-de-passe-ent"
def _jeton(sub, secret=SECRET):
def b64(d: bytes) -> str:
return base64.urlsafe_b64encode(d).rstrip(b"=").decode()
header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
payload = b64(json.dumps({
"sub": sub, "name": "Testeur", "iat": int(time.time()),
"exp": int(time.time()) + 3600, "jti": f"jti-{sub}",
}).encode())
sig = hmac.new(secret.encode(), f"{header}.{payload}".encode(),
hashlib.sha256).digest()
return f"{header}.{payload}.{b64(sig)}"
@pytest.fixture(autouse=True)
def _env(monkeypatch):
monkeypatch.setenv("DRACO_SESSION_SECRET", SECRET)
monkeypatch.setattr(draco, "_refresh_revocations", lambda: None)
draco._revoked_cache.clear()
yield
draco._revoked_cache.clear()
@pytest.fixture()
def compte_lie(client):
with get_conn() as conn:
cur = conn.execute(
"INSERT INTO users (login, email, password_hash, full_name, is_active) "
"VALUES (?, ?, ?, ?, 1)",
("enttesteur", "[email protected]", hash_password(MDP),
"Testeur Test"))
uid = cur.lastrowid
conn.execute(
"INSERT INTO draco_links (sub, user_id) VALUES (?, ?)",
(SUB_TESTEUR, uid))
conn.commit()
return uid
def test_racine_sans_session_cookie_valide_renvoyee_au_pont(client, compte_lie):
anon(client)
client.cookies.set("draco_session", _jeton(SUB_TESTEUR))
r = client.get("/", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "/auth/draco?next=/workspaces"
def test_chaine_complete_session_ouverte(client, compte_lie):
anon(client)
client.cookies.set("draco_session", _jeton(SUB_TESTEUR))
r = client.get("/auth/draco", follow_redirects=False)
assert r.status_code == 302
assert r.headers["location"] == "/workspaces"
assert "flowdeck_session" in r.headers.get("set-cookie", "")
# La racine ne sert plus la page publique.
r2 = client.get("/", follow_redirects=False)
assert "Notion-style" not in r2.text
def test_racine_sans_cookie_page_publique_inchangee(client, compte_lie):
anon(client)
r = client.get("/", follow_redirects=False)
assert r.status_code == 200
assert "Notion-style" in r.text
def test_racine_cookie_invalide_page_publique_inchangee(client, compte_lie):
anon(client)
client.cookies.set("draco_session", _jeton(SUB_TESTEUR, secret="mauvais"))
r = client.get("/", follow_redirects=False)
assert r.status_code == 200
assert "Notion-style" in r.text
def test_formulaire_local_jamais_intercepte(client, compte_lie):
anon(client)
client.cookies.set("draco_session", _jeton(SUB_TESTEUR))
r = client.get("/auth/login?provider=local", follow_redirects=False)
assert r.status_code == 200
+222
View File
@@ -0,0 +1,222 @@
"""FlowDeck — porte centrale unique (phase 2, programme « compte
partout »), basée sur les réglages d'accès de la phase 1 (#146).
Contrat :
- réglage « porte centrale » **distinct** de celui d'inscription,
**désactivé par défaut** à la livraison ;
- mode actif : l'écran de connexion ne présente que l'entrée
centrale (bouton ObsiGate SSO) et les identifiants locaux
ordinaires sont refusés — comptes historiques inclus ; les
sessions déjà établies ne sont pas touchées ;
- **porte de secours** : UN compte administrateur local désigné,
identifiants posés par les variables d'environnement
``FLOWDECK_SECOURS_ADMIN_LOGIN`` /
``FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE``, authentification locale
uniquement par le chemin discret ``?secours=1`` (drapeau
``secours`` à l'envoi) — jamais proposé sur l'écran principal ;
- mode inactif : comportement actuel intégralement préservé.
"""
from __future__ import annotations
import pytest
from conftest import anon_csrf
from app.services import reglages_acces
LOGIN_SECOURS = "admin-secours"
MDP_SECOURS = "mot-de-passe-de-secours-designe"
def _activer():
reglages_acces.set_reglages_acces(porte_centrale_active=True)
def _desactiver():
reglages_acces.set_reglages_acces(porte_centrale_active=False)
def _creer_compte_local(login: str, mot_de_passe: str, admin: int = 0) -> None:
from app.db import get_conn
from app.password_utils import hash_password
with get_conn() as conn:
conn.execute(
"INSERT OR REPLACE INTO users (login, full_name, email,"
" password_hash, is_admin) VALUES (?, ?, ?, ?, ?)",
(login, login, f"{login}@test.dev",
hash_password(mot_de_passe), admin),
)
conn.commit()
@pytest.fixture
def secours_env(monkeypatch):
"""Pose les variables d'environnement de la porte de secours."""
monkeypatch.setenv("FLOWDECK_SECOURS_ADMIN_LOGIN", LOGIN_SECOURS)
monkeypatch.setenv("FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE", MDP_SECOURS)
_creer_compte_local(LOGIN_SECOURS, "ancien-mot-de-passe-local", admin=1)
return monkeypatch
# ── Défaut et distinction des réglages ──────────────────────────────
def test_porte_centrale_desactivee_par_defaut_dans_le_code():
from app.config import settings
assert settings.porte_centrale_active is False
def test_reglage_distinct_de_celui_d_inscription(client):
reglages_acces.set_reglages_acces(
inscription_locale_ouverte=True, porte_centrale_active=False)
reglages = reglages_acces.get_reglages_acces()
assert reglages["inscription_locale_ouverte"] is True
assert reglages["porte_centrale_active"] is False
# ── Mode actif : le local ordinaire est refusé ──────────────────────
def test_mode_actif_refuse_compte_ordinaire(client):
_creer_compte_local("ordinaire", "secret123")
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": "ordinaire", "password": "secret123"})
assert r.status_code == 403
assert "ObsiGate (SSO)" in r.json()["error"]
def test_mode_actif_refuse_compte_local_historique(client):
"""Compte créé par inscription publique (quand elle était
ouverte), puis porte activée : son mot de passe local ne passe
plus."""
reglages_acces.set_reglages_acces(
inscription_locale_ouverte=True, porte_centrale_active=False)
r = client.post("/auth/register", json={
"email": "[email protected]", "password": "secret123",
"name": "Historique"})
assert r.status_code == 200
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": "[email protected]", "password": "secret123"})
assert r.status_code == 403
def test_mode_actif_ecran_connexion_que_l_entree_centrale(client):
_activer()
r = anon_csrf(client).get("/auth/login?provider=local")
assert r.status_code == 200
# L'entrée centrale, bien visible…
assert "Login with ObsiGate (SSO)" in r.text
# … et rien d'autre : ni formulaire local, ni onglet Register,
# ni lien vers le chemin discret de secours.
assert 'id="login-form"' not in r.text
assert 'id="password"' not in r.text
assert 'id="tab-register"' not in r.text
assert "secours" not in r.text
def test_mode_actif_inscription_aussi_fermee(client):
reglages_acces.set_reglages_acces(inscription_locale_ouverte=True)
_activer()
r = client.post("/auth/register", json={
"email": "[email protected]", "password": "secret123"})
assert r.status_code == 403
def test_mode_actif_session_deja_etablie_non_touchee(client):
_creer_compte_local("deja-dedans", "secret123")
c = anon_csrf(client)
r = c.post("/auth/local-login", json={
"email": "deja-dedans", "password": "secret123"})
assert r.status_code == 200
_activer()
r = c.get("/auth/user")
assert r.json()["authenticated"] is True
# ── Porte de secours administrateur ─────────────────────────────────
def test_secours_connexion_admin_par_le_chemin_discret(client, secours_env):
_activer()
# L'écran discret rend le formulaire, marqué « secours ».
r = anon_csrf(client).get("/auth/login?provider=local&secours=1")
assert r.status_code == 200
assert 'id="login-form"' in r.text
assert "secours:true" in r.text
# L'authentification elle-même : identifiants d'environnement.
r = anon_csrf(client).post("/auth/local-login", json={
"email": LOGIN_SECOURS, "password": MDP_SECOURS,
"secours": True})
assert r.status_code == 200
assert r.json()["status"] == "ok"
assert "flowdeck_session" in r.headers.get("set-cookie", "")
def test_secours_mauvais_mot_de_passe_environnement_refuse(client, secours_env):
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": LOGIN_SECOURS, "password": "pas-le-bon",
"secours": True})
assert r.status_code == 401
def test_secours_sans_drapeau_refuse_meme_pour_l_admin(client, secours_env):
"""Sans le drapeau du chemin discret, même les identifiants de
secours sont traités comme un envoi ordinaire : refusés."""
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": LOGIN_SECOURS, "password": MDP_SECOURS})
assert r.status_code == 403
def test_secours_compte_non_admin_refuse(client, secours_env, monkeypatch):
"""Le compte désigné doit être administrateur : un compte
ordinaire désigné par les variables ne passe pas."""
_creer_compte_local("pas-admin", "secret123", admin=0)
monkeypatch.setenv("FLOWDECK_SECOURS_ADMIN_LOGIN", "pas-admin")
monkeypatch.setenv("FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE", "secret123")
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": "pas-admin", "password": "secret123", "secours": True})
assert r.status_code == 401
def test_secours_sans_variables_environnement_fermee(client, monkeypatch):
monkeypatch.delenv("FLOWDECK_SECOURS_ADMIN_LOGIN", raising=False)
monkeypatch.delenv("FLOWDECK_SECOURS_ADMIN_MOT_DE_PASSE", raising=False)
_creer_compte_local("admin-local", "secret123", admin=1)
_activer()
r = anon_csrf(client).post("/auth/local-login", json={
"email": "admin-local", "password": "secret123", "secours": True})
assert r.status_code == 401
# ── Mode inactif : comportement actuel intégralement préservé ───────
def test_mode_inactif_connexion_locale_comme_avant(client):
_desactiver()
_creer_compte_local("normal", "secret123")
r = anon_csrf(client).post("/auth/local-login", json={
"email": "normal", "password": "secret123"})
assert r.status_code == 200
assert r.json()["status"] == "ok"
def test_mode_inactif_ecran_connexion_complet(client):
_desactiver()
r = anon_csrf(client).get("/auth/login?provider=local")
assert r.status_code == 200
assert 'id="login-form"' in r.text
assert "Login with ObsiGate (SSO)" in r.text
def test_api_admin_aller_retour_porte_centrale(client):
_desactiver()
r = client.put("/api/admin/reglages-acces",
json={"porte_centrale_active": True})
assert r.status_code == 200
assert r.json()["porte_centrale_active"] is True
r = client.get("/api/admin/reglages-acces")
assert r.json()["porte_centrale_active"] is True
_desactiver()