282 lines
13 KiB
Python
282 lines
13 KiB
Python
"""
|
|
Pruebas de la entrada con contraseña (cartera/acceso.py y su parte del
|
|
servidor): solo cuando Caudal va en un servidor; en tu equipo sigue igual.
|
|
|
|
python3 -m unittest discover -s pruebas -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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 acceso as modulo # noqa: E402
|
|
from cartera import servidor # noqa: E402
|
|
from cartera.acceso import Acceso, ErrorAcceso, Espera # noqa: E402
|
|
|
|
try:
|
|
import cryptography # noqa: F401
|
|
HAY_CIFRADO = True
|
|
except ImportError:
|
|
HAY_CIFRADO = False
|
|
SIN_CIFRADO = "falta cryptography: la contraseña guarda una llave y sin él no se puede crear"
|
|
|
|
CLAVE = "una frase que recuerdo"
|
|
|
|
|
|
@unittest.skipUnless(HAY_CIFRADO, SIN_CIFRADO)
|
|
class TestContraseña(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.ahora = 1000.0
|
|
self.a = Acceso(Path(self.tmp.name), reloj=lambda: self.ahora)
|
|
|
|
def test_se_crea_una_vez_y_no_se_guarda_en_claro(self):
|
|
self.assertFalse(self.a.configurado())
|
|
self.a.crear(CLAVE)
|
|
self.assertTrue(self.a.configurado())
|
|
guardado = self.a.ruta.read_text(encoding="utf-8")
|
|
self.assertNotIn(CLAVE, guardado)
|
|
self.assertEqual(self.a.ruta.stat().st_mode & 0o777, 0o600)
|
|
with self.assertRaises(ErrorAcceso):
|
|
self.a.crear("otra contraseña distinta") # no se pisa
|
|
self.a.comprobar(CLAVE)
|
|
|
|
def test_una_contraseña_corta_no_vale(self):
|
|
with self.assertRaises(ErrorAcceso):
|
|
self.a.crear("corta")
|
|
self.assertFalse(self.a.configurado())
|
|
|
|
def test_la_equivocada_no_entra_y_tras_varios_fallos_hay_que_esperar(self):
|
|
self.a.crear(CLAVE)
|
|
for _ in range(modulo.FALLOS_LIBRES):
|
|
with self.assertRaises(ErrorAcceso) as e:
|
|
self.a.comprobar("no es esta")
|
|
self.assertNotIsInstance(e.exception, Espera)
|
|
with self.assertRaises(Espera):
|
|
self.a.comprobar(CLAVE) # ni la buena, hasta que pase la espera
|
|
self.ahora += 2
|
|
self.a.comprobar(CLAVE)
|
|
with self.assertRaises(ErrorAcceso) as e: # acertar pone el contador a cero
|
|
self.a.comprobar("no es esta")
|
|
self.assertNotIsInstance(e.exception, Espera)
|
|
|
|
def test_la_espera_se_dobla_pero_tiene_tope(self):
|
|
self.a.crear(CLAVE)
|
|
for _ in range(modulo.FALLOS_LIBRES + 30):
|
|
self.ahora += modulo.ESPERA_MAXIMA + 1
|
|
with self.assertRaises(ErrorAcceso):
|
|
self.a.comprobar("no es esta")
|
|
self.ahora += modulo.ESPERA_MAXIMA + 1
|
|
self.a.comprobar(CLAVE)
|
|
|
|
def test_cambiarla_cierra_las_sesiones_abiertas(self):
|
|
self.a.crear(CLAVE)
|
|
sid, _ = self.a.abrir()
|
|
with self.assertRaises(ErrorAcceso):
|
|
self.a.cambiar("no es esta", "otra contraseña nueva")
|
|
self.a.cambiar(CLAVE, "otra contraseña nueva")
|
|
self.assertIsNone(self.a.sesion(sid))
|
|
self.a.comprobar("otra contraseña nueva")
|
|
|
|
def test_si_se_olvida_se_restablece_sin_tocar_nada_mas(self):
|
|
self.a.crear(CLAVE)
|
|
otro = Path(self.tmp.name) / "datos.txt"
|
|
otro.write_text("mis datos")
|
|
nueva = self.a.restablecer()
|
|
self.a.comprobar(nueva)
|
|
with self.assertRaises(ErrorAcceso):
|
|
self.a.comprobar(CLAVE)
|
|
self.assertEqual(otro.read_text(), "mis datos")
|
|
|
|
def test_la_contraseña_guarda_una_llave_que_sobrevive_al_cambio_pero_no_al_olvido(self):
|
|
llave = self.a.crear(CLAVE)
|
|
self.assertEqual(len(llave), 32)
|
|
self.assertEqual(self.a.comprobar(CLAVE), llave)
|
|
self.assertNotIn(llave.hex(), self.a.ruta.read_text(encoding="utf-8"))
|
|
self.a.cambiar(CLAVE, "otra contraseña nueva")
|
|
self.assertEqual(self.a.comprobar("otra contraseña nueva"), llave)
|
|
nueva = self.a.restablecer()
|
|
self.assertNotEqual(self.a.comprobar(nueva), llave) # sin la vieja no se abre su sobre
|
|
|
|
def test_tras_cambiarla_falta_una_copia_nueva_hasta_que_se_hace(self):
|
|
self.a.crear(CLAVE)
|
|
self.assertFalse(self.a.copia_pendiente()) # recién creada no hay copias viejas que valgan menos
|
|
self.a.anotar_copia()
|
|
mia, _ = self.a.abrir()
|
|
otra, _ = self.a.abrir()
|
|
self.a.cambiar(CLAVE, "otra contraseña nueva", conservar=mia)
|
|
self.assertTrue(self.a.copia_pendiente())
|
|
self.assertIsNotNone(self.a.sesion(mia)) # quien la cambia sigue dentro
|
|
self.assertIsNone(self.a.sesion(otra)) # los demás, fuera
|
|
self.a.anotar_copia()
|
|
self.assertFalse(self.a.copia_pendiente())
|
|
self.a.restablecer()
|
|
self.assertTrue(self.a.copia_pendiente())
|
|
|
|
def test_una_sesion_caduca(self):
|
|
self.a.crear(CLAVE)
|
|
sid, token = self.a.abrir()
|
|
self.assertEqual(self.a.sesion(sid)["token"], token)
|
|
self.ahora += modulo.DURACION + 1
|
|
self.assertIsNone(self.a.sesion(sid))
|
|
self.assertIsNone(self.a.sesion("inventada"))
|
|
|
|
|
|
@unittest.skipUnless(HAY_CIFRADO, SIN_CIFRADO)
|
|
class TestServidorConContraseña(unittest.TestCase):
|
|
"""Peticiones de verdad contra Caudal arrancado como en un servidor."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls._tmp = tempfile.TemporaryDirectory()
|
|
cls.app = servidor.arrancar(Path(cls._tmp.name) / "datos", 0, "127.0.0.1")
|
|
threading.Thread(target=cls.app.servidor.serve_forever, daemon=True).start()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.app.servidor.shutdown()
|
|
cls.app.servidor.server_close()
|
|
cls.app.cartera.cerrar()
|
|
cls._tmp.cleanup()
|
|
|
|
def pedir(self, metodo, ruta, cuerpo=None, **cabeceras):
|
|
c = http.client.HTTPConnection("127.0.0.1", self.app.puerto, timeout=10)
|
|
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()
|
|
texto = r.read().decode()
|
|
c.close()
|
|
try:
|
|
datos = json.loads(texto)
|
|
except ValueError:
|
|
datos = texto
|
|
return r.status, datos, r.getheader("Set-Cookie") or ""
|
|
|
|
def test_1_sin_sesion_solo_se_ve_la_puerta(self):
|
|
estado, pagina, _ = self.pedir("GET", "/")
|
|
self.assertEqual(estado, 200)
|
|
self.assertIn('name="token" content=""', pagina) # la página no lleva nada privado
|
|
estado, datos, _ = self.pedir("GET", "/api/estado")
|
|
self.assertEqual((estado, datos.get("sin_sesion")), (401, True))
|
|
self.assertEqual(self.pedir("POST", "/api/movimiento", {"x": 1})[0], 401)
|
|
estado, datos, _ = self.pedir("GET", "/api/acceso")
|
|
self.assertEqual((estado, datos["configurado"], datos["dentro"], datos["token"]), (200, False, False, None))
|
|
|
|
def test_2_crear_la_contraseña_y_entrar(self):
|
|
# Sin la cabecera propia de la aplicación, ni se intenta (otra web no puede ponerla)
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": CLAVE})[0], 403)
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": "corta"}, **{"X-Caudal": "1"})[0], 400)
|
|
estado, datos, cookie = self.pedir("POST", "/api/acceso/crear", {"clave": CLAVE}, **{"X-Caudal": "1"})
|
|
self.assertEqual(estado, 200)
|
|
self.assertIn("HttpOnly", cookie)
|
|
self.assertIn("SameSite=Strict", cookie)
|
|
type(self).cookie, type(self).token = cookie.split(";")[0], datos["token"]
|
|
# Con la cookie y el token, dentro; con solo la cookie, no
|
|
self.assertEqual(self.pedir("GET", "/api/estado", Cookie=self.cookie, **{"X-Token": self.token})[0], 200)
|
|
self.assertEqual(self.pedir("GET", "/api/estado", Cookie=self.cookie)[0], 401)
|
|
self.assertEqual(self.pedir("GET", "/api/estado", **{"X-Token": self.token})[0], 401)
|
|
estado, datos, _ = self.pedir("GET", "/api/acceso", Cookie=self.cookie)
|
|
self.assertEqual((datos["configurado"], datos["dentro"], datos["token"]), (True, True, self.token))
|
|
self.assertTrue(self.pedir("GET", "/api/estado", Cookie=self.cookie, **{"X-Token": self.token})[1]["servidor"])
|
|
|
|
def test_3_no_se_puede_crear_otra_ni_entrar_con_la_equivocada(self):
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/crear", {"clave": "otra distinta más"}, **{"X-Caudal": "1"})[0], 400)
|
|
estado, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": "no es esta"}, **{"X-Caudal": "1"})
|
|
self.assertEqual((estado, cookie), (400, ""))
|
|
estado, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": CLAVE}, **{"X-Caudal": "1"})
|
|
self.assertEqual(estado, 200)
|
|
self.assertNotEqual(datos["token"], self.token) # cada sesión, su token
|
|
|
|
def test_4_salir_cierra_la_sesion_pero_no_apaga_caudal(self):
|
|
cab = {"Cookie": self.cookie, "X-Token": self.token}
|
|
self.assertEqual(self.pedir("POST", "/api/salir", {}, **cab)[0], 200) # no apaga nada
|
|
self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 200)
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/salir", {}, Cookie=self.cookie, **{"X-Caudal": "1"})[0], 200)
|
|
self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 401)
|
|
|
|
def test_5_cambiar_la_contraseña_desde_dentro(self):
|
|
_, datos, cookie = self.pedir("POST", "/api/acceso/entrar", {"clave": CLAVE}, **{"X-Caudal": "1"})
|
|
cab = {"Cookie": cookie.split(";")[0], "X-Token": datos["token"]}
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": "no", "nueva": "nueva contraseña"}, **cab)[0], 400)
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": CLAVE, "nueva": "nueva contraseña"}, **cab)[0], 200)
|
|
estado, datos, _ = self.pedir("GET", "/api/estado", **cab) # quien la cambia sigue dentro…
|
|
self.assertEqual((estado, datos["copia_pendiente"]), (200, True)) # …y se le avisa de la copia
|
|
# Un «_sesion» mandado desde fuera no sirve para conservar la sesión de otro
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/cambiar", {"actual": "nueva contraseña", "nueva": "nueva contraseña", "_sesion": "x"}, **cab)[0], 200)
|
|
self.assertEqual(self.pedir("GET", "/api/estado", **cab)[0], 200)
|
|
self.assertEqual(self.pedir("POST", "/api/acceso/entrar", {"clave": "nueva contraseña"}, **{"X-Caudal": "1"})[0], 200)
|
|
|
|
|
|
class TestLigero(unittest.TestCase):
|
|
"""La interfaz viaja comprimida y el navegador no la vuelve a bajar si no ha cambiado."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls._tmp = tempfile.TemporaryDirectory()
|
|
cls.app = servidor.arrancar(Path(cls._tmp.name) / "datos", 0, "127.0.0.1")
|
|
threading.Thread(target=cls.app.servidor.serve_forever, daemon=True).start()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.app.servidor.shutdown()
|
|
cls.app.servidor.server_close()
|
|
cls.app.cartera.cerrar()
|
|
cls._tmp.cleanup()
|
|
|
|
def pedir(self, ruta, **cabeceras):
|
|
c = http.client.HTTPConnection("127.0.0.1", self.app.puerto, timeout=10)
|
|
c.request("GET", ruta, headers=cabeceras)
|
|
r = c.getresponse()
|
|
cuerpo = r.read()
|
|
c.close()
|
|
return r, cuerpo
|
|
|
|
def test_la_interfaz_se_comprime_si_el_navegador_lo_admite(self):
|
|
import gzip
|
|
original = (servidor.WEB / "app.js").read_bytes()
|
|
r, cuerpo = self.pedir("/web/app.js", **{"Accept-Encoding": "gzip, deflate"})
|
|
self.assertEqual(r.getheader("Content-Encoding"), "gzip")
|
|
self.assertLess(len(cuerpo), len(original) / 3)
|
|
self.assertEqual(gzip.decompress(cuerpo), original)
|
|
r, cuerpo = self.pedir("/web/app.js") # y sin comprimir para quien no lo pida
|
|
self.assertIsNone(r.getheader("Content-Encoding"))
|
|
self.assertEqual(cuerpo, original)
|
|
|
|
def test_lo_que_no_ha_cambiado_no_se_vuelve_a_bajar_y_los_datos_nunca_se_guardan(self):
|
|
r, _ = self.pedir("/web/app.js")
|
|
marca = r.getheader("ETag")
|
|
self.assertEqual(r.getheader("Cache-Control"), "no-cache")
|
|
r, cuerpo = self.pedir("/web/app.js", **{"If-None-Match": marca})
|
|
self.assertEqual((r.status, cuerpo), (304, b""))
|
|
r, _ = self.pedir("/web/app.js", **{"If-None-Match": '"otra"'})
|
|
self.assertEqual(r.status, 200)
|
|
r, _ = self.pedir("/api/acceso")
|
|
self.assertEqual(r.getheader("Cache-Control"), "no-store")
|
|
|
|
|
|
class TestEnTuEquipoSigueIgual(unittest.TestCase):
|
|
def test_sin_modo_servidor_no_hay_contraseña_ni_rutas_de_acceso(self):
|
|
with tempfile.TemporaryDirectory() as carpeta:
|
|
app = servidor.arrancar(Path(carpeta) / "datos", 0)
|
|
try:
|
|
self.assertIsNone(app.acceso)
|
|
self.assertFalse(app.estado()["servidor"])
|
|
self.assertFalse((Path(carpeta) / "acceso.json").exists())
|
|
self.assertEqual(app.servidor.server_address[0], "127.0.0.1")
|
|
finally:
|
|
app.servidor.server_close()
|
|
app.cartera.cerrar()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|