caudal/programa/cartera/copia.py

"""
copia.py — La copia de seguridad: todo lo tuyo en un solo fichero cifrado,
para guardarlo donde quieras y restaurarlo en cualquier Caudal.

Qué lleva: las inversiones, las operaciones, los precios y las automáticas;
los ajustes (monederos de bitcoin, extractos) y, si se pide, los
justificantes. No lleva la contraseña de entrada ni nada del servidor.

Cómo va cifrado (ver sobres.py): el contenido se cifra con una llave, y esa
llave viaja dentro del fichero en uno o más «sobres» que solo abre quien sabe
el secreto de cada uno:

  · En un servidor la llave es la de tu instalación y su sobre se cierra con
    tu contraseña de Caudal: la copia se hace con un clic y se abre con la
    contraseña que tenías cuando la hiciste.
  · En tu equipo, sin contraseña de entrada, se pide una para la copia.

Sin el secreto no se puede abrir, y no hay forma de recuperarlo: quien lo
olvide pierde esa copia, no sus datos.

  fichero = «CAUDAL2\n» + largo (4 bytes) + cabecera JSON con los sobres
            + nonce (12) + cifrado (un .tar.gz) con su sello

Restaurar SUSTITUYE lo que hay por lo de la copia (no mezcla). Antes de tocar
nada se comprueba que la copia se abre y que sus datos se leen bien; quien
llama guarda además una copia local de lo que había.

Necesita el paquete «cryptography»; el resto de Caudal funciona sin él.
"""

from __future__ import annotations

import datetime as dt
import io
import json
import re
import tarfile
import tempfile
from pathlib import Path

from . import adjuntos, almacen, sobres
from .datos import FICHERO_VERSION, FICHEROS, Datos, ErrorDatos, leer
from .sincronizar import AJUSTES
from .sobres import ErrorCifrado

MARCA = b"CAUDAL2\n"
MINIMO = 12                             # caracteres de la contraseña, cuando la copia lleva la suya
MAXIMO = 150 * 1024 * 1024              # una copia más grande no se puede subir para restaurarla
EXTENSION = ".caudal"
_DATOS = {f"{n}.jsonl" for n in FICHEROS} | {FICHERO_VERSION}
_ADJUNTO = re.compile(r"^[0-9a-f]{64}$")       # los justificantes se llaman como su huella


class ErrorCopia(Exception):
    """Lo que se le dice al usuario tal cual."""


def cifrar(contenido: bytes, llave: bytes, sobres_llave: list[dict]) -> bytes:
    """El fichero de la copia: el contenido cifrado con esa llave, y sus sobres delante."""
    cabecera = json.dumps({"sobres": sobres_llave}, separators=(",", ":")).encode()
    prefijo = MARCA + len(cabecera).to_bytes(4, "big") + cabecera
    try:
        return prefijo + sobres.cifrar(llave, contenido, prefijo)       # la cabecera va sellada: no se puede cambiar
    except ErrorCifrado as e:
        raise ErrorCopia(str(e)) from None


def con_clave(contenido: bytes, clave: str) -> bytes:
    """Una copia con su propia contraseña (cuando no hay contraseña de Caudal)."""
    if len(clave) < MINIMO:
        raise ErrorCopia(f"La contraseña de la copia tiene que tener al menos {MINIMO} caracteres.")
    llave = sobres.llave_nueva()
    try:
        return cifrar(contenido, llave, [sobres.envolver(llave, clave)])
    except ErrorCifrado as e:
        raise ErrorCopia(str(e)) from None


def descifrar(fichero: bytes, secreto: str) -> bytes:
    """El contenido, si el secreto (contraseña o clave de recuperación) abre alguno de sus sobres."""
    no_es = ErrorCopia("Ese fichero no es una copia de seguridad de Caudal.")
    if not fichero.startswith(MARCA) or len(fichero) < len(MARCA) + 4:
        raise no_es
    largo = int.from_bytes(fichero[len(MARCA):len(MARCA) + 4], "big")
    fin = len(MARCA) + 4 + largo
    try:
        cabecera = json.loads(fichero[len(MARCA) + 4:fin])
        los_sobres = cabecera["sobres"]
    except (ValueError, KeyError, TypeError):
        raise no_es from None
    try:
        llave = sobres.abrir_alguno(los_sobres if isinstance(los_sobres, list) else [], secreto)
        if llave is None:
            raise ErrorCopia("No se puede abrir: esa no es la contraseña que tenías cuando hiciste esta copia.")
        return sobres.descifrar(llave, fichero[fin:], fichero[:fin])
    except ErrorCifrado as e:
        raise ErrorCopia("El fichero de la copia está dañado." if "cryptography" not in str(e) else str(e)) from None


