"""Pruebas de lo que afirman los apuntes, sobre capturas de la topología tcp/tcp.clab.yml. Las mediciones son ruidosas: se comprueban tendencias con margen, nunca cifras exactas. """ import os import re import statistics import subprocess import unittest import scenarios from netlab import CAPTURES, HERE, wait_until from pcaptools import read_capture from tcpsum import summarize LAB = scenarios.LAB scenarios.VERBOSE = False CLIENT = "10.210.1.10" def capture(name, node="a"): return read_capture(os.path.join(CAPTURES, f"{name}-{node}.pcap")) def data_of(segments): return [s for s in segments if s.src == CLIENT and s.len] def from_server(segments): return [s for s in segments if s.src != CLIENT] def grew(before, after, name): return after.get(name, 0) - before.get(name, 0) def setUpModule(): unittest.addModuleCleanup(LAB.destroy) LAB.deploy() scenarios.reset() if not wait_until(lambda: LAB.sh("a", "ping", "-c", "1", "-W", "2", scenarios.SERVER).returncode == 0, seconds=30): raise RuntimeError("la topología no quedó lista") class CleanTransfer(unittest.TestCase): @classmethod def setUpClass(cls): cls.output = scenarios.scenario_clean() cls.segments = capture("clean") cls.info = summarize(cls.segments) def test_all_bytes_arrive(self): self.assertIn("recibidos 2000000 bytes", self.output) def test_three_way_handshake(self): syn, synack, third = self.segments[:3] self.assertTrue(syn.syn and not syn.has_ack) self.assertTrue(synack.syn and synack.has_ack) self.assertTrue(third.has_ack and not third.syn) # El SYN gasta un número de secuencia: el ACK es el número inicial más uno. self.assertEqual(third.ack, 1) self.assertEqual(synack.ack, 1) self.assertEqual(third.seq_raw, syn.seq_raw + 1) def test_mss_and_size_of_the_data(self): self.assertEqual(self.segments[0].mss, 1500 - 20 - 20) self.assertEqual(self.info["largest"], 1460 - 12) # 12 bytes de la opción de timestamps self.assertTrue(self.segments[0].ts) def test_sequence_counts_bytes(self): data = data_of(self.segments) self.assertEqual(data[0].seq, 1) for current, following in zip(data, data[1:]): self.assertEqual(following.seq, current.seq + current.len) def test_ack_is_the_next_byte_expected(self): limits = {s.seq + s.len for s in data_of(self.segments)} | {1} limits.add(max(limits) + 1) # el FIN del cliente también gasta un número for ack in from_server(self.segments): if ack.has_ack and not ack.fin and not ack.syn: self.assertIn(ack.ack, limits) def test_relative_is_not_the_real_number(self): first = data_of(self.segments)[0] self.assertEqual(first.seq, 1) self.assertEqual(first.seq_raw, self.segments[0].seq_raw + 1) self.assertNotEqual(first.seq_raw, 1) def test_window_scale(self): self.assertEqual(self.segments[0].wscale, 7) self.assertEqual(self.segments[0].win, self.segments[0].win_field) # el SYN no se escala later = from_server(self.segments)[5] self.assertEqual(later.win, later.win_field * 128) def test_no_loss_means_no_retransmissions(self): self.assertEqual(self.info["resent"], 0) self.assertEqual(self.info["dup_acks"], 0) def test_fewer_acks_than_data_segments(self): self.assertLess(self.info["acks"], 0.95 * self.info["data_segments"]) self.assertGreater(self.info["acks"], 0.4 * self.info["data_segments"]) def test_closes_with_fin(self): self.assertEqual(self.info["fins"], 2) self.assertEqual(self.info["rsts"], 0) def test_receiver_sees_the_same_bytes(self): self.assertEqual(len(data_of(capture("clean", "b"))), self.info["data_segments"]) def test_time_sequence_graph(self): png = os.path.join(CAPTURES, "clean-a.png") done = subprocess.run(["python3", "timeseq.py", os.path.join(CAPTURES, "clean-a.pcap"), png], cwd=HERE, capture_output=True, text=True, timeout=120) self.assertEqual(done.returncode, 0, done.stderr) with open(png, "rb") as file: self.assertEqual(file.read(8)[1:4], b"PNG") def test_initial_sequence_numbers_change(self): scenarios.scenario_mss() self.assertNotEqual(capture("mss")[0].seq_raw, self.segments[0].seq_raw) class SmallerMtu(unittest.TestCase): def test_mss_follows_the_mtu(self): scenarios.scenario_mss() segments = capture("mss") info = summarize(segments) self.assertEqual(info["syn"].mss, 1400 - 40) self.assertEqual(info["synack"].mss, 1500 - 40) self.assertEqual(info["largest"], 1360 - 12) class OneDrop(unittest.TestCase): @classmethod def setUpClass(cls): cls.output, cls.before, cls.after = scenarios.scenario_drop() cls.a = capture("drop", "a") cls.b = capture("drop", "b") cls.resent = [s for s in data_of(cls.a) if s.resent] def test_all_bytes_arrive(self): self.assertIn("recibidos 2000000 bytes", self.output) def test_lost_segment_goes_twice_and_arrives_once(self): self.assertGreaterEqual(len(self.resent), 1) lost = self.resent[0].seq self.assertEqual(len([s for s in data_of(self.a) if s.seq == lost]), 2) self.assertEqual(len([s for s in data_of(self.b) if s.seq == lost]), 1) def test_retransmission_comes_after_duplicate_acks(self): retransmission = self.resent[0] earlier = [s for s in from_server(self.a) if s.frame < retransmission.frame and s.dup_ack] self.assertGreaterEqual(len(earlier), 3) self.assertTrue(all(s.ack == retransmission.seq for s in earlier)) self.assertTrue(retransmission.fast) def test_duplicate_acks_carry_sack(self): duplicates = [s for s in from_server(self.a) if s.dup_ack] self.assertTrue(duplicates) lost = self.resent[0].seq for segment in duplicates[:3]: self.assertGreater(int(segment.sack_le.split(",")[0]), lost) def test_ack_jumps_once_the_hole_is_filled(self): retransmission = self.resent[0] later = [s for s in from_server(self.a) if s.frame > retransmission.frame and s.ack > retransmission.seq] self.assertGreater(later[0].ack - retransmission.seq, 10 * retransmission.len) def test_no_timeout(self): self.assertEqual(grew(self.before, self.after, "TcpExtTCPTimeouts"), 0) self.assertGreaterEqual(grew(self.before, self.after, "TcpExtTCPSackRecovery"), 1) class ClassicDrop(unittest.TestCase): @classmethod def setUpClass(cls): cls.output = scenarios.scenario_classic() cls.segments = capture("classic") def test_sack_was_not_negotiated(self): self.assertFalse(self.segments[0].sack_perm) self.assertFalse(any(s.sack_le for s in self.segments)) def test_retransmits_on_the_third_duplicate(self): retransmission = next(s for s in data_of(self.segments) if s.resent) earlier = [s for s in from_server(self.segments) if s.frame < retransmission.frame and s.dup_ack] # El kernel puede atender varios ACK juntos: al menos tres, y no muchos más. self.assertGreaterEqual(len(earlier), 3) self.assertLessEqual(len(earlier), 8) self.assertIn("recibidos 2000000 bytes", self.output) class RetransmissionTimeout(unittest.TestCase): @classmethod def setUpClass(cls): cls.before, cls.after, cls.samples = scenarios.scenario_rto() cls.segments = capture("rto") resent = [s for s in data_of(cls.segments) if s.resent] counts = {} for segment in resent: counts[segment.seq] = counts.get(segment.seq, []) + [segment.time] cls.times = max(counts.values(), key=len) def test_timer_expired_several_times(self): self.assertGreaterEqual(grew(self.before, self.after, "TcpExtTCPTimeouts"), 2) self.assertGreaterEqual(len(self.times), 3) def test_interval_doubles(self): gaps = [b - a for a, b in zip(self.times, self.times[1:])] for shorter, longer in zip(gaps, gaps[1:]): self.assertGreater(longer / shorter, 1.5) self.assertLess(longer / shorter, 2.6) def test_a_probe_goes_first(self): self.assertGreaterEqual(grew(self.before, self.after, "TcpExtTCPLossProbes"), 1) def test_rto_of_linux_is_rtt_plus_a_floor(self): text = self.samples["before"] rto = float(re.search(r"\brto:([\d.]+)", text).group(1)) rtt = float(re.search(r"\brtt:([\d.]+)", text).group(1)) self.assertGreater(rto - rtt, 150) self.assertLess(rto - rtt, 320) def test_backoff_is_visible_in_ss(self): self.assertRegex(self.samples["during"], r"backoff:\d+") def test_transfer_recovers(self): self.assertIn("recibidos 60000 bytes", LAB.out("b", "cat", "/tmp/server.out")) class RtoFloor(unittest.TestCase): def test_gap_follows_rto_min(self): rows = scenarios.scenario_rtomin() gaps = {} for name, text in rows.items(): rto = float(re.search(r"\brto:([\d.]+)", text).group(1)) rtt = float(re.search(r"\brtt:([\d.]+)", text).group(1)) gaps[name] = rto - rtt self.assertLess(gaps["default"], 320) self.assertGreater(gaps["500ms"], 430) class TailLoss(unittest.TestCase): def test_a_probe_goes_out_with_no_ack_to_trigger_it(self): # Después de la sonda, el dato puede salir por RACK o por timeout: cambia entre corridas. output, before, after = scenarios.scenario_tail() self.assertIn("recibidos 7000 bytes", output) self.assertGreaterEqual(grew(before, after, "TcpExtTCPLossProbes"), 1) segments = capture("tail") probe = next(s for s in segments if s.resent) self.assertTrue(probe.fin) # la sonda repite lo último que se envió: el FIN self.assertEqual(len([s for s in segments if s.dup_ack and s.frame < probe.frame]), 0) class RandomLoss(unittest.TestCase): def test_transfer_survives_and_slows_down(self): output, before, after = scenarios.scenario_loss() self.assertIn("recibidos 3000000 bytes", output) rate = float(re.search(r"\(([\d.]+) Mbit/s\)", output).group(1)) self.assertLess(rate, 12) info = summarize(capture("loss")) self.assertGreater(info["resent"], 5) self.assertGreater(info["dup_acks"], info["resent"]) self.assertGreater(grew(before, after, "TcpRetransSegs"), 5) class Reordering(unittest.TestCase): @classmethod def setUpClass(cls): cls.modern_output, before, after = scenarios.scenario_reorder() cls.modern_resent = grew(before, after, "TcpRetransSegs") cls.info = summarize(capture("reorder")) cls.classic_output, before, after = scenarios.scenario_reorder_classic() cls.classic_resent = grew(before, after, "TcpRetransSegs") def test_all_bytes_arrive(self): self.assertIn("recibidos 3000000 bytes", self.modern_output) self.assertIn("recibidos 3000000 bytes", self.classic_output) def test_duplicate_acks_without_loss(self): self.assertGreater(self.info["dup_acks"], 100) self.assertGreater(self.info["dup_reached"][3], 10) def test_classic_tcp_retransmits_more(self): # Nada se perdió: todo lo retransmitido es trabajo de más, y el TCP clásico hace más. self.assertGreater(self.classic_resent, self.modern_resent) self.assertGreater(self.classic_resent, 5) def test_capture_is_complete(self): self.assertEqual(self.info["new_bytes"], 3000000) class ZeroWindow(unittest.TestCase): @classmethod def setUpClass(cls): cls.output = scenarios.scenario_zero() cls.segments = capture("zero") def test_all_bytes_arrive(self): self.assertIn("recibidos 300000 bytes", self.output) def test_receiver_announces_zero(self): zero = [s for s in from_server(self.segments) if s.zero_window] self.assertGreaterEqual(len(zero), 2) self.assertTrue(all(s.win == 0 for s in zero)) def test_probes_have_no_data_and_old_sequence(self): probes = [s for s in self.segments if s.probe] self.assertGreaterEqual(len(probes), 2) zero = next(s for s in self.segments if s.zero_window) for probe in probes: self.assertEqual(probe.len, 0) self.assertEqual(probe.seq, zero.ack - 1) def test_probe_interval_grows(self): times = [s.time for s in self.segments if s.probe] gaps = [b - a for a, b in zip(times, times[1:])] for shorter, longer in zip(gaps, gaps[1:]): self.assertGreater(longer, 1.4 * shorter) def test_sender_never_passes_the_window(self): limit = None for segment in self.segments: if segment.src != CLIENT and segment.has_ack and segment.win is not None: limit = segment.ack + segment.win elif segment.src == CLIENT and segment.len and limit is not None: self.assertLessEqual(segment.seq + segment.len, limit) def test_transfer_waits_for_the_update(self): last_zero = [s for s in self.segments if s.zero_window][-1] update = next(s for s in from_server(self.segments) if s.frame > last_zero.frame and s.win > 0) self.assertGreater(update.time, 2.5) # el servidor pasó 3 s sin leer self.assertTrue(any(s.frame > update.frame for s in data_of(self.segments))) class WindowLimitsThroughput(unittest.TestCase): @classmethod def setUpClass(cls): cls.results = scenarios.scenario_limit() cls.rates = {size: float(re.search(r"\(([\d.]+) Mbit/s\)", text).group(1)) for size, text in cls.results.items()} def test_more_buffer_more_throughput(self): self.assertLess(self.rates[8192] * 2, self.rates[32768]) self.assertLess(self.rates[32768] * 1.5, self.rates[262144]) def test_rate_is_about_window_over_rtt(self): window = summarize(capture("limit-8192"))["max_window"] predicted = window * 8 / 0.040 / 1e6 self.assertGreater(self.rates[8192], 0.4 * predicted) self.assertLess(self.rates[8192], 1.5 * predicted) def test_large_buffer_is_limited_by_the_link(self): self.assertLess(self.rates[262144], 22) self.assertGreater(self.rates[262144], 8) def test_the_limit_is_not_loss(self): # Puede salir alguna sonda de más, pero la ventana explica la velocidad, no la pérdida. info = summarize(capture("limit-8192")) self.assertLess(info["resent"], 0.01 * info["data_segments"]) class ConnectResults(unittest.TestCase): @classmethod def setUpClass(cls): cls.results = scenarios.scenario_ports() def seconds(self, name): return float(re.search(r"en ([\d.]+) s", self.results[name]).group(1)) def test_closed_port_is_refused_at_once(self): self.assertIn("ECONNREFUSED", self.results["closed"]) self.assertLess(self.seconds("closed"), 1) def test_filtered_port_times_out(self): self.assertIn("ETIMEDOUT", self.results["filtered"]) self.assertGreater(self.seconds("filtered"), 5) def syn_gaps(self, name): times = [s.time for s in read_capture(os.path.join(CAPTURES, f"{name}-a.pcap")) if s.syn] return [b - a for a, b in zip(times, times[1:])] def test_syn_retransmissions_start_linear_in_this_kernel(self): gaps = self.syn_gaps("filtered") self.assertGreaterEqual(len(gaps), 4) for gap in gaps[:4]: self.assertAlmostEqual(gap, 1, delta=0.4) def test_syn_retransmissions_double_without_the_linear_phase(self): gaps = self.syn_gaps("filtered-exp") self.assertEqual(len(gaps), 2) self.assertAlmostEqual(gaps[0], 1, delta=0.4) self.assertAlmostEqual(gaps[1], 2, delta=0.6) self.assertIn("ETIMEDOUT", self.results["filtered_exp"]) def test_rejected_by_the_router(self): self.assertIn("EHOSTUNREACH", self.results["rejected"]) self.assertLess(self.seconds("rejected"), 1) def test_no_route(self): self.assertIn("ENETUNREACH", self.results["no_route"]) def test_open_port(self): self.assertIn("devolvió 0", self.results["open"]) class SynQueue(unittest.TestCase): @classmethod def setUpClass(cls): cls.results = scenarios.scenario_syn() def test_queue_fills_up_to_the_backlog(self): for result in self.results.values(): self.assertGreaterEqual(result["pending"], 10) self.assertLessEqual(result["pending"], 30) def test_without_cookies_a_legitimate_client_cannot_enter(self): result = self.results["off"] self.assertIn("ETIMEDOUT", result["probe"]) self.assertGreater(grew(result["before"], result["after"], "TcpExtTCPReqQFullDrop"), 0) def test_with_cookies_the_same_client_enters(self): result = self.results["on"] self.assertIn("devolvió 0", result["probe"]) self.assertGreater(grew(result["before"], result["after"], "TcpExtSyncookiesSent"), 0) self.assertGreaterEqual(grew(result["before"], result["after"], "TcpExtSyncookiesRecv"), 1) class DelayedAck(unittest.TestCase): def test_ack_waits_when_the_application_is_not_reading(self): scenarios.scenario_delack() segments = capture("delack", "b") pending = {} delays = [] for segment in segments: if segment.src == CLIENT and segment.len: pending[segment.seq + segment.len] = segment.time elif segment.src != CLIENT and segment.ack in pending: delays.append(segment.time - pending[segment.ack]) self.assertGreater(len(delays), 20) self.assertLess(delays[0], 0.01) # al comienzo, Linux responde de inmediato tail = statistics.median(delays[-10:]) self.assertGreater(tail, 0.02) self.assertLess(tail, 0.2) self.assertLess(max(delays), 0.5) # el tope del RFC 9293 class HalfClose(unittest.TestCase): def test_server_answers_after_the_client_fin(self): scenarios.scenario_halfclose() segments = capture("halfclose") client_fin = next(s for s in segments if s.fin and s.src == CLIENT) answer = next(s for s in from_server(segments) if s.len) self.assertGreater(answer.frame, client_fin.frame) self.assertGreater(next(s for s in from_server(segments) if s.fin).frame, answer.frame) class TimeWait(unittest.TestCase): def test_the_last_ack_is_repeated_from_time_wait(self): _, states = scenarios.scenario_timewait() segments = capture("timewait") fins = [s for s in from_server(segments) if s.fin] self.assertEqual(len(fins), 2) # el servidor no recibió el último ACK y repite su FIN last_acks = [s for s in segments if s.src == CLIENT and not s.fin and not s.syn and s.seq == 2] self.assertEqual(len(last_acks), 2) self.assertGreater(last_acks[1].frame, fins[1].frame) port = segments[0].sport match = re.search(rf":{port}\s.*timewait,(\d+)sec", states) self.assertIsNotNone(match, states) self.assertLessEqual(int(match.group(1)), 60) self.assertGreater(int(match.group(1)), 50) def test_the_side_that_closes_first_keeps_the_state(self): without, with_reuse = scenarios.scenario_bindfail() self.assertIn("timewait", without[0]) self.assertIn("Address already in use", without[1]) self.assertIn("timewait", with_reuse[0]) self.assertNotIn("Address already in use", with_reuse[1]) class Resets(unittest.TestCase): def test_abort_sends_rst_and_the_peer_sees_an_error(self): output = scenarios.scenario_abort() self.assertIn("Connection reset by peer", output) segments = capture("abort") self.assertTrue(any(s.rst for s in segments)) self.assertFalse(any(s.fin for s in segments)) def test_closed_port_answers_rst(self): output = scenarios.scenario_closed() self.assertIn("ECONNREFUSED", output) segments = capture("closed") self.assertEqual(len(segments), 2) self.assertTrue(segments[1].rst and segments[1].has_ack) class Statistics(unittest.TestCase): def test_ss_shows_rtt_rto_and_cwnd(self): rows = scenarios.scenario_ss() for text in rows: self.assertRegex(text, r"\brto:\d+") self.assertRegex(text, r"\bcwnd:\d+") rtt = float(re.search(r"\brtt:([\d.]+)", text).group(1)) self.assertGreater(rtt, 38) # nunca menos que el retardo del enlace minimum = float(re.search(r"\bminrtt:([\d.]+)", rows[-1]).group(1)) self.assertLess(minimum, 60) if __name__ == "__main__": unittest.main()