"""Servidor didáctico de loopback; HTTP sin TLS. No desplegar en Internet.""" from contextlib import closing from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import hmac import json import os from pathlib import Path import re import sqlite3 from nucleo import numero, validar_config def preparar(ruta): with closing(sqlite3.connect(ruta)) as db, db: db.execute("CREATE TABLE IF NOT EXISTS config (id INTEGER PRIMARY KEY, payload TEXT NOT NULL)") db.execute("CREATE TABLE IF NOT EXISTS muestras (id TEXT PRIMARY KEY, payload TEXT NOT NULL, recibido TEXT NOT NULL)") db.execute("INSERT OR IGNORE INTO config VALUES (1, ?)", (json.dumps(dict(version=1, on=28, off=26)),)) def validar_registro(r): claves = {"schema", "device", "boot", "seq", "tick", "temperature", "quality", "output", "config_version"} if not isinstance(r, dict) or set(r) != claves: raise ValueError("Esquema de muestra inválido") if type(r["schema"]) is not int or r["schema"] != 1 or r["device"] != "medidor-01": raise ValueError("Esquema o identidad no admitidos") if not isinstance(r["boot"], str) or not re.fullmatch(r"[A-Za-z0-9-]{1,40}", r["boot"]): raise ValueError("Sesión inválida") for k in ("seq", "tick", "config_version"): if type(r[k]) is not int or not 1 <= r[k] <= 1000000000000: raise ValueError("Contador inválido") if type(r["output"]) is not bool: raise ValueError("Salida inválida") v = r["temperature"] if r["quality"] == "valid": if not numero(v) or not -40 <= v <= 125: raise ValueError("Medición válida fuera de rango") elif r["quality"] == "invalid": if v is not None or r["output"]: raise ValueError("Una lectura inválida exige null y salida apagada en este banco") else: raise ValueError("Calidad inválida") return f"{r['device']}/{r['boot']}/{r['seq']}" def manejador(ruta, token_equipo, token_admin): class API(BaseHTTPRequestHandler): def setup(self): super().setup() self.connection.settimeout(3) def log_message(self, *_): pass # No registrar cabeceras ni credenciales. def responder(self, estado, datos): cuerpo = json.dumps(datos, allow_nan=False).encode("utf-8") self.send_response(estado) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(cuerpo))) self.end_headers() self.wfile.write(cuerpo) def autorizado(self, token): recibido = self.headers.get("Authorization", "").encode("utf-8") return hmac.compare_digest(recibido, ("Bearer " + token).encode("utf-8")) def cuerpo_json(self): if self.headers.get("Content-Type", "").split(";")[0].strip() != "application/json": raise ValueError("Content-Type debe ser application/json") longitud = self.headers.get("Content-Length", "") if not longitud.isascii() or not longitud.isdigit() or not 1 <= int(longitud) <= 4096: raise ValueError("Cuerpo requerido de hasta 4096 bytes") return json.loads(self.rfile.read(int(longitud)).decode("utf-8")) def do_GET(self): if self.path == "/": contenido = Path(__file__).with_name("panel.html").read_bytes() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(contenido))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(contenido) return token = token_equipo if self.path == "/api/config" else token_admin if not self.autorizado(token): self.responder(401, dict(error="Sin autorización")) return with closing(sqlite3.connect(ruta)) as db: config = json.loads(db.execute("SELECT payload FROM config WHERE id=1").fetchone()[0]) if self.path == "/api/config": self.responder(200, config) elif self.path == "/api/estado": total = db.execute("SELECT count(*) FROM muestras").fetchone()[0] filas = db.execute("SELECT payload, recibido FROM muestras ORDER BY rowid DESC LIMIT 20").fetchall() self.responder(200, dict(config=config, total=total, historia=[dict(registro=json.loads(p), recibido=t) for p, t in filas])) else: self.responder(404, dict(error="Ruta inexistente")) def do_POST(self): if self.path not in ("/api/config", "/api/muestras"): self.responder(404, dict(error="Ruta inexistente")) return token = token_admin if self.path == "/api/config" else token_equipo if not self.autorizado(token): self.responder(401, dict(error="Sin autorización")) return try: datos = self.cuerpo_json() if self.path == "/api/config": datos = validar_config(datos) else: clave = validar_registro(datos) payload = json.dumps(datos, sort_keys=True, allow_nan=False) with closing(sqlite3.connect(ruta)) as db, db: db.execute("BEGIN IMMEDIATE") if self.path == "/api/config": anterior = json.loads(db.execute("SELECT payload FROM config WHERE id=1").fetchone()[0]) if datos["version"] <= anterior["version"]: self.responder(409, dict(error="Se requiere una versión superior")) return db.execute("UPDATE config SET payload=? WHERE id=1", (payload,)) respuesta = dict(guardada=datos["version"]) else: fila = db.execute("SELECT payload FROM muestras WHERE id=?", (clave,)).fetchone() if fila and fila[0] != payload: self.responder(409, dict(error="Identidad de registro reutilizada con otro contenido")) return if not fila: if db.execute("SELECT count(*) FROM muestras").fetchone()[0] >= 10000: self.responder(503, dict(error="Almacenamiento didáctico lleno")) return db.execute("INSERT INTO muestras VALUES (?, ?, ?)", (clave, payload, datetime.now(timezone.utc).isoformat())) respuesta = dict(boot=datos["boot"], seq=datos["seq"], duplicado=bool(fila)) # El acuse se envía después de completar la transacción. self.responder(200, respuesta) except (ValueError, UnicodeError): self.responder(400, dict(error="JSON o campos inválidos")) except sqlite3.Error: self.responder(503, dict(error="Almacenamiento no disponible")) return API def main(): equipo = os.environ.get("IOT_DEVICE_TOKEN", "") admin = os.environ.get("IOT_ADMIN_TOKEN", "") if len(equipo) < 16 or len(admin) < 16 or equipo == admin: raise SystemExit("Definí IOT_DEVICE_TOKEN e IOT_ADMIN_TOKEN: distintos y de al menos 16 caracteres.") ruta = Path(__file__).with_name("proyecto.sqlite3") preparar(ruta) servidor = ThreadingHTTPServer(("127.0.0.1", 8765), manejador(ruta, equipo, admin)) print("Panel: http://127.0.0.1:8765 — solo loopback; detener con Ctrl+C.") try: servidor.serve_forever() except KeyboardInterrupt: pass finally: servidor.server_close() if __name__ == "__main__": main()