199 lines
7.2 KiB
Python
199 lines
7.2 KiB
Python
"""User accounts: password hashing and bootstrap (0.24.0, P2).
|
|
|
|
Passwords are hashed with PBKDF2-HMAC-SHA256 (stdlib) with a
|
|
per-user salt; the stored string carries its parameters
|
|
("pbkdf2$<iterations>$<salt_b64>$<hash_b64>") so the cost can
|
|
be raised later without invalidating existing hashes.
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import unicodedata
|
|
|
|
from app import models
|
|
|
|
_ITERATIONS = 260_000
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, _ITERATIONS
|
|
)
|
|
return "pbkdf2${}${}${}".format(
|
|
_ITERATIONS,
|
|
base64.b64encode(salt).decode("ascii"),
|
|
base64.b64encode(digest).decode("ascii"),
|
|
)
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
scheme, iterations, salt_b64, hash_b64 = stored.split("$")
|
|
if scheme != "pbkdf2":
|
|
return False
|
|
salt = base64.b64decode(salt_b64)
|
|
expected = base64.b64decode(hash_b64)
|
|
digest = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, int(iterations)
|
|
)
|
|
return hmac.compare_digest(digest, expected)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def ensure_bootstrap_admin() -> dict | None:
|
|
"""Create the first admin from the env credentials.
|
|
|
|
Runs when the users table is empty and KANBANTE_AUTH_USER /
|
|
_PASSWORD are configured (decision D6): Bruno's existing
|
|
login becomes the initial database admin, with no lockout.
|
|
The env path keeps working as an emergency fallback either
|
|
way (see app.main auth).
|
|
"""
|
|
from app.config import get_settings
|
|
|
|
if models.count_users() > 0:
|
|
return None
|
|
settings = get_settings()
|
|
if not settings.auth_enabled:
|
|
return None
|
|
user_id = models.create_user(
|
|
settings.auth_user,
|
|
hash_password(settings.auth_password),
|
|
role="admin",
|
|
display_name=settings.auth_user,
|
|
)
|
|
return models.get_user(user_id)
|
|
|
|
|
|
# ── Correspondance draco par utilisateur (HUB-16, 0.46.0) ────
|
|
# La correspondance identité draco ↔ compte Kanbante vit dans la
|
|
# colonne users.draco_sub (UNIQUE, présente depuis la 0.24.0) :
|
|
# models.get_user_by_draco(sub) est la résolution canonique.
|
|
# resolve_draco_user() est l'unique point d'entrée du pont SSO
|
|
# (app.main.draco_bridge) et du contrôle draco de /api/search :
|
|
#
|
|
# 1. correspondance existante → son compte (None si désactivé) ;
|
|
# 2. amorçage : s'il n'existe qu'UN seul compte et que c'est un
|
|
# admin actif sans liaison, le premier sub valide s'y lie —
|
|
# le compte d'amorçage (celui de Bruno, créé depuis
|
|
# l'environnement) garde ainsi son accès sans verrouillage ;
|
|
# 3. sinon, si provisionner=True : création JIT d'un compte
|
|
# propre à cette identité (voir provision_draco_user).
|
|
#
|
|
# Un sub inconnu ne désigne JAMAIS un compte existant non lié
|
|
# (ni l'opérateur d'environnement, ni un compte dont le nom
|
|
# ressemblerait au nom affiché draco) : la liaison d'un compte
|
|
# préexistant est un geste d'administrateur (/admin, voir
|
|
# docs/correspondance-draco.md) ou la première visite
|
|
# doublement connectée (session native + cookie draco).
|
|
|
|
|
|
def _slug_draco(valeur: str) -> str:
|
|
texte = unicodedata.normalize("NFKD", valeur or "")
|
|
texte = texte.encode("ascii", "ignore").decode("ascii").lower()
|
|
return re.sub(r"[^a-z0-9]+", "-", texte).strip("-")[:40].strip("-")
|
|
|
|
|
|
def _nom_utilisateur_draco(nom_affiche: str, sub: str) -> str:
|
|
"""Nom d'utilisateur libre dérivé des claims (jamais réutilisé).
|
|
|
|
Base : le nom affiché slugifié (« Alice Exemple » →
|
|
« alice-exemple ») ; repli « draco-<sub slugifié> ». Si le nom
|
|
est déjà pris — y compris par un compte préexistant NON lié,
|
|
qui ne doit pas être adopté — un suffixe -2, -3… est ajouté :
|
|
le JIT crée toujours un compte nouveau, distinct.
|
|
"""
|
|
base = _slug_draco(nom_affiche)
|
|
if len(base) < 3:
|
|
base = _slug_draco(f"draco-{sub}") or "draco"
|
|
candidat, n = base, 2
|
|
while models.get_user_by_username(candidat) is not None:
|
|
candidat = f"{base}-{n}"
|
|
n += 1
|
|
return candidat
|
|
|
|
|
|
def provision_draco_user(sub: str, nom_affiche: str = "") -> dict | None:
|
|
"""Création JIT du compte Kanbante d'une identité draco (HUB-16).
|
|
|
|
Rôle par défaut sûr : membre (« member », l'utilisateur
|
|
standard de Kanbante) — jamais admin. Le mot de passe local
|
|
est un secret aléatoire haché, jamais divulgué : l'accès
|
|
normal passe par draco ; un administrateur peut poser un mot
|
|
de passe local depuis /admin si besoin. Le compte rejoint
|
|
l'espace Principal comme membre, comme tout nouvel
|
|
utilisateur créé par l'administration.
|
|
"""
|
|
sub = str(sub or "")
|
|
if not sub:
|
|
return None
|
|
existant = models.get_user_by_draco(sub)
|
|
if existant is not None:
|
|
return existant
|
|
nom = (nom_affiche or "").strip()[:80]
|
|
for _tentative in range(5):
|
|
username = _nom_utilisateur_draco(nom, sub)
|
|
try:
|
|
user_id = models.create_user(
|
|
username,
|
|
hash_password(secrets.token_urlsafe(32)),
|
|
role="member",
|
|
display_name=nom,
|
|
draco_sub=sub,
|
|
)
|
|
break
|
|
except sqlite3.IntegrityError:
|
|
# Course sur draco_sub ou username : la correspondance
|
|
# gagnante fait foi ; sinon, réessayer un autre nom.
|
|
existant = models.get_user_by_draco(sub)
|
|
if existant is not None:
|
|
return existant
|
|
else:
|
|
return models.get_user_by_draco(sub)
|
|
espaces = models.list_workspaces()
|
|
if espaces:
|
|
models.set_member(espaces[0]["id"], user_id, "member")
|
|
models.log_admin(user_id, "draco.jit", f"{username} (sub {sub})")
|
|
return models.get_user(user_id)
|
|
|
|
|
|
def resolve_draco_user(claims: dict | None, provisionner: bool = False) -> dict | None:
|
|
"""Compte Kanbante désigné par des claims draco valides.
|
|
|
|
provisionner=False : correspondance et amorçage seulement
|
|
(surfaces qui ne doivent pas créer de compte, p. ex.
|
|
/api/search). provisionner=True : création JIT si aucune
|
|
correspondance n'existe (pont /auth/draco — la « première
|
|
connexion draco » du critère « un compte fonctionne
|
|
partout »). Un compte désactivé résout à None et n'est
|
|
jamais contourné par un JIT (la correspondance existe).
|
|
"""
|
|
sub = str((claims or {}).get("sub") or "")
|
|
if not sub:
|
|
return None
|
|
lie = models.get_user_by_draco(sub)
|
|
if lie is not None:
|
|
return lie if lie["active"] else None
|
|
if models.count_users() == 1:
|
|
seul = models.list_users()[0]
|
|
if (
|
|
seul["role"] == "admin"
|
|
and seul["active"]
|
|
and not seul.get("draco_sub")
|
|
):
|
|
# Compte d'amorçage : liaison au premier passage.
|
|
models.update_user(seul["id"], draco_sub=sub)
|
|
models.log_admin(seul["id"], "draco.link.amorcage", sub)
|
|
return models.get_user(seul["id"])
|
|
if provisionner:
|
|
cree = provision_draco_user(sub, str((claims or {}).get("name") or ""))
|
|
if cree is not None and cree["active"]:
|
|
return cree
|
|
return None
|