diff --git a/.env b/.env index fe14b29..da685c1 100644 --- a/.env +++ b/.env @@ -3,6 +3,7 @@ DB_HOST=162.19.78.131 DB_USER=excel_auth DB_PASSWORD=%n#%3Lay1MPa$%kR^5@ DB_NAME=Sondes +AUTH_USERS=[{"user":"Michel","pass":"921#%!LC3^71509PiyK"}] # paramètres mail SMTP_HOST=smtp.mail.ovh.net diff --git a/app/tracker.py b/app/tracker.py index fc763d7..f8db109 100644 --- a/app/tracker.py +++ b/app/tracker.py @@ -3,31 +3,47 @@ # Streamlit — Gestion de la table MySQL Sondes.tracker # ------------------------------------------------------------- # Schéma attendu : id (PK), address, lieu, repere, mise_en_service (DATE), res_bits, date (timestamp) +# Authentification intégrée via .env (AUTH_USERS JSON) # ------------------------------------------------------------- import os import re import time +import json +import hmac +from hashlib import sha256 # (non utilisé directement mais laissé si besoin d'évolution) +from typing import Optional +from datetime import date + import pandas as pd import streamlit as st import mysql.connector as mysql from contextlib import contextmanager from dotenv import load_dotenv -from datetime import date # ========================== # Configuration / Constantes # ========================== +load_dotenv() # lit .env si présent + TABLE_DB = os.getenv("MYSQL_DB", "Sondes") TABLE_NAME = "tracker" COL_ID = "id" COL_ADDRESS = "address" COL_LIEU = "lieu" COL_REPERE = "repere" -COL_MES = "mise_en_service" # 🆕 DATE +COL_MES = "mise_en_service" # DATE COL_RESBITS = "res_bits" COL_DATE = "date" +DB_CFG = dict( + host=os.getenv("DB_HOST"), + user=os.getenv("DB_USER"), + password=os.getenv("DB_PASSWORD"), + database=os.getenv("DB_NAME"), + port=int(os.getenv("MYSQL_PORT", "3306")), +) + # Regex d'une ROM code DS18B20 au format {0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0xCRC} ROM_REGEX = re.compile(r"^\{(?:0x[0-9A-Fa-f]{2},){7}0x[0-9A-Fa-f]{2}\}$") @@ -39,19 +55,55 @@ RES_MAP = { 12: {"precision": 0.0625,"tconv_ms": 750}, } +# ================== +# Authentification via .env (AUTH_USERS) +# ================== +AUTH_USERS_RAW = os.getenv("AUTH_USERS", "[]") + +def _load_users() -> dict: + try: + data = json.loads(AUTH_USERS_RAW) + return {str(d.get("user", "")).strip(): str(d.get("pass", "")) for d in data if d.get("user") and d.get("pass")} + except Exception: + return {} + +USERS = _load_users() + +def _constant_time_equals(a: str, b: str) -> bool: + return hmac.compare_digest(a.encode("utf-8"), b.encode("utf-8")) + +def verify_credentials(username: str, password: str) -> bool: + if not username or not password: + return False + expected = USERS.get(username.strip()) + if expected is None: + return False + # NOTE : stockage en clair pour l'exemple. En prod, préférer un hash (bcrypt/argon2). + return _constant_time_equals(password, expected) + +def require_login() -> Optional[str]: + if st.session_state.get("auth_ok") and st.session_state.get("auth_user"): + return st.session_state.get("auth_user") + + st.markdown("

🔒 Accès restreint

