Helgen-Services/hbox-docs
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
6e4178cb51
hbox-docs/backend/acceso.py

404 lines
20 KiB
Python
Raw Blame History

"""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}
Reference in New Issue View Git Blame Copy Permalink