346 lines
15 KiB
Python
346 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
acceso.py — La contraseña con la que se entra cuando Caudal está en un servidor.
|
||
|
||
En tu propio equipo Caudal solo escucha en ese equipo y se abre con una
|
||
dirección secreta: no hace falta contraseña. En un servidor de casa (Start9,
|
||
Umbrel) está en la red, así que se entra con una contraseña que se crea la
|
||
primera vez. Un solo usuario por instalación.
|
||
|
||
Es la única contraseña que hay que recordar: sirve para entrar y, además,
|
||
cierra el «sobre» de la llave con la que se cifran las copias de seguridad
|
||
(ver sobres.py). Al entrar, esa llave queda en memoria para poder hacer una
|
||
copia con un clic.
|
||
|
||
Qué se guarda y dónde: junto a los datos, en «acceso.json» (solo lo lee tu
|
||
usuario), la contraseña pasada por scrypt con su sal y el sobre de la llave.
|
||
Nunca la contraseña.
|
||
|
||
Si se olvida, se puede restablecer desde el servidor sin perder los datos (lo
|
||
que no se recupera es la llave: las copias viejas siguen pidiendo la
|
||
contraseña que había cuando se hicieron):
|
||
|
||
python3 -m cartera.acceso --datos <carpeta de datos> restablecer
|
||
|
||
que pone una contraseña nueva al azar y la enseña (es lo que hará el botón de
|
||
la ficha de la aplicación en el servidor).
|
||
|
||
Las sesiones viven en memoria: al reiniciar el servidor hay que volver a
|
||
entrar. Tras varios intentos fallidos seguidos hay que esperar, cada vez más.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import secrets
|
||
import sys
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from . import sobres
|
||
|
||
FICHERO = "acceso.json"
|
||
MINIMO = 12 # caracteres: por debajo no se admite
|
||
RECOMENDADO = (20, 32) # lo que se aconseja en pantalla; vale cualquier carácter
|
||
N, R, P, LARGO = 2 ** 15, 8, 1, 32 # scrypt: ~33 MB y unas décimas de segundo por intento
|
||
DURACION = 7 * 24 * 3600 # segundos que vale una sesión
|
||
FALLOS_LIBRES = 5 # intentos seguidos antes de hacer esperar
|
||
ESPERA_MAXIMA = 300
|
||
|
||
|
||
RECUPERACION = "recuperacion" # el tipo de sobre que se abre con la clave de recuperación
|
||
ENLACE = "enlace" # el de un equipo enlazado para sincronizar (otro Caudal, el del USB)
|
||
ALFABETO = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # sin 0/O ni 1/I: se va a copiar a mano
|
||
|
||
|
||
def _normal(recuperacion: str) -> str:
|
||
"""La clave de recuperación como se guarda: sin guiones ni espacios y en mayúsculas."""
|
||
return "".join(c for c in str(recuperacion or "").upper() if c.isalnum())
|
||
|
||
|
||
class ErrorAcceso(Exception):
|
||
"""Lo que se le dice al usuario tal cual (contraseña corta, no coincide…)."""
|
||
|
||
|
||
class Espera(ErrorAcceso):
|
||
"""Demasiados intentos: hay que esperar."""
|
||
|
||
|
||
def _ahora() -> str:
|
||
return time.strftime("%Y-%m-%dT%H:%M:%S") + f".{int(time.time() * 1000) % 1000:03d}"
|
||
|
||
|
||
def _huella(clave: str, sal: bytes, n: int = N, r: int = R, p: int = P) -> bytes:
|
||
return hashlib.scrypt(clave.encode("utf-8"), salt=sal, n=n, r=r, p=p, dklen=LARGO, maxmem=2 ** 27)
|
||
|
||
|
||
class Acceso:
|
||
def __init__(self, carpeta: Path, reloj=time.time):
|
||
self.ruta = Path(carpeta) / FICHERO
|
||
self._reloj = reloj
|
||
self._lock = threading.Lock()
|
||
self._sesiones: dict[str, dict] = {} # id de sesión → {token, caduca}
|
||
self._fallos = 0
|
||
self._ultimo_fallo = 0.0
|
||
|
||
# ── la contraseña ──
|
||
def _leer(self) -> dict | None:
|
||
try:
|
||
return json.loads(self.ruta.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
def configurado(self) -> bool:
|
||
return self._leer() is not None
|
||
|
||
def _guardar(self, clave: str, llave: bytes | None = None, cambio: bool = False) -> bytes:
|
||
"""Guarda la contraseña (su huella) y el sobre de la llave. Sin llave, hace una nueva.
|
||
«cambio» apunta cuándo dejó de valer la anterior, para avisar de que las
|
||
copias de seguridad hechas hasta entonces siguen pidiendo la vieja."""
|
||
if len(clave) < MINIMO:
|
||
raise ErrorAcceso(f"La contraseña tiene que tener al menos {MINIMO} caracteres.")
|
||
if len(clave) > 200:
|
||
raise ErrorAcceso("Esa contraseña es demasiado larga.")
|
||
sal = secrets.token_bytes(16)
|
||
llave_dada = llave is not None
|
||
llave = llave or sobres.llave_nueva()
|
||
registro = {"version": 2, "n": N, "r": R, "p": P,
|
||
"sal": base64.b64encode(sal).decode(), "huella": base64.b64encode(_huella(clave, sal)).decode(),
|
||
"sobres": [sobres.envolver(llave, clave)]}
|
||
antes = self._leer() or {}
|
||
if llave_dada:
|
||
# Misma llave: los sobres de la clave de recuperación y de los equipos enlazados siguen valiendo
|
||
registro["sobres"] += [x for x in antes.get("sobres") or [] if x.get("tipo") in (RECUPERACION, ENLACE)]
|
||
registro["cifrado"] = bool(antes.get("cifrado")) and llave_dada
|
||
registro["cambiada"] = _ahora() if cambio else antes.get("cambiada")
|
||
registro["ultima_copia"] = antes.get("ultima_copia")
|
||
self.ruta.parent.mkdir(parents=True, exist_ok=True)
|
||
temporal = self.ruta.with_suffix(".tmp")
|
||
temporal.write_text(json.dumps(registro), encoding="utf-8")
|
||
temporal.chmod(0o600)
|
||
os.replace(temporal, self.ruta)
|
||
return llave
|
||
|
||
def _reescribir(self, registro: dict) -> None:
|
||
temporal = self.ruta.with_suffix(".tmp")
|
||
temporal.write_text(json.dumps(registro), encoding="utf-8")
|
||
temporal.chmod(0o600)
|
||
os.replace(temporal, self.ruta)
|
||
|
||
# ── el cifrado de los datos y la clave de recuperación ──
|
||
def cifrado(self) -> bool:
|
||
"""Si los datos del disco van cifrados con la llave (ver almacen.py)."""
|
||
return bool((self._leer() or {}).get("cifrado"))
|
||
|
||
def poner_cifrado(self, activo: bool) -> None:
|
||
with self._lock:
|
||
reg = self._leer()
|
||
if reg is None:
|
||
raise ErrorAcceso("Antes hay que crear la contraseña.")
|
||
reg["cifrado"] = bool(activo)
|
||
self._reescribir(reg)
|
||
|
||
def tiene_recuperacion(self) -> bool:
|
||
return any(x.get("tipo") == RECUPERACION for x in self.sobres())
|
||
|
||
def recuperacion_nueva(self, clave: str) -> tuple[bytes, str]:
|
||
"""(llave, clave de recuperación recién hecha). La clave de recuperación
|
||
es un segundo sobre para la misma llave: quien olvide la contraseña
|
||
entra con ella. Solo se enseña esta vez; la anterior deja de valer."""
|
||
llave = self.comprobar(clave)
|
||
letras = "".join(secrets.choice(ALFABETO) for _ in range(32)) # 160 bits
|
||
recuperacion = "-".join(letras[i:i + 4] for i in range(0, 32, 4))
|
||
with self._lock:
|
||
reg = self._leer()
|
||
reg["sobres"] = [x for x in reg["sobres"] if x.get("tipo") != RECUPERACION] \
|
||
+ [sobres.envolver(llave, _normal(recuperacion), RECUPERACION)]
|
||
self._reescribir(reg)
|
||
return llave, recuperacion
|
||
|
||
# ── equipos enlazados para sincronizar (ver enlace.py) ──
|
||
def enlace_nuevo(self, llave: bytes, nombre: str) -> str:
|
||
"""Da a otro Caudal una credencial propia para sincronizar con este: un
|
||
sobre más para la misma llave, que se abre con un secreto al azar que
|
||
solo conoce ese equipo. No es la contraseña y se puede retirar."""
|
||
secreto = secrets.token_urlsafe(32)
|
||
with self._lock:
|
||
reg = self._leer()
|
||
if reg is None:
|
||
raise ErrorAcceso("Antes hay que crear la contraseña.")
|
||
sobre = sobres.envolver(llave, secreto, ENLACE)
|
||
sobre.update({"nombre": str(nombre or "")[:60], "creado": _ahora()})
|
||
reg["sobres"] = [*reg.get("sobres", []), sobre]
|
||
self._reescribir(reg)
|
||
return secreto
|
||
|
||
def enlaces(self) -> list[dict]:
|
||
return [{"nombre": x.get("nombre") or "", "creado": x.get("creado") or ""} for x in self.sobres() if x.get("tipo") == ENLACE]
|
||
|
||
def olvidar_enlaces(self) -> int:
|
||
with self._lock:
|
||
reg = self._leer() or {}
|
||
antes = len(reg.get("sobres") or [])
|
||
reg["sobres"] = [x for x in reg.get("sobres") or [] if x.get("tipo") != ENLACE]
|
||
self._reescribir(reg)
|
||
return antes - len(reg["sobres"])
|
||
|
||
def llave_de_enlace(self, secreto: str) -> bytes:
|
||
"""La llave, si ese secreto es el de un equipo enlazado. Cuenta los fallos como la contraseña."""
|
||
with self._lock:
|
||
self._frenar()
|
||
llave = sobres.abrir_alguno([x for x in self.sobres() if x.get("tipo") == ENLACE], str(secreto or ""))
|
||
if llave is None:
|
||
self._fallos += 1
|
||
self._ultimo_fallo = self._reloj()
|
||
raise ErrorAcceso("Este equipo ya no está enlazado con el servidor: vuelve a enlazarlo.")
|
||
self._fallos = 0
|
||
return llave
|
||
|
||
def recuperar(self, recuperacion: str, nueva: str) -> bytes:
|
||
"""Entra con la clave de recuperación y pone una contraseña nueva. La
|
||
llave es la misma: los datos y las copias de seguridad siguen valiendo."""
|
||
with self._lock:
|
||
self._frenar()
|
||
de_recuperacion = [x for x in self.sobres() if x.get("tipo") == RECUPERACION]
|
||
llave = sobres.abrir_alguno(de_recuperacion, _normal(recuperacion))
|
||
if llave is None:
|
||
self._fallos += 1
|
||
self._ultimo_fallo = self._reloj()
|
||
raise ErrorAcceso("Esa no es la clave de recuperación.")
|
||
self._fallos = 0
|
||
self._guardar(nueva, llave, cambio=True)
|
||
self._sesiones.clear()
|
||
return llave
|
||
|
||
def sobres(self) -> list[dict]:
|
||
"""Los sobres de la llave, para que una copia de seguridad se pueda abrir en otra instalación."""
|
||
return list((self._leer() or {}).get("sobres") or [])
|
||
|
||
def crear(self, clave: str) -> bytes:
|
||
"""La primera vez. Si ya hay contraseña, no se pisa. Devuelve la llave."""
|
||
with self._lock:
|
||
if self.configurado():
|
||
raise ErrorAcceso("Ya hay una contraseña. Entra con ella.")
|
||
return self._guardar(clave)
|
||
|
||
def _coincide(self, clave: str) -> bool:
|
||
reg = self._leer()
|
||
if reg is None:
|
||
return False
|
||
try:
|
||
esperada = base64.b64decode(reg["huella"])
|
||
dada = _huella(clave, base64.b64decode(reg["sal"]), reg["n"], reg["r"], reg["p"])
|
||
except (KeyError, ValueError, TypeError):
|
||
return False
|
||
return hmac.compare_digest(esperada, dada)
|
||
|
||
def _frenar(self) -> None:
|
||
"""Tras varios fallos seguidos, cada intento nuevo espera el doble que el anterior."""
|
||
if self._fallos < FALLOS_LIBRES:
|
||
return
|
||
espera = min(2 ** (self._fallos - FALLOS_LIBRES), ESPERA_MAXIMA)
|
||
falta = self._ultimo_fallo + espera - self._reloj()
|
||
if falta > 0:
|
||
raise Espera(f"Demasiados intentos. Espera {int(falta) + 1} segundos y vuelve a probar.")
|
||
|
||
def comprobar(self, clave: str) -> bytes:
|
||
"""La llave, si la contraseña es la buena; si no, ErrorAcceso. Cuenta los fallos seguidos."""
|
||
clave = str(clave or "")
|
||
with self._lock:
|
||
self._frenar()
|
||
if self._coincide(clave):
|
||
self._fallos = 0
|
||
llave = sobres.abrir_alguno(self.sobres(), clave)
|
||
# Un acceso.json de antes de las llaves, o con el sobre perdido: se le hace una
|
||
return llave or self._guardar(clave)
|
||
self._fallos += 1
|
||
self._ultimo_fallo = self._reloj()
|
||
raise ErrorAcceso("Esa no es la contraseña.")
|
||
|
||
def cambiar(self, actual: str, nueva: str, conservar: str = "") -> None:
|
||
"""Contraseña nueva para la MISMA llave: no hay que volver a cifrar nada.
|
||
Se cierran las demás sesiones; la de quien la cambia («conservar») sigue."""
|
||
llave = self.comprobar(actual)
|
||
with self._lock:
|
||
self._guardar(nueva, llave, cambio=True)
|
||
self._sesiones = {k: v for k, v in self._sesiones.items() if k == conservar}
|
||
|
||
def anotar_copia(self) -> None:
|
||
"""Se acaba de hacer una copia de seguridad con la contraseña de ahora."""
|
||
with self._lock:
|
||
reg = self._leer()
|
||
if reg is not None:
|
||
reg["ultima_copia"] = _ahora()
|
||
temporal = self.ruta.with_suffix(".tmp")
|
||
temporal.write_text(json.dumps(reg), encoding="utf-8")
|
||
temporal.chmod(0o600)
|
||
os.replace(temporal, self.ruta)
|
||
|
||
def copia_pendiente(self) -> bool:
|
||
"""La contraseña cambió y aún no hay ninguna copia de seguridad hecha con la nueva."""
|
||
reg = self._leer() or {}
|
||
return bool(reg.get("cambiada")) and (reg.get("ultima_copia") or "") < reg["cambiada"]
|
||
|
||
def restablecer(self) -> str:
|
||
"""Pone una contraseña nueva al azar y la devuelve. Para quien la olvidó
|
||
y tiene acceso al servidor; los datos no se tocan. La llave es nueva:
|
||
sin la contraseña vieja no hay forma de abrir el sobre de la anterior."""
|
||
if self.cifrado():
|
||
raise ErrorAcceso("Los datos están cifrados con la contraseña: sin ella solo se entra con la clave de "
|
||
"recuperación (en la pantalla de entrada, «He olvidado la contraseña»).")
|
||
nueva = "-".join(secrets.token_hex(3) for _ in range(4))
|
||
with self._lock:
|
||
self._guardar(nueva, cambio=True)
|
||
self._sesiones.clear()
|
||
self._fallos = 0
|
||
return nueva
|
||
|
||
# ── las sesiones ──
|
||
def abrir(self) -> tuple[str, str]:
|
||
"""Una sesión nueva: (id, que va en la cookie; token, que va en cada petición)."""
|
||
sid, token = secrets.token_urlsafe(32), secrets.token_urlsafe(24)
|
||
with self._lock:
|
||
ahora = self._reloj()
|
||
self._sesiones = {k: v for k, v in self._sesiones.items() if v["caduca"] > ahora}
|
||
self._sesiones[sid] = {"token": token, "caduca": ahora + DURACION}
|
||
return sid, token
|
||
|
||
def sesion(self, sid: str) -> dict | None:
|
||
with self._lock:
|
||
s = self._sesiones.get(sid or "")
|
||
if s and s["caduca"] > self._reloj():
|
||
return s
|
||
self._sesiones.pop(sid or "", None)
|
||
return None
|
||
|
||
def cerrar(self, sid: str) -> None:
|
||
with self._lock:
|
||
self._sesiones.pop(sid or "", None)
|
||
|
||
|
||
def main() -> int:
|
||
p = argparse.ArgumentParser(description="La contraseña de entrada de Caudal en un servidor")
|
||
p.add_argument("--datos", required=True, help="la carpeta de datos (la contraseña se guarda junto a ella)")
|
||
p.add_argument("orden", choices=["restablecer", "estado"])
|
||
args = p.parse_args()
|
||
acceso = Acceso(Path(args.datos).resolve().parent)
|
||
if args.orden == "estado":
|
||
print("Hay contraseña." if acceso.configurado() else "Aún no se ha creado la contraseña.")
|
||
return 0
|
||
try:
|
||
print(f"Contraseña nueva: {acceso.restablecer()}")
|
||
except ErrorAcceso as e:
|
||
print(f"✗ {e}")
|
||
return 1
|
||
print("Entra con ella y cámbiala en «Más › Cambiar contraseña». Si Caudal estaba abierto, reinícialo.")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|