""" Pruebas del cifrado de los datos en el disco (cartera/almacen.py, la clave de recuperación de acceso.py y su parte del servidor): opcional y solo cuando Caudal va en un servidor. python3 -m unittest discover -s pruebas -v """ from __future__ import annotations import base64 import http.client import json import sys import tempfile import threading import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from cartera import almacen, btc, servidor, sobres # noqa: E402 from cartera.acceso import ErrorAcceso # noqa: E402 from cartera.operaciones import ErrorOperacion # noqa: E402 try: import cryptography # noqa: F401 HAY_CIFRADO = True except ImportError: HAY_CIFRADO = False CLAVE = "mi contraseña de Caudal" SECRETO = "Fondo Muy Reconocible" # un nombre que no debe verse en el disco cifrado @unittest.skipUnless(HAY_CIFRADO, "falta cryptography") class TestAlmacen(unittest.TestCase): def setUp(self): self.addCleanup(almacen.poner_llave, None) def test_sin_cifrado_todo_pasa_tal_cual(self): self.assertEqual(almacen.sellar(b"hola"), b"hola") self.assertEqual(almacen.abrir(b"hola"), b"hola") def test_con_la_llave_activa_se_cifra_y_se_lee_igual(self): almacen.poner_llave(sobres.llave_nueva(), True) sellado = almacen.sellar(b"hola") self.assertTrue(almacen.esta_cifrado(sellado)) self.assertNotIn(b"hola", sellado) self.assertEqual(almacen.abrir(sellado), b"hola") self.assertEqual(almacen.abrir(b"viejo sin cifrar"), b"viejo sin cifrar") # conviven mientras se cambia def test_sin_llave_o_con_otra_no_se_lee_y_no_es_un_error_corriente(self): almacen.poner_llave(sobres.llave_nueva(), True) sellado = almacen.sellar(b"hola") for llave in (None, sobres.llave_nueva()): almacen.poner_llave(llave) with self.assertRaises(almacen.Bloqueado) as e: almacen.abrir(sellado) # quien se traga OSError/ValueError para seguir con un ajuste vacío no debe tragarse este self.assertNotIsInstance(e.exception, (OSError, ValueError)) @unittest.skipUnless(HAY_CIFRADO, "falta cryptography") class TestDatosCifrados(unittest.TestCase): """La vida entera: cifrar, reiniciar, entrar, olvidar la contraseña, descifrar.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.addCleanup(almacen.poner_llave, None) self.carpeta = Path(self.tmp.name) / "datos" self.app = self.arrancar() self.app.entrar(self.app.acceso.crear(CLAVE)) c = self.app.cartera fondo = c.guardar_activo({"nombre": SECRETO, "tipo": "Fondo"}) mov = c.guardar_movimiento({"activo": fondo["id"], "tipo": "Compra", "fecha": "2025-01-10", "cantidad": 10, "precio": 100}) c.adjuntar(mov["id"], "factura.pdf", f"%PDF {SECRETO}".encode()) self.app.fiscal_criterios({"opciones": {"territorio": "comun"}}) btc.guardar(self.carpeta, {**btc.cargar(self.carpeta), "nota": SECRETO}) self.app.guardar_sesion(1234.5) self.app._copia_datos() def arrancar(self) -> servidor.App: app = servidor.arrancar(self.carpeta, 0, "127.0.0.1") self.addCleanup(lambda: (app.servidor.server_close(), app.cartera and app.cartera.cerrar())) return app def reiniciar(self) -> servidor.App: """Como apagar y encender el servidor: la llave se va de la memoria.""" self.app.servidor.server_close() self.app.cartera.cerrar() almacen.poner_llave(None) self.app = self.arrancar() return self.app def en_disco(self) -> list[Path]: return [f for f in Path(self.tmp.name).rglob("*") if f.is_file() and f.name not in ("acceso.json", ".cerrojo")] def nombres(self): return [a["nombre"] for a in self.app.estado()["activos"]] def test_al_activarlo_no_queda_nada_legible_en_el_disco(self): self.assertTrue(any(SECRETO.encode() in f.read_bytes() for f in self.en_disco())) # antes sí se lee r = self.app.cifrado_poner({"activar": True, "clave": CLAVE}) self.assertTrue(r["cifrado"]) self.assertRegex(r["recuperacion"], r"^([A-Z2-9]{4}-){7}[A-Z2-9]{4}$") ficheros = self.en_disco() self.assertGreaterEqual(len(ficheros), 9) # 4 tablas, versión, justificante, 2 ajustes, sesión, copia local for f in ficheros: bruto = f.read_bytes() self.assertTrue(almacen.esta_cifrado(bruto), f.name) self.assertNotIn(SECRETO.encode(), bruto, f.name) self.assertNotIn(SECRETO, (Path(self.tmp.name) / "acceso.json").read_text()) # y Caudal sigue funcionando igual, y lo que se escribe después también va cifrado self.assertEqual(self.nombres(), [SECRETO]) self.app.cartera.guardar_activo({"nombre": "Otra " + SECRETO, "tipo": "Accion"}) self.assertNotIn(SECRETO.encode(), (self.carpeta / "activos.jsonl").read_bytes()) self.assertEqual(self.app.estado()["adjuntos_faltan"], []) self.assertEqual((self.app.estado()["cifrado"], self.app.estado()["recuperacion"]), (True, True)) def test_hace_falta_la_contraseña_para_activarlo(self): with self.assertRaises(ErrorAcceso): self.app.cifrado_poner({"activar": True, "clave": "no es esta"}) self.assertFalse(self.app.acceso.cifrado()) def test_tras_reiniciar_esta_bloqueada_hasta_que_se_entra(self): self.app.cifrado_poner({"activar": True, "clave": CLAVE}) app = self.reiniciar() self.assertIsNone(app.cartera) # no se ha abierto nada with self.assertRaises(ErrorAcceso): app.acceso.comprobar("no es esta") app.entrar(app.acceso.comprobar(CLAVE)) self.assertEqual(self.nombres(), [SECRETO]) self.assertEqual(app._ajustes()["fiscal"]["territorio"], "comun") self.assertEqual(btc.cargar(self.carpeta)["nota"], SECRETO) self.assertEqual(app.ultima_sesion["valor"], 1234.5) def test_quien_olvida_la_contraseña_entra_con_la_clave_de_recuperacion(self): recuperacion = self.app.cifrado_poner({"activar": True, "clave": CLAVE})["recuperacion"] app = self.reiniciar() with self.assertRaises(ErrorAcceso): # desde el servidor ya no se puede restablecer app.acceso.restablecer() with self.assertRaises(ErrorAcceso): app.acceso.recuperar("AAAA-AAAA-AAAA-AAAA-AAAA-AAAA-AAAA-AAAA", "contraseña nueva larga") # escrita a mano: en minúsculas y con espacios en vez de guiones app.entrar(app.acceso.recuperar(recuperacion.lower().replace("-", " "), "contraseña nueva larga")) self.assertEqual(self.nombres(), [SECRETO]) app = self.reiniciar() with self.assertRaises(ErrorAcceso): app.acceso.comprobar(CLAVE) # la vieja ya no vale app.entrar(app.acceso.comprobar("contraseña nueva larga")) self.assertEqual(self.nombres(), [SECRETO]) self.assertTrue(app.acceso.tiene_recuperacion()) # y la clave de recuperación sigue sirviendo def test_una_clave_de_recuperacion_nueva_anula_la_anterior(self): vieja = self.app.cifrado_poner({"activar": True, "clave": CLAVE})["recuperacion"] nueva = self.app.recuperacion_nueva({"clave": CLAVE})["recuperacion"] self.assertNotEqual(vieja, nueva) with self.assertRaises(ErrorAcceso): self.app.acceso.recuperar(vieja, "contraseña nueva larga") self.app.acceso.recuperar(nueva, "contraseña nueva larga") def test_cambiar_la_contraseña_no_obliga_a_cifrar_otra_vez(self): self.app.cifrado_poner({"activar": True, "clave": CLAVE}) antes = (self.carpeta / "activos.jsonl").read_bytes() self.app.acceso.cambiar(CLAVE, "contraseña nueva larga") self.assertEqual((self.carpeta / "activos.jsonl").read_bytes(), antes) app = self.reiniciar() app.entrar(app.acceso.comprobar("contraseña nueva larga")) self.assertEqual(self.nombres(), [SECRETO]) self.assertTrue(app.acceso.cifrado()) def test_al_desactivarlo_vuelve_todo_en_claro_y_arranca_sin_bloqueo(self): self.app.cifrado_poner({"activar": True, "clave": CLAVE}) with self.assertRaises(ErrorAcceso): self.app.cifrado_poner({"activar": False, "clave": "no es esta"}) self.assertFalse(self.app.cifrado_poner({"activar": False, "clave": CLAVE})["cifrado"]) for f in self.en_disco(): self.assertFalse(almacen.esta_cifrado(f.read_bytes()), f.name) app = self.reiniciar() self.assertIsNotNone(app.cartera) self.assertEqual([a["nombre"] for a in app.cartera.datos.vivos("activos")], [SECRETO]) def test_la_copia_de_seguridad_de_datos_cifrados_se_restaura_en_cualquier_caudal(self): recuperacion = self.app.cifrado_poner({"activar": True, "clave": CLAVE})["recuperacion"] _, fichero = self.app.copia_descargar(self.app.copia_crear({})["id"]) self.assertNotIn(SECRETO.encode(), fichero) self.app.cartera.cerrar() almacen.poner_llave(None) from cartera.operaciones import Cartera otra = servidor.App(Cartera(Path(self.tmp.name) / "otra" / "datos")) # un Caudal sin cifrado ni contraseña self.addCleanup(otra.cartera.cerrar) cuerpo = {"fichero": base64.b64encode(fichero).decode(), "confirmar": True} for secreto in (CLAVE, recuperacion.replace("-", "")): # con la contraseña o con la de recuperación otra.copia_restaurar({**cuerpo, "clave": secreto}) self.assertEqual([a["nombre"] for a in otra.estado()["activos"]], [SECRETO]) self.assertIn(SECRETO.encode(), (otra.cartera.carpeta / "activos.jsonl").read_bytes()) # allí, en claro self.assertEqual(otra.estado()["adjuntos_faltan"], []) def test_una_copia_local_cifrada_se_abre_a_mano_sin_arrancar_caudal(self): import tarfile recuperacion = self.app.cifrado_poner({"activar": True, "clave": CLAVE})["recuperacion"] local = next((Path(self.tmp.name) / "backups").glob("*.tar.gz")) self.assertTrue(almacen.esta_cifrado(local.read_bytes())) almacen.poner_llave(None) # como desde una terminal, sin Caudal en marcha with self.assertRaises(almacen.Bloqueado): almacen.descifrar_a_mano(self.carpeta, local, "no es esta") salida = almacen.descifrar_a_mano(self.carpeta, local, CLAVE) self.assertEqual(salida.name, local.name.replace(".tar.gz", ".claro.tar.gz")) with tarfile.open(salida) as tar: self.assertIn(SECRETO, tar.extractfile("activos.jsonl").read().decode()) self.assertTrue(almacen.esta_cifrado(local.read_bytes())) # el original no se toca with self.assertRaises(almacen.Bloqueado): # y no pisa una salida que ya exista almacen.descifrar_a_mano(self.carpeta, local, CLAVE) otra = almacen.descifrar_a_mano(self.carpeta, local, recuperacion.lower(), Path(self.tmp.name) / "con-recuperacion.tar.gz") self.assertEqual(otra.read_bytes(), salida.read_bytes()) almacen.poner_llave(self.app.llave, True) def test_en_tu_equipo_no_hay_cifrado(self): from cartera.operaciones import Cartera local = servidor.App(Cartera(Path(self.tmp.name) / "local" / "datos")) self.addCleanup(local.cartera.cerrar) with self.assertRaises(ErrorOperacion): local.cifrado_poner({"activar": True, "clave": CLAVE}) @unittest.skipUnless(HAY_CIFRADO, "falta cryptography") class TestServidorBloqueado(unittest.TestCase): """Por la red: donde el disco no va cifrado, Caudal cifra desde el principio.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.addCleanup(almacen.poner_llave, None) self.addCleanup(setattr, servidor, "CIFRAR_DE_ENTRADA", servidor.CIFRAR_DE_ENTRADA) servidor.CIFRAR_DE_ENTRADA = True self.levantar() def levantar(self): self.app = servidor.arrancar(Path(self.tmp.name) / "datos", 0, "127.0.0.1") threading.Thread(target=self.app.servidor.serve_forever, daemon=True).start() self.addCleanup(self.apagar, self.app) @staticmethod def apagar(app): app.servidor.shutdown() app.servidor.server_close() if app.cartera is not None: app.cartera.cerrar() app.cartera = None def pedir(self, metodo, ruta, cuerpo=None, **cabeceras): c = http.client.HTTPConnection("127.0.0.1", self.app.puerto, timeout=20) if cuerpo is not None: cabeceras.setdefault("Content-Type", "application/json") c.request(metodo, ruta, json.dumps(cuerpo) if cuerpo is not None else None, cabeceras) r = c.getresponse() datos = json.loads(r.read().decode()) cookie = (r.getheader("Set-Cookie") or "").split(";")[0] c.close() return r.status, datos, cookie def test_crear_la_contraseña_cifra_y_da_la_clave_de_recuperacion_y_al_reiniciar_hay_que_entrar(self): estado, datos, cookie = self.pedir("POST", "/api/acceso/crear", {"clave": CLAVE}, **{"X-Caudal": "1"}) self.assertEqual(estado, 200) recuperacion = datos["recuperacion"] cab = {"Cookie": cookie, "X-Token": datos["token"]} self.assertEqual(self.pedir("POST", "/api/activo", {"nombre": SECRETO, "tipo": "Fondo"}, **cab)[0], 200) self.assertTrue(almacen.esta_cifrado((Path(self.tmp.name) / "datos" / "activos.jsonl").read_bytes())) # se apaga y se enciende self.apagar(self.app) almacen.poner_llave(None) self.levantar() estado, datos, _ = self.pedir("GET", "/api/acceso") self.assertEqual((datos["cifrado"], datos["recuperable"], datos["dentro"]), (True, True, False)) self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 401) # la sesión de antes ya no vale self.assertEqual(self.pedir("POST", "/api/acceso/entrar", {"clave": "no es esta"}, **{"X-Caudal": "1"})[0], 400) self.assertIsNone(self.app.cartera) # con la clave de recuperación y una contraseña nueva estado, datos, cookie = self.pedir("POST", "/api/acceso/recuperar", {"recuperacion": recuperacion, "nueva": "contraseña nueva larga"}, **{"X-Caudal": "1"}) self.assertEqual(estado, 200) estado, datos, _ = self.pedir("GET", "/api/estado", Cookie=cookie, **{"X-Token": datos["token"]}) self.assertEqual((estado, [a["nombre"] for a in datos["activos"]]), (200, [SECRETO])) if __name__ == "__main__": unittest.main()