191 lines
9.6 KiB
Python
191 lines
9.6 KiB
Python
"""
|
|
Pruebas de la copia de seguridad cifrada (cartera/copia.py y su parte del
|
|
servidor): todo en un fichero que solo se abre con su contraseña.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from cartera import copia, servidor # noqa: E402
|
|
from cartera.operaciones import Cartera, ErrorOperacion # noqa: E402
|
|
|
|
try:
|
|
import cryptography # noqa: F401
|
|
HAY_CIFRADO = True
|
|
except ImportError:
|
|
HAY_CIFRADO = False
|
|
|
|
CLAVE = "la contraseña de mi copia"
|
|
|
|
|
|
@unittest.skipUnless(HAY_CIFRADO, "falta cryptography: solo hace falta para las copias cifradas")
|
|
class TestCifrado(unittest.TestCase):
|
|
def test_lo_cifrado_solo_se_abre_con_su_contraseña(self):
|
|
fichero = copia.con_clave(b"mis datos", CLAVE)
|
|
self.assertNotIn(b"mis datos", fichero)
|
|
self.assertEqual(copia.descifrar(fichero, CLAVE), b"mis datos")
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
copia.descifrar(fichero, "otra contraseña")
|
|
self.assertNotEqual(fichero, copia.con_clave(b"mis datos", CLAVE)) # llave, sal y nonce nuevos cada vez
|
|
|
|
def test_una_llave_con_dos_sobres_se_abre_con_cualquiera_de_los_dos_secretos(self):
|
|
from cartera import sobres
|
|
llave = sobres.llave_nueva()
|
|
fichero = copia.cifrar(b"mis datos", llave, [sobres.envolver(llave, CLAVE),
|
|
sobres.envolver(llave, "clave-de-recuperacion", "recuperacion")])
|
|
self.assertEqual(copia.descifrar(fichero, CLAVE), b"mis datos")
|
|
self.assertEqual(copia.descifrar(fichero, "clave-de-recuperacion"), b"mis datos")
|
|
|
|
def test_no_se_pueden_cambiar_los_sobres_de_una_copia(self):
|
|
from cartera import sobres
|
|
llave = sobres.llave_nueva()
|
|
fichero = copia.cifrar(b"mis datos", llave, [sobres.envolver(llave, CLAVE)])
|
|
# Quien conozca la llave no puede fabricar otra copia con otra cabecera y el mismo contenido
|
|
otro = copia.cifrar(b"x", llave, [sobres.envolver(llave, "contraseña del atacante")])
|
|
largo = lambda f: len(copia.MARCA) + 4 + int.from_bytes(f[len(copia.MARCA):len(copia.MARCA) + 4], "big") # noqa: E731
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
copia.descifrar(otro[:largo(otro)] + fichero[largo(fichero):], "contraseña del atacante")
|
|
|
|
def test_un_fichero_tocado_o_que_no_es_una_copia_se_rechaza(self):
|
|
fichero = bytearray(copia.con_clave(b"mis datos", CLAVE))
|
|
fichero[-1] ^= 1
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
copia.descifrar(bytes(fichero), CLAVE)
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
copia.descifrar(b"esto es un pdf cualquiera, largo de sobra para pasar el minimo de bytes", CLAVE)
|
|
|
|
def test_una_contraseña_corta_no_vale(self):
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
copia.con_clave(b"x", "corta")
|
|
|
|
|
|
@unittest.skipUnless(HAY_CIFRADO, "falta cryptography: solo hace falta para las copias cifradas")
|
|
class TestCopiaCompleta(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.app = self.nueva("origen")
|
|
c = self.app.cartera
|
|
self.fondo = c.guardar_activo({"nombre": "Mi fondo", "tipo": "Fondo"})
|
|
self.mov = c.guardar_movimiento({"activo": self.fondo["id"], "tipo": "Compra", "fecha": "2025-01-10",
|
|
"cantidad": 10, "precio": 100})
|
|
c.adjuntar(self.mov["id"], "factura.pdf", b"%PDF-1.4 una factura")
|
|
self.app.fiscal_criterios({"opciones": {"territorio": "comun"}})
|
|
|
|
def nueva(self, nombre) -> servidor.App:
|
|
cartera = Cartera(Path(self.tmp.name) / nombre / "datos")
|
|
self.addCleanup(cartera.cerrar)
|
|
return servidor.App(cartera)
|
|
|
|
def copia(self, **mas):
|
|
r = self.app.copia_crear({"clave": CLAVE, **mas})
|
|
nombre, fichero = self.app.copia_descargar(r["id"])
|
|
return r, nombre, fichero
|
|
|
|
def restaurar(self, app, fichero, clave=CLAVE, **mas):
|
|
return app.copia_restaurar({"fichero": base64.b64encode(fichero).decode(), "clave": clave, **mas})
|
|
|
|
def test_la_copia_lleva_todo_y_se_entrega_una_sola_vez(self):
|
|
r, nombre, fichero = self.copia()
|
|
self.assertTrue(nombre.startswith("caudal-") and nombre.endswith(".caudal"))
|
|
self.assertEqual((r["bytes"], r["justificantes"]), (len(fichero), 1))
|
|
self.assertNotIn(b"Mi fondo", fichero)
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.app.copia_descargar(r["id"])
|
|
abierta = copia.abrir(fichero, CLAVE)
|
|
self.assertEqual([a["nombre"] for a in abierta["datos"].vivos("activos")], ["Mi fondo"])
|
|
self.assertEqual(len(abierta["adjuntos"]), 1)
|
|
self.assertIn("extractos.json", abierta["ajustes"])
|
|
|
|
def test_sin_justificantes_si_no_se_piden(self):
|
|
r, _, fichero = self.copia(justificantes=False)
|
|
self.assertEqual((r["justificantes"], copia.abrir(fichero, CLAVE)["adjuntos"]), (0, {}))
|
|
|
|
def test_restaurar_en_una_instalacion_nueva_lo_deja_todo_como_estaba(self):
|
|
_, _, fichero = self.copia()
|
|
otra = self.nueva("destino")
|
|
vista = self.restaurar(otra, fichero)
|
|
self.assertEqual((vista["restaurada"], vista["inversiones"], vista["operaciones"], vista["justificantes"]), (False, 1, 1, 1))
|
|
self.assertEqual(otra.estado()["activos"], []) # mirar no cambia nada
|
|
self.assertTrue(self.restaurar(otra, fichero, confirmar=True)["restaurada"])
|
|
e = otra.estado()
|
|
self.assertEqual([a["nombre"] for a in e["activos"]], ["Mi fondo"])
|
|
self.assertEqual(e["movimientos"][0]["importe"], 1000)
|
|
self.assertEqual(e["adjuntos_faltan"], []) # los justificantes también han llegado
|
|
self.assertEqual(otra._ajustes()["fiscal"]["territorio"], "comun")
|
|
# y sigue ahí al volver a abrir la carpeta
|
|
otra.cartera.cerrar()
|
|
reabierta = Cartera(otra.cartera.carpeta)
|
|
self.addCleanup(reabierta.cerrar)
|
|
self.assertEqual(len(reabierta.datos.vivos("movimientos")), 1)
|
|
|
|
def test_restaurar_sustituye_y_guarda_antes_una_copia_local_de_lo_que_habia(self):
|
|
_, _, fichero = self.copia()
|
|
self.app.cartera.guardar_activo({"nombre": "Añadido después", "tipo": "Accion"})
|
|
self.restaurar(self.app, fichero, confirmar=True)
|
|
self.assertEqual([a["nombre"] for a in self.app.estado()["activos"]], ["Mi fondo"])
|
|
locales = list((self.app.cartera.carpeta.parent / "backups").glob("*.tar.gz"))
|
|
self.assertEqual(len(locales), 1)
|
|
with tarfile.open(locales[0]) as tar:
|
|
self.assertIn("Añadido después", tar.extractfile("activos.jsonl").read().decode())
|
|
|
|
def test_con_la_contraseña_equivocada_no_se_toca_nada(self):
|
|
_, _, fichero = self.copia()
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
self.restaurar(self.app, fichero, "no es esta", confirmar=True)
|
|
self.assertEqual(len(self.app.estado()["activos"]), 1)
|
|
|
|
def test_una_copia_con_rutas_raras_no_escribe_fuera_de_su_sitio(self):
|
|
bruto = io.BytesIO()
|
|
with tarfile.open(fileobj=bruto, mode="w:gz") as tar:
|
|
for nombre, contenido in (("datos/activos.jsonl", b""), ("datos/version.json", json.dumps({"version": 2}).encode()),
|
|
("../fuera.txt", b"x"), ("datos/../../fuera.txt", b"x"), ("ajustes/acceso.json", b"{}"),
|
|
("adjuntos/../../fuera", b"x")):
|
|
info = tarfile.TarInfo(nombre)
|
|
info.size = len(contenido)
|
|
tar.addfile(info, io.BytesIO(contenido))
|
|
otra = self.nueva("destino")
|
|
self.restaurar(otra, copia.con_clave(bruto.getvalue(), CLAVE), confirmar=True)
|
|
raiz = Path(self.tmp.name)
|
|
self.assertFalse(list(raiz.rglob("fuera*")))
|
|
self.assertFalse((otra.cartera.carpeta.parent / "acceso.json").exists())
|
|
|
|
def test_en_un_servidor_la_copia_va_con_la_contraseña_de_caudal_sin_pedir_otra(self):
|
|
from cartera.acceso import Acceso
|
|
acceso = Acceso(self.app.cartera.carpeta.parent)
|
|
self.app.acceso, self.app.llave = acceso, acceso.crear("mi contraseña de Caudal")
|
|
r = self.app.copia_crear({}) # sin contraseña en la petición
|
|
_, vieja = self.app.copia_descargar(r["id"])
|
|
otra = self.nueva("destino")
|
|
self.assertEqual(self.restaurar(otra, vieja, "mi contraseña de Caudal")["inversiones"], 1)
|
|
# Al cambiar la contraseña, las copias nuevas piden la nueva y las viejas, la de entonces
|
|
acceso.cambiar("mi contraseña de Caudal", "otra contraseña distinta")
|
|
self.assertEqual(acceso.comprobar("otra contraseña distinta"), self.app.llave) # misma llave
|
|
_, nueva = self.app.copia_descargar(self.app.copia_crear({})["id"])
|
|
self.assertEqual(self.restaurar(otra, nueva, "otra contraseña distinta")["inversiones"], 1)
|
|
self.assertEqual(self.restaurar(otra, vieja, "mi contraseña de Caudal")["inversiones"], 1)
|
|
for fichero, clave in ((nueva, "mi contraseña de Caudal"), (vieja, "otra contraseña distinta")):
|
|
with self.assertRaises(copia.ErrorCopia):
|
|
self.restaurar(otra, fichero, clave)
|
|
|
|
def test_en_la_cartera_de_ejemplo_no_hay_copias(self):
|
|
self.app.ejemplo(True)
|
|
self.addCleanup(lambda: self.app.ejemplo(False))
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.app.copia_crear({"clave": CLAVE})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|