"""Entrada al sitio con página propia (27-oct-2026). Reemplaza la ventana gris de HTTP Basic de nginx. Por qué (INC-0001, reporte de Adriana del 26-oct: «después de restablecer la contraseña ya no puedo entrar»): con HTTP Basic el navegador guarda la contraseña anterior y la vuelve a mandar sola; el gestor de contraseñas del teléfono llena la vieja en la ventana gris porque Mi cuenta no tenía campo de usuario; no hay forma real de cerrar sesión; un usuario con mayúscula inicial («Adriana») no existe para nginx; y el servidor no sabía quién falló ni por qué. Ahora la entrada es una página nuestra, con usuario (o correo) y contraseña, que dice qué pasó y deja cada intento registrado (sin la contraseña) para el vigía y el agente diario. Cómo funciona: nginx pregunta a /api/_auth en cada petición (auth_request), también por páginas y archivos. Responde 204 con X-Hbox-User si hay sesión válida (cookie firmada «hbox_sesion») y 401 si no. Sin sesión, nginx manda a /entrar.html (las API reciben 401 en JSON). HTTP Basic ya no entra, salvo desde las IP de ACCESO_BASIC_IPS (vacío por omisión): los scripts del servidor (vigía, tablero.py, el agente) hablan con 127.0.0.1:8081 directo. Revisión de seguridad del 27-oct: sin Basic abierto (fuerza bruta sin límite), redirección solo a este sitio, bloqueo por usuario + IP (nadie bloquea a otro), POST/PATCH/DELETE solo desde este sitio (Origin/Sec-Fetch-Site), cualquier error al revisar la sesión = 401, secreto creado de forma atómica. Las contraseñas siguen en el mismo htpasswd ($apr1$): Mi cuenta, Equipo y cuentas.py no cambian. Rutas (las de «sin sesión» las deja pasar nginx sin auth_request y SIN X-Remote-User): GET /api/_auth solo nginx (la ruta pública /api/_auth se bloquea en nginx) POST /api/entrar {usuario, password, recordar} sin sesión. usuario = el del sitio o su correo de Equipo POST /api/salir sin sesión: borra la cookie GET /api/sesion sin sesión: {sesion, user} (la página de entrada salta si ya hay sesión) GET /api/accesos/intentos?dias=7 administradores: intentos por usuario (sin contraseñas) La cookie lleva usuario ~ vence ~ versión ~ firma (HMAC-SHA256). versión = huella del hash del htpasswd de esa persona: si su contraseña cambia (restablecimiento, Mi cuenta o cuentas.py) sus sesiones viejas dejan de valer y la próxima vez ve la página de entrada. Quien cambia la suya en Mi cuenta recibe una cookie nueva en ese momento. Secreto: SESION_SECRETO (entorno) o DATOS_BASE/.sesion-secreto (se crea solo, modo 600). Cambiarlo cierra todas las sesiones. Intentos: DATOS_BASE/acceso/intentos-AAAA-MM.jsonl — cuándo, a quién corresponde (o, si el usuario no existe, solo sus 3 primeras letras y su largo: a veces se teclea la contraseña en el campo del usuario), resultado (ok · usuario-no-existe · contrasena-incorrecta · sin-datos), pistas (bloq-mayus, mayuscula-inicial, espacios-alrededor: se calculan al fallar, nunca se le dicen a la persona), IP y navegador. Los intentos bloqueados no se anotan. Límite: 8 fallas de un mismo usuario desde una misma IP, o 20 fallas de una IP, en 15 minutos → espera (además del limit_req de nginx). Así nadie puede dejar fuera a otra persona tecleando mal su usuario a propósito. """ import base64 import datetime import hashlib import hmac import json import os import re import time import unicodedata from flask import jsonify, request COOKIE = "hbox_sesion" DIAS_SESION = int(os.environ.get("SESION_DIAS", "30")) DIAS_RENOVAR = 10 # si a la sesión le quedan menos días, /api/whoami la renueva VENTANA_S = 15 * 60 MAX_FALLAS_USUARIO = int(os.environ.get("ENTRAR_MAX_FALLAS_USUARIO", "8")) # por usuario Y por IP BASIC_IPS = {x.strip() for x in os.environ.get("ACCESO_BASIC_IPS", "").split(",") if x.strip()} MAX_FALLAS_IP = int(os.environ.get("ENTRAR_MAX_FALLAS_IP", "20")) COOKIE_INSEGURA = os.environ.get("SESION_COOKIE_INSEGURA", "") == "1" # solo pruebas locales por http SEP = "~" def _plano(s): """' Adriana ' → 'adriana'; 'Jesús' → 'jesus'. Sin acentos ni espacios.""" s = unicodedata.normalize("NFKD", str(s or "")).encode("ascii", "ignore").decode().lower() return re.sub(r"\s+", "", s) def registrar(app, g): """g = globals() de app.py (HTPASSWD, apr1, _leer_htpasswd, DATOS_BASE, EQUIPO_DIR, ADMINS…).""" cache = {"clave": None, "usuarios": {}} carpeta = os.path.join(g["DATOS_BASE"], "acceso") # ------------------------------------------------------------ secreto de las cookies def _secreto(): s = os.environ.get("SESION_SECRETO", "").strip() if s: return s.encode() ruta = os.path.join(g["DATOS_BASE"], ".sesion-secreto") def leer(): try: with open(ruta, encoding="utf-8") as f: return f.read().strip() except OSError: return "" actual = leer() if len(actual) < 32: # no existe, o quedó vacío/corto: se crea completo y se publica de un golpe tmp = ruta + ".tmp-" + os.urandom(4).hex() fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as f: f.write(base64.urlsafe_b64encode(os.urandom(32)).decode()) try: if actual: # corto o vacío: se reemplaza os.replace(tmp, ruta) else: os.link(tmp, ruta) # atómico; si otro proceso ganó, se usa el suyo except FileExistsError: pass finally: if os.path.exists(tmp): os.unlink(tmp) actual = leer() if len(actual) < 32: raise RuntimeError("No pude crear el secreto de sesión en " + ruta) return actual.encode() SECRETO = {"v": None} def secreto(): if SECRETO["v"] is None: SECRETO["v"] = _secreto() return SECRETO["v"] # ------------------------------------------------------------ usuarios del htpasswd def usuarios(): """{usuario en minúsculas: (usuario tal cual, hash)}; se relee solo si el archivo cambió.""" try: st = os.stat(g["HTPASSWD"]) except OSError: return {} clave = (st.st_mtime_ns, st.st_size, st.st_ino) if cache["clave"] != clave: d = {} for linea in g["_leer_htpasswd"]() or []: if ":" in linea and not linea.lstrip().startswith("#"): u, h = linea.split(":", 1) d[u.strip().lower()] = (u.strip(), h.strip()) cache.update(clave=clave, usuarios=d) return cache["usuarios"] def verificar(h, password): if not password: return False if h.startswith("$apr1$"): partes = h.split("$") return len(partes) >= 4 and hmac.compare_digest(g["apr1"](password, partes[2]), h) if h.startswith("{SHA}"): calc = "{SHA}" + base64.b64encode(hashlib.sha1(password.encode("utf-8")).digest()).decode() return hmac.compare_digest(calc, h) if h.startswith(("$2y$", "$2b$", "$2a$")): # htpasswd -B (bcrypt), si alguien lo usó a mano try: import bcrypt # noqa: PLC0415 return bcrypt.checkpw(password.encode("utf-8"), h.replace("$2y$", "$2b$", 1).encode()) except Exception: # noqa: BLE001 return False return False def version(h): return hashlib.sha256(h.encode()).hexdigest()[:12] def por_correo(texto): """Correo de Equipo (correo o correo_alt) → usuario. None si no hay o si dos personas lo comparten.""" t = str(texto or "").strip().lower() if "@" not in t: return None hallados = {m["id"] for m in g["_listar_json"](g["EQUIPO_DIR"]) if m.get("id") and m.get("activo", True) is not False and t in (str(m.get("correo") or "").strip().lower(), str(m.get("correo_alt") or "").strip().lower())} return hallados.pop() if len(hallados) == 1 else None # ------------------------------------------------------------ cookie firmada def firmar(u, vence, ver): return hmac.new(secreto(), f"{u}|{vence}|{ver}".encode(), hashlib.sha256).hexdigest()[:40] def emitir(resp, u, dias=DIAS_SESION): reg = usuarios().get(u.lower()) if not reg: return resp vence = int(time.time()) + dias * 86400 ver = version(reg[1]) valor = SEP.join((reg[0], str(vence), ver, firmar(reg[0], vence, ver))) resp.set_cookie(COOKIE, valor, max_age=dias * 86400, path="/", httponly=True, secure=not COOKIE_INSEGURA, samesite="Lax") return resp def leer_cookie(): """(usuario, vence) si la cookie es válida hoy; (None, 0) si no.""" v = request.cookies.get(COOKIE) or "" partes = v.split(SEP) if len(partes) != 4: return None, 0 u, vence, ver, firma = partes try: vence_i = int(vence) except ValueError: return None, 0 b = lambda x: str(x).encode("utf-8", "replace") if vence_i < time.time() or not hmac.compare_digest(b(firma), b(firmar(u, vence_i, ver))): return None, 0 reg = usuarios().get(u.lower()) if not reg or not hmac.compare_digest(b(version(reg[1])), b(ver)): # contraseña cambiada o acceso quitado return None, 0 return reg[0], vence_i def leer_basic(): """Authorization: Basic solo desde las IP de ACCESO_BASIC_IPS (vacío por omisión = nunca). Sin esto, cualquiera podría probar contraseñas sin límite por /api/_auth. Los scripts del servidor usan 127.0.0.1:8081 directo.""" if not BASIC_IPS or ip_cliente() not in BASIC_IPS: return None h = request.headers.get("Authorization", "") if not h.lower().startswith("basic "): return None if (request.headers.get("User-Agent") or "").startswith("Mozilla/"): return None try: crudo = base64.b64decode(h[6:].strip()).decode("utf-8") except Exception: # noqa: BLE001 try: crudo = base64.b64decode(h[6:].strip()).decode("latin-1") except Exception: # noqa: BLE001 return None u, _, pw = crudo.partition(":") reg = usuarios().get(u.strip().lower()) return reg[0] if reg and verificar(reg[1], pw) else None # ------------------------------------------------------------ registro de intentos def ip_cliente(): return (request.headers.get("X-Real-IP") or (request.headers.get("X-Forwarded-For") or "").split(",")[0] or request.remote_addr or "").strip()[:60] def ruta_intentos(cuando=None): cuando = cuando or datetime.datetime.now() return os.path.join(carpeta, f"intentos-{cuando:%Y-%m}.jsonl") def clave_de(usuario, tecleado): """Usuario real (minúsculas) o, si no existe, una huella de lo tecleado: cuenta los intentos sin guardar el texto.""" return (usuario or "").lower() or ("?" + hashlib.sha1(_plano(tecleado).encode()).hexdigest()[:10]) def enmascarar(tecleado): t = _plano(tecleado) return (t[:3] + "…(" + str(len(t)) + ")") if t else "" def anotar(tecleado, usuario, resultado, pistas=()): os.makedirs(carpeta, exist_ok=True) r = {"cuando": g["ahora"](), "tecleado": "" if usuario else enmascarar(tecleado), "usuario": usuario or "", "clave": clave_de(usuario, tecleado), "resultado": resultado, "pistas": list(pistas), "ip": ip_cliente(), "navegador": (request.headers.get("User-Agent") or "")[:160]} with open(ruta_intentos(), "a", encoding="utf-8") as f: f.write(json.dumps(r, ensure_ascii=False) + "\n") def intentos_desde(desde_iso, max_bytes=400_000): """Intentos con cuando >= desde_iso, de este mes y del anterior (lee el final de cada archivo).""" hoy = datetime.date.today() mes_ant = (hoy.replace(day=1) - datetime.timedelta(days=1)) out = [] for d in (mes_ant, hoy): ruta = ruta_intentos(datetime.datetime(d.year, d.month, 1)) if not os.path.isfile(ruta): continue with open(ruta, "rb") as f: f.seek(0, 2) tam = f.tell() f.seek(max(0, tam - max_bytes)) texto = f.read().decode("utf-8", "ignore") for linea in texto.splitlines(): try: r = json.loads(linea) except Exception: # noqa: BLE001 continue if str(r.get("cuando", "")) >= desde_iso: out.append(r) return out def bloqueado(clave, ip): """Por usuario Y por IP (8) y por IP (20) en 15 min. No hay bloqueo global por usuario: si lo hubiera, cualquiera podría dejar fuera a una persona tecleando mal su usuario a propósito.""" desde = (datetime.datetime.now() - datetime.timedelta(seconds=VENTANA_S)).replace(microsecond=0).isoformat() fallas = [r for r in intentos_desde(desde, 2_000_000) if r.get("resultado") not in ("ok", "bloqueado")] n_ui = sum(1 for r in fallas if r.get("ip") == ip and r.get("clave") == clave) n_ip = sum(1 for r in fallas if ip and r.get("ip") == ip) return n_ui >= MAX_FALLAS_USUARIO or n_ip >= MAX_FALLAS_IP # ------------------------------------------------------------ rutas @app.get("/api/_auth") def _auth(): try: u, _ = leer_cookie() u = u or leer_basic() except Exception as e: # noqa: BLE001 (cookie rara, htpasswd ilegible…): mejor la página de entrada que «caído» app.logger.warning("acceso: /api/_auth falló, respondo 401: %s", e) u = None if not u: return ("", 401) # sin WWW-Authenticate: nginx manda a /entrar.html r = app.response_class(status=204) r.headers["X-Hbox-User"] = u return r @app.get("/api/sesion") def sesion(): u, vence = leer_cookie() return jsonify(ok=True, sesion=bool(u), user=u or "", vence=datetime.datetime.fromtimestamp(vence).replace(microsecond=0).isoformat() if u else None) @app.post("/api/entrar") def entrar(): d = request.get_json(silent=True) or request.form or {} tecleado = str(d.get("usuario") or "") password = str(d.get("password") or "") recordar = g["_bool"](d.get("recordar"), True) u = _plano(tecleado) if "@" in u: u = por_correo(u) or u ip = ip_cliente() if not u or not password: anotar(tecleado, "", "sin-datos") return jsonify(ok=False, error="Escribe tu usuario y tu contraseña."), 400 reg0 = usuarios().get(u) if bloqueado(clave_de(reg0[0] if reg0 else "", tecleado), ip): return jsonify(ok=False, codigo="bloqueado", error="Demasiados intentos seguidos. Espera 15 minutos y vuelve a intentar, " "o repórtalo abajo para que te ayudemos."), 429 reg = usuarios().get(u) if not reg: anotar(tecleado, "", "usuario-no-existe") return jsonify(ok=False, codigo="usuario", error="No encuentro ese usuario. Es tu nombre en minúsculas y sin acentos, como te lo " "dieron (por ejemplo «adriana» o «jesush»), o tu correo de Helgen."), 401 if not verificar(reg[1], password): pistas = [] if password != password.strip() and verificar(reg[1], password.strip()): pistas.append("espacios-alrededor") if verificar(reg[1], password.swapcase()): pistas.append("bloq-mayus") if password[:1].isupper() and verificar(reg[1], password[:1].lower() + password[1:]): pistas.append("mayuscula-inicial") anotar(tecleado, reg[0], "contrasena-incorrecta", pistas) return jsonify(ok=False, codigo="contrasena", error="La contraseña no coincide. Si te la restablecieron, usa la nueva tal como te la " "pasaron (respeta mayúsculas); si tu navegador la llena solo, bórrala y escríbela."), 401 anotar(tecleado, reg[0], "ok") resp = jsonify(ok=True, user=reg[0]) return emitir(resp, reg[0], DIAS_SESION if recordar else 1) @app.post("/api/salir") def salir(): resp = jsonify(ok=True) resp.set_cookie(COOKIE, "", max_age=0, expires=0, path="/", httponly=True, secure=not COOKIE_INSEGURA, samesite="Lax") return resp @app.get("/api/accesos/intentos") def accesos_intentos(): if not g["is_admin"](): return jsonify(ok=False, error="Solo un administrador puede ver los intentos de entrada."), 403 dias = max(1, min(60, g["_entero"](request.args.get("dias"), 7))) desde = (datetime.datetime.now() - datetime.timedelta(days=dias)).replace(microsecond=0).isoformat() filas = intentos_desde(desde, 2_000_000) por = {} for r in filas: clave = r.get("usuario") or ("? " + (r.get("tecleado") or r.get("clave") or "?")) p = por.setdefault(clave, {"ok": 0, "fallas": 0, "ultimo_ok": None, "ultima_falla": None, "motivos": {}, "pistas": {}}) if r.get("resultado") == "ok": p["ok"] += 1 p["ultimo_ok"] = r["cuando"] else: p["fallas"] += 1 p["ultima_falla"] = r["cuando"] p["motivos"][r.get("resultado")] = p["motivos"].get(r.get("resultado"), 0) + 1 for x in r.get("pistas") or []: p["pistas"][x] = p["pistas"].get(x, 0) + 1 return jsonify(ok=True, desde=desde, total=len(filas), por_usuario=por, ultimos=list(reversed(filas[-50:]))) # ------------------------------------------------------------ solo desde este sitio (contra CSRF) INSEGUROS = {"POST", "PUT", "PATCH", "DELETE"} @app.before_request def _mismo_sitio(): """Un POST/PATCH/DELETE a la API tiene que venir de una página de este sitio. Los navegadores mandan Sec-Fetch-Site y Origin; los scripts del servidor (curl a 127.0.0.1:8081) no mandan ninguno y pasan.""" if request.method not in INSEGUROS or not request.path.startswith("/api/"): return None sfs = (request.headers.get("Sec-Fetch-Site") or "").lower() if sfs and sfs not in ("same-origin", "none"): return jsonify(ok=False, error="Petición desde otro sitio: no se acepta."), 403 origen = request.headers.get("Origin") if origen: from urllib.parse import urlsplit # noqa: PLC0415 if (urlsplit(origen).hostname or "").lower() != (request.host or "").split(":")[0].lower(): return jsonify(ok=False, error="Petición desde otro sitio: no se acepta."), 403 return None # ------------------------------------------------------------ después de cada respuesta @app.after_request def _despues(resp): try: if request.path == "/api/password" and request.method == "POST" and resp.status_code == 200: j = resp.get_json(silent=True) or {} yo = g["current_user"]() if j.get("ok") and not j.get("reset") and yo and str(j.get("usuario", "")).lower() == yo.lower(): emitir(resp, yo) # cambió la suya: sesión nueva con la versión nueva, no lo saca elif request.path == "/api/whoami" and resp.status_code == 200: u, vence = leer_cookie() if u and vence - time.time() < DIAS_RENOVAR * 86400: emitir(resp, u) except Exception as e: # noqa: BLE001 app.logger.warning("acceso: no se pudo renovar la sesión: %s", e) return resp g["_acceso"] = {"usuarios": usuarios, "verificar": verificar, "intentos_desde": intentos_desde, "plano": _plano}