"""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