Michel/Gestion_sondes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cosmétique Tracker et sécurité

Browse Source
This commit is contained in:
Michel
2025-08-26 14:09:52 +02:00
parent cd6bb7b5cf
commit 088e7a4821
2 changed files with 95 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
app/tracker.py
View File

@@ -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("<h2>🔒 Accès restreint</h2>", 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('<meta http-equiv="refresh" content="0; url=/logout">', unsafe_allow_html=True)
st.stop()
st.success("Déconnecté.")
time.sleep(0.3)
st.rerun()
# Vue principale (liste / édition)