", unsafe_allow_html=True) + with st.form("login_form", clear_on_submit=False): + username = st.text_input("Utilisateur") + password = st.text_input("Mot de passe", type="password") + ok = st.form_submit_button("Se connecter") + if ok: + if verify_credentials(username, password): + st.session_state["auth_ok"] = True + st.session_state["auth_user"] = username.strip() + st.success("Connexion réussie.") + time.sleep(0.3) + st.rerun() + else: + st.error("Identifiants invalides.") + st.stop() + # ================== # Accès Base de Données # ================== -load_dotenv() # lit .env si présent - -DB_CFG = dict( - host=os.getenv("DB_HOST"), - user=os.getenv("DB_USER"), - password=os.getenv("DB_PASSWORD"), - database=os.getenv("DB_NAME"), - port=int(os.getenv("MYSQL_PORT", "3306")), -) - @contextmanager def get_conn(): conn = mysql.connect(**DB_CFG) @@ -90,7 +142,7 @@ def insert_tracker(address: str, lieu: str, res_bits: int, repere: str | None = address, lieu, (repere.strip() if repere and str(repere).strip() else None), - mise_en_service, # mysql-connector accepte datetime.date + mise_en_service, res_bits, )) conn.commit() @@ -127,38 +179,53 @@ def delete_tracker(row_id: int) -> None: # Utilitaires UI # -------------- -def is_valid_rom(address: str) -> bool: - return bool(ROM_REGEX.match(address.strip())) - - def rom_help() -> str: + """Message d'aide sur le format des adresses ROM DS18B20.""" return ( - "Format attendu : `{0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0x12}` (8 octets en hex).\n" + "Format attendu : `{0x28,0xFF,0xAA,0xBB,0xCC,0xDD,0xEE,0x12}` " + "(8 octets en hex).\n" "Le premier octet (famille) est souvent 0x28 pour DS18B20." ) +def is_valid_rom(address: str) -> bool: + """Vérifie que l'adresse saisie correspond bien au format attendu.""" + import re + ROM_REGEX = re.compile(r"^\{(?:0x[0-9A-Fa-f]{2},){7}0x[0-9A-Fa-f]{2}\}$") + return bool(ROM_REGEX.match(address.strip())) def res_label(bits: int) -> str: + """Retourne un label lisible pour la résolution DS18B20.""" + RES_MAP = { + 9: {"precision": 0.5, "tconv_ms": 94}, + 10: {"precision": 0.25, "tconv_ms": 188}, + 11: {"precision": 0.125, "tconv_ms": 375}, + 12: {"precision": 0.0625,"tconv_ms": 750}, + } info = RES_MAP.get(bits) if not info: return f"{bits} bits (inconnu)" return f"{bits} bits (±{info['precision']}°C, {info['tconv_ms']} ms)" - # ================== # Application Streamlit # ================== -st.set_page_config(page_title="Gestion des sondes — tracker", page_icon="🌡️", layout="wide") +st.set_page_config(page_title="Tracker", page_icon="🌡️", layout="wide") -st.title("🌡️ Gestion de Sondes.tracker") +# 🔐 Exige la connexion avant d'afficher l'app +user = require_login() + +st.title("🌡️ Gestion du parc sondes en stock ou installées") with st.expander("Paramètres de connexion (lecture seule)"): st.write({k: ("***" if k in {"password"} else v) for k, v in DB_CFG.items()}) - st.caption("Configurez ces valeurs via le fichier .env (MYSQL_HOST, MYSQL_DB, MYSQL_USER, MYSQL_PASSWORD, MYSQL_PORT)") + st.caption("Configurez ces valeurs via le fichier .env (MYSQL_HOST, MYSQL_DB, MYSQL_USER, MYSQL_PASSWORD, MYSQL_PORT, AUTH_USERS)") # Barre latérale — Filtres & Actions st.sidebar.header("Filtres & Actions") +# Info utilisateur connecté +st.sidebar.caption(f"Connecté en tant que **{user}**") + # Récupération de la liste des lieux existants _all = fetch_trackers() lieux = sorted([x for x in _all[COL_LIEU].dropna().unique()]) if not _all.empty else [] @@ -200,17 +267,17 @@ with st.sidebar.form("add_form", clear_on_submit=True): time.sleep(0.6) st.rerun() -# Bouton de déconnexion (EXIT) +# Bouton de déconnexion (session Streamlit) st.sidebar.divider() -st.sidebar.subheader("Sécurité") -if st.sidebar.button("EXIT / Déconnexion", type="primary"): +if st.sidebar.button("Se déconnecter"): for _k in list(st.session_state.keys()): try: del st.session_state[_k] except Exception: pass - st.markdown('', unsafe_allow_html=True) - st.stop() + st.success("Déconnecté.") + time.sleep(0.3) + st.rerun() # Vue principale (liste / édition)