289 lines
15 KiB
Python
289 lines
15 KiB
Python
"""
|
|
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()
|