709 lines
35 KiB
Python
709 lines
35 KiB
Python
"""
|
|
Pruebas de la sincronización entre dos Caudal (cartera/enlace.py): el de un
|
|
servidor y otro que se enlaza con él. Con dos instalaciones de verdad, una
|
|
atendiendo por la red, y sobre todo los casos en que mezclar podría romper algo.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import socket
|
|
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, enlace, servidor # noqa: E402
|
|
from cartera.datos import huella_datos # noqa: E402
|
|
from cartera.enlace import ErrorEnlace # noqa: E402
|
|
from cartera.operaciones import Cartera, ErrorOperacion # noqa: E402
|
|
|
|
try:
|
|
import cryptography # noqa: F401
|
|
import requests # noqa: F401
|
|
HAY_TODO = True
|
|
except ImportError:
|
|
HAY_TODO = False
|
|
|
|
CLAVE = "mi contraseña de Caudal"
|
|
|
|
|
|
class DosCaudales(unittest.TestCase):
|
|
"""Base: un Caudal en un servidor y otro enlazado con él."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.addCleanup(almacen.poner_llave, None)
|
|
# El servidor (como en Start9) …
|
|
self.srv = servidor.arrancar(Path(self.tmp.name) / "servidor" / "datos", 0, "127.0.0.1")
|
|
threading.Thread(target=self.srv.servidor.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.apagar)
|
|
self.srv.entrar(self.srv.acceso.crear(CLAVE))
|
|
self.url = f"http://127.0.0.1:{self.srv.puerto}"
|
|
# … y el otro Caudal (como el del USB)
|
|
self.usb = servidor.App(Cartera(Path(self.tmp.name) / "usb" / "datos"))
|
|
self.addCleanup(self.usb.cartera.cerrar)
|
|
self.usb.enlace_configurar({"url": self.url, "clave": CLAVE, "nombre": "USB"})
|
|
|
|
def apagar(self):
|
|
self.srv.servidor.shutdown()
|
|
self.srv.servidor.server_close()
|
|
if self.srv.cartera is not None:
|
|
self.srv.cartera.cerrar()
|
|
|
|
def sincronizar(self, **mas):
|
|
return self.usb.enlace_sincronizar(mas)
|
|
|
|
def iguales(self):
|
|
self.assertEqual(huella_datos(self.usb.cartera.datos), huella_datos(self.srv.cartera.datos))
|
|
|
|
def fondo(self, app, nombre="Fondo", **mas):
|
|
a = app.cartera.guardar_activo({"nombre": nombre, "tipo": "Fondo", **mas})
|
|
m = app.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-01-10", "cantidad": 10, "precio": 100})
|
|
return a, m
|
|
|
|
def nombres(self, app):
|
|
return sorted(a["nombre"] for a in app.cartera.datos.vivos("activos"))
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestEnlazarConCodigo(DosCaudales):
|
|
"""Enlazar sin teclear la contraseña: el servidor enseña un código de pocos minutos."""
|
|
|
|
def otro(self, nombre="otro"):
|
|
app = servidor.App(Cartera(Path(self.tmp.name) / nombre / "datos"))
|
|
self.addCleanup(app.cartera.cerrar)
|
|
return app
|
|
|
|
def test_con_el_codigo_se_enlaza_y_sincroniza_sin_contraseña(self):
|
|
self.fondo(self.srv, "Del servidor")
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
self.assertRegex(codigo, r"^[A-Z2-9]{5}-[A-Z2-9]{5}$")
|
|
movil = self.otro()
|
|
movil.enlace_configurar({"url": self.url, "codigo": codigo.lower().replace("-", " "), "nombre": "Móvil"}) # se teclee como se teclee
|
|
movil.enlace_sincronizar({})
|
|
self.assertEqual(self.nombres(movil), ["Del servidor"])
|
|
self.assertEqual([e["nombre"] for e in self.srv.acceso.enlaces()], ["USB", "Móvil"])
|
|
|
|
def test_el_codigo_vale_una_sola_vez(self):
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
self.otro("uno").enlace_configurar({"url": self.url, "codigo": codigo})
|
|
with self.assertRaises(ErrorEnlace) as e:
|
|
self.otro("dos").enlace_configurar({"url": self.url, "codigo": codigo})
|
|
self.assertIn("pide uno nuevo", str(e.exception))
|
|
|
|
def test_pedir_otro_anula_el_anterior(self):
|
|
viejo = self.srv.enlace_codigo()["codigo"]
|
|
nuevo = self.srv.enlace_codigo()["codigo"]
|
|
with self.assertRaises(ErrorEnlace):
|
|
self.otro("uno").enlace_configurar({"url": self.url, "codigo": viejo})
|
|
self.otro("dos").enlace_configurar({"url": self.url, "codigo": nuevo})
|
|
|
|
def test_caduca(self):
|
|
ahora = [1000.0]
|
|
self.srv.acceso._reloj = lambda: ahora[0]
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
ahora[0] += 601
|
|
with self.assertRaises(ErrorEnlace):
|
|
self.otro().enlace_configurar({"url": self.url, "codigo": codigo})
|
|
|
|
def test_varios_fallos_lo_anulan_aunque_luego_se_acierte(self):
|
|
ahora = [1000.0]
|
|
self.srv.acceso._reloj = lambda: ahora[0]
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
for n in range(5):
|
|
ahora[0] += 400 if n else 0 # dejando pasar la espera que imponen los fallos... sin caducar
|
|
self.srv.acceso._codigo["caduca"] = ahora[0] + 600
|
|
with self.assertRaises(ErrorEnlace):
|
|
self.otro(f"malo{n}").enlace_configurar({"url": self.url, "codigo": "AAAAA-AAAAA"})
|
|
ahora[0] += 400
|
|
with self.assertRaises(ErrorEnlace):
|
|
self.otro("bueno").enlace_configurar({"url": self.url, "codigo": codigo})
|
|
|
|
def test_en_el_disco_no_queda_rastro_del_codigo(self):
|
|
codigo = self.srv.enlace_codigo()["codigo"].replace("-", "")
|
|
for f in (Path(self.tmp.name) / "servidor").rglob("*"):
|
|
if f.is_file():
|
|
self.assertNotIn(codigo.encode(), f.read_bytes(), f.name)
|
|
|
|
def test_con_el_servidor_bloqueado_no_hay_codigo_que_valga(self):
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
self.srv.llave = None # como tras reiniciar con los datos cifrados, sin que nadie haya entrado
|
|
with self.assertRaises(ErrorEnlace):
|
|
self.otro().enlace_configurar({"url": self.url, "codigo": codigo})
|
|
|
|
def test_sin_la_cabecera_de_caudal_no_se_atiende(self):
|
|
import requests
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
r = requests.post(f"{self.url}/api/enlace/canje", json={"codigo": codigo}, timeout=10)
|
|
self.assertEqual(r.status_code, 400)
|
|
self.otro().enlace_configurar({"url": self.url, "codigo": codigo}) # y el código sigue valiendo
|
|
|
|
def test_el_qr_lleva_direccion_y_codigo(self):
|
|
filas = self.srv.dibujar_qr({"texto": f"caudal://enlazar?u={self.url}&c=ABCDE-FGHJK"})["filas"]
|
|
self.assertEqual({len(f) for f in filas}, {len(filas)})
|
|
self.assertEqual(len(filas) % 4, 1) # 17 + 4·versión
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestSincronizar(DosCaudales):
|
|
# ── lo normal ──
|
|
def test_enlazar_no_guarda_la_contraseña_y_la_credencial_es_propia(self):
|
|
guardado = (Path(self.tmp.name) / "usb" / "enlace.json").read_text()
|
|
self.assertNotIn(CLAVE, guardado)
|
|
self.assertEqual([e["nombre"] for e in self.srv.acceso.enlaces()], ["USB"])
|
|
self.assertEqual(self.usb.estado()["enlace"]["url"], self.url)
|
|
self.assertEqual(len(self.srv.estado()["equipos"]), 1)
|
|
|
|
def test_con_la_contraseña_equivocada_no_se_enlaza(self):
|
|
otro = servidor.App(Cartera(Path(self.tmp.name) / "otro" / "datos"))
|
|
self.addCleanup(otro.cartera.cerrar)
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
otro.enlace_configurar({"url": self.url, "clave": "no es esta nunca"})
|
|
self.assertEqual(otro.estado()["enlace"], {})
|
|
|
|
def test_la_primera_vez_sube_todo_y_simular_no_cambia_nada(self):
|
|
self.fondo(self.usb, "Mi fondo")
|
|
s = self.sincronizar(simular=True)
|
|
self.assertTrue(s["simulacro"])
|
|
self.assertEqual((s["resumen"]["llegan_aqui"], s["resumen"]["llegan_alli"]), (2, 0))
|
|
self.assertEqual(self.nombres(self.srv), []) # el simulacro no toca nada
|
|
r = self.sincronizar()
|
|
self.assertEqual(self.nombres(self.srv), ["Mi fondo"])
|
|
self.assertEqual(len(self.srv.cartera.datos.vivos("movimientos")), 1)
|
|
self.iguales()
|
|
self.assertEqual(self.sincronizar()["resumen"]["llegan_aqui"], 0) # repetir no hace nada
|
|
|
|
def test_lo_apuntado_en_el_servidor_llega_al_abrir_el_otro(self):
|
|
self.fondo(self.usb, "De antes")
|
|
self.sincronizar()
|
|
self.fondo(self.srv, "Nuevo en el servidor")
|
|
self.srv.cartera.guardar_movimiento({"activo": self.srv.cartera.datos.vivos("activos")[0]["id"], "tipo": "Compra",
|
|
"fecha": "2025-06-01", "cantidad": 1, "precio": 110})
|
|
r = self.sincronizar()["resumen"]
|
|
self.assertEqual((r["llegan_alli"], r["llegan_aqui"]), (3, 0))
|
|
self.assertEqual(self.nombres(self.usb), ["De antes", "Nuevo en el servidor"])
|
|
self.iguales()
|
|
# y antes de cambiar lo suyo guardó una copia local
|
|
self.assertTrue(list((Path(self.tmp.name) / "usb" / "backups").glob("*.tar.gz")))
|
|
|
|
def test_lo_apuntado_en_los_dos_lados_acaba_en_los_dos(self):
|
|
a, _ = self.fondo(self.usb, "Común")
|
|
self.sincronizar()
|
|
self.usb.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-03-01", "cantidad": 2, "precio": 100})
|
|
self.srv.cartera.guardar_movimiento({"activo": a["id"], "tipo": "Compra", "fecha": "2025-04-01", "cantidad": 3, "precio": 100})
|
|
self.sincronizar()
|
|
for app in (self.usb, self.srv):
|
|
self.assertEqual(sorted(m["cantidad"] for m in app.cartera.datos.vivos("movimientos")), [2, 3, 10])
|
|
self.iguales()
|
|
|
|
def test_lo_borrado_en_un_lado_no_resucita(self):
|
|
a, m = self.fondo(self.usb, "Común")
|
|
self.sincronizar()
|
|
self.srv.cartera.borrar_movimiento(m["id"])
|
|
self.sincronizar()
|
|
self.assertEqual(self.usb.cartera.datos.vivos("movimientos"), [])
|
|
self.sincronizar()
|
|
self.assertEqual(self.srv.cartera.datos.vivos("movimientos"), [])
|
|
self.iguales()
|
|
|
|
def test_los_ajustes_y_los_justificantes_viajan_en_los_dos_sentidos(self):
|
|
_, m = self.fondo(self.usb, "Con factura")
|
|
self.usb.cartera.adjuntar(m["id"], "factura.pdf", b"%PDF del usb")
|
|
self.usb.fiscal_criterios({"opciones": {"territorio": "navarra"}})
|
|
r = self.sincronizar()
|
|
self.assertEqual(r["justificantes"]["subidos"], 1)
|
|
self.assertEqual(self.srv.estado()["adjuntos_faltan"], [])
|
|
self.assertEqual(self.srv._ajustes()["fiscal"]["territorio"], "navarra")
|
|
_, m2 = self.fondo(self.srv, "Del servidor")
|
|
self.srv.cartera.adjuntar(m2["id"], "otra.pdf", b"%PDF del servidor")
|
|
self.assertEqual(self.sincronizar()["justificantes"]["bajados"], 1)
|
|
self.assertEqual(self.usb.estado()["adjuntos_faltan"], [])
|
|
|
|
# ── lo que podría romper la mezcla ──
|
|
def test_dos_inversiones_distintas_con_el_mismo_identificador_no_se_pisan(self):
|
|
self.fondo(self.usb, "Común")
|
|
self.sincronizar()
|
|
del_usb, _ = self.fondo(self.usb, "Solo en el USB")
|
|
del_srv, _ = self.fondo(self.srv, "Solo en el servidor")
|
|
self.assertEqual(del_usb["id"], del_srv["id"]) # chocan: las dos son «F2»
|
|
s = self.sincronizar(simular=True)["resumen"]
|
|
self.assertEqual([x["nombre"] for x in s["renombradas"]], ["Solo en el USB"])
|
|
self.sincronizar()
|
|
for app in (self.usb, self.srv):
|
|
activos = {a["nombre"]: a["id"] for a in app.cartera.datos.vivos("activos")}
|
|
self.assertEqual(sorted(activos), ["Común", "Solo en el USB", "Solo en el servidor"])
|
|
self.assertEqual(len(set(activos.values())), 3)
|
|
# cada operación sigue a SU inversión
|
|
por_activo = {m["activo"] for m in app.cartera.datos.vivos("movimientos")}
|
|
self.assertEqual(por_activo, set(activos.values()))
|
|
self.iguales()
|
|
self.assertEqual(self.sincronizar()["resumen"]["renombradas"], []) # y ya no vuelve a chocar
|
|
|
|
def test_los_lugares_añadidos_en_cada_lado_se_conservan(self):
|
|
btc = self.usb.cartera.guardar_activo({"nombre": "Bitcoin", "tipo": "Cripto", "codigo": "bitcoin", "unidad": "BTC"})
|
|
self.sincronizar()
|
|
self.usb.cartera.guardar_lugar(btc["id"], {"nombre": "Monedero", "clase": "monedero"})
|
|
self.srv.cartera.guardar_lugar(btc["id"], {"nombre": "Exchange", "clase": "exchange"})
|
|
self.sincronizar()
|
|
for app in (self.usb, self.srv):
|
|
self.assertEqual(sorted(x["nombre"] for x in app.cartera.activo(btc["id"])["lugares"]), ["Exchange", "Monedero"])
|
|
self.iguales()
|
|
|
|
def test_la_misma_linea_de_extracto_apuntada_en_los_dos_no_se_duplica(self):
|
|
a, _ = self.fondo(self.usb, "Común")
|
|
self.sincronizar()
|
|
linea = {"activo": a["id"], "tipo": "Compra", "fecha": "2025-05-05", "cantidad": 4, "precio": 100, "extracto": "csv:abc123"}
|
|
self.usb.cartera.guardar_movimiento(dict(linea))
|
|
self.srv.cartera.guardar_movimiento(dict(linea))
|
|
r = self.sincronizar()["resumen"]
|
|
self.assertEqual(r["duplicadas"], 1)
|
|
for app in (self.usb, self.srv):
|
|
self.assertEqual(sorted(m["cantidad"] for m in app.cartera.datos.vivos("movimientos")), [4, 10])
|
|
self.iguales()
|
|
|
|
def test_dos_compras_iguales_apuntadas_a_mano_no_se_tocan(self):
|
|
a, _ = self.fondo(self.usb, "Común")
|
|
self.sincronizar()
|
|
compra = {"activo": a["id"], "tipo": "Compra", "fecha": "2025-05-05", "cantidad": 4, "precio": 100}
|
|
self.usb.cartera.guardar_movimiento(dict(compra))
|
|
self.srv.cartera.guardar_movimiento(dict(compra))
|
|
self.assertEqual(self.sincronizar()["resumen"]["duplicadas"], 0) # sin extracto no se puede saber si son la misma
|
|
self.assertEqual(len(self.srv.cartera.datos.vivos("movimientos")), 3)
|
|
|
|
def test_avisa_si_son_dos_carteras_que_no_tienen_nada_que_ver(self):
|
|
self.fondo(self.usb, "Una cartera")
|
|
self.srv.cartera.guardar_activo({"nombre": "Otra cartera", "tipo": "Accion"})
|
|
self.assertTrue(self.sincronizar(simular=True)["resumen"]["carteras_distintas"])
|
|
|
|
def test_si_lo_que_llega_no_vale_no_se_mezcla_nada(self):
|
|
self.fondo(self.srv, "Del servidor")
|
|
antes = huella_datos(self.srv.cartera.datos)
|
|
e = enlace.Enlace(self.usb.carpeta)
|
|
bueno = enlace.empaquetar(self.usb.cartera.datos, {})
|
|
for malo in ({**bueno, "version": bueno["version"] + 1},
|
|
{**bueno, "datos": {**bueno["datos"], "movimientos": [{"id": "m1", "activo": "F1", "tipo": "NoExiste", "fecha": "2025-01-01"}]}},
|
|
{**bueno, "datos": {**bueno["datos"], "activos": [{"id": "F9", "nombre": "x", "tipo": "Fondo"}] * 2}},
|
|
{"datos": "esto no es"}):
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
e.llamar("mezclar", malo)
|
|
self.assertEqual(huella_datos(self.srv.cartera.datos), antes)
|
|
|
|
def test_un_equipo_olvidado_ya_no_puede_sincronizar(self):
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.assertEqual(self.srv.enlace_olvidar_equipos({})["olvidados"], 1)
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
self.sincronizar()
|
|
self.assertEqual(self.nombres(self.srv), [])
|
|
|
|
def test_en_la_cartera_de_ejemplo_no_se_sincroniza(self):
|
|
self.usb.ejemplo(True)
|
|
self.addCleanup(lambda: self.usb.ejemplo(False))
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.sincronizar()
|
|
|
|
def test_con_el_servidor_cifrado_y_recien_reiniciado_tambien_sincroniza(self):
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.sincronizar()
|
|
self.srv.cifrado_poner({"activar": True, "clave": CLAVE})
|
|
self.fondo(self.srv, "Nuevo en el servidor")
|
|
# se apaga y se enciende: los datos del servidor quedan cerrados
|
|
puerto = self.srv.puerto
|
|
self.apagar()
|
|
almacen.poner_llave(None)
|
|
self.srv = servidor.arrancar(Path(self.tmp.name) / "servidor" / "datos", puerto, "127.0.0.1")
|
|
threading.Thread(target=self.srv.servidor.serve_forever, daemon=True).start()
|
|
self.assertIsNone(self.srv.cartera)
|
|
self.sincronizar() # la credencial del equipo los abre
|
|
self.assertEqual(self.nombres(self.usb), ["Mi fondo", "Nuevo en el servidor"])
|
|
self.assertTrue(almacen.esta_cifrado((Path(self.tmp.name) / "servidor" / "datos" / "activos.jsonl").read_bytes()))
|
|
# (Que en el otro equipo sigan en claro no se puede comprobar aquí: en esta prueba los dos
|
|
# Caudal van en el mismo proceso y comparten la llave en memoria; de verdad son dos máquinas.)
|
|
|
|
|
|
class TorFalso(threading.Thread):
|
|
"""Un SOCKS5 de mentira: lleva cualquier dirección .onion al servidor de la prueba."""
|
|
|
|
def __init__(self, puerto_destino: int):
|
|
super().__init__(daemon=True)
|
|
self.puerto_destino = puerto_destino
|
|
self.pedidas: list[str] = []
|
|
self.sock = socket.socket()
|
|
self.sock.bind(("127.0.0.1", 0))
|
|
self.sock.listen(8)
|
|
self.puerto = self.sock.getsockname()[1]
|
|
|
|
def run(self):
|
|
while True:
|
|
try:
|
|
cliente, _ = self.sock.accept()
|
|
except OSError:
|
|
return
|
|
threading.Thread(target=self._atender, args=(cliente,), daemon=True).start()
|
|
|
|
def parar(self):
|
|
self.sock.close()
|
|
|
|
def _atender(self, cliente: socket.socket):
|
|
try:
|
|
cliente.recv(3)
|
|
cliente.sendall(b"\x05\x00")
|
|
cabecera = cliente.recv(5)
|
|
nombre = cliente.recv(cabecera[4]).decode()
|
|
cliente.recv(2)
|
|
self.pedidas.append(nombre)
|
|
if not nombre.endswith(".onion") or nombre.startswith("apagada"):
|
|
cliente.sendall(b"\x05\x04\x00\x01" + bytes(6)) # destino inalcanzable
|
|
return
|
|
destino = socket.create_connection(("127.0.0.1", self.puerto_destino))
|
|
cliente.sendall(b"\x05\x00\x00\x01" + bytes(6))
|
|
|
|
def pasar(de, a):
|
|
try:
|
|
while True:
|
|
trozo = de.recv(65536)
|
|
if not trozo:
|
|
break
|
|
a.sendall(trozo)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
for s in (de, a):
|
|
try:
|
|
s.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
hilo = threading.Thread(target=pasar, args=(destino, cliente), daemon=True)
|
|
hilo.start()
|
|
pasar(cliente, destino)
|
|
hilo.join(5)
|
|
destino.close()
|
|
except (OSError, IndexError):
|
|
pass
|
|
finally:
|
|
cliente.close()
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestFueraDeCasa(DosCaudales):
|
|
"""Con una dirección .onion de reserva: se usa cuando la de casa no responde."""
|
|
|
|
ONION = "http://caudaldeprueba234567caudaldeprueba234567caudaldeprueba23.onion"
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.tor = TorFalso(self.srv.puerto)
|
|
self.tor.start()
|
|
self.addCleanup(self.tor.parar)
|
|
config = btc.cargar(self.usb.cartera.carpeta)
|
|
config["proxy"] = {"host": "127.0.0.1", "puerto": self.tor.puerto}
|
|
btc.guardar(self.usb.cartera.carpeta, config)
|
|
self.e = enlace.Enlace(self.usb.cartera.carpeta)
|
|
|
|
def fuera_de_casa(self):
|
|
"""Como llevarse el USB: la dirección de casa deja de responder."""
|
|
cerrado = socket.socket()
|
|
cerrado.bind(("127.0.0.1", 0))
|
|
puerto = cerrado.getsockname()[1]
|
|
cerrado.close()
|
|
self.e.guardar({**self.e.config(), "url": f"http://127.0.0.1:{puerto}"})
|
|
|
|
def test_al_ponerla_se_prueba_y_solo_entonces_se_guarda(self):
|
|
r = self.usb.enlace_configurar({"url_tor": self.ONION.replace("http://", "")}) # sin «http://» también vale
|
|
self.assertEqual(r["url_tor"], self.ONION)
|
|
self.assertEqual(self.usb.estado()["enlace"]["url_tor"], self.ONION)
|
|
self.assertEqual(self.tor.pedidas, [self.ONION.split("//")[1]])
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
self.usb.enlace_configurar({"url_tor": "http://apagada234567.onion"}) # no responde: no se guarda
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
self.usb.enlace_configurar({"url_tor": "https://ejemplo.com"}) # no es una .onion
|
|
self.assertEqual(self.e.config()["url_tor"], self.ONION)
|
|
self.usb.enlace_configurar({"url_tor": ""})
|
|
self.assertNotIn("url_tor", self.e.config())
|
|
|
|
def test_en_casa_no_se_toca_tor(self):
|
|
self.usb.enlace_configurar({"url_tor": self.ONION})
|
|
self.tor.pedidas.clear()
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.sincronizar()
|
|
self.assertEqual(self.tor.pedidas, [])
|
|
self.iguales()
|
|
|
|
def test_fuera_de_casa_se_sincroniza_por_tor(self):
|
|
self.usb.enlace_configurar({"url_tor": self.ONION})
|
|
self.fondo(self.usb, "Apuntado de viaje")
|
|
self.fondo(self.srv, "Apuntado en casa")
|
|
self.fuera_de_casa()
|
|
self.tor.pedidas.clear()
|
|
self.assertTrue(self.sincronizar(simular=True)["simulacro"])
|
|
self.sincronizar()
|
|
self.assertTrue(self.tor.pedidas)
|
|
self.assertEqual(self.nombres(self.srv), ["Apuntado de viaje", "Apuntado en casa"])
|
|
self.iguales()
|
|
|
|
def test_sin_direccion_de_reserva_fuera_de_casa_lo_dice(self):
|
|
self.fuera_de_casa()
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "No llego al servidor"):
|
|
self.sincronizar()
|
|
|
|
def test_si_no_se_llega_por_ninguna_lo_dice_y_no_cambia_nada(self):
|
|
self.usb.enlace_configurar({"url_tor": self.ONION})
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.fuera_de_casa()
|
|
self.e.guardar({**self.e.config(), "url_tor": "http://apagada234567.onion"})
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "ni por la dirección de casa ni por la .onion"):
|
|
self.sincronizar()
|
|
self.assertEqual(self.nombres(self.srv), [])
|
|
|
|
def test_un_servidor_anterior_que_no_conoce_la_prueba_tambien_vale(self):
|
|
de_verdad = enlace.atender
|
|
enlace.atender = lambda app, orden, cuerpo, prefijos: (_ for _ in ()).throw(enlace.ErrorEnlace(enlace.NO_ENTIENDO)) \
|
|
if orden == "hola" else de_verdad(app, orden, cuerpo, prefijos)
|
|
self.addCleanup(setattr, enlace, "atender", de_verdad)
|
|
self.assertEqual(self.usb.enlace_configurar({"url_tor": self.ONION})["url_tor"], self.ONION)
|
|
|
|
def test_tambien_se_puede_enlazar_directamente_por_la_onion(self):
|
|
self.usb.enlace_configurar({"desenlazar": True})
|
|
self.usb.enlace_configurar({"url": self.ONION, "clave": CLAVE, "nombre": "USB de viaje"})
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.sincronizar()
|
|
self.assertEqual(self.nombres(self.srv), ["Mi fondo"])
|
|
|
|
def _puerto_cerrado(self):
|
|
cerrado = socket.socket()
|
|
cerrado.bind(("127.0.0.1", 0))
|
|
puerto = cerrado.getsockname()[1]
|
|
cerrado.close()
|
|
return puerto
|
|
|
|
def test_el_qr_con_onion_en_casa_enlaza_por_casa_y_la_deja_de_reserva(self):
|
|
self.usb.enlace_configurar({"desenlazar": True})
|
|
self.tor.pedidas.clear()
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
r = self.usb.enlace_configurar({"url": self.url, "url_tor": self.ONION.replace("http://", ""), "codigo": codigo, "nombre": "Móvil"})
|
|
self.assertEqual(r["url_tor"], self.ONION)
|
|
self.assertEqual(self.tor.pedidas, []) # en casa no se toca Tor
|
|
self.assertEqual(self.e.config()["url_tor"], self.ONION)
|
|
self.fuera_de_casa()
|
|
self.fondo(self.usb, "De viaje")
|
|
self.sincronizar() # y fuera de casa ya va por la .onion
|
|
self.assertEqual(self.nombres(self.srv), ["De viaje"])
|
|
|
|
def test_el_qr_leido_fuera_de_casa_enlaza_por_la_onion(self):
|
|
self.usb.enlace_configurar({"desenlazar": True})
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
casa = f"http://127.0.0.1:{self._puerto_cerrado()}"
|
|
self.usb.enlace_configurar({"url": casa, "url_tor": self.ONION, "codigo": codigo, "nombre": "Móvil"})
|
|
self.assertTrue(self.tor.pedidas)
|
|
self.assertEqual(self.e.config()["url"], casa)
|
|
self.fondo(self.usb, "Mi fondo")
|
|
self.sincronizar()
|
|
self.assertEqual(self.nombres(self.srv), ["Mi fondo"])
|
|
|
|
def test_si_no_llega_por_ninguna_no_queda_enlazado(self):
|
|
self.usb.enlace_configurar({"desenlazar": True})
|
|
codigo = self.srv.enlace_codigo()["codigo"]
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "ni por la dirección de casa ni por la .onion"):
|
|
self.usb.enlace_configurar({"url": f"http://127.0.0.1:{self._puerto_cerrado()}", "url_tor": "http://apagada234567.onion",
|
|
"codigo": codigo})
|
|
self.assertEqual(self.e.config(), {})
|
|
self.otro_con(codigo) # el código no se ha gastado
|
|
|
|
def otro_con(self, codigo):
|
|
app = servidor.App(Cartera(Path(self.tmp.name) / "otro" / "datos"))
|
|
self.addCleanup(app.cartera.cerrar)
|
|
app.enlace_configurar({"url": self.url, "codigo": codigo})
|
|
|
|
def test_una_onion_que_no_es_onion_no_se_acepta_al_enlazar(self):
|
|
self.usb.enlace_configurar({"desenlazar": True})
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "dirección .onion"):
|
|
self.usb.enlace_configurar({"url": self.url, "url_tor": "https://ejemplo.com", "clave": CLAVE})
|
|
|
|
def test_con_la_credencial_retirada_la_onion_no_se_guarda(self):
|
|
self.srv.enlace_olvidar_equipos({})
|
|
with self.assertRaises(enlace.ErrorEnlace):
|
|
self.usb.enlace_configurar({"url_tor": self.ONION})
|
|
self.assertNotIn("url_tor", self.e.config())
|
|
|
|
|
|
def _certificado(carpeta: Path, nombre: str) -> tuple[str, str]:
|
|
"""Un certificado firmado por sí mismo, como autoridad y como servidor a la vez."""
|
|
import datetime as dt
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
clave = ec.generate_private_key(ec.SECP256R1())
|
|
quien = x509.Name([x509.NameAttribute(x509.NameOID.COMMON_NAME, nombre)])
|
|
ahora = dt.datetime.now(dt.timezone.utc)
|
|
cert = (x509.CertificateBuilder().subject_name(quien).issuer_name(quien).public_key(clave.public_key())
|
|
.serial_number(x509.random_serial_number()).not_valid_before(ahora - dt.timedelta(days=1))
|
|
.not_valid_after(ahora + dt.timedelta(days=2))
|
|
.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
|
|
.sign(clave, hashes.SHA256()))
|
|
ruta_cert, ruta_clave = carpeta / f"{nombre}.crt", carpeta / f"{nombre}.key"
|
|
ruta_cert.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
|
ruta_clave.write_bytes(clave.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
|
|
serialization.NoEncryption()))
|
|
return str(ruta_cert), str(ruta_clave)
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestCertificadoElegidoComoFichero(DosCaudales):
|
|
"""En el teléfono no hay una ruta que escribir: el certificado raíz llega con su contenido."""
|
|
|
|
def otro(self):
|
|
app = servidor.App(Cartera(Path(self.tmp.name) / "movil" / "datos"))
|
|
self.addCleanup(app.cartera.cerrar)
|
|
return app
|
|
|
|
def test_se_guarda_junto_a_los_datos_y_se_usa(self):
|
|
ruta, _ = _certificado(Path(self.tmp.name), "startos")
|
|
pem = Path(ruta).read_text()
|
|
movil = self.otro()
|
|
movil.enlace_configurar({"url": self.url, "codigo": self.srv.enlace_codigo()["codigo"], "certificado_pem": pem})
|
|
e = enlace.Enlace(movil.cartera.carpeta)
|
|
guardado = Path(e.config()["certificado"])
|
|
self.assertEqual(guardado.name, enlace.FICHERO_CERTIFICADO)
|
|
self.assertEqual(guardado.parent, e.ruta.parent)
|
|
self.assertEqual(guardado.read_text().strip(), pem.strip())
|
|
movil.enlace_configurar({"desenlazar": True}) # al desenlazar se va con lo demás
|
|
self.assertFalse(guardado.exists())
|
|
|
|
def test_lo_que_no_es_un_certificado_no_se_acepta(self):
|
|
movil = self.otro()
|
|
for malo in ("hola", "-----BEGIN CERTIFICATE-----\nnada\n-----END CERTIFICATE-----"):
|
|
with self.assertRaises(ErrorEnlace):
|
|
movil.enlace_configurar({"url": self.url, "codigo": "ABCDE-FGHJK", "certificado_pem": malo})
|
|
self.assertFalse((enlace.Enlace(movil.cartera.carpeta).ruta.parent / enlace.FICHERO_CERTIFICADO).exists())
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestOnionConHttps(unittest.TestCase):
|
|
"""Una .onion con https, como las de StartOS: un mismo puerto reparte varios
|
|
servicios, así que hay que decir en el saludo a cuál se va (SNI)."""
|
|
|
|
def setUp(self):
|
|
import http.server
|
|
import json
|
|
import ssl
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
carpeta = Path(self.tmp.name)
|
|
self.suya, clave = _certificado(carpeta, "servidor")
|
|
self.otra, _ = _certificado(carpeta, "otra-autoridad")
|
|
visto = self.visto = {}
|
|
|
|
class Manejador(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
visto["host"] = self.headers.get("Host")
|
|
self.rfile.read(int(self.headers["Content-Length"]))
|
|
cuerpo = json.dumps({"ok": True}).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Length", str(len(cuerpo)))
|
|
self.send_header("Set-Cookie", "sesion=abc; HttpOnly")
|
|
self.end_headers()
|
|
self.wfile.write(cuerpo)
|
|
|
|
self.https = http.server.HTTPServer(("127.0.0.1", 0), Manejador)
|
|
contexto = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
contexto.load_cert_chain(self.suya, clave)
|
|
contexto.sni_callback = lambda s, nombre, c: visto.__setitem__("sni", nombre)
|
|
self.https.socket = contexto.wrap_socket(self.https.socket, server_side=True)
|
|
threading.Thread(target=self.https.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.https.server_close)
|
|
self.addCleanup(self.https.shutdown)
|
|
self.tor = TorFalso(self.https.server_address[1])
|
|
self.tor.start()
|
|
self.addCleanup(self.tor.parar)
|
|
|
|
def pedir(self, certificado: str):
|
|
sesion = enlace._SesionTor(("127.0.0.1", self.tor.puerto), certificado)
|
|
r = sesion.post("https://caudaldeprueba234567.onion/api/enlace", json={"orden": "hola"}, headers={}, timeout=(8, 30))
|
|
return r, sesion
|
|
|
|
def test_dice_a_que_servicio_va_y_guarda_la_sesion(self):
|
|
r, sesion = self.pedir(self.suya)
|
|
self.assertEqual((r.status_code, r.json()), (200, {"ok": True}))
|
|
self.assertEqual(self.visto, {"sni": "caudaldeprueba234567.onion", "host": "caudaldeprueba234567.onion"})
|
|
self.assertEqual(sesion.cookies, {"sesion": "abc"})
|
|
|
|
def test_sin_certificado_raiz_tambien_va_porque_tor_ya_garantiza_el_destino(self):
|
|
self.assertEqual(self.pedir("")[0].status_code, 200)
|
|
|
|
def test_con_el_certificado_de_otra_autoridad_no_sigue(self):
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "no es de la autoridad"):
|
|
self.pedir(self.otra)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestDireccionesDelQR(unittest.TestCase):
|
|
"""Las direcciones que lleva el QR de enlazar: la de casa (nunca un .local si hay otra) y la .onion."""
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.datos = Path(self.tmp.name) / "datos"
|
|
self.datos.mkdir()
|
|
viejo = os.environ.pop(enlace.VARIABLE_DIRECCIONES, None)
|
|
self.addCleanup(lambda: os.environ.__setitem__(enlace.VARIABLE_DIRECCIONES, viejo) if viejo is not None
|
|
else os.environ.pop(enlace.VARIABLE_DIRECCIONES, None))
|
|
|
|
def test_de_las_de_startos_elige_la_ip_de_casa_y_la_onion_sin_cifrar(self):
|
|
urls = ["https://servidor.local:50717", "http://servidor.local:50716", "https://10.8.0.5:50717",
|
|
"https://192.168.1.20:50717", "http://192.168.1.20:50716", "https://[fe80::1]:50717",
|
|
"https://abcdefghijklmnop.onion", "http://abcdefghijklmnop.onion"]
|
|
self.assertEqual(enlace.elegir_direcciones(urls), {"casa": "https://192.168.1.20:50717", "tor": "http://abcdefghijklmnop.onion"})
|
|
|
|
def test_un_local_solo_si_no_hay_otra(self):
|
|
self.assertEqual(enlace.elegir_direcciones(["https://caudal.local:8443"])["casa"], "https://caudal.local:8443")
|
|
self.assertEqual(enlace.elegir_direcciones(["https://caudal.local:8443", "https://caudal.midominio.es"])["casa"],
|
|
"https://caudal.midominio.es")
|
|
self.assertEqual(enlace.elegir_direcciones([]), {"casa": "", "tor": ""})
|
|
self.assertEqual(enlace.elegir_direcciones(["http://127.0.0.1:8080", "https://8.8.8.8"]), {"casa": "", "tor": ""})
|
|
|
|
def test_las_del_sistema_llegan_por_la_variable(self):
|
|
os.environ[enlace.VARIABLE_DIRECCIONES] = '["https://192.168.1.50:8443", "http://xyz.onion"]'
|
|
d = enlace.direcciones_de_enlace(self.datos)
|
|
self.assertEqual((d["casa"], d["tor"]), ("https://192.168.1.50:8443", "http://xyz.onion"))
|
|
os.environ[enlace.VARIABLE_DIRECCIONES] = "esto no es json"
|
|
self.assertEqual(enlace.direcciones_de_enlace(self.datos)["casa"], "")
|
|
|
|
def test_lo_escrito_a_mano_manda_y_vacio_vuelve_a_las_del_sistema(self):
|
|
os.environ[enlace.VARIABLE_DIRECCIONES] = '["https://192.168.1.50:8443", "http://xyz.onion"]'
|
|
d = enlace.poner_direcciones_de_enlace(self.datos, "192.168.1.60:8080", "")
|
|
self.assertEqual((d["casa"], d["tor"]), ("http://192.168.1.60:8080", "http://xyz.onion"))
|
|
d = enlace.poner_direcciones_de_enlace(self.datos, "", "")
|
|
self.assertEqual(d["casa"], "https://192.168.1.50:8443")
|
|
self.assertFalse((self.datos.parent / enlace.FICHERO_DIRECCIONES).exists())
|
|
|
|
def test_no_acepta_un_local_ni_una_onion_falsa(self):
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, ".local"):
|
|
enlace.poner_direcciones_de_enlace(self.datos, "https://umbrel.local:3000", "")
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, ".onion"):
|
|
enlace.poner_direcciones_de_enlace(self.datos, "", "https://ejemplo.com")
|
|
with self.assertRaisesRegex(enlace.ErrorEnlace, "no puede ser una .onion"):
|
|
enlace.poner_direcciones_de_enlace(self.datos, "http://xyz.onion", "")
|
|
|
|
|
|
@unittest.skipUnless(HAY_TODO, "faltan cryptography o requests")
|
|
class TestDireccionesEnElServidor(DosCaudales):
|
|
|
|
def test_el_codigo_viene_con_las_direcciones_y_se_pueden_cambiar(self):
|
|
self.assertEqual(self.srv.enlace_codigo()["direcciones"]["casa"], "")
|
|
self.srv.enlace_direcciones({"casa": "https://192.168.1.50:8443", "tor": "xyz.onion"})
|
|
d = self.srv.enlace_codigo()["direcciones"]
|
|
self.assertEqual((d["casa"], d["tor"]), ("https://192.168.1.50:8443", "http://xyz.onion"))
|
|
with self.assertRaises(ErrorOperacion):
|
|
self.usb.enlace_direcciones({}) # solo en un servidor
|