"""Lo que dicen los apuntes sobre TLS, migración y DASH, medido en la red c - mb - s.""" import time import unittest import scenarios from netlab import make_pki, wait_until from scenarios import NET, RTT_MS, SERVER def setUpModule(): make_pki() NET.destroy() NET.deploy() if not wait_until(lambda: NET.sh("c", "ping", "-c1", "-W1", SERVER).returncode == 0, seconds=40): NET.destroy() raise RuntimeError("c no alcanza a s") def tearDownModule(): NET.destroy() class HandshakeRounds(unittest.TestCase): """Se comparan RTT, con márgenes: otras cosas pueden estar corriendo en la máquina.""" @classmethod def setUpClass(cls): scenarios.start_server() cls.tls13 = scenarios.measure("--tlsv1.3") cls.tls12 = scenarios.measure("--tlsv1.2", "--tls-max", "1.2") scenarios.stop_server() def test_tcp_handshake_takes_about_one_rtt(self): self.assertGreater(self.tls13[0], RTT_MS * 0.9) self.assertLess(self.tls13[0], RTT_MS * 2) def test_tls13_adds_one_rtt(self): extra = self.tls13[1] - self.tls13[0] self.assertGreater(extra, RTT_MS * 0.9) self.assertLess(extra, RTT_MS * 1.8) def test_tls12_adds_about_two(self): extra = self.tls12[1] - self.tls12[0] self.assertGreater(extra, RTT_MS * 1.8) self.assertLess(extra, RTT_MS * 3) def test_first_byte_is_three_rtt_with_tls13(self): self.assertGreater(self.tls13[2], RTT_MS * 2.8) self.assertLess(self.tls13[2], RTT_MS * 5) class Observer(unittest.TestCase): @classmethod def setUpClass(cls): NET.sh("c", "rm", "-f", "/w13/captures/keys.log") scenarios.start_server() cls.path = scenarios.capture("test-tls") NET.sh("c", "sh", "-c", f"SSLKEYLOGFILE=/w13/captures/keys.log curl -s --max-time 5 --tlsv1.3 --http1.1 " f"--resolve {scenarios.RESOLVE} --cacert /w13/build/pki/ca.crt -o /dev/null {scenarios.URL}") scenarios.stop_capture() scenarios.stop_server() cls.keys = "tls.keylog_file:/w13/captures/keys.log" def count(self, display, *options): return scenarios.tshark(self.path, *options, "-Y", display, "-T", "fields", "-e", "frame.number").split() def test_sni_is_readable_without_keys(self): out = scenarios.tshark(self.path, "-Y", "tls.handshake.type == 1", "-T", "fields", "-e", "tls.handshake.extensions_server_name") self.assertIn(scenarios.NAME, out) def test_certificate_is_hidden_without_keys(self): self.assertEqual(self.count("tls.handshake.type == 11"), []) self.assertEqual(self.count("tls.handshake.type == 20"), []) self.assertEqual(self.count("http.request"), []) def test_server_hello_is_in_the_clear(self): self.assertTrue(self.count("tls.handshake.type == 2")) def test_the_keylog_shows_certificate_finished_and_request(self): for display in ("tls.handshake.type == 11", "tls.handshake.type == 15", "tls.handshake.type == 20", "http.request"): self.assertTrue(self.count(display, "-o", self.keys), display) class Peek(unittest.TestCase): def test_sni_peek_sees_the_name(self): NET.daemon("s", f"exec /w13/bin/sni_peek {scenarios.PORT} 1 >/tmp/peek.log 2>&1") ok = wait_until(lambda: "escuchando" in NET.out("s", "cat", "/tmp/peek.log"), seconds=10, step=0.2) self.assertTrue(ok) NET.sh("c", "curl", "-s", "--max-time", "2", "--resolve", scenarios.RESOLVE, "-k", scenarios.URL) wait_until(lambda: "SNI" in NET.out("s", "cat", "/tmp/peek.log"), seconds=5, step=0.2) self.assertIn(f"SNI: {scenarios.NAME}", NET.out("s", "cat", "/tmp/peek.log")) class Certificates(unittest.TestCase): def result(self, cert, *args): scenarios.start_server(cert) return NET.sh("c", "curl", "-sS", "--max-time", "5", "--resolve", scenarios.RESOLVE, *args, "-o", "/dev/null", scenarios.URL) def tearDown(self): scenarios.stop_server() def test_good_with_our_ca(self): self.assertEqual(self.result("good", "--cacert", "/w13/build/pki/ca.crt").returncode, 0) def test_good_without_our_ca_is_rejected(self): done = self.result("good") self.assertEqual(done.returncode, 60) self.assertIn("local issuer", done.stderr) def test_self_signed_is_rejected(self): done = self.result("selfsigned", "--cacert", "/w13/build/pki/ca.crt") self.assertEqual(done.returncode, 60) self.assertIn("self-signed", done.stderr) def test_wrong_name_is_rejected(self): done = self.result("wrongname", "--cacert", "/w13/build/pki/ca.crt") self.assertEqual(done.returncode, 60) self.assertIn("subject name", done.stderr) def test_expired_is_rejected(self): done = self.result("expired", "--cacert", "/w13/build/pki/ca.crt") self.assertEqual(done.returncode, 60) self.assertIn("expired", done.stderr) class Migration(unittest.TestCase): @classmethod def setUpClass(cls): NET.daemon("s", "exec python3 -u /w13/cid_echo.py serve 7000 >/tmp/cid.log 2>&1") wait_until(lambda: NET.sh("s", "pgrep", "-f", "cid_echo").returncode == 0, seconds=10, step=0.2) time.sleep(0.5) @classmethod def tearDownClass(cls): NET.sh("s", "pkill", "-f", "cid_echo") def tearDown(self): NET.sh("c", "python3", "/w13/cid_echo.py", "restore") def test_tcp_stops_after_the_move(self): out = NET.out("c", "python3", "/w13/cid_echo.py", "tcp", SERVER, "7000") self.assertIn("5: mensaje-5", out) self.assertIn("respondidos: 6 de 12", out) def test_connection_id_keeps_going_after_the_move(self): out = NET.out("c", "python3", "/w13/cid_echo.py", "udp", SERVER, "7000") self.assertIn("respondidos: 12 de 12", out) self.assertIn("10.213.3.20", NET.out("s", "cat", "/tmp/cid.log")) class RealDash(unittest.TestCase): """La caída dura 12 s, con un buffer de hasta 10 s: lo que importa es la tendencia.""" def rows(self, strategy): rows = [] output = scenarios.dash_run(strategy) for line in output.splitlines(): parts = line.split() if len(parts) == 6 and parts[0].isdigit(): rows.append((int(parts[0]), float(parts[1][:-1]), int(parts[2][:-1]), float(parts[5][:-1]))) self.assertGreater(len(rows), 20, output) return rows def test_last_sample_lowers_quality_during_the_cut_and_recovers(self): rows = self.rows("last") before = [rate for _, moment, rate, _ in rows if 3 < moment < 9] during = [rate for _, moment, rate, _ in rows if 14 < moment < 21] self.assertTrue(before and during, rows) self.assertGreater(min(before), max(during), rows) self.assertGreater(rows[-1][2], max(during), rows) def test_always_the_top_quality_freezes(self): rows = self.rows("fixed") self.assertGreater(sum(row[3] for row in rows), 2, rows) if __name__ == "__main__": unittest.main()