250 lines
8.6 KiB
Python
250 lines
8.6 KiB
Python
"""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()
|