kingame21/BioTech.ai

1# auth.py2# Database SQLite per autenticazione utenti.3#4# FILE DB: data/biomech.db  (creato automaticamente)5#6# TABELLE:7#   users   → credenziali (PBKDF2-HMAC-SHA256, salt random per-utente)8#   storico → sessioni di allenamento JSON per utente9#10# MIGRAZIONE AUTOMATICA:11#   Se trova users/users.json (vecchio formato), lo importa e lo rinomina12#   .migrated. Al primo login l'hash viene aggiornato al nuovo formato.13# ─────────────────────────────────────────────────────────────────────────────14import os15import json16import uuid17import sqlite318import hashlib19import secrets20from contextlib import contextmanager21from datetime import datetime22from typing import Optional, List, Dict, Any23 24BASE_DIR = os.path.dirname(os.path.abspath(__file__))25DATA_DIR = os.path.join(BASE_DIR, "data")26DB_PATH  = os.path.join(DATA_DIR, "biomech.db")27os.makedirs(DATA_DIR, exist_ok=True)28 29_OLD_USERS_FILE = os.path.join(BASE_DIR, "users", "users.json")30 31 32# ── Connessione ───────────────────────────────────────────────────────────────33@contextmanager34def _conn():35    con = sqlite3.connect(DB_PATH, timeout=10, check_same_thread=False)36    con.row_factory = sqlite3.Row37    con.execute("PRAGMA journal_mode=WAL")  # sicuro per accessi concorrenti38    con.execute("PRAGMA foreign_keys=ON")39    try:40        yield con41        con.commit()42    except Exception:43        con.rollback()44        raise45    finally:46        con.close()47 48 49# ── Schema ────────────────────────────────────────────────────────────────────50def _init_db():51    with _conn() as con:52        con.executescript("""53            CREATE TABLE IF NOT EXISTS users (54                id         TEXT PRIMARY KEY,55                username   TEXT UNIQUE NOT NULL,56                salt       TEXT NOT NULL,57                pin_hash   TEXT NOT NULL,58                created_at TEXT NOT NULL59            );60            CREATE TABLE IF NOT EXISTS storico (61                id         INTEGER PRIMARY KEY AUTOINCREMENT,62                user_id    TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,63                data_json  TEXT NOT NULL,64                created_at TEXT NOT NULL65            );66            CREATE INDEX IF NOT EXISTS idx_storico_user67                ON storico(user_id, created_at);68        """)69    _migrate_from_json()70 71 72# ── Hashing (PBKDF2-HMAC-SHA256) ─────────────────────────────────────────────73def _make_salt() -> str:74    return secrets.token_hex(32)  # 256 bit random per-utente75 76 77def _hash_pin(pin: str, salt: str) -> str:78    dk = hashlib.pbkdf2_hmac("sha256", pin.encode(), salt.encode(), 100_000)79    return dk.hex()80 81 82# ── Registrazione ─────────────────────────────────────────────────────────────83def register_user(username: str, pin: str) -> str:84    username = username.strip().lower()85    if len(username) < 3:86        raise ValueError("Username troppo corto (min 3 caratteri).")87    if len(pin) < 4:88        raise ValueError("PIN deve avere almeno 4 cifre.")89 90    user_id = str(uuid.uuid4())91    salt    = _make_salt()92    now     = datetime.now().isoformat()93 94    try:95        with _conn() as con:96            con.execute(97                "INSERT INTO users (id, username, salt, pin_hash, created_at) "98                "VALUES (?, ?, ?, ?, ?)",99                (user_id, username, salt, _hash_pin(pin, salt), now),100            )101    except sqlite3.IntegrityError:102        raise ValueError("Username già esistente.")103 104    return user_id105 106 107# ── Login ─────────────────────────────────────────────────────────────────────108def login_user(username: str, pin: str) -> Optional[str]:109    """110    Ritorna user_id se le credenziali sono corrette, None altrimenti.111    Gestisce anche gli utenti migrati dal vecchio JSON (hash legacy).112    """113    username = username.strip().lower()114 115    with _conn() as con:116        row = con.execute(117            "SELECT id, salt, pin_hash FROM users WHERE username = ?",118            (username,),119        ).fetchone()120 121    if not row:122        return None123 124    user_id, salt, pin_hash = row["id"], row["salt"], row["pin_hash"]125 126    # Utenti migrati dal vecchio JSON: salt="LEGACY", hash prefissato "legacy:"127    if salt == "LEGACY" and pin_hash.startswith("legacy:"):128        old_hash = pin_hash[len("legacy:"):]129        computed = hashlib.sha256(130            (pin + "biomech_salt_v1").encode()131        ).hexdigest()132        if computed != old_hash:133            return None134        # Upgrade silenzioso al nuovo formato PBKDF2135        new_salt = _make_salt()136        with _conn() as con:137            con.execute(138                "UPDATE users SET salt=?, pin_hash=? WHERE id=?",139                (new_salt, _hash_pin(pin, new_salt), user_id),140            )141        return user_id142 143    # Login normale144    if _hash_pin(pin, salt) != pin_hash:145        return None146    return user_id147 148 149# ── Storico ───────────────────────────────────────────────────────────────────150def carica_storico_user(user_id: str) -> List[Dict[str, Any]]:151    if not user_id:152        return []153    with _conn() as con:154        rows = con.execute(155            "SELECT data_json FROM storico WHERE user_id=? ORDER BY created_at ASC",156            (user_id,),157        ).fetchall()158    result = []159    for r in rows:160        try:161            result.append(json.loads(r["data_json"]))162        except Exception:163            pass164    return result165 166 167def user_exists(user_id: str) -> bool:168    """Ritorna True se user_id esiste nella tabella users."""169    if not user_id:170        return False171    with _conn() as con:172        row = con.execute(173            "SELECT 1 FROM users WHERE id=?", (user_id,)174        ).fetchone()175    return row is not None176 177 178def salva_storico_user(user_id: str, storico: List[Dict[str, Any]]) -> None:179    """Sovrascrive l'intero storico dell'utente."""180    if not user_id:181        return182    # Verifica che l'utente esista nel DB prima di salvare183    # (evita FOREIGN KEY constraint failed se user_id è obsoleto)184    if not user_exists(user_id):185        print(f"[WARN] salva_storico: user_id {user_id!r} non trovato nel DB, skip.")186        return187    now = datetime.now().isoformat()188    try:189        with _conn() as con:190            con.execute("DELETE FROM storico WHERE user_id=?", (user_id,))191            con.executemany(192                "INSERT INTO storico (user_id, data_json, created_at) VALUES (?,?,?)",193                [(user_id, json.dumps(s, ensure_ascii=False), now) for s in storico],194            )195    except sqlite3.IntegrityError as ex:196        print(f"[WARN] salva_storico IntegrityError: {ex}")197 198 199def reset_storico_user(user_id: str) -> None:200    """Cancella solo lo storico, l'account rimane."""201    if not user_id:202        return203    with _conn() as con:204        con.execute("DELETE FROM storico WHERE user_id=?", (user_id,))205 206 207def delete_user(user_id: str) -> None:208    """Elimina account + storico (CASCADE)."""209    with _conn() as con:210        con.execute("DELETE FROM users WHERE id=?", (user_id,))211 212 213# ── Migrazione da JSON ────────────────────────────────────────────────────────214def _migrate_from_json():215    if not os.path.exists(_OLD_USERS_FILE):216        return217    migrated = _OLD_USERS_FILE + ".migrated"218    if os.path.exists(migrated):219        return  # già fatto220 221    try:222        with open(_OLD_USERS_FILE, "r", encoding="utf-8") as f:223            old_users: Dict[str, Any] = json.load(f)224    except Exception:225        return226 227    print(f"[auth] Migrazione {len(old_users)} utenti da JSON → SQLite...")228    now = datetime.now().isoformat()229 230    for username, info in old_users.items():231        uid      = info.get("user_id", str(uuid.uuid4()))232        old_hash = info.get("pin_hash", "")233        try:234            with _conn() as con:235                con.execute(236                    "INSERT OR IGNORE INTO users (id, username, salt, pin_hash, created_at) "237                    "VALUES (?,?,?,?,?)",238                    (uid, username.lower(), "LEGACY", "legacy:" + old_hash, now),239                )240        except Exception as ex:241            print(f"[auth] Skip {username}: {ex}")242 243    os.rename(_OLD_USERS_FILE, migrated)244    print("[auth] Migrazione completata → users.json.migrated")245 246 247# ── Avvio automatico ──────────────────────────────────────────────────────────248_init_db()