Leopoldo Rodríguez
4cc18b517f
backend/cuentas.py: restablece en bloque (todos menos leo y claude) y da de alta a varios desde backend/cuentas/2026-10-23_alta-sitios.json (jesush, miguel, joel, antonio, danielg); la contraseña la pide en la terminal. Miembros con sitio, correo y correo_alt; marca contrasena_inicial con aviso en el sitio hasta que la persona la cambie. Los ids de responsable se recalculan con el equipo de hoy. Lote 2026-10-23_usuarios-campo. Pruebas: tests-cuentas.sh (47). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016X4CAujMuib2tDqt9JRg6X
316 lines
14 KiB
Python
316 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Cuentas del sitio en bloque (23-oct-2026): restablecer contraseñas y dar de alta a varios de una vez.
|
|
|
|
Se corre EN EL SERVIDOR como leo (el dueño de /etc/nginx/hbox-docs.htpasswd), junto al app.py desplegado:
|
|
|
|
cd /srv/projects/drogheda/hbox-docs/backend
|
|
venv/bin/python3 cuentas.py lista # quién tiene acceso (sin contraseñas)
|
|
venv/bin/python3 cuentas.py todo cuentas/2026-10-23_alta-sitios.json # altas + datos + restablecer a todos
|
|
venv/bin/python3 cuentas.py alta cuentas/ARCHIVO.json # solo las altas y los datos del archivo
|
|
venv/bin/python3 cuentas.py restablecer [--excepto leo,claude] [--solo ana,beto]
|
|
|
|
La contraseña se pide en la terminal (dos veces, sin eco): nunca va en un archivo ni como argumento, porque se vería
|
|
en `ps`. Para pruebas existe --password-stdin (la lee de la entrada estándar). Antes de escribir muestra el plan y pide
|
|
«si» (o --si). Usa las mismas funciones y rutas que el servicio (lee las Environment= de actas-hbox.service): hash
|
|
$apr1$, el candado del archivo y los miembros del equipo. Antes de escribir deja un respaldo del archivo de usuarios en
|
|
equipo/_respaldos/ (modo 600, contiene los hashes: no lo copies a otro lado).
|
|
|
|
«todo» y «restablecer» tocan a todos los que tienen acceso, salvo --excepto (por omisión leo y claude) y salvo las
|
|
cuentas sin miembro activo, que se listan para que decidas si se quitan (Equipo → «Quitar acceso»). A quien se le
|
|
restablece o se le da de alta le queda la marca «contraseña inicial»: el sitio le muestra un aviso para cambiarla en
|
|
Mi cuenta y Equipo (admin) muestra quién falta.
|
|
|
|
Archivo de altas (JSON):
|
|
{"alta": [{"usuario": "jesush", "nombre": "…", "area": "campo", "sitio": "timsa", "rol": "…", "correo": "…"}],
|
|
"actualizar": [{"usuario": "jesus", "nombre": "Jesús Romero", "correo": "…"}]}
|
|
En «actualizar» (y en las altas de quien ya era miembro) correo, correo_alt, sitio, rol y teléfono solo se llenan si
|
|
están vacíos; el nombre solo si el nuevo completa al actual («Jesús» → «Jesús Romero»). Nada se borra.
|
|
"""
|
|
import argparse
|
|
import getpass
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import unicodedata
|
|
|
|
AQUI = os.path.dirname(os.path.abspath(__file__))
|
|
EXCEPTO = "leo,claude"
|
|
CAMPOS_SI_VACIO = ("correo", "correo_alt", "sitio", "rol", "telefono")
|
|
|
|
|
|
def _entorno_del_servicio():
|
|
"""Las mismas rutas que el servicio: las Environment= de actas-hbox.service, sin pisar lo que ya venga fijado."""
|
|
ruta = os.path.join(AQUI, "actas-hbox.service")
|
|
if not os.path.isfile(ruta):
|
|
return
|
|
with open(ruta, encoding="utf-8") as f:
|
|
for linea in f:
|
|
linea = linea.strip()
|
|
if linea.startswith("Environment="):
|
|
k, _, v = linea[len("Environment="):].partition("=")
|
|
if k.strip():
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
|
|
_entorno_del_servicio()
|
|
sys.path.insert(0, AQUI)
|
|
import app as A # noqa: E402 (después de fijar el entorno: app.py lee las rutas al importarse)
|
|
|
|
|
|
def _plano(s):
|
|
return unicodedata.normalize("NFKD", str(s or "")).encode("ascii", "ignore").decode().lower().strip()
|
|
|
|
|
|
def _nombre_sitio(s):
|
|
return {i: c for i, _, c in A.SITIOS_TAREA}.get(s, s or "—")
|
|
|
|
|
|
def _logins():
|
|
lineas = A._leer_htpasswd()
|
|
if lineas is None:
|
|
sys.exit(f"No encuentro el archivo de usuarios: {A.HTPASSWD}")
|
|
return [l.split(":", 1)[0].strip() for l in lineas if ":" in l and not l.lstrip().startswith("#")]
|
|
|
|
|
|
def _leer_archivo(ruta):
|
|
if not ruta:
|
|
return {"alta": [], "actualizar": []}
|
|
with open(ruta, encoding="utf-8") as f:
|
|
d = json.load(f)
|
|
alta, act = d.get("alta") or [], d.get("actualizar") or []
|
|
vistos = set()
|
|
for p in alta + act:
|
|
u = str(p.get("usuario") or "").strip().lower()
|
|
if not A.USUARIO_RE.match(u):
|
|
sys.exit(f"Usuario inválido en {ruta}: {u!r} (minúsculas, sin espacios ni acentos)")
|
|
if u in vistos:
|
|
sys.exit(f"{u} aparece dos veces en {ruta}")
|
|
vistos.add(u)
|
|
p["usuario"] = u
|
|
datos = {k: v for k, v in p.items() if k != "usuario"}
|
|
_, err = A._miembro_valido(datos, {"nombre": u, "area": "produccion"} if p in act else None)
|
|
if err:
|
|
sys.exit(f"{u}: {err}")
|
|
return {"alta": alta, "actualizar": act}
|
|
|
|
|
|
def _cambios_de_datos(base, nuevo):
|
|
"""Qué se llena de «nuevo» en el miembro «base» sin pisar nada que ya tenga."""
|
|
c = {}
|
|
for k in CAMPOS_SI_VACIO:
|
|
v = str(nuevo.get(k) or "").strip()
|
|
if v and not str(base.get(k) or "").strip():
|
|
c[k] = v
|
|
n = str(nuevo.get("nombre") or "").strip()
|
|
if n and _plano(n) != _plano(base.get("nombre")) and _plano(n).startswith(_plano(base.get("nombre"))):
|
|
c["nombre"] = n
|
|
return c
|
|
|
|
|
|
def hacer_plan(accion, archivo, excepto, solo):
|
|
logins = _logins()
|
|
presentes = {u.lower() for u in logins}
|
|
miembros = A._miembros()
|
|
d = _leer_archivo(archivo)
|
|
excepto = {x.strip().lower() for x in (excepto or "").split(",") if x.strip()}
|
|
solo = {x.strip().lower() for x in (solo or "").split(",") if x.strip()}
|
|
plan = {"crear": [], "restablecer": [], "omitidos": [], "sin_miembro": [], "datos": [], "logins": len(logins)}
|
|
if accion in ("todo", "alta"):
|
|
for p in d["alta"]:
|
|
if p["usuario"] in presentes: # ya tenía acceso: solo se completan sus datos
|
|
base = miembros.get(p["usuario"])
|
|
c = _cambios_de_datos(base, p) if base else {k: v for k, v in p.items() if k != "usuario"}
|
|
if c:
|
|
plan["datos"].append((p["usuario"], c, base is None))
|
|
else:
|
|
plan["crear"].append(p)
|
|
for p in d["actualizar"]:
|
|
base = miembros.get(p["usuario"])
|
|
if not base:
|
|
plan["omitidos"].append((p["usuario"], "no es miembro del equipo: no se actualiza"))
|
|
continue
|
|
c = _cambios_de_datos(base, p)
|
|
if c:
|
|
plan["datos"].append((p["usuario"], c, False))
|
|
if accion in ("todo", "restablecer"):
|
|
nuevos = {p["usuario"] for p in plan["crear"]}
|
|
for u in sorted(presentes):
|
|
if u in nuevos:
|
|
continue
|
|
if solo and u not in solo:
|
|
continue
|
|
if u in excepto:
|
|
plan["omitidos"].append((u, "excepto"))
|
|
continue
|
|
m = miembros.get(u)
|
|
if not m or m.get("activo", True) is False:
|
|
plan["sin_miembro"].append(u)
|
|
continue
|
|
plan["restablecer"].append(u)
|
|
for u in sorted(solo - presentes):
|
|
plan["omitidos"].append((u, "no tiene acceso al sitio"))
|
|
return plan
|
|
|
|
|
|
def mostrar_plan(plan):
|
|
print(f"Archivo de usuarios: {A.HTPASSWD} ({plan['logins']} cuentas)")
|
|
print(f"Miembros del equipo: {A.EQUIPO_DIR}")
|
|
if plan["crear"]:
|
|
print(f"\nAltas nuevas ({len(plan['crear'])}):")
|
|
for p in plan["crear"]:
|
|
print(f" + {p['usuario']:<10} {p.get('nombre', '')} · {p.get('area', 'produccion')} · {_nombre_sitio(p.get('sitio'))}"
|
|
+ (f" · {p['correo']}" if p.get("correo") else " · (sin correo)"))
|
|
if plan["restablecer"]:
|
|
print(f"\nContraseña restablecida ({len(plan['restablecer'])}): " + ", ".join(plan["restablecer"]))
|
|
if plan["datos"]:
|
|
print(f"\nDatos de miembros ({len(plan['datos'])}; solo se llena lo vacío):")
|
|
for u, c, nuevo in plan["datos"]:
|
|
print(f" ~ {u:<10} " + ("(miembro nuevo) " if nuevo else "") + "; ".join(f"{k} → {v}" for k, v in c.items()))
|
|
for u, por in plan["omitidos"]:
|
|
print(f" · no se toca {u}: {por}")
|
|
if plan["sin_miembro"]:
|
|
print("\nCuentas con acceso pero sin miembro activo (no se tocan; si ya no se usan, quítalas en Equipo → "
|
|
"«Quitar acceso» o con htpasswd -D): " + ", ".join(plan["sin_miembro"]))
|
|
if not (plan["crear"] or plan["restablecer"] or plan["datos"]):
|
|
print("\nNada que hacer.")
|
|
|
|
|
|
def pedir_password(desde_stdin=False):
|
|
if desde_stdin:
|
|
pw = sys.stdin.readline().rstrip("\n")
|
|
confirma = pw
|
|
else:
|
|
pw = getpass.getpass("Contraseña para todos (no se ve al escribir): ")
|
|
confirma = getpass.getpass("Repítela: ")
|
|
if pw != confirma:
|
|
sys.exit("Las dos contraseñas no coinciden. No se cambió nada.")
|
|
if len(pw) < A.PASS_MIN:
|
|
sys.exit(f"La contraseña debe tener al menos {A.PASS_MIN} caracteres. No se cambió nada.")
|
|
if pw != pw.strip():
|
|
sys.exit("La contraseña empieza o termina con un espacio; escríbela sin espacios a los lados. No se cambió nada.")
|
|
return pw
|
|
|
|
|
|
def _respaldo():
|
|
carpeta = os.path.join(A.EQUIPO_DIR, "_respaldos")
|
|
os.makedirs(carpeta, mode=0o700, exist_ok=True)
|
|
destino = os.path.join(carpeta, "htpasswd-" + A.ahora().replace(":", "").replace("-", "").replace("T", "-"))
|
|
shutil.copyfile(A.HTPASSWD, destino)
|
|
os.chmod(destino, 0o600)
|
|
return destino
|
|
|
|
|
|
def aplicar(plan, password, yo):
|
|
sal = lambda: "".join(A.ITOA64[b % 64] for b in os.urandom(8))
|
|
resp = _respaldo()
|
|
nuevos = {p["usuario"]: p for p in plan["crear"]}
|
|
reset = set(plan["restablecer"])
|
|
with A._candado_htpasswd():
|
|
lineas = A._leer_htpasswd()
|
|
presentes = {l.split(":", 1)[0].strip().lower() for l in lineas if ":" in l}
|
|
salida = []
|
|
for l in lineas:
|
|
if ":" in l and l.split(":", 1)[0].strip().lower() in reset:
|
|
u = l.split(":", 1)[0].strip()
|
|
l = u + ":" + A.apr1(password, sal())
|
|
salida.append(l)
|
|
salida = [l for l in salida if l.strip()]
|
|
for u in nuevos:
|
|
if u not in presentes: # alguien la creó mientras tanto: no se duplica
|
|
salida.append(u + ":" + A.apr1(password, sal()))
|
|
A._escribir_htpasswd(salida)
|
|
hecho = {"creados": [], "restablecidos": [], "datos": []}
|
|
for u, p in nuevos.items():
|
|
base = A._leer_json(A._ruta_registro(A.EQUIPO_DIR, u))
|
|
datos = {k: v for k, v in p.items() if k != "usuario"}
|
|
if base:
|
|
t, err = A._miembro_valido(_cambios_de_datos(base, p), base)
|
|
else:
|
|
t, err = A._miembro_valido(datos)
|
|
if not err:
|
|
t.update({"id": u, "creado_por": yo, "creado_en": A.ahora(), "historial": []})
|
|
if err:
|
|
print(f" ! {u}: {err} (el acceso sí quedó creado)")
|
|
continue
|
|
t["id"] = u
|
|
t["activo"] = True
|
|
t["contrasena_inicial"] = True
|
|
t["contrasena_cambiada_en"] = A.ahora()
|
|
A._hist(t, yo, "acceso al sitio creado (cuentas.py); queda la contraseña inicial hasta que la cambie")
|
|
A._guardar_miembro(t)
|
|
A._registrar_acceso(yo, "alta", u)
|
|
hecho["creados"].append(u)
|
|
for u in sorted(reset):
|
|
A._marcar_contrasena(u, True, yo, "contraseña restablecida en bloque (cuentas.py): queda la inicial hasta que la cambie")
|
|
A._registrar_acceso(yo, "contraseña restablecida", u)
|
|
hecho["restablecidos"].append(u)
|
|
for u, c, nuevo in plan["datos"]:
|
|
base = A._leer_json(A._ruta_registro(A.EQUIPO_DIR, u))
|
|
if nuevo or not base: # tenía acceso pero no era miembro
|
|
t, err = A._miembro_valido(dict(c, nombre=c.get("nombre") or u))
|
|
if not err:
|
|
t.update({"id": u, "activo": True, "creado_por": yo, "creado_en": A.ahora(), "historial": []})
|
|
else:
|
|
t, err = A._miembro_valido(c, base)
|
|
if err:
|
|
print(f" ! {u}: {err}")
|
|
continue
|
|
A._hist(t, yo, "datos (cuentas.py): " + "; ".join(f"{k}: {A._corto((base or {}).get(k))} → {A._corto(v)}" for k, v in c.items()))
|
|
A._guardar_miembro(t)
|
|
hecho["datos"].append(u)
|
|
return hecho, resp
|
|
|
|
|
|
def lista():
|
|
miembros = A._miembros()
|
|
logins = _logins()
|
|
print(f"{A.HTPASSWD}: {len(logins)} cuentas\n")
|
|
print(f"{'usuario':<11} {'nombre':<24} {'sitio':<6} {'correo':<28} contraseña")
|
|
for u in sorted(logins, key=str.lower):
|
|
m = miembros.get(u.lower()) or {}
|
|
estado = ("inicial" if m.get("contrasena_inicial") else "propia" if "contrasena_inicial" in m else "—")
|
|
if not m:
|
|
estado = "sin miembro"
|
|
elif m.get("activo", True) is False:
|
|
estado += " · inactivo"
|
|
print(f"{u:<11} {str(m.get('nombre') or '')[:24]:<24} {_nombre_sitio(m.get('sitio')) if m.get('sitio') else '':<6} "
|
|
f"{str(m.get('correo') or '')[:28]:<28} {estado}")
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description="Cuentas del sitio en bloque (ver el encabezado de cuentas.py).")
|
|
ap.add_argument("accion", choices=("lista", "todo", "alta", "restablecer"))
|
|
ap.add_argument("archivo", nargs="?", help="JSON con «alta» y «actualizar» (todo, alta)")
|
|
ap.add_argument("--excepto", default=EXCEPTO, help=f"no restablecer a estos (por omisión {EXCEPTO})")
|
|
ap.add_argument("--solo", default="", help="restablecer solo a estos, separados por coma")
|
|
ap.add_argument("--si", action="store_true", help="no preguntar antes de escribir")
|
|
ap.add_argument("--password-stdin", action="store_true", help="leer la contraseña de la entrada estándar (pruebas)")
|
|
a = ap.parse_args(argv)
|
|
if a.accion == "lista":
|
|
return lista()
|
|
if a.accion in ("todo", "alta") and not a.archivo:
|
|
ap.error(f"«{a.accion}» necesita el archivo de altas")
|
|
plan = hacer_plan(a.accion, a.archivo, a.excepto, a.solo)
|
|
mostrar_plan(plan)
|
|
if not (plan["crear"] or plan["restablecer"] or plan["datos"]):
|
|
return 0
|
|
if not a.si:
|
|
if input("\n¿Continuar? Escribe si: ").strip().lower() not in ("si", "sí"):
|
|
print("No se cambió nada.")
|
|
return 1
|
|
password = pedir_password(a.password_stdin) if (plan["crear"] or plan["restablecer"]) else None
|
|
yo = os.environ.get("SUDO_USER") or getpass.getuser()
|
|
hecho, resp = aplicar(plan, password, yo)
|
|
print(f"\nListo. Respaldo del archivo anterior: {resp}")
|
|
for k, v in hecho.items():
|
|
if v:
|
|
print(f" {k}: {', '.join(v)}")
|
|
if hecho["creados"] or hecho["restablecidos"]:
|
|
print("\nCada quien entra con su usuario y esa contraseña, y la cambia en Mi cuenta (el sitio se lo recuerda).")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|