Spaces:
Running
Running
| """ | |
| SESSION PERSISTANTE ADMIN (cookie "Rester connecte") | |
| ===================================================== | |
| Evite de se reconnecter a chaque nouvelle session Streamlit (onglet rouvert, | |
| inactivite, redeploiement du Space) tant que l'utilisateur ne s'est pas | |
| deconnecte lui-meme. | |
| Jeton = base64url(JSON {u, exp, n, p}) + "." + HMAC-SHA256. | |
| - Signe avec SESSION_SECRET, sinon une cle derivee de SIGNATURE_SECRET (pas de | |
| nouveau secret obligatoire). Sans aucun secret : fonction desactivee. | |
| - Survit aux redeploiements : rien n'est garde en memoire, tout est dans le | |
| jeton signe et dans la feuille de revocation. | |
| - `p` = empreinte du mot de passe stocke : changer le mot de passe d'un compte | |
| invalide tous ses cookies. | |
| - Deconnexion : le jeton est inscrit dans la feuille Sessions_Revoquees (refuse | |
| ensuite partout) et le cookie est efface du navigateur. | |
| - Le role et les modules autorises sont relus a chaque restauration (un droit | |
| retire s'applique immediatement). | |
| Limite assumee : le cookie est pose par JavaScript, il ne peut donc pas etre | |
| HttpOnly. Il est Secure et SameSite=Lax, et l'app n'affiche aucun HTML brut | |
| issu des donnees (voir module Ontology). | |
| """ | |
| import base64 | |
| import hashlib | |
| import hmac | |
| import json | |
| import secrets | |
| import time | |
| import gspread | |
| from core.email_client import get_credential | |
| NOM_COOKIE = "vortex_session" | |
| DUREE_SECONDES = 30 * 24 * 3600 # 30 jours, renouveles a l'usage | |
| RENOUVELER_SOUS = 15 * 24 * 3600 # nouveau jeton quand il reste moins de 15 jours | |
| FEUILLE_REVOCATIONS = "Sessions_Revoquees" | |
| HEADER_REVOCATIONS = ["Token_Hash", "Utilisateur", "Date_Revocation", "Motif"] | |
| def _cle(): | |
| secret = get_credential("SESSION_SECRET") | |
| if secret: | |
| return str(secret).encode("utf-8") | |
| base = get_credential("SIGNATURE_SECRET") | |
| if base: | |
| return hmac.new(str(base).encode("utf-8"), b"vortex-session-v1", hashlib.sha256).digest() | |
| return None | |
| def actif() -> bool: | |
| return _cle() is not None | |
| def _b64(data: bytes) -> str: | |
| return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") | |
| def _b64_dec(texte: str) -> bytes: | |
| return base64.urlsafe_b64decode(texte + "=" * (-len(texte) % 4)) | |
| def _empreinte_mdp(valeur_stockee) -> str: | |
| return hashlib.sha256(str(valeur_stockee or "").encode("utf-8")).hexdigest()[:16] | |
| def emettre_jeton(utilisateur, mot_de_passe_stocke, maintenant=None): | |
| """Jeton signe pour `utilisateur`, ou None si aucun secret n'est configure.""" | |
| cle = _cle() | |
| if cle is None: | |
| return None | |
| charge = {"u": utilisateur, "exp": int((maintenant or time.time()) + DUREE_SECONDES), | |
| "n": secrets.token_hex(8), "p": _empreinte_mdp(mot_de_passe_stocke)} | |
| corps = _b64(json.dumps(charge, separators=(",", ":")).encode("utf-8")) | |
| return f"{corps}.{_b64(hmac.new(cle, corps.encode('ascii'), hashlib.sha256).digest())}" | |
| def lire_jeton(jeton, utilisateurs: dict, maintenant=None): | |
| """Charge du jeton {u, exp, ...} s'il est authentique, non expire, et que le compte | |
| existe avec le meme mot de passe ; sinon None.""" | |
| cle = _cle() | |
| if cle is None or not jeton or jeton.count(".") != 1: | |
| return None | |
| corps, signature = jeton.split(".") | |
| attendue = _b64(hmac.new(cle, corps.encode("ascii"), hashlib.sha256).digest()) | |
| if not hmac.compare_digest(signature, attendue): | |
| return None | |
| try: | |
| charge = json.loads(_b64_dec(corps)) | |
| except (ValueError, json.JSONDecodeError): | |
| return None | |
| utilisateur = charge.get("u") | |
| if int(charge.get("exp", 0)) <= (maintenant or time.time()) or utilisateur not in utilisateurs: | |
| return None | |
| if not hmac.compare_digest(str(charge.get("p", "")), _empreinte_mdp(utilisateurs[utilisateur])): | |
| return None | |
| return charge | |
| def a_renouveler(charge, maintenant=None) -> bool: | |
| return int(charge["exp"]) - (maintenant or time.time()) < RENOUVELER_SOUS | |
| def hash_jeton(jeton) -> str: | |
| return hashlib.sha256(str(jeton).encode("utf-8")).hexdigest() | |
| def est_revoque(sh, jeton) -> bool: | |
| """Lecture directe (pas de cache) : une deconnexion doit etre prise en compte tout de suite.""" | |
| try: | |
| valeurs = sh.worksheet(FEUILLE_REVOCATIONS).get_all_values() | |
| except gspread.exceptions.WorksheetNotFound: | |
| return False | |
| return hash_jeton(jeton) in {ligne[0] for ligne in valeurs[1:] if ligne} | |
| def revoquer(sh, jeton, utilisateur, motif="DECONNEXION"): | |
| try: | |
| ws = sh.worksheet(FEUILLE_REVOCATIONS) | |
| except gspread.exceptions.WorksheetNotFound: | |
| ws = sh.add_worksheet(FEUILLE_REVOCATIONS, 1000, len(HEADER_REVOCATIONS)) | |
| ws.append_row(HEADER_REVOCATIONS) | |
| ws.append_row([hash_jeton(jeton), utilisateur, time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()), motif]) | |
| def script_poser(jeton) -> str: | |
| """HTML a rendre (components.html, hauteur 0) pour poser le cookie sur la page principale.""" | |
| valeur = json.dumps(f"{NOM_COOKIE}={jeton}; Path=/; Max-Age={DUREE_SECONDES}; Secure; SameSite=Lax") | |
| return f"<script>try {{ window.parent.document.cookie = {valeur}; }} catch (e) {{}}</script>" | |
| def script_effacer() -> str: | |
| valeur = json.dumps(f"{NOM_COOKIE}=; Path=/; Max-Age=0; Secure; SameSite=Lax") | |
| return f"<script>try {{ window.parent.document.cookie = {valeur}; }} catch (e) {{}}</script>" | |