caudal/programa/cartera/almacen.py

"""
almacen.py — Los ficheros de datos, cifrados en el disco cuando así se pide.

Es opcional y solo se ofrece cuando Caudal va en un servidor. Con el cifrado
activado, todo lo que Caudal guarda (inversiones, operaciones, ajustes,
justificantes, copias locales) queda en el disco cifrado con la llave de la
instalación, que solo sale del sobre que cierra tu contraseña (ver sobres.py
y acceso.py). Quien se lleve el disco, o una copia de él, no puede leer nada.

Cómo funciona, para que el resto del programa casi no se entere:

  · leer(ruta) devuelve el contenido en claro. Si el fichero lleva la marca de
    cifrado lo descifra con la llave puesta; si no la lleva, lo devuelve tal
    cual. Así conviven ficheros cifrados y sin cifrar mientras se cambia.
  · sellar(contenido) devuelve lo que hay que escribir en el disco: cifrado si
    hay llave y el cifrado está activado; si no, el contenido tal cual.

Si un fichero está cifrado y no hay llave, o no es la suya, se lanza
«Bloqueado», que no es un error de lectura corriente A PROPÓSITO: quien lee
un ajuste y, si falla, sigue con uno vacío, acabaría escribiendo ese vacío
encima de los datos cifrados.

La llave vive en memoria, en este módulo, desde que se entra con la
contraseña hasta que el servidor se apaga.

Para abrir a mano un fichero cifrado (una copia local de «backups/», por
ejemplo) sin arrancar Caudal:

  python3 -m cartera.almacen --datos <carpeta de datos> descifrar <fichero> [-o <salida>]

Pide la contraseña de Caudal, o la clave de recuperación, y escribe el
contenido en claro en la salida (por defecto, junto al original con «.claro»
delante de la extensión). El fichero cifrado no se toca.
"""

from __future__ import annotations

import argparse
import getpass
import json
import os
import sys
import tempfile
from pathlib import Path

from . import sobres

MARCA = b"CAUDAL-CIFRADO-1\n"

_llave: bytes | None = None
_activo = False


class Bloqueado(Exception):
    """El fichero está cifrado y no se puede abrir con lo que hay (falta la llave o no es la suya)."""


def poner_llave(llave: bytes | None, activo: bool = False) -> None:
    """La llave con la que leer lo cifrado y, si «activo», con la que cifrar lo que se escriba."""
    global _llave, _activo
    _llave, _activo = llave, bool(llave) and activo


def activo() -> bool:
    return _activo


def esta_cifrado(bruto: bytes) -> bool:
    return bruto.startswith(MARCA)


def abrir(bruto: bytes) -> bytes:
    """El contenido en claro de lo leído del disco."""
    if not esta_cifrado(bruto):
        return bruto
    if _llave is None:
        raise Bloqueado("Los datos están cifrados: hay que entrar con la contraseña para abrirlos.")
    try:
        return sobres.descifrar(_llave, bruto[len(MARCA):], MARCA)
    except sobres.ErrorCifrado:
        raise Bloqueado("Un fichero de datos no se puede descifrar: no es de esta instalación o está dañado.") from None


def sellar(contenido: bytes) -> bytes:
    """Lo que hay que escribir en el disco para guardar ese contenido."""
    if not _activo:
        return contenido
    return MARCA + sobres.cifrar(_llave, contenido, MARCA)


def leer(ruta: Path) -> bytes:
    return abrir(Path(ruta).read_bytes())


def leer_texto(ruta: Path) -> str:
    return leer(ruta).decode("utf-8")


def escribir(ruta: Path, contenido: bytes) -> None:
    """Escribe el fichero entero o no lo toca, y solo lo lee su dueño."""
    ruta = Path(ruta)
    fd, tmp = tempfile.mkstemp(dir=ruta.parent, prefix=f".{ruta.name}.", suffix=".tmp")
    try:
        with os.fdopen(fd, "wb") as f:
            f.write(sellar(contenido))
            f.flush()
            os.fsync(f.fileno())
        try:
            os.chmod(tmp, 0o600)
        except OSError:
            pass                                # exFAT (el USB): sin permisos
        os.replace(tmp, ruta)
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise


def reescribir(ficheros: list[Path]) -> int:
    """Vuelve a guardar cada fichero tal como toca ahora (cifrado o en claro).
    Devuelve cuántos ha cambiado. Primero comprueba que TODOS se pueden leer:
    si alguno no, no se toca ninguno."""
    contenidos = [(f, f.read_bytes()) for f in ficheros if f.is_file()]
    claros = [(f, bruto, abrir(bruto)) for f, bruto in contenidos]
    cambiados = 0
    for f, bruto, claro in claros:
        if esta_cifrado(bruto) != _activo:
            escribir(f, claro)
            cambiados += 1
    return cambiados


def descifrar_a_mano(carpeta_datos: Path, fichero: Path, secreto: str, salida: Path | None = None) -> Path:
    """Escribe en claro un fichero cifrado por esta instalación. «secreto» es
    la contraseña de Caudal o la clave de recuperación. Devuelve dónde lo ha dejado."""
    fichero = Path(fichero)
    bruto = fichero.read_bytes()
    if not esta_cifrado(bruto):
        raise Bloqueado("Ese fichero no está cifrado: se puede abrir tal cual.")
    try:
        registro = json.loads((Path(carpeta_datos).resolve().parent / "acceso.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        raise Bloqueado("No encuentro junto a esos datos el fichero de la contraseña (acceso.json).") from None
    # La clave de recuperación se guarda sin guiones y en mayúsculas: se prueba también así
    limpio = "".join(c for c in secreto.upper() if c.isalnum())
    llave = sobres.abrir_alguno(registro.get("sobres") or [], secreto) or sobres.abrir_alguno(registro.get("sobres") or [], limpio)
    if llave is None:
        raise Bloqueado("Esa no es la contraseña ni la clave de recuperación de esta instalación.")
    try:
        claro = sobres.descifrar(llave, bruto[len(MARCA):], MARCA)
    except sobres.ErrorCifrado:
        raise Bloqueado("El fichero no se puede descifrar: no es de esta instalación o está dañado.") from None
    if salida is None:
        nombre = fichero.name.split(".", 1)
        salida = fichero.with_name(nombre[0] + ".claro" + ("." + nombre[1] if len(nombre) > 1 else ""))
    salida = Path(salida)
    if salida.exists():
        raise Bloqueado(f"Ya existe {salida}: bórralo o elige otra salida con -o.")
    salida.write_bytes(claro)
    try:
        os.chmod(salida, 0o600)
    except OSError:
        pass
    return salida


def main() -> int:
    p = argparse.ArgumentParser(description="Abrir a mano un fichero de datos cifrado por Caudal")
    p.add_argument("--datos", required=True, help="la carpeta de datos de la instalación (junto a ella está acceso.json)")
    p.add_argument("orden", choices=["descifrar"])
    p.add_argument("fichero")
    p.add_argument("-o", "--salida", help="dónde escribir el contenido en claro")
    args = p.parse_args()
    secreto = getpass.getpass("Contraseña de Caudal o clave de recuperación: ")
    try:
        salida = descifrar_a_mano(Path(args.datos), Path(args.fichero), secreto, Path(args.salida) if args.salida else None)
    except (Bloqueado, sobres.ErrorCifrado, OSError) as e:
        print(f"✗ {e}")
        return 1
    print(f"✓ En claro en {salida}. Bórralo cuando termines: ahí ya no está protegido.")
    return 0


if __name__ == "__main__":
    sys.exit(main())