206 lines
7.8 KiB
Python
206 lines
7.8 KiB
Python
"""Coordinación de varias tarjetas: registro de sesiones por identidad,
|
|
autenticación, límites por tarjeta y retención de logs.
|
|
|
|
El estado por tarjeta (buffers, sockets, calibración) vive en HubManager;
|
|
aquí solo se coordina. Nunca se reservan buffers para identidades
|
|
desconocidas: solo se instancian tarjetas registradas.
|
|
"""
|
|
import logging
|
|
import hashlib
|
|
import os
|
|
import time
|
|
from typing import Dict, Optional
|
|
|
|
from app.config import (
|
|
LOGS_DIR, MAX_DEVICES, WS_RATE_PER_SEC, WS_RATE_BURST,
|
|
is_valid_device_id,
|
|
)
|
|
from app.hub import HubManager
|
|
|
|
log = logging.getLogger("tracker_hub.fleet")
|
|
|
|
|
|
class _TokenBucket:
|
|
__slots__ = ("rate", "burst", "tokens", "ts")
|
|
|
|
def __init__(self, rate: float, burst: int):
|
|
self.rate = rate
|
|
self.burst = float(burst)
|
|
self.tokens = float(burst)
|
|
self.ts = time.monotonic()
|
|
|
|
def allow(self) -> bool:
|
|
now = time.monotonic()
|
|
self.tokens = min(self.burst, self.tokens + (now - self.ts) * self.rate)
|
|
self.ts = now
|
|
if self.tokens >= 1.0:
|
|
self.tokens -= 1.0
|
|
return True
|
|
return False
|
|
|
|
|
|
class FleetManager:
|
|
def __init__(self, store, rate: float = WS_RATE_PER_SEC, burst: int = WS_RATE_BURST):
|
|
self.store = store
|
|
self._rate = rate
|
|
self._burst = burst
|
|
self._hubs: Dict[str, HubManager] = {}
|
|
self._buckets: Dict[str, _TokenBucket] = {}
|
|
self._credentials = {}
|
|
|
|
@property
|
|
def hubs(self) -> Dict[str, HubManager]:
|
|
return self._hubs
|
|
|
|
# ---------------------------- identidad ----------------------------
|
|
|
|
def is_registered(self, device_id: str) -> bool:
|
|
if not is_valid_device_id(device_id):
|
|
return False
|
|
return self.store.get_device(device_id) is not None
|
|
|
|
def get(self, device_id: str) -> Optional[HubManager]:
|
|
"""Devuelve el hub de una tarjeta REGISTRADA (instanciando bajo
|
|
demanda; buffers vacíos hasta recibir telemetría). Nunca crea estado
|
|
para IDs desconocidos."""
|
|
if not self.is_registered(device_id):
|
|
return None
|
|
h = self._hubs.get(device_id)
|
|
if h is None:
|
|
h = HubManager(device_id=device_id, store=self.store)
|
|
self._hubs[device_id] = h
|
|
else:
|
|
h._load_settings()
|
|
return h
|
|
|
|
def bind_session(self, device_id, connection_id, token):
|
|
self._credentials[device_id] = (connection_id, hashlib.sha256(token.encode()).hexdigest())
|
|
self._hubs[device_id].authorize_session = lambda: self.session_valid(device_id, connection_id)
|
|
|
|
def forget_session(self, device_id, connection_id):
|
|
if self._credentials.get(device_id, (None,))[0] == connection_id:
|
|
self._credentials.pop(device_id, None)
|
|
|
|
def session_valid(self, device_id, connection_id):
|
|
record = self._credentials.get(device_id)
|
|
if record is None or record[0] != connection_id:
|
|
return False
|
|
try:
|
|
return self.store.credential_active(device_id, record[1])
|
|
except Exception:
|
|
return False
|
|
|
|
def get_or_create(self, device_id: str) -> Optional[HubManager]:
|
|
return self.get(device_id)
|
|
|
|
def authenticate(self, device_id: str, token: str) -> bool:
|
|
"""Verifica ID + token (solo para ese ID y canal). Rechazo genérico:
|
|
no se revela si la identidad existe."""
|
|
if not is_valid_device_id(device_id) or not token:
|
|
return False
|
|
try:
|
|
return self.store.verify_token(device_id, token)
|
|
except Exception as e:
|
|
log.error("auth_check_failed", extra={"event": "auth_check_failed",
|
|
"device_id": device_id,
|
|
"error_type": type(e).__name__})
|
|
return False
|
|
|
|
# ----------------------------- límites -----------------------------
|
|
|
|
def check_rate(self, device_id: str) -> bool:
|
|
if not self.is_registered(device_id):
|
|
return False
|
|
b = self._buckets.get(device_id)
|
|
if b is None:
|
|
b = self._buckets[device_id] = _TokenBucket(self._rate, self._burst)
|
|
return b.allow()
|
|
|
|
# --------------------- revocación (polling) ------------------------
|
|
|
|
async def poll_credentials(self):
|
|
"""Sondea credenciales de las tarjetas conectadas. La revocación o
|
|
deshabilitación desde la CLI (proceso separado) cierra la sesión
|
|
activa en este intervalo."""
|
|
closed = []
|
|
for device_id, h in list(self._hubs.items()):
|
|
if not h.esp32_connected:
|
|
continue
|
|
try:
|
|
connection_id = h.connection_id
|
|
active = self.session_valid(device_id, connection_id)
|
|
except Exception as e:
|
|
log.error("credential_poll_failed", extra={"event": "credential_poll_failed",
|
|
"device_id": device_id,
|
|
"error_type": type(e).__name__})
|
|
active = False
|
|
if not active:
|
|
log.warning("credential_revoked_close", extra={"event": "credential_revoked_close",
|
|
"device_id": device_id})
|
|
try:
|
|
await h.force_close(connection_id)
|
|
except Exception:
|
|
pass
|
|
closed.append(device_id)
|
|
self.forget_session(device_id, connection_id)
|
|
return closed
|
|
|
|
# --------------------------- retención logs -------------------------
|
|
|
|
def prune_logs(self, retention_days: int) -> int:
|
|
"""Borra logs diarios por tarjeta mayores a retention_days. NUNCA
|
|
toca los archivos legacy en la raíz de LOGS_DIR."""
|
|
if retention_days <= 0:
|
|
return 0
|
|
cutoff = time.time() - retention_days * 86400
|
|
removed = 0
|
|
try:
|
|
entries = os.listdir(LOGS_DIR)
|
|
except Exception:
|
|
return 0
|
|
for name in entries:
|
|
p = os.path.join(LOGS_DIR, name)
|
|
if not os.path.isdir(p) or os.path.islink(p):
|
|
continue
|
|
if not is_valid_device_id(name):
|
|
continue
|
|
try:
|
|
for fn in os.listdir(p):
|
|
fp = os.path.join(p, fn)
|
|
if (fn.startswith("log_") and fn.endswith(".txt")
|
|
and os.path.isfile(fp) and os.path.getmtime(fp) < cutoff):
|
|
try:
|
|
os.remove(fp)
|
|
removed += 1
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return removed
|
|
|
|
# ------------------------------ listado -----------------------------
|
|
|
|
def list_devices(self):
|
|
out = []
|
|
for row in self.store.list_devices():
|
|
h = self._hubs.get(row["device_id"])
|
|
pcb = h.pcb_version if h is not None else None
|
|
if pcb is None:
|
|
try:
|
|
s = self.store.get_settings(row["device_id"])
|
|
pcb = s["pcb_version"] if s else None
|
|
except Exception:
|
|
pcb = None
|
|
out.append({
|
|
"device_id": row["device_id"],
|
|
"alias": row["alias"],
|
|
"enabled": row["enabled"],
|
|
"esp32_connected": bool(h.esp32_connected) if h is not None else False,
|
|
"last_seen": h.last_seen.isoformat() if (h and h.last_seen) else None,
|
|
"last_telemetry_at": h.last_telemetry_at.isoformat() if (h and h.last_telemetry_at) else None,
|
|
"firmware": h.esp32_firmware_version if h is not None else None,
|
|
"pcb_version": pcb,
|
|
"created_at": row["created_at"],
|
|
})
|
|
return out
|