def crear(carpeta: Path, clave: str | None = None, con_justificantes: bool = True,
          llave: bytes | None = None, sobres_llave: list[dict] | None = None) -> tuple[bytes, dict]:
    """(fichero cifrado, resumen) con todo lo de esa carpeta de datos. Se cifra
    con la llave de la instalación y sus sobres, o con una contraseña propia."""
    carpeta = Path(carpeta)
    resumen = {"aplicacion": "caudal", "creada": dt.datetime.now().isoformat(timespec="seconds"),
               "justificantes": 0, "ajustes": []}
    bruto = io.BytesIO()
    with tarfile.open(fileobj=bruto, mode="w:gz") as tar:
        def meter(nombre: str, contenido: bytes):
            info = tarfile.TarInfo(nombre)
            info.size = len(contenido)
            tar.addfile(info, io.BytesIO(contenido))
        for nombre in sorted(_DATOS):
            if (carpeta / nombre).is_file():
                meter(f"datos/{nombre}", almacen.leer(carpeta / nombre))
        for nombre in AJUSTES:
            if (carpeta.parent / nombre).is_file():
                meter(f"ajustes/{nombre}", almacen.leer(carpeta.parent / nombre))
                resumen["ajustes"].append(nombre)
        if con_justificantes and adjuntos.carpeta_de(carpeta).is_dir():
            for f in sorted(adjuntos.carpeta_de(carpeta).iterdir()):
                if f.is_file() and _ADJUNTO.match(f.name):
                    meter(f"adjuntos/{f.name}", almacen.leer(f))
                    resumen["justificantes"] += 1
        meter("copia.json", json.dumps(resumen, ensure_ascii=False).encode())
    fichero = cifrar(bruto.getvalue(), llave, sobres_llave or []) if llave else con_clave(bruto.getvalue(), clave or "")
    if len(fichero) > MAXIMO:
        raise ErrorCopia("La copia pasa de 150 MB por los justificantes. Hazla sin justificantes y guarda esos ficheros aparte.")
    return fichero, {**resumen, "bytes": len(fichero)}


def abrir(fichero: bytes, clave: str) -> dict:
    """Descifra y comprueba una copia sin tocar nada. Devuelve lo que trae:
    {datos: Datos, ajustes: {nombre: bytes}, adjuntos: {huella: bytes}, resumen}."""
    try:
        tar = tarfile.open(fileobj=io.BytesIO(descifrar(fichero, clave)), mode="r:gz")
    except tarfile.TarError:
        raise ErrorCopia("La copia se abre, pero su contenido no se entiende.") from None
    datos, ajustes, justificantes, resumen = {}, {}, {}, {}
    with tar:
        for m in tar.getmembers():
            if not m.isfile():
                continue
            grupo, _, nombre = m.name.partition("/")
            contenido = tar.extractfile(m).read()
            # Solo lo que una copia puede traer; cualquier otra ruta se ignora
            if m.name == "copia.json":
                resumen = json.loads(contenido)
            elif grupo == "datos" and nombre in _DATOS:
                datos[nombre] = contenido
            elif grupo == "ajustes" and nombre in AJUSTES:
                json.loads(contenido)
                ajustes[nombre] = contenido
            elif grupo == "adjuntos" and _ADJUNTO.match(nombre):
                justificantes[nombre] = contenido
    if not datos:
        raise ErrorCopia("La copia no trae datos.")
    with tempfile.TemporaryDirectory() as tmp:
        for nombre, contenido in datos.items():
            (Path(tmp) / nombre).write_bytes(contenido)
        try:
            leidos: Datos = leer(Path(tmp))          # valida y pone al día un formato anterior
        except ErrorDatos as e:
            raise ErrorCopia(f"Los datos de la copia no se pueden leer: {e}") from None
    return {"datos": leidos, "ajustes": ajustes, "adjuntos": justificantes, "resumen": resumen}