"""Lo que se puede comprobar sin red: criptografía de juguete, simuladores, QUIC y DNS.""" import os import struct import tempfile import unittest import crypto_toys import ctr_flip import dash_sim import geo_dns import hol_sim import quic_initial from netlab import HERE class Crypto(unittest.TestCase): def test_dh_agrees_and_small_primes_break(self): big_a, _, secret_a, secret_b = crypto_toys.dh_shared(2087, 5, 731, 1289) self.assertEqual(secret_a, secret_b) # Probando exponentes se recupera un exponente equivalente al secreto. found = crypto_toys.dh_break(2087, 5, big_a) self.assertEqual(pow(5, found, 2087), big_a) def test_rsa_round_trip(self): n, e, d = crypto_toys.rsa_keys(61, 53, 17) self.assertEqual((n, d), (3233, 2753)) for message in (0, 1, 65, 3232): self.assertEqual(crypto_toys.rsa_apply(crypto_toys.rsa_apply(message, e, n), d, n), message) self.assertEqual(crypto_toys.rsa_apply(crypto_toys.rsa_apply(message, d, n), e, n), message) def test_ctr_flip_changes_the_text_and_not_the_rest(self): cipher = ctr_flip.aes_ctr(ctr_flip.MESSAGE) self.assertNotEqual(cipher, ctr_flip.MESSAGE) tampered = ctr_flip.flip(cipher, ctr_flip.MESSAGE.index(b"0100"), ord("0"), ord("9")) self.assertEqual(ctr_flip.aes_ctr(tampered), b"pagar 9100 soles a Ana") class QuicInitial(unittest.TestCase): def test_keys_match_rfc9001_appendix_a(self): key, iv, hp = quic_initial.client_initial_keys(bytes.fromhex("8394c8f03e515708")) self.assertEqual(key.hex(), "1f369613dd76d5467730efcbe3b1a22d") self.assertEqual(iv.hex(), "fa044b2f42a3fd3b46fb255c") self.assertEqual(hp.hex(), "9f50449e04a0e810283a1e9933adedd2") def test_decrypts_the_packet_of_rfc9001_appendix_a(self): with open(os.path.join(HERE, "rfc9001_a2.hex"), encoding="ascii") as file: packet = bytes.fromhex(file.read()) self.assertEqual(len(packet), 1200) info = quic_initial.unprotect_initial(packet) self.assertEqual(info["packet_number"], 2) self.assertEqual(info["frames"][:16].hex(), "060040f1010000ed0303ebf8fa56f129") hello = quic_initial.reassemble(quic_initial.crypto_frames(info["frames"])) self.assertEqual(quic_initial.client_hello_names(hello)[0], "example.com") def test_reads_the_sni_of_a_real_curl(self): packets = quic_initial.catch("secreto.redes.test") # El primer vuelo se rellena hasta 1200 bytes: lo exige QUIC, contra la amplificación. self.assertTrue(all(len(packet) >= 1200 for packet in packets)) pieces = [] for packet in packets: pieces += quic_initial.crypto_frames(quic_initial.unprotect_initial(packet)["frames"]) sni, alpn = quic_initial.client_hello_names(quic_initial.reassemble(pieces)) self.assertEqual(sni, "secreto.redes.test") self.assertIn("h3", alpn) class QuicFromPcap(unittest.TestCase): def test_reads_the_sni_from_a_capture_and_stops_at_the_server_cid(self): packets = quic_initial.catch("captura.redes.test") later = bytearray(packets[0]) later[6] ^= 0xff # otro DCID: como cuando el cliente ya usa el ID del servidor with tempfile.TemporaryDirectory() as folder: path = os.path.join(folder, "q.pcap") with open(path, "wb") as file: # Formato pcap con enlace "raw IP" (101): IPv4 + UDP, sin Ethernet. file.write(struct.pack(" 0 for name in "ABC")) def test_quic_delays_only_the_stream_with_the_hole(self): waits = hol_sim.delayed_by_stream(12, 3, "quic") self.assertGreater(waits["A"][0], 0) self.assertEqual(waits["B"], [0, 0]) self.assertEqual(waits["C"], [0, 0]) def test_one_stream_is_the_same_in_both(self): self.assertEqual(hol_sim.deliver(12, 3, "tcp", "A"), hol_sim.deliver(12, 3, "quic", "A")) def test_without_loss_nobody_waits(self): for mode in ("tcp", "quic"): self.assertTrue(all(value == [0, 0] for value in hol_sim.delayed_by_stream(12, None, mode).values())) class Dash(unittest.TestCase): def test_top_quality_freezes_and_last_sample_barely(self): _, frozen_top = dash_sim.simulate(dash_sim.fixed_top) _, frozen_last = dash_sim.simulate(dash_sim.last_sample) self.assertGreater(frozen_top, 10) self.assertLess(frozen_last, 1) def test_last_sample_follows_the_network(self): rows, _ = dash_sim.simulate(dash_sim.last_sample) during = [row["bitrate"] for row in rows if 50 < row["start"] < 65] self.assertTrue(during) self.assertTrue(all(rate <= 1_000_000 for rate in during)) self.assertEqual(rows[-1]["bitrate"], 5_000_000) def test_last_sample_jumps_on_an_oscillating_network(self): rows, _ = dash_sim.simulate(dash_sim.last_sample, network=dash_sim.OSCILLATING) self.assertGreater(dash_sim.score(rows)[1], 10) def query(name, qtype=1): labels = b"".join(bytes([len(part)]) + part.encode() for part in name.split(".")) return struct.pack("!HHHHHH", 0x1234, 0x0100, 1, 0, 0, 0) + labels + b"\x00" + struct.pack("!HH", qtype, 1) class GeoDns(unittest.TestCase): def answer(self, source): return geo_dns.respond(query(geo_dns.NAME), source) def test_the_source_decides_the_replica(self): self.assertTrue(self.answer("10.213.11.10").endswith(bytes([10, 213, 21, 10]))) self.assertTrue(self.answer("10.213.12.10").endswith(bytes([10, 213, 22, 10]))) def test_keeps_the_query_id_and_sets_the_authoritative_flag(self): reply = self.answer("10.213.11.10") self.assertEqual(reply[:2], b"\x12\x34") self.assertEqual(struct.unpack("!H", reply[2:4])[0] & 0x8400, 0x8400) def test_other_names_and_types_get_an_error_code(self): self.assertEqual(geo_dns.respond(query("otro.redes.test"), "10.213.11.10")[3] & 0x0f, 3) self.assertEqual(geo_dns.respond(query(geo_dns.NAME, 28), "10.213.11.10")[3] & 0x0f, 4) def test_garbage_gets_no_answer(self): self.assertIsNone(geo_dns.respond(b"\x00" * 5, "10.213.11.10")) self.assertIsNone(geo_dns.respond(b"\x00" * 12 + b"\x3f" + b"a" * 4, "10.213.11.10")) if __name__ == "__main__": unittest.main()