Solo para leer y verificar. No puedes copiar, usar, modificar ni redistribuir este código. Aviso completo en la página principal.
tests/_pcap_builder.py
| 1 | """Builds classic pcap files for the pcap test suite and the capture test suite: Ethernet, Linux cooked v2 and raw-IP frames, IPv4 and |
| 2 | IPv6, TCP conversations whose segments can be split anywhere (a marker split across two packets), a REAL TLS ClientHello (made by Python's |
| 3 | ssl module through memory BIOs, so the SNI parser reads what a browser sends), and TLS application-data records that look like the real thing |
| 4 | (a record header, then random bytes: the analyser only reads headers and searches bytes). Checksums are zero: nothing here verifies them.""" |
| 5 | from __future__ import annotations |
| 6 | |
| 7 | import os |
| 8 | import socket |
| 9 | import ssl |
| 10 | import struct |
| 11 | |
| 12 | ETHERNET, RAW, SLL2 = 1, 101, 276 |
| 13 | SYN, ACK, PSH, FIN = 0x02, 0x10, 0x08, 0x01 |
| 14 | |
| 15 | |
| 16 | def _ip4(src: str, dst: str, payload: bytes) -> bytes: |
| 17 | header = struct.pack(">BBHHHBBH4s4s", 0x45, 0, 20 + len(payload), 0, 0x4000, 64, 6, 0, socket.inet_aton(src), socket.inet_aton(dst)) |
| 18 | return header + payload |
| 19 | |
| 20 | |
| 21 | def _ip6(src: str, dst: str, payload: bytes) -> bytes: |
| 22 | header = struct.pack(">IHBB16s16s", 0x60000000, len(payload), 6, 64, socket.inet_pton(socket.AF_INET6, src), socket.inet_pton(socket.AF_INET6, dst)) |
| 23 | return header + payload |
| 24 | |
| 25 | |
| 26 | def tcp_frame(src: str, sport: int, dst: str, dport: int, seq: int, flags: int, payload: bytes = b"", *, link: int = ETHERNET) -> bytes: |
| 27 | tcp = struct.pack(">HHIIBBHHH", sport, dport, seq & 0xFFFFFFFF, 0, 5 << 4, flags, 65535, 0, 0) + payload |
| 28 | v6 = ":" in src |
| 29 | ip = _ip6(src, dst, tcp) if v6 else _ip4(src, dst, tcp) |
| 30 | kind = 0x86DD if v6 else 0x0800 |
| 31 | if link == ETHERNET: |
| 32 | return b"\x02\x00\x00\x00\x00\x01" + b"\x02\x00\x00\x00\x00\x02" + struct.pack(">H", kind) + ip |
| 33 | if link == SLL2: |
| 34 | return struct.pack(">HHIHBB8s", kind, 0, 1, 1, 0, 6, b"\x02\x00\x00\x00\x00\x01\x00\x00") + ip |
| 35 | return ip |
| 36 | |
| 37 | |
| 38 | def pcap_file(frames: list, link: int = ETHERNET, endian: str = "<", nano: bool = False) -> bytes: |
| 39 | magic = {("<", False): b"\xd4\xc3\xb2\xa1", (">", False): b"\xa1\xb2\xc3\xd4", ("<", True): b"\x4d\x3c\xb2\xa1", (">", True): b"\xa1\xb2\x3c\x4d"}[(endian, nano)] |
| 40 | out = bytearray(magic + struct.pack(endian + "HHIIII", 2, 4, 0, 0, 262144, link)) |
| 41 | for index, frame in enumerate(frames): |
| 42 | out += struct.pack(endian + "IIII", 1700000000 + index, 0, len(frame), len(frame)) + frame |
| 43 | return bytes(out) |
| 44 | |
| 45 | |
| 46 | class Conversation: |
| 47 | """One TCP connection, client -> server, with a handshake and segments cut wherever the test says.""" |
| 48 | |
| 49 | def __init__(self, client=("203.0.113.9", 51000), server=("10.120.1.4", 443), *, link: int = ETHERNET, c_seq: int = 1000, s_seq: int = 9000): |
| 50 | self.client, self.server, self.link = client, server, link |
| 51 | self.c_seq, self.s_seq = c_seq, s_seq |
| 52 | self.frames: list = [] |
| 53 | |
| 54 | def _frame(self, from_client: bool, flags: int, payload: bytes = b"") -> None: |
| 55 | src, dst = (self.client, self.server) if from_client else (self.server, self.client) |
| 56 | seq = self.c_seq if from_client else self.s_seq |
| 57 | self.frames.append(tcp_frame(src[0], src[1], dst[0], dst[1], seq, flags, payload, link=self.link)) |
| 58 | consumed = len(payload) + (1 if flags & (SYN | FIN) else 0) |
| 59 | if from_client: |
| 60 | self.c_seq += consumed |
| 61 | else: |
| 62 | self.s_seq += consumed |
| 63 | |
| 64 | def handshake(self) -> "Conversation": |
| 65 | self._frame(True, SYN) |
| 66 | self._frame(False, SYN | ACK) |
| 67 | self._frame(True, ACK) |
| 68 | return self |
| 69 | |
| 70 | def send(self, data: bytes, *, from_client: bool = True, cuts: tuple = ()) -> "Conversation": |
| 71 | """Sends data in segments cut at the given offsets (default: one segment per 1400 bytes).""" |
| 72 | points = sorted(set(cuts)) or list(range(1400, len(data), 1400)) |
| 73 | start = 0 |
| 74 | for point in [*points, len(data)]: |
| 75 | if point > start: |
| 76 | self._frame(from_client, PSH | ACK, data[start:point]) |
| 77 | start = point |
| 78 | return self |
| 79 | |
| 80 | def reorder(self, first: int, second: int) -> "Conversation": |
| 81 | self.frames[first], self.frames[second] = self.frames[second], self.frames[first] |
| 82 | return self |
| 83 | |
| 84 | def duplicate(self, index: int) -> "Conversation": |
| 85 | self.frames.append(self.frames[index]) |
| 86 | return self |
| 87 | |
| 88 | def pcap(self, **kwargs) -> bytes: |
| 89 | return pcap_file(self.frames, self.link, **kwargs) |
| 90 | |
| 91 | |
| 92 | def client_hello(server_name: str) -> bytes: |
| 93 | """The first flight of a real TLS client (Python's ssl through memory BIOs) for the given SNI name.""" |
| 94 | context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
| 95 | context.check_hostname = False |
| 96 | context.verify_mode = ssl.CERT_NONE |
| 97 | incoming, outgoing = ssl.MemoryBIO(), ssl.MemoryBIO() |
| 98 | tls = context.wrap_bio(incoming, outgoing, server_hostname=server_name) |
| 99 | try: |
| 100 | tls.do_handshake() |
| 101 | except ssl.SSLWantReadError: |
| 102 | pass |
| 103 | return outgoing.read() |
| 104 | |
| 105 | |
| 106 | def app_data_records(total: int, *, avoid: tuple = ()) -> bytes: |
| 107 | """TLS 1.2-style application-data records (type 23) of random bytes, `total` bytes of ciphertext in all.""" |
| 108 | out = bytearray() |
| 109 | remaining = total |
| 110 | while remaining > 0: |
| 111 | size = min(remaining, 1200) |
| 112 | while True: |
| 113 | body = os.urandom(size) |
| 114 | if not any(marker in body for marker in avoid): |
| 115 | break |
| 116 | out += b"\x17\x03\x03" + struct.pack(">H", size) + body |
| 117 | remaining -= size |
| 118 | return bytes(out) |