Spaces:
Running
Running
File size: 4,888 Bytes
70e641d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """Capa de datos de usuarios.
Diferencias de seguridad frente a la versión PHP:
- La BD SQLite vive en instance/ FUERA del directorio servido (no es descargable).
- Sin credenciales por defecto: si se configura MySQL, usuario/clave vienen de entorno.
- Hash de contraseña con scrypt (stdlib), sal aleatoria por usuario.
"""
from __future__ import annotations
import hashlib
import hmac
import secrets
import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager
from .config import obtener_config
_ESQUEMA = """
CREATE TABLE IF NOT EXISTS usuarios (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nombre TEXT NOT NULL,
apellido TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
password TEXT NOT NULL,
creado_en DATETIME DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS intentos_login (
email TEXT NOT NULL,
ip TEXT NOT NULL,
momento DATETIME DEFAULT CURRENT_TIMESTAMP
);
-- Persistencia OPCIONAL de resultados de analizador (sólo con lab_persistir=true; útil sólo
-- con volumen persistente). Clave = muestra_id normalizada; último gana (INSERT OR REPLACE).
CREATE TABLE IF NOT EXISTS resultados_lab (
muestra_id TEXT PRIMARY KEY,
momento DATETIME,
recibido_en DATETIME DEFAULT CURRENT_TIMESTAMP,
payload_json TEXT NOT NULL
);
"""
def inicializar_db() -> None:
cfg = obtener_config()
cfg.db_path.parent.mkdir(parents=True, exist_ok=True)
with _conexion() as con:
con.executescript(_ESQUEMA)
@contextmanager
def _conexion() -> Iterator[sqlite3.Connection]:
cfg = obtener_config()
con = sqlite3.connect(cfg.db_path)
con.row_factory = sqlite3.Row
try:
yield con
con.commit()
finally:
con.close()
# --- Hash de contraseñas (scrypt, stdlib) ---
def hash_password(password: str) -> str:
sal = secrets.token_bytes(16)
dk = hashlib.scrypt(password.encode(), salt=sal, n=2**14, r=8, p=1, dklen=32)
return f"scrypt${sal.hex()}${dk.hex()}"
def verificar_password(password: str, almacenado: str) -> bool:
try:
algo, sal_hex, hash_hex = almacenado.split("$")
if algo != "scrypt":
return False
sal = bytes.fromhex(sal_hex)
dk = hashlib.scrypt(password.encode(), salt=sal, n=2**14, r=8, p=1, dklen=32)
return hmac.compare_digest(dk.hex(), hash_hex)
except (ValueError, AttributeError):
return False
# --- Operaciones de usuario ---
def buscar_usuario(email: str) -> sqlite3.Row | None:
with _conexion() as con:
cur = con.execute(
"SELECT id, nombre, apellido, email, password FROM usuarios WHERE email = ? LIMIT 1",
(email,),
)
return cur.fetchone()
def crear_usuario(nombre: str, apellido: str, email: str, password: str) -> None:
with _conexion() as con:
con.execute(
"INSERT INTO usuarios (nombre, apellido, email, password) VALUES (?, ?, ?, ?)",
(nombre, apellido, email, hash_password(password)),
)
# --- Registro de intentos de login (para throttling) ---
def registrar_intento(email: str, ip: str) -> None:
with _conexion() as con:
con.execute("INSERT INTO intentos_login (email, ip) VALUES (?, ?)", (email, ip))
# Poda oportunista: `limpiar_intentos` sólo corre tras un login correcto, así que los
# intentos fallidos contra emails que nunca aciertan crecerían sin límite. Una hora cubre
# de sobra cualquier ventana de throttling configurada.
con.execute("DELETE FROM intentos_login WHERE momento < datetime('now', '-1 hour')")
def intentos_recientes(email: str, ip: str, ventana_s: int) -> int:
with _conexion() as con:
cur = con.execute(
"SELECT COUNT(*) AS n FROM intentos_login "
"WHERE (email = ? OR ip = ?) AND momento > datetime('now', ?)",
(email, ip, f"-{ventana_s} seconds"),
)
return int(cur.fetchone()["n"])
def limpiar_intentos(email: str) -> None:
with _conexion() as con:
con.execute("DELETE FROM intentos_login WHERE email = ?", (email,))
# --- Persistencia opcional de resultados de laboratorio ---
def guardar_resultado_lab(muestra_id: str, momento: str, payload_json: str) -> None:
with _conexion() as con:
con.execute(
"INSERT OR REPLACE INTO resultados_lab (muestra_id, momento, payload_json) VALUES (?, ?, ?)",
(muestra_id, momento, payload_json),
)
def cargar_resultados_lab(limite: int = 500) -> list[str]:
"""Devuelve los payloads JSON más recientes, para recargar el almacén en proceso al arrancar."""
with _conexion() as con:
cur = con.execute(
"SELECT payload_json FROM resultados_lab ORDER BY recibido_en DESC LIMIT ?",
(limite,),
)
return [row["payload_json"] for row in cur.fetchall()]
|