"""CLI de provisionamiento local de tarjetas (proceso separado del hub). Uso: python -m app.devices add --device-id esp32-001122aabbcc --alias "Cuarto sur" python -m app.devices list python -m app.devices rename --device-id ... --alias ... python -m app.devices enable --device-id ... python -m app.devices disable --device-id ... python -m app.devices rotate --device-id ... [--grace-hours N] python -m app.devices revoke --device-id ... python -m app.devices retire-previous --device-id ... python -m app.devices migrate-primary --device-id ... El token generado se entrega UNA sola vez, solo a un archivo local protegido (0600) bajo STATE_DIR/provision/. Nunca se imprime en stdout ni se guarda en claro (la BD guarda solo SHA-256). """ import argparse import json import os import shutil import stat import sys import time from app.config import ( DATABASE_PATH, PROVISION_DIR, FIRMWARE_DIR, PCB_STATE_FILE, DEFAULT_PCB_VERSION, PYRANO_MV_PER_WM2, AZ_DEV_DEFAULT, is_valid_device_id, ) from app.device_store import DeviceStore, StoreError def _store() -> DeviceStore: return DeviceStore(DATABASE_PATH) def _require_device_id(args) -> str: device_id = getattr(args, "device_id", None) if not device_id or not is_valid_device_id(device_id): print("ERROR: --device-id inválido (formato esperado: esp32-<12 hex minúsculos>)") sys.exit(2) return device_id def _write_token_file(device_id: str, token: str, label: str): """Canal local seguro: archivo nuevo (nunca sobreescribe) con permisos 0600. Solo imprime la ruta, NUNCA el token.""" os.makedirs(PROVISION_DIR, exist_ok=True) path = os.path.join(PROVISION_DIR, f"{device_id}.{label}.token") if os.path.exists(path): print(f"ERROR: ya existe {path}; se usa un canal nuevo por operación (no se sobreescribe).") sys.exit(1) try: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) except FileExistsError: print(f"ERROR: ya existe {path}.") sys.exit(1) with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(token + "\n") try: os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) except Exception: pass print(f"Token guardado (una sola vez) en: {path}") print("ADVERTENCIA: verifica los permisos del archivo (0600 / ACL en Windows).") print("El token NO se puede recuperar: cópialo a tu ESP32 ahora y borra este archivo tras installarlo.") def cmd_add(args): device_id = _require_device_id(args) store = _store() try: token = store.add_device(device_id, args.alias) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) _write_token_file(device_id, token, "v1") print(f"Alta completada: {device_id} (alias: {args.alias!r})") def cmd_list(args): store = _store() rows = store.list_devices() if not rows: print("Sin tarjetas registradas.") return print(f"{'DEVICE_ID':<22} {'ENABLED':<8} {'TOKEN_V':<8} {'ALIAS'}") for r in rows: print(f"{r['device_id']:<22} {'si' if r['enabled'] else 'no':<8} " f"{r['token_version']:<8} {r['alias']}") def cmd_rename(args): device_id = _require_device_id(args) store = _store() try: store.set_alias(device_id, args.alias) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) print(f"Alias actualizado: {device_id} -> {args.alias!r}") def cmd_enable(args): _set_enabled(args, True) def cmd_disable(args): _set_enabled(args, False) def _set_enabled(args, enabled: bool): device_id = _require_device_id(args) store = _store() try: store.set_enabled(device_id, enabled) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) print(f"{'Habilitada' if enabled else 'Deshabilitada'}: {device_id} " "(la sesión activa se cierra en el próximo sondeo del hub)") def cmd_rotate(args): device_id = _require_device_id(args) store = _store() try: token = store.rotate_token(device_id, args.grace_hours) v = store.get_token_version(device_id) or 2 except StoreError as e: print(f"ERROR: {e}") sys.exit(1) _write_token_file(device_id, token, f"v{v}") print("Rotación completada: el token anterior sigue válido hasta su caducidad " "(o hasta 'retire-previous' / 'revoke').") def cmd_revoke(args): device_id = _require_device_id(args) store = _store() try: store.revoke_tokens(device_id) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) print(f"Revocación urgente: {device_id}. La sesión activa se cierra en el " "próximo sondeo del hub y ambos tokens quedan inválidos.") def cmd_retire_previous(args): device_id = _require_device_id(args) store = _store() try: store.retire_previous(device_id) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) print(f"Token anterior revocado: {device_id} (solo el token actual queda válido).") def cmd_migrate_primary(args): device_id = _require_device_id(args) store = _store() pcb = DEFAULT_PCB_VERSION if os.path.exists(PCB_STATE_FILE): try: with open(PCB_STATE_FILE, "r", encoding="utf-8") as f: v = json.load(f).get("version") if v not in ("V1", "V2"): raise ValueError("PCB legacy inválida") pcb = v except Exception: print("ERROR: PCB legacy inválida; migración cancelada sin cambiar ajustes.") sys.exit(1) # Respaldo ANTES de la importación (idempotente por timestamp). try: bak = f"{PCB_STATE_FILE}.bak-{time.strftime('%Y%m%d-%H%M%S')}" shutil.copy2(PCB_STATE_FILE, bak) print(f"Respaldo legacy: {bak}") except Exception as e: print(f"ERROR: no se pudo respaldar el PCB legacy ({type(e).__name__}); migración cancelada.") sys.exit(1) # La calibración legacy vive en memoria del proceso antiguo; aquí se # importan los defaults y el operador aplica explícitamente si difieren. try: imported, token = store.migrate_legacy(device_id, pcb, PYRANO_MV_PER_WM2, AZ_DEV_DEFAULT) except StoreError as e: print(f"ERROR: {e}") sys.exit(1) if not imported: print("Migración ya realizada (idempotente): no se repite. " "La primaria ya tiene su estado importado.") return print(f"Migración legacy -> {device_id} completada (PCB={pcb}, calibración=defaults).") if token is not None: _write_token_file(device_id, token, "v1") print("La primaria no existía: se creó con su token de provisionamiento.") else: print("La primaria ya estaba dada de alta: se aplicó solo su estado.") def main(argv=None): p = argparse.ArgumentParser(prog="app.devices", description="Provisionamiento local de tarjetas") sub = p.add_subparsers(dest="cmd", required=True) sp = sub.add_parser("add", help="Alta de tarjeta (genera token, se entrega una vez)") sp.add_argument("--device-id", required=True) sp.add_argument("--alias", default="") sp.set_defaults(fn=cmd_add) sp = sub.add_parser("list", help="Lista tarjetas (sin secretos)") sp.set_defaults(fn=cmd_list) sp = sub.add_parser("rename", help="Renombra alias") sp.add_argument("--device-id", required=True) sp.add_argument("--alias", required=True) sp.set_defaults(fn=cmd_rename) sp = sub.add_parser("enable", help="Habilita tarjeta") sp.add_argument("--device-id", required=True) sp.set_defaults(fn=cmd_enable) sp = sub.add_parser("disable", help="Deshabilita tarjeta") sp.add_argument("--device-id", required=True) sp.set_defaults(fn=cmd_disable) sp = sub.add_parser("rotate", help="Rota token (el anterior caduca en grace-hours)") sp.add_argument("--device-id", required=True) sp.add_argument("--grace-hours", type=int, default=None) sp.set_defaults(fn=cmd_rotate) sp = sub.add_parser("revoke", help="Revocación urgente de ambos tokens") sp.add_argument("--device-id", required=True) sp.set_defaults(fn=cmd_revoke) sp = sub.add_parser("retire-previous", help="Revoca solo el token anterior") sp.add_argument("--device-id", required=True) sp.set_defaults(fn=cmd_retire_previous) sp = sub.add_parser("migrate-primary", help="Importa estado legacy a una primaria explícita") sp.add_argument("--device-id", required=True) sp.set_defaults(fn=cmd_migrate_primary) args = p.parse_args(argv) args.fn(args) if __name__ == "__main__": main()