caudal/programa/cartera/acceso.py

#!/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
# Código de enlace: lo enseña el servidor (escrito y en un QR) para enlazar otro equipo sin teclear la contraseña
CODIGO_SEGUNDOS = 600               # lo que dura
CODIGO_INTENTOS = 5                 # fallos que aguanta antes de anularse
CODIGO_ALFABETO = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"     # sin letras ni cifras que se confundan al copiarlas
CODIGO_LARGO = 10


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
        self._codigo: dict | None = None            # el código de enlace vigente, solo en memoria

    # ── 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 codigo_de_enlace(self) -> dict:
        """Un código de un solo uso y de pocos minutos para enlazar otro equipo.
        Vive solo en memoria: en el disco no queda nada que se pueda adivinar
        con calma. Pedir otro anula el anterior."""
        codigo = "".join(secrets.choice(CODIGO_ALFABETO) for _ in range(CODIGO_LARGO))
        with self._lock:
            self._codigo = {"codigo": codigo, "caduca": self._reloj() + CODIGO_SEGUNDOS, "fallos": 0}
        return {"codigo": f"{codigo[:5]}-{codigo[5:]}", "segundos": CODIGO_SEGUNDOS}

    def canjear_codigo(self, dado: str, llave: bytes | None, nombre: str) -> str:
        """Cambia el código por la credencial del equipo que lo trae. Vale una vez."""
        limpio = "".join(c for c in str(dado or "").upper() if c.isalnum())
        with self._lock:
            self._frenar()
            vigente = self._codigo
            if vigente and self._reloj() > vigente["caduca"]:
                vigente = self._codigo = None
            if not vigente or not llave or not hmac.compare_digest(vigente["codigo"], limpio):
                self._fallos += 1
                self._ultimo_fallo = self._reloj()
                if vigente:
                    vigente["fallos"] += 1
                    if vigente["fallos"] >= CODIGO_INTENTOS:
                        self._codigo = None
                raise ErrorAcceso("Ese código no vale o ya ha caducado: pide uno nuevo en el Caudal del servidor.")
            self._codigo = None
            self._fallos = 0
        return self.enlace_nuevo(llave, nombre)

    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())