From 0d23b5e3a252a7a671f1af879dc8ade9be716c8e Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 22:53:39 -0500 Subject: [PATCH] refactor(protocols): finish the shared value helpers and move TLS onto them (#2614) - ProtocolBase: add _read_octets, _read_uint, _read_sint, _read_mac_addr, _read_text, _make_octets, _make_sint, _uint_size and _make_mac_addr; _read_vector reads through them and _make_uint takes alias=. - Ethernet uses the base MAC pair, and its dead PAT_MAC_ADDR goes, as does PCAP-NG's; PCAP-NG and 802.11 keep only the overrides that change behaviour. - protocol-member-order.rst: ARP is the worked example and the reference. - TLS: drop _take, with octets and messages unchanged; every maker and the record header pack through _make_pack, so an oversize value is StructError. - SCTP: chunk flags and invalid-stream reserved use _make_octets, naming the chunk or cause in name=; OSPF.make checks packet_length with _make_uint. TLS/DTLS/SSL, link, helper, #2296 and member-order tests pass. --- .../conventions/protocol-member-order.rst | 30 +- .../pcapkit/protocols/link/ethernet.rst | 2 - .../pcapkit/protocols/link/ieee802_11.rst | 3 +- docs/source/pcapkit/protocols/protocol.rst | 8 +- pcapkit/protocols/application/dtls.py | 13 + pcapkit/protocols/application/ospf.py | 6 +- pcapkit/protocols/application/ssl.py | 187 +++++-------- pcapkit/protocols/application/tls.py | 8 + pcapkit/protocols/link/ethernet.py | 43 +-- pcapkit/protocols/link/ieee802_11.py | 61 ++--- pcapkit/protocols/misc/pcapng.py | 33 +-- pcapkit/protocols/protocol.py | 257 +++++++++++++++++- pcapkit/protocols/transport/sctp.py | 48 +--- .../test_docs_internal_refs_1940_unit.py | 2 +- .../test_protocol_member_order_2490_unit.py | 12 +- .../test_rfc_constants_one_home_1739_unit.py | 18 +- .../test_ospf_dead_pattern_1844_unit.py | 47 ++-- .../test_ospf_ngap_roundtrip_unit.py | 2 +- .../test_tls_extensions_2514_unit.py | 119 ++++++++ ...maker_invalid_format_messages_2296_unit.py | 14 +- .../test_protocol_value_helpers_2514_unit.py | 257 +++++++++++++++++- 21 files changed, 837 insertions(+), 333 deletions(-) diff --git a/docs/source/contributing/conventions/protocol-member-order.rst b/docs/source/contributing/conventions/protocol-member-order.rst index de577a04f8..bc48c2edc9 100644 --- a/docs/source/contributing/conventions/protocol-member-order.rst +++ b/docs/source/contributing/conventions/protocol-member-order.rst @@ -31,7 +31,7 @@ The Order :class:`~pcapkit.protocols.internet.mh.MH`, does not count either: it is a specific reader, ruled on :issue:`2490`. Each helper's ``_read_`` and ``_make_`` methods stay together, with the reader first: - ``_read_mac_addr()`` then ``_make_mac_addr()``. + ``_read_addr_resolve()`` then ``_make_addr_resolve()``. c. **Specific readers and makers.** The option, type and parameter methods, all ``_read_*()`` before all ``_make_*()``. Each set opens with the registry's default, its fallback for a code with no entry of its own, as @@ -56,25 +56,37 @@ before its ``_make_xxx()``, and subgroup b, with a table of known exceptions Example ~~~~~~~ -The utilities of :class:`~pcapkit.protocols.link.ethernet.Ethernet`, reduced to -its member names: +The utilities of :class:`~pcapkit.protocols.link.arp.ARP`, reduced to its member +names. ``read()`` resolves a hardware address before a protocol address, so the +hardware pair comes first: .. code-block:: python - class Ethernet(Link[Data_Ethernet, Schema_Ethernet]): + class ARP(Link[Data_ARP, Schema_ARP]): # Utilities. - def _make_data(cls, data, ...): # a. data first + def _make_data(cls, data): # a. data first ... - def _read_mac_addr(self, ...): # b. helper: reader ... + def _read_addr_resolve(self, addr, htype): # b. first helper: reader ... ... - def _make_mac_addr(self, ...): # ... then its maker + def _make_addr_resolve(self, addr, htype): # ... then its maker + ... + + def _read_proto_resolve(self, addr, ptype): # b. next helper: reader ... + ... + + def _make_proto_resolve(self, addr, ptype): # ... then its maker ... The Reference ~~~~~~~~~~~~~ -:class:`~pcapkit.protocols.link.ethernet.Ethernet` is the reference for the -utilities order. +:class:`~pcapkit.protocols.link.arp.ARP` is the reference for the utilities +order. Its utilities are ``_make_data()`` and the helpers that ``read()`` and +``make()`` name, and nothing else, and there is more than one helper, so the +first-use order of subgroup b shows as well as each reader preceding its maker. +Chosen on :issue:`2614`, when the MAC address pair of +:class:`~pcapkit.protocols.link.ethernet.Ethernet`, the reference before it, +moved to the base class. diff --git a/docs/source/pcapkit/protocols/link/ethernet.rst b/docs/source/pcapkit/protocols/link/ethernet.rst index ca677eb284..47a2cfec88 100644 --- a/docs/source/pcapkit/protocols/link/ethernet.rst +++ b/docs/source/pcapkit/protocols/link/ethernet.rst @@ -38,8 +38,6 @@ below: .. automethod:: __index__ - .. automethod:: _make_mac_addr - Header Schemas -------------- diff --git a/docs/source/pcapkit/protocols/link/ieee802_11.rst b/docs/source/pcapkit/protocols/link/ieee802_11.rst index cc9fb573e0..5ac7d83ff4 100644 --- a/docs/source/pcapkit/protocols/link/ieee802_11.rst +++ b/docs/source/pcapkit/protocols/link/ieee802_11.rst @@ -70,14 +70,13 @@ Scope .. automethod:: _make_data .. automethod:: _read_frame_control .. automethod:: _make_frame_control - .. automethod:: _read_mac_addr - .. automethod:: _make_mac_addr .. automethod:: _read_seq_ctrl .. automethod:: _make_seq_ctrl .. automethod:: _read_qos_control .. automethod:: _make_qos_control .. automethod:: _read_ht_control .. automethod:: _make_ht_control + .. automethod:: _make_mac_addr .. autodata:: pcapkit.protocols.link.ieee802_11.SUBTYPE :no-value: diff --git a/docs/source/pcapkit/protocols/protocol.rst b/docs/source/pcapkit/protocols/protocol.rst index cacbe1cd29..31624f7dca 100644 --- a/docs/source/pcapkit/protocols/protocol.rst +++ b/docs/source/pcapkit/protocols/protocol.rst @@ -83,6 +83,9 @@ Internal Definitions below. .. automethod:: _make_payload .. automethod:: _packing_target + .. automethod:: _read_mac_addr + .. automethod:: _make_mac_addr + .. automethod:: _lookup_registry .. automethod:: _lookup_next_layer .. automethod:: _decode_next_layer @@ -198,6 +201,8 @@ Internal Definitions _make_index _make_payload _packing_target + _read_mac_addr + _make_mac_addr _lookup_registry _lookup_next_layer _decode_next_layer @@ -223,7 +228,8 @@ Internal Definitions :meth:`~Protocol.expand_comp`, :meth:`~Protocol._read_packet`, :meth:`~Protocol._get_payload`, :meth:`~Protocol._make_data`, :meth:`~Protocol._make_index`, :meth:`~Protocol._make_payload`, - :meth:`~Protocol._packing_target`, + :meth:`~Protocol._packing_target`, :meth:`~Protocol._read_mac_addr`, + :meth:`~Protocol._make_mac_addr`, :meth:`~Protocol._lookup_registry`, :meth:`~Protocol._lookup_next_layer`, :meth:`~Protocol._decode_next_layer`, :meth:`~Protocol._import_next_layer`, diff --git a/pcapkit/protocols/application/dtls.py b/pcapkit/protocols/application/dtls.py index 021e0af242..023c74e254 100644 --- a/pcapkit/protocols/application/dtls.py +++ b/pcapkit/protocols/application/dtls.py @@ -499,12 +499,20 @@ def make(self, Raises: ProtocolError: If ``cid`` is missing or invalid for a ``tls12_cid`` record, or given for any other. + StructError: If a record-header field, or a field of ``content``, + does not fit its width. + FieldValueError: If a unified-header record's ``epoch`` does not + fit its two bits. """ self._learn_payload_connection_ids(payload) if ciphertext: if cid is not None: raise ProtocolError(f'{self.alias}: invalid connection ID: only a tls12_cid record carries one') + # NOTE: packed first for the reason below; ``epoch`` is a bit sub-field the schema refuses itself. + self._make_pack(sequence_number, size=2 if sequence_long else 1) + if length_present: + self._make_pack(len(content), size=2) return Schema_DTLSCiphertext( flags={ 'fixed': _UNIFIED_PREFIX, @@ -524,6 +532,11 @@ def make(self, reversed=content_type_reversed, pack=False)) fragment = data if data is not None else self._make_tls_content(content_type_, content, dtls=True) + # NOTE: the schema masks a field to its width, so each is packed here first, to + # refuse a wider value with the StructError the content makers raise. + for value, size in ((int(content_type_), 1), (version, 2), (epoch, 2), (len(fragment), 2)): + self._make_pack(value, size=size) + self._make_pack(sequence_number, size=6) if int(content_type_) == Enum_ContentType.tls12_cid: if cid is None or not _is_cid(cid): raise ProtocolError(f'{self.alias}: invalid connection ID: {cid!r} ' diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index f5fa473afc..77648f96c8 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -326,9 +326,9 @@ def make(self, if packet_length is None: packet_length = 24 + len(payload) - elif (not isinstance(packet_length, int) or isinstance(packet_length, bool) - or not 0 <= packet_length <= 0xFFFF): - raise ProtocolError(f'OSPF: invalid packet length: {packet_length!r}') + else: + # NOTE: ``alias`` reads ``_version``, which only ``read()`` sets. + packet_length = self._make_uint(packet_length, bits=16, name='packet length', alias='OSPF') schema = Schema_OSPF( version=version, diff --git a/pcapkit/protocols/application/ssl.py b/pcapkit/protocols/application/ssl.py index 06eb989a64..bf0746d66d 100644 --- a/pcapkit/protocols/application/ssl.py +++ b/pcapkit/protocols/application/ssl.py @@ -158,32 +158,6 @@ def _negotiated_versions(message: 'Any') -> 'tuple[int, ...]': return () -def _take(data: 'bytes', pos: 'int', size: 'int', *, what: 'str', alias: 'str') -> 'bytes': - """Slice ``size`` octets out of ``data`` at ``pos``, refusing a short read. - - Plain slicing silently returns fewer octets, which would let a truncated - message parse short rather than fail. - - Args: - data: Buffer being read. - pos: Offset to start at. - size: Number of octets to take. - what: Name of the field being read, for the error message. - alias: Protocol alias, for the error message. - - Returns: - The requested slice. - - Raises: - ProtocolError: If fewer than ``size`` octets remain from ``pos``. - - """ - chunk = data[pos:pos + size] - if len(chunk) != size: - raise ProtocolError(f'{alias}: invalid format: {what} runs past the end of the buffer') - return chunk - - def _read_key_share_entry(proto: 'SSL', data: 'bytes', pos: 'int') -> 'tuple[Data_KeyShareEntry, int]': """Read one ``KeyShareEntry`` (:rfc:`9846#section-4.3.8`) at ``pos``. @@ -199,7 +173,7 @@ def _read_key_share_entry(proto: 'SSL', data: 'bytes', pos: 'int') -> 'tuple[Dat ProtocolError: If the entry runs past the end of ``data``. """ - group = int.from_bytes(_take(data, pos, 2, what='key_share.group', alias=proto.alias), 'big') + group = proto._read_uint(data, pos, size=2, what='key_share.group') # pylint: disable=protected-access key_exchange, end = proto._read_vector(data, pos + 2, size=2, # pylint: disable=protected-access what='key_share.key_exchange') return Data_KeyShareEntry(group=group, key_exchange=key_exchange), end @@ -207,8 +181,9 @@ def _read_key_share_entry(proto: 'SSL', data: 'bytes', pos: 'int') -> 'tuple[Dat def _make_key_share_entry(proto: 'SSL', entry: 'Data_KeyShareEntry') -> 'bytes': """Write one ``KeyShareEntry`` -- the inverse of :func:`_read_key_share_entry`.""" + group = proto._make_pack(int(entry.group), size=2) # pylint: disable=protected-access key_exchange = proto._make_vector(entry.key_exchange, size=2) # pylint: disable=protected-access - return int(entry.group).to_bytes(2, 'big') + key_exchange + return group + key_exchange def _hello_type(random: 'bytes') -> 'Enum_HandshakeType': @@ -420,7 +395,7 @@ def _read_ext_server_name(self, ext_type: 'Enum_ExtensionType', data: 'bytes', * names = [] # type: list[Data_ServerName] pos = 0 while pos < len(body): - name_type = _take(body, pos, 1, what='server_name.name_type', alias=alias)[0] + name_type = self._read_uint(body, pos, size=1, what='server_name.name_type') if name_type != _HOST_NAME: raise ProtocolError(f'{alias}: invalid format: server_name.name_type {name_type} ' 'is not host_name') @@ -558,7 +533,8 @@ def _read_ext_supported_versions(self, ext_type: 'Enum_ExtensionType', data: 'by if len(data) != 2: raise ProtocolError(f'{alias}: invalid format: supported_versions.selected_version ' 'must be exactly two octets') - return Data_SupportedVersionsServerHello(selected_version=int.from_bytes(data, 'big')) + return Data_SupportedVersionsServerHello(selected_version=self._read_uint( + data, size=2, what='supported_versions.selected_version')) return data def _read_ext_key_share(self, ext_type: 'Enum_ExtensionType', data: 'bytes', *, # pylint: disable=unused-argument @@ -612,7 +588,8 @@ def _read_ext_key_share(self, ext_type: 'Enum_ExtensionType', data: 'bytes', *, if len(data) != 2: raise ProtocolError(f'{alias}: invalid format: key_share.selected_group ' 'must be exactly two octets') - return Data_KeyShareHelloRetryRequest(selected_group=int.from_bytes(data, 'big')) + return Data_KeyShareHelloRetryRequest(selected_group=self._read_uint( + data, size=2, what='key_share.selected_group')) if msg_type == Enum_HandshakeType.server_hello: entry, end = _read_key_share_entry(self, data, 0) if end != len(data): @@ -689,8 +666,8 @@ def _make_ext_server_name(self, ext_type: 'Enum_ExtensionType', data: 'Optional[ """ names = data.server_name_list if data is not None else server_name_list - return self._make_vector(b''.join(bytes([int(name.name_type)]) + self._make_vector(name.name, size=2) - for name in names), size=2) + return self._make_vector(b''.join(self._make_pack(int(name.name_type), size=1) + + self._make_vector(name.name, size=2) for name in names), size=2) def _make_ext_supported_groups(self, ext_type: 'Enum_ExtensionType', data: 'Optional[Data_NamedGroupList]' = None, *, # pylint: disable=unused-argument,line-too-long msg_type: 'Enum_HandshakeType', # pylint: disable=unused-argument @@ -788,7 +765,7 @@ def _make_ext_supported_versions(self, ext_type: 'Enum_ExtensionType', # pylint if isinstance(data, Data_SupportedVersionsClientHello): return self._make_vector(self._make_uints(data.versions, size=2), size=1) if isinstance(data, Data_SupportedVersionsServerHello): - return int(data.selected_version).to_bytes(2, 'big') + return self._make_pack(int(data.selected_version), size=2) raise ProtocolError(f'{self.alias}: [Extension {ext_type}] no supported_versions structure ' f'for {msg_type!r}') @@ -832,7 +809,7 @@ def _make_ext_key_share(self, ext_type: 'Enum_ExtensionType', # pylint: disable return self._make_vector(b''.join(_make_key_share_entry(self, entry) for entry in data.client_shares), size=2) if isinstance(data, Data_KeyShareHelloRetryRequest): - return int(data.selected_group).to_bytes(2, 'big') + return self._make_pack(int(data.selected_group), size=2) if isinstance(data, Data_KeyShareServerHello): return _make_key_share_entry(self, data.server_share) raise ProtocolError(f'{self.alias}: [Extension {ext_type}] no key_share structure for {msg_type!r}') @@ -1036,7 +1013,7 @@ def _read_tls_content(self, content_type: 'Enum_ContentType', fragment: 'bytes', return fragment if len(fragment) != 2: raise ProtocolError(f'{self.alias}: invalid format: alert record must be exactly two octets') - level, description = _take(fragment, 0, 2, what='alert', alias=self.alias) + level, description = self._read_octets(fragment, size=2, what='alert') return Data_Alert(level=level, description=description) if content_type == Enum_ContentType.change_cipher_spec: if not self._tls_content_is_cleartext(content_type, fragment, dtls=dtls, epoch=epoch): @@ -1083,9 +1060,9 @@ def _read_tls_handshake_messages(self, fragment: 'bytes', *, pos = 0 total = len(fragment) while pos < total: - header = _take(fragment, pos, 4, what='handshake header', alias=self.alias) - msg_type = Enum_HandshakeType.get(header[0]) - length = int.from_bytes(header[1:4], 'big') + code = self._read_uint(fragment, pos, size=1, what='handshake header') + length = self._read_uint(fragment, pos + 1, size=3, what='handshake header') + msg_type = Enum_HandshakeType.get(code) pos += 4 is_fragmented = False @@ -1093,17 +1070,17 @@ def _read_tls_handshake_messages(self, fragment: 'bytes', *, # :class:`DTLSHandshake` declares these as plain ``int``. message_seq = fragment_offset = fragment_length = 0 if dtls: - frag_header = _take(fragment, pos, 8, what='DTLS fragment header', alias=self.alias) + what = 'DTLS fragment header' + message_seq = self._read_uint(fragment, pos, size=2, what=what) + fragment_offset = self._read_uint(fragment, pos + 2, size=3, what=what) + fragment_length = self._read_uint(fragment, pos + 5, size=3, what=what) pos += 8 - message_seq = int.from_bytes(frag_header[0:2], 'big') - fragment_offset = int.from_bytes(frag_header[2:5], 'big') - fragment_length = int.from_bytes(frag_header[5:8], 'big') is_fragmented = bool(fragment_offset) or fragment_length != length body_length = fragment_length else: body_length = length - body_bytes = _take(fragment, pos, body_length, what='handshake body', alias=self.alias) + body_bytes = self._read_octets(fragment, pos, size=body_length, what='handshake body') pos += body_length if not is_fragmented and msg_type == Enum_HandshakeType.client_hello: @@ -1183,11 +1160,11 @@ def _make_tls_handshake_messages(self, messages: 'tuple[Any, ...]', *, else: fragment_length = len(body_bytes) - out += bytes([int(message.msg_type)]) + length.to_bytes(3, 'big') + out += self._make_pack(int(message.msg_type), size=1) + self._make_pack(length, size=3) if dtls: - out += (message.message_seq.to_bytes(2, 'big') - + message.fragment_offset.to_bytes(3, 'big') - + fragment_length.to_bytes(3, 'big')) + out += (self._make_pack(message.message_seq, size=2) + + self._make_pack(message.fragment_offset, size=3) + + self._make_pack(fragment_length, size=3)) out += body_bytes return bytes(out) @@ -1240,50 +1217,31 @@ def _read_tls_client_hello(self, body: 'bytes', *, dtls: 'bool') -> 'Data_Client """ alias = self.alias - version = int.from_bytes(_take(body, 0, 2, what='client_hello.version', alias=alias), 'big') - random = _take(body, 2, 32, what='client_hello.random', alias=alias) - pos = 34 - - session_id_len = _take(body, pos, 1, what='client_hello.session_id length', alias=alias)[0] - pos += 1 - session_id = _take(body, pos, session_id_len, what='client_hello.session_id', alias=alias) - pos += session_id_len + version = self._read_uint(body, size=2, what='client_hello.version') + random = self._read_octets(body, 2, size=32, what='client_hello.random') + session_id, pos = self._read_vector(body, 34, size=1, what='client_hello.session_id') cookie = None # type: Optional[bytes] if dtls: - cookie_len = _take(body, pos, 1, what='client_hello.cookie length', alias=alias)[0] - pos += 1 - cookie = _take(body, pos, cookie_len, what='client_hello.cookie', alias=alias) - pos += cookie_len + cookie, pos = self._read_vector(body, pos, size=1, what='client_hello.cookie') - cipher_suites_len = int.from_bytes( - _take(body, pos, 2, what='client_hello.cipher_suites length', alias=alias), 'big') + what = 'client_hello.cipher_suites' + cipher_suites_len = self._read_uint(body, pos, size=2, what=f'{what} length') pos += 2 - # NOTE: a ``CipherSuite`` is 2 octets (:rfc:`9846#section-4.2.2`): an odd length - # would decode a lone octet and re-pad it on rebuild, so it is refused. + # NOTE: a ``CipherSuite`` is 2 octets (:rfc:`9846#section-4.2.2`), so an odd length is + # refused before the short read is checked, rather than decoded and re-padded on rebuild. if cipher_suites_len % 2: - raise ProtocolError( - f'{alias}: invalid format: client_hello.cipher_suites length ' - f'{cipher_suites_len} is not a multiple of 2') - cipher_suites_raw = _take(body, pos, cipher_suites_len, what='client_hello.cipher_suites', alias=alias) + raise ProtocolError(f'{alias}: invalid format: {what} length {cipher_suites_len} is not a multiple of 2') + cipher_suites = self._read_uints(self._read_octets(body, pos, size=cipher_suites_len, what=what), + size=2, what=what) pos += cipher_suites_len - cipher_suites = tuple( - int.from_bytes(cipher_suites_raw[i:i + 2], 'big') for i in range(0, cipher_suites_len, 2) - ) - compression_len = _take(body, pos, 1, what='client_hello.compression_methods length', alias=alias)[0] - pos += 1 - compression_methods = tuple( - _take(body, pos, compression_len, what='client_hello.compression_methods', alias=alias)) - pos += compression_len + compression_methods_raw, pos = self._read_vector(body, pos, size=1, what='client_hello.compression_methods') + compression_methods = tuple(compression_methods_raw) extensions = None # type: Optional[tuple[Data_Extension, ...]] if pos < len(body): - ext_total_len = int.from_bytes( - _take(body, pos, 2, what='client_hello.extensions length', alias=alias), 'big') - pos += 2 - extensions_raw = _take(body, pos, ext_total_len, what='client_hello.extensions', alias=alias) - pos += ext_total_len + extensions_raw, pos = self._read_vector(body, pos, size=2, what='client_hello.extensions') extensions = self._read_tls_extensions( extensions_raw, msg_type=cast('Enum_HandshakeType', Enum_HandshakeType.client_hello)) @@ -1323,32 +1281,21 @@ def _read_tls_server_hello(self, body: 'bytes') -> 'Data_ServerHello': ProtocolError: If ``body`` is malformed. """ - alias = self.alias - version = int.from_bytes(_take(body, 0, 2, what='server_hello.version', alias=alias), 'big') - random = _take(body, 2, 32, what='server_hello.random', alias=alias) - pos = 34 - - session_id_len = _take(body, pos, 1, what='server_hello.session_id length', alias=alias)[0] - pos += 1 - session_id = _take(body, pos, session_id_len, what='server_hello.session_id', alias=alias) - pos += session_id_len + version = self._read_uint(body, size=2, what='server_hello.version') + random = self._read_octets(body, 2, size=32, what='server_hello.random') + session_id, pos = self._read_vector(body, 34, size=1, what='server_hello.session_id') - cipher_suite = int.from_bytes(_take(body, pos, 2, what='server_hello.cipher_suite', alias=alias), 'big') - pos += 2 - compression_method = _take(body, pos, 1, what='server_hello.compression_method', alias=alias)[0] - pos += 1 + cipher_suite = self._read_uint(body, pos, size=2, what='server_hello.cipher_suite') + compression_method = self._read_uint(body, pos + 2, size=1, what='server_hello.compression_method') + pos += 3 extensions = None # type: Optional[tuple[Data_Extension, ...]] if pos < len(body): - ext_total_len = int.from_bytes( - _take(body, pos, 2, what='server_hello.extensions length', alias=alias), 'big') - pos += 2 - extensions_raw = _take(body, pos, ext_total_len, what='server_hello.extensions', alias=alias) - pos += ext_total_len + extensions_raw, pos = self._read_vector(body, pos, size=2, what='server_hello.extensions') extensions = self._read_tls_extensions(extensions_raw, msg_type=_hello_type(random)) if pos != len(body): - raise ProtocolError(f'{alias}: invalid format: server_hello carries excess octets') + raise ProtocolError(f'{self.alias}: invalid format: server_hello carries excess octets') return Data_ServerHello( version=version, random=random, session_id=session_id, cipher_suite=cipher_suite, @@ -1418,11 +1365,11 @@ def _read_tls_extensions(self, data: 'bytes', *, msg_type: 'Enum_HandshakeType') pos = 0 total = len(data) while pos < total: - header = _take(data, pos, 4, what='extension header', alias=self.alias) - ext_type = Enum_ExtensionType.get(int.from_bytes(header[0:2], 'big')) - ext_len = int.from_bytes(header[2:4], 'big') + code = self._read_uint(data, pos, size=2, what='extension header') + ext_len = self._read_uint(data, pos + 2, size=2, what='extension header') + ext_type = Enum_ExtensionType.get(code) pos += 4 - ext_data = _take(data, pos, ext_len, what='extension data', alias=self.alias) + ext_data = self._read_octets(data, pos, size=ext_len, what='extension data') pos += ext_len name = self._lookup_registry(self.__extension__, ext_type) @@ -1459,9 +1406,9 @@ def _make_tls_content(self, content_type: 'Enum_ContentType', content: 'Any', *, if content_type == Enum_ContentType.handshake: return self._make_tls_handshake_messages(content, dtls=dtls) if content_type == Enum_ContentType.alert: - return bytes([int(content.level), int(content.description)]) + return self._make_uints((content.level, content.description), size=1) if content_type == Enum_ContentType.change_cipher_spec: - return bytes([int(content.msg)]) + return self._make_pack(int(content.msg), size=1) return bytes(content) def _make_tls_client_hello(self, data: 'Data_ClientHello', *, dtls: 'bool') -> 'bytes': @@ -1478,24 +1425,20 @@ def _make_tls_client_hello(self, data: 'Data_ClientHello', *, dtls: 'bool') -> ' """ out = bytearray() - out += int(data.version).to_bytes(2, 'big') + out += self._make_pack(int(data.version), size=2) out += data.random - out += bytes([len(data.session_id)]) + data.session_id + out += self._make_vector(data.session_id, size=1) if dtls: - cookie = data.cookie if data.cookie is not None else b'' - out += bytes([len(cookie)]) + cookie - - cipher_suites = b''.join(int(suite).to_bytes(2, 'big') for suite in data.cipher_suites) - out += len(cipher_suites).to_bytes(2, 'big') + cipher_suites + out += self._make_vector(data.cookie if data.cookie is not None else b'', size=1) - compression_methods = bytes(int(method) for method in data.compression_methods) - out += bytes([len(compression_methods)]) + compression_methods + out += self._make_vector(self._make_uints(data.cipher_suites, size=2), size=2) + out += self._make_vector(self._make_uints(data.compression_methods, size=1), size=1) if data.extensions is not None: extensions_raw = self._make_tls_extensions( data.extensions, msg_type=cast('Enum_HandshakeType', Enum_HandshakeType.client_hello)) - out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw + out += self._make_vector(extensions_raw, size=2) return bytes(out) def _make_tls_server_hello(self, data: 'Data_ServerHello') -> 'bytes': @@ -1510,15 +1453,15 @@ def _make_tls_server_hello(self, data: 'Data_ServerHello') -> 'bytes': """ out = bytearray() - out += int(data.version).to_bytes(2, 'big') + out += self._make_pack(int(data.version), size=2) out += data.random - out += bytes([len(data.session_id)]) + data.session_id - out += int(data.cipher_suite).to_bytes(2, 'big') - out += bytes([int(data.compression_method)]) + out += self._make_vector(data.session_id, size=1) + out += self._make_pack(int(data.cipher_suite), size=2) + out += self._make_pack(int(data.compression_method), size=1) if data.extensions is not None: extensions_raw = self._make_tls_extensions(data.extensions, msg_type=_hello_type(data.random)) - out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw + out += self._make_vector(extensions_raw, size=2) return bytes(out) def _make_tls_encrypted_extensions(self, data: 'Data_EncryptedExtensions') -> 'bytes': @@ -1571,5 +1514,5 @@ def _make_tls_extensions(self, extensions: 'Iterable[Data_Extension | tuple[Enum meth = name[1] ext_data = meth(ext_type, data, msg_type=msg_type, **kwargs) - out += int(ext_type).to_bytes(2, 'big') + self._make_vector(ext_data, size=2) + out += self._make_pack(int(ext_type), size=2) + self._make_vector(ext_data, size=2) return bytes(out) diff --git a/pcapkit/protocols/application/tls.py b/pcapkit/protocols/application/tls.py index b9a655399b..78c9789a46 100644 --- a/pcapkit/protocols/application/tls.py +++ b/pcapkit/protocols/application/tls.py @@ -260,12 +260,20 @@ def make(self, Returns: Constructed packet data. + Raises: + StructError: If a record-header field, or a field of ``content``, + does not fit its width. + """ content_type_ = cast('Enum_ContentType', self._make_index( content_type, content_type_default, namespace=content_type_namespace, reversed=content_type_reversed, pack=False)) fragment = data if data is not None else self._make_tls_content(content_type_, content, dtls=False) + # NOTE: the schema masks a field to its width, so each is packed here first, to + # refuse a wider value with the StructError the content makers raise. + for value, size in ((int(content_type_), 1), (version, 2), (len(fragment), 2)): + self._make_pack(value, size=size) return Schema_TLS( content_type=content_type_, version=version, diff --git a/pcapkit/protocols/link/ethernet.py b/pcapkit/protocols/link/ethernet.py index 6e0b9352fa..d828ee7ae1 100644 --- a/pcapkit/protocols/link/ethernet.py +++ b/pcapkit/protocols/link/ethernet.py @@ -25,7 +25,6 @@ .. [*] https://en.wikipedia.org/wiki/Ethernet """ -import re from typing import TYPE_CHECKING from pcapkit.const.reg.ethertype import EtherType as Enum_EtherType @@ -33,7 +32,6 @@ from pcapkit.protocols.data.link.ethernet import Ethernet as Data_Ethernet from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.ethernet import Ethernet as Schema_Ethernet -from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -47,9 +45,6 @@ __all__ = ['Ethernet'] -# Ethernet address pattern -PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}') - class Ethernet(Link[Data_Ethernet, Schema_Ethernet], schema=Schema_Ethernet, data=Data_Ethernet): @@ -158,7 +153,7 @@ def make(self, Raises: ProtocolError: If ``dst`` or ``src`` is rejected by - :meth:`_make_mac_addr`. + :meth:`~pcapkit.protocols.protocol.Protocol._make_mac_addr`. """ _type = self._make_index(type, type_default, namespace=type_namespace, @@ -213,39 +208,3 @@ def _make_data(cls, data: 'Data_Ethernet') -> 'dict[str, Any]': # type: ignore[ 'type': data.type, 'payload': cls._make_payload(data), } - - def _read_mac_addr(self, addr: 'bytes') -> 'str': - """Read MAC address. - - Args: - addr: MAC address. - - Returns: - Colon (``:``) separated *hex* encoded MAC address. - - """ - return addr.hex(':') - - def _make_mac_addr(self, addr: 'str | bytes | bytearray') -> 'bytes': - """Make MAC address. - - Args: - addr: MAC address. - - Returns: - MAC address packed as 6 octets. - - Raises: - ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes` - or :class:`bytearray` (a :obj:`bool` included), or is not a - ``:``- or ``-``-separated *hex* encoded MAC address. - - """ - if not isinstance(addr, (str, bytes, bytearray)): - raise ProtocolError(f'{self.alias}: invalid MAC address: must be a str, bytes or ' - f'bytearray, not {addr!r}') - _addr = addr.encode() if isinstance(addr, str) else addr - - if PAT_MAC_ADDR.fullmatch(_addr) is not None: - return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode()) - raise ProtocolError(f'{self.alias}: invalid MAC address: {addr!r}') diff --git a/pcapkit/protocols/link/ieee802_11.py b/pcapkit/protocols/link/ieee802_11.py index 052d1c89f6..d2e297d97f 100644 --- a/pcapkit/protocols/link/ieee802_11.py +++ b/pcapkit/protocols/link/ieee802_11.py @@ -69,7 +69,6 @@ from pcapkit.protocols.data.link.ieee802_11 import HTControl as Data_HTControl from pcapkit.protocols.data.link.ieee802_11 import QoSControl as Data_QoSControl from pcapkit.protocols.data.link.ieee802_11 import SequenceControl as Data_SequenceControl -from pcapkit.protocols.link.ethernet import PAT_MAC_ADDR from pcapkit.protocols.link.link import Link from pcapkit.protocols.schema.link.ieee802_11 import IEEE802_11 as Schema_IEEE802_11 from pcapkit.protocols.schema.link.ieee802_11 import Layout, layout @@ -499,43 +498,6 @@ def _make_frame_control(self, *, version: 'int', type: 'int', subtype: 'int', # return version | type << 2 | subtype << 4 | low << 8 | high << 12 - def _read_mac_addr(self, addr: 'bytes') -> 'str': - """Read MAC address. - - Args: - addr: MAC address. - - Returns: - Colon (``:``) separated *hex* encoded MAC address. - - """ - return addr.hex(':') - - def _make_mac_addr(self, addr: 'Optional[str | bytes | bytearray]') -> 'bytes': - """Make MAC address. - - Args: - addr: MAC address; :obj:`None` makes all zeros. - - Returns: - MAC address packed as 6 octets. - - Raises: - ProtocolError: If ``addr`` is not a ``:``- or ``-``-separated *hex* - encoded MAC address. - - """ - if addr is None: - return bytes(6) - if not isinstance(addr, (str, bytes, bytearray)): - raise ProtocolError(f'{self.alias}: invalid MAC address: must be a str, bytes or ' - f'bytearray, not {addr!r}') - _addr = addr.encode() if isinstance(addr, str) else bytes(addr) - - if PAT_MAC_ADDR.fullmatch(_addr) is not None: - return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode()) - raise ProtocolError(f'{self.alias}: invalid MAC address: {addr!r}') - def _read_seq_ctrl(self, value: 'int') -> 'Data_SequenceControl': """Read the Sequence Control field (IEEE Std 802.11-2020 §9.2.4.4). @@ -643,3 +605,26 @@ def _make_ht_control(self, value: 'Optional[int]') -> 'int': if value is None: return 0 return self._make_uint(value, bits=32, name='ht_control') + + def _make_mac_addr(self, addr: 'Optional[str | bytes | bytearray]', *, size: 'Optional[int]' = 6, + name: 'str' = 'MAC address', alias: 'Optional[str]' = None) -> 'bytes': + """Make MAC address. + + Args: + addr: MAC address; :obj:`None` makes all zeros. + size: Octets in the address, as for :meth:`Protocol._make_mac_addr + `. + name: Name of the address, for the error message. + alias: Prefix of the error message, :attr:`alias` by default. + + Returns: + MAC address packed as ``size`` octets, 6 by default. + + Raises: + ProtocolError: If ``addr`` is not a ``:``- or ``-``-separated *hex* + encoded MAC address. + + """ + if addr is None: + return bytes(size or 0) + return super()._make_mac_addr(addr, size=size, name=name, alias=alias) diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index c277a91458..7599176d3d 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -260,8 +260,6 @@ _FlagsEnum = TypeVar('_FlagsEnum', 'PacketDirection', 'PacketReception') _S = TypeVar('_S', bound='Schema | bytes') -# Ethernet address pattern -PAT_MAC_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){5}[0-9a-f]{2}') # EUI address pattern PAT_EUI_ADDR = re.compile(rb'(?i)(?:[0-9a-f]{2}[:-]){7}[0-9a-f]{2}') @@ -2335,18 +2333,6 @@ def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]' ) return (ts_info >> 32) & 0xFFFF_FFFF, ts_info & 0xFFFF_FFFF - def _read_mac_addr(self, addr: 'bytes') -> 'str': - """Read MAC address. - - Args: - addr: MAC address. - - Returns: - Colon (``:``) separated *hex* encoded MAC address. - - """ - return addr.hex(':') - def _read_eui_addr(self, addr: 'bytes') -> 'str': """Read EUI address. @@ -2359,14 +2345,19 @@ def _read_eui_addr(self, addr: 'bytes') -> 'str': """ return addr.hex(':') - def _make_mac_addr(self, addr: 'str | bytes | bytearray') -> 'bytes': + def _make_mac_addr(self, addr: 'str | bytes | bytearray', *, size: 'Optional[int]' = 6, + name: 'str' = 'MAC address', alias: 'Optional[str]' = None) -> 'bytes': """Make MAC address. Args: addr: MAC address. + size: Octets in the address, as for :meth:`Protocol._make_mac_addr + `. + name: Name of the address, for the error message. + alias: Prefix of the error message, ``'PCAP-NG'`` by default. Returns: - MAC address packed as 6 octets. + MAC address packed as ``size`` octets, 6 by default. Raises: ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes` @@ -2374,14 +2365,8 @@ def _make_mac_addr(self, addr: 'str | bytes | bytearray') -> 'bytes': ``:``- or ``-``-separated *hex* encoded MAC address. """ - if not isinstance(addr, (str, bytes, bytearray)): - raise ProtocolError(f'PCAP-NG: invalid MAC address: must be a str, bytes or ' - f'bytearray, not {addr!r}') - _addr = addr.encode() if isinstance(addr, str) else addr - - if PAT_MAC_ADDR.fullmatch(_addr) is not None: - return bytes.fromhex(_addr.replace(b':', b'').replace(b'-', b'').decode()) - raise ProtocolError(f'PCAP-NG: invalid MAC address: {addr!r}') + prefix = 'PCAP-NG' if alias is None else alias + return super()._make_mac_addr(addr, size=size, name=name, alias=prefix) def _make_eui_addr(self, addr: 'str | bytes | bytearray') -> 'bytes': """Make EUI address. diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 7771185542..40a550161f 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -1870,6 +1870,81 @@ def _read_binary(self, size: 'int' = 1) -> 'str': raise StructError(f'{self.__class__.__name__}: unpack failed') return ''.join(bin(byte)[2:].zfill(8) for byte in mem) + def _read_octets(self, data: 'bytes', pos: 'int' = 0, *, size: 'int', what: 'str') -> 'bytes': + """Read exactly ``size`` octets out of ``data`` at ``pos``. + + A plain slice comes back short past the end of ``data``, which would + let a truncated field parse rather than fail. + + Args: + data: buffer being read + pos: offset of the octets in ``data`` + size: octets to read + what: name of the field, for the error message + + Returns: + The ``size`` octets at ``pos``. + + Raises: + ProtocolError: If fewer than ``size`` octets remain from ``pos``. + The message names :attr:`alias`. + + .. versionadded:: 1.6.0 + + """ + octets = data[pos:pos + size] + if len(octets) != size: + raise ProtocolError(f'{self.alias}: invalid format: {what} runs past the end of the buffer') + return octets + + def _read_uint(self, data: 'bytes', pos: 'int' = 0, *, size: 'int', what: 'str', + lilendian: 'bool' = False) -> 'int': + """Read a ``size``-octet unsigned integer out of ``data`` at ``pos``. + + :meth:`_read_unpack` reads the file; this reads a buffer already in + hand. The inverse is :meth:`_make_pack`, not :meth:`_make_uint`, which + checks a value and packs nothing. + + Args: + data: buffer being read + pos: offset of the integer in ``data`` + size: octets in the integer + what: name of the field, for the error message + lilendian: little-endian flag + + Returns: + The integer. + + Raises: + ProtocolError: If fewer than ``size`` octets remain from ``pos``, + as :meth:`_read_octets` raises it. + + .. versionadded:: 1.6.0 + + """ + octets = self._read_octets(data, pos, size=size, what=what) + return int.from_bytes(octets, 'little' if lilendian else 'big') + + @staticmethod + def _read_sint(value: 'int', bits: 'int') -> 'int': + """Read ``value``, an unsigned ``bits``-bit field, as a two's-complement one. + + It takes the unsigned integer a field has already read, such as a + :class:`~pcapkit.corekit.fields.strings.BitField` sub-field, rather + than a buffer. + + Args: + value: the unsigned value read from the field + bits: width of the field, in bits + + Returns: + The signed value the same bits denote. + + .. versionadded:: 1.6.0 + + """ + return value - (1 << bits) if value & (1 << (bits - 1)) else value + def _read_vector(self, data: 'bytes', pos: 'int' = 0, *, size: 'int' = 1, what: 'str') -> 'tuple[bytes, int]': """Read a length-prefixed vector out of ``data``. @@ -1894,15 +1969,9 @@ def _read_vector(self, data: 'bytes', pos: 'int' = 0, *, size: 'int' = 1, .. versionadded:: 1.6.0 """ - prefix = data[pos:pos + size] - if len(prefix) != size: - raise ProtocolError(f'{self.alias}: invalid format: {what} length runs past the end of the buffer') start = pos + size - end = start + int.from_bytes(prefix, 'big') - octets = data[start:end] - if len(octets) != end - start: - raise ProtocolError(f'{self.alias}: invalid format: {what} runs past the end of the buffer') - return octets, end + end = start + self._read_uint(data, pos, size=size, what=f'{what} length') + return self._read_octets(data, start, size=end - start, what=what), end def _read_whole_vector(self, data: 'bytes', *, size: 'int' = 1, what: 'str') -> 'bytes': """Read the one length-prefixed vector that ``data`` holds. @@ -1954,6 +2023,48 @@ def _read_uints(self, data: 'bytes', *, size: 'int' = 1, what: 'str') -> 'tuple[ f'is not a multiple of {size}') return tuple(int.from_bytes(data[index:index + size], 'big') for index in range(0, len(data), size)) + def _read_mac_addr(self, addr: 'bytes') -> 'str': + """Read a MAC address, an EUI-64 one included. + + Args: + addr: the address octets + + Returns: + Colon (``:``) separated *hex* encoded address. + + .. versionadded:: 1.6.0 + + """ + return addr.hex(':') + + def _read_text(self, data: 'bytes') -> 'tuple[str, Optional[str]]': + """Decode text, recording how to encode it back. + + The text is decoded by :meth:`decode`. Where the result does not + encode back to ``data`` as UTF-8 -- which ``make()`` uses by default -- + the detected charset is returned with it, or ``'latin-1'`` (decoding + again) if that does not reproduce ``data`` either, so that the original + octets survive a rebuild. + + Args: + data: raw text + + Returns: + The decoded text and the charset to encode it with, the latter + :obj:`None` for UTF-8. + + .. versionadded:: 1.6.0 + + """ + value = self.decode(data) + for charset in (None, detect(data)): + try: + if value.encode(charset or 'utf-8') == data: + return value, charset + except (LookupError, UnicodeError): + pass + return data.decode('latin-1'), 'latin-1' + @overload # pragma: no cover def _read_packet(self, length: 'Optional[int]' = ..., *, header: 'None' = ...) -> 'bytes': ... @overload # pragma: no cover @@ -2036,7 +2147,39 @@ def _make_pack(cls, integer: 'int', *, size: 'int' = 1, raise StructError(f'{cls.__name__}: pack failed') from error return buf - def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': + def _make_octets(self, value: 'Any', *, size: 'int', name: 'str', + alias: 'Optional[str]' = None) -> 'bytes': + """Check that ``value`` is exactly ``size`` octets. + + The octets :meth:`_read_octets` reads are written as they are, so + their size is all there is to check. + + Args: + value: the value given for the field + size: octets in the field + name: name of the field, for the error message + alias: prefix of the error message, for a caller whose + :attr:`alias` raises before ``read()`` sets it; :attr:`alias` + by default + + Returns: + ``value`` as :class:`bytes`. + + Raises: + ProtocolError: If ``value`` is not :class:`bytes` or + :class:`bytearray`, or is not ``size`` octets long. The + message names ``alias``. + + .. versionadded:: 1.6.0 + + """ + if not isinstance(value, (bytes, bytearray)) or len(value) != size: + unit = 'octet' if size == 1 else 'octets' + raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: ' + f'must be {size} {unit}, not {value!r}') + return bytes(value) + + def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str', alias: 'Optional[str]' = None) -> 'int': """Check that ``value`` is an unsigned integer that fits in ``bits`` bits. A :obj:`bool` passes only for a one-bit field, so that a flag given @@ -2046,6 +2189,9 @@ def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': value: the value given for the field bits: width of the field, in bits name: name of the field, for the error message + alias: prefix of the error message, for a caller whose + :attr:`alias` raises before ``read()`` sets it; :attr:`alias` + by default Returns: ``value`` as a plain :class:`int`. @@ -2053,17 +2199,65 @@ def _make_uint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': Raises: ProtocolError: If ``value`` is not an :class:`int`, is a :obj:`bool` for a field wider than one bit, is negative, or - does not fit in ``bits`` bits. The message names - :attr:`alias`. + does not fit in ``bits`` bits. The message names ``alias``. .. versionadded:: 1.6.0 """ if (not isinstance(value, int) or (isinstance(value, bool) and bits != 1) or not 0 <= value < (1 << bits)): - raise ProtocolError(f'{self.alias}: invalid {name}: {value!r}') + raise ProtocolError(f'{self.alias if alias is None else alias}: invalid {name}: {value!r}') return int(value) + def _make_sint(self, value: 'Any', *, bits: 'int', name: 'str') -> 'int': + """Check that ``value`` is a signed integer that fits in ``bits`` bits. + + The inverse of :meth:`_read_sint`. A :obj:`bool` is refused whatever + ``bits`` is, since no signed field holds a flag. + + Args: + value: the value given for the field + bits: width of the field, in bits + name: name of the field, for the error message + + Returns: + The unsigned integer whose ``bits`` bits denote ``value`` in + two's complement. + + Raises: + ProtocolError: If ``value`` is not an :class:`int`, is a + :obj:`bool`, or does not fit in ``bits`` bits as two's + complement. The message names :attr:`alias`. + + .. versionadded:: 1.6.0 + + """ + bound = 1 << (bits - 1) + if not isinstance(value, int) or isinstance(value, bool) or not -bound <= value < bound: + raise ProtocolError(f'{self.alias}: invalid {name}: {value!r}') + return value & ((1 << bits) - 1) + + @staticmethod + def _uint_size(value: 'int', *, minimum: 'int' = 0) -> 'int': + """Count the octets a non-negative integer needs. + + ``value.bit_length()`` is rounded up to whole octets: ``// 8`` alone + floors, and counts one octet too few for a width that is not a + multiple of eight. + + Args: + value: the integer, not negative + minimum: the fewest octets to count, for a field that cannot be + empty although ``0`` needs none + + Returns: + The octets ``value`` needs, and at least ``minimum``. + + .. versionadded:: 1.6.0 + + """ + return max(minimum, (value.bit_length() + 7) // 8) + @classmethod def _make_vector(cls, octets: 'bytes', *, size: 'int' = 1) -> 'bytes': """Write ``octets`` as a length-prefixed vector. @@ -2111,6 +2305,45 @@ def _make_uints(cls, values: 'Iterable[int]', *, size: 'int' = 1) -> 'bytes': """ return b''.join(cls._make_pack(int(value), size=size) for value in values) + def _make_mac_addr(self, addr: 'str | bytes | bytearray', *, size: 'Optional[int]' = 6, + name: 'str' = 'MAC address', alias: 'Optional[str]' = None) -> 'bytes': + """Make a MAC address, or an EUI-64 one with ``size=8``. + + The inverse of :meth:`_read_mac_addr`. + + Args: + addr: ``:``- or ``-``-separated *hex* encoded address + size: octets in the address, or :data:`None` for any number of + them, none included + name: name of the address, for the error message + alias: prefix of the error message, for a caller whose + :attr:`alias` raises before ``read()`` sets it or is not the + prefix its messages carry; :attr:`alias` by default + + Returns: + The address octets. + + Raises: + ProtocolError: If ``addr`` is not a :class:`str`, :class:`bytes` + or :class:`bytearray` (a :obj:`bool` included), or is not a + ``:``- or ``-``-separated *hex* encoded address of ``size`` + octets. The message names ``alias``. + + .. versionadded:: 1.6.0 + + """ + prefix = self.alias if alias is None else alias + if not isinstance(addr, (str, bytes, bytearray)): + raise ProtocolError(f'{prefix}: invalid {name}: must be a str, bytes or ' + f'bytearray, not {addr!r}') + text = addr.encode() if isinstance(addr, str) else bytes(addr) + octets = text.replace(b'-', b':').split(b':') if text else [] + if ((size is None or len(octets) == size) + and all(len(octet) == 2 and all(chr(digit) in string.hexdigits for digit in octet) + for octet in octets)): + return bytes.fromhex(b''.join(octets).decode()) + raise ProtocolError(f'{prefix}: invalid {name}: {addr!r}') + @overload # pragma: no cover @classmethod def _make_index(cls, name: 'int | StdlibEnum | AenumEnum', *, pack: 'Literal[False]' = ...) -> 'int': ... diff --git a/pcapkit/protocols/transport/sctp.py b/pcapkit/protocols/transport/sctp.py index 31f9e16b15..a4c2ace65d 100644 --- a/pcapkit/protocols/transport/sctp.py +++ b/pcapkit/protocols/transport/sctp.py @@ -1913,9 +1913,7 @@ def _make_chunk_donone(self, code: 'Enum_Chunk', chunk: 'Optional[Data_UnknownCh flags = chunk.flags value = chunk.value - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') return Schema_UnknownChunk( type=code, @@ -2037,9 +2035,7 @@ def _make_chunk_init(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITChunk]' init_tsn = chunk.init_tsn parameters = chunk.parameters - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2099,9 +2095,7 @@ def _make_chunk_init_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_INITACK init_tsn = chunk.init_tsn parameters = chunk.parameters - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2161,9 +2155,7 @@ def _make_chunk_sack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_SACKChunk]' gap_blocks = list(chunk.gap_blocks) dup_tsn = list(chunk.dup_tsn) - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') blocks = [] # type: list[Schema_GapAckBlock] for block in gap_blocks or []: @@ -2212,9 +2204,7 @@ def _make_chunk_heartbeat(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Heartb flags = chunk.flags parameters = chunk.parameters - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2254,9 +2244,7 @@ def _make_chunk_heartbeat_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_He flags = chunk.flags parameters = chunk.parameters - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') if parameters is not None: parameters_value = self._make_sctp_parameters(parameters) @@ -2337,9 +2325,7 @@ def _make_chunk_shutdown(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Shutdow flags = chunk.flags cum_tsn_ack = chunk.cum_tsn_ack - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') return Schema_ShutdownChunk( type=code, @@ -2369,9 +2355,7 @@ def _make_chunk_shutdown_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Shu if chunk is not None: flags = chunk.flags - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') return Schema_ShutdownACKChunk( type=code, @@ -2403,9 +2387,7 @@ def _make_chunk_error(self, code: 'Enum_Chunk', chunk: 'Optional[Data_ErrorChunk flags = chunk.flags error = chunk.error - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') if error is not None: error_value = self._make_sctp_causes(error) @@ -2446,9 +2428,7 @@ def _make_chunk_cookie_echo(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cook flags = chunk.flags cookie = chunk.cookie - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') return Schema_CookieEchoChunk( type=code, @@ -2478,9 +2458,7 @@ def _make_chunk_cookie_ack(self, code: 'Enum_Chunk', chunk: 'Optional[Data_Cooki if chunk is not None: flags = chunk.flags - if len(flags) != 1: - raise ProtocolError(f'{self.alias}: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + flags = self._make_octets(flags, size=1, name=f'[Chunk {code}] flags') return Schema_CookieACKChunk( type=code, @@ -3603,9 +3581,7 @@ def _make_cause_invalid_stream(self, code: 'Enum_CauseCode', cause: 'Optional[Da stream_id = cause.stream_id reserved = cause.reserved - if len(reserved) != 2: - raise ProtocolError(f'{self.alias}: [Cause {code}] invalid reserved: ' - f'must be two octets, not {reserved!r}') + reserved = self._make_octets(reserved, size=2, name=f'[Cause {code}] reserved') return Schema_InvalidStreamIdentifierCause( code=code, diff --git a/tests/project/test_docs_internal_refs_1940_unit.py b/tests/project/test_docs_internal_refs_1940_unit.py index 20f80a45b9..2ca7641d46 100644 --- a/tests/project/test_docs_internal_refs_1940_unit.py +++ b/tests/project/test_docs_internal_refs_1940_unit.py @@ -79,11 +79,11 @@ ('pcapkit/protocols/internet/ipv6.rst', 'pcapkit.protocols.internet.ipv6.IPv6._read_ip_hextet'), ('pcapkit/protocols/link/arp.rst', 'pcapkit.protocols.link.arp.ARP._make_addr_resolve'), ('pcapkit/protocols/link/arp.rst', 'pcapkit.protocols.link.arp.ARP._make_proto_resolve'), - ('pcapkit/protocols/link/ethernet.rst', 'pcapkit.protocols.link.ethernet.Ethernet._make_mac_addr'), ('pcapkit/protocols/misc/pcap.rst', 'pcapkit.protocols.misc.pcap.header.Header._read_protos'), ('pcapkit/protocols/misc/pcapng.rst', 'pcapkit.protocols.misc.pcapng.PCAPNG.PACKET_TYPES'), ('pcapkit/protocols/misc/pcapng.rst', 'pcapkit.protocols.misc.pcapng.PCAPNG._make_mac_addr'), ('pcapkit/protocols/misc/pcapng.rst', 'pcapkit.protocols.schema.misc.pcapng.OptionEnumField.post_process'), + ('pcapkit/protocols/protocol.rst', 'pcapkit.protocols.protocol.Protocol._make_mac_addr'), ('pcapkit/protocols/protocol.rst', 'pcapkit.protocols.protocol._declared_keywords'), ('pcapkit/utilities/exceptions.rst', 'pcapkit.utilities.exceptions.BaseError.__init__'), ('pcapkit/utilities/index.rst', 'pcapkit.utilities.compat.cached_property'), diff --git a/tests/project/test_protocol_member_order_2490_unit.py b/tests/project/test_protocol_member_order_2490_unit.py index f304f1104c..3c8e01ea38 100644 --- a/tests/project/test_protocol_member_order_2490_unit.py +++ b/tests/project/test_protocol_member_order_2490_unit.py @@ -240,7 +240,7 @@ def audit_tree(root: 'pathlib.Path' = ROOT) -> 'dict[str, dict[str, str]]': class TestAudit(unittest.TestCase): """The audit finds each rule broken in a planted class body, and nothing in a sound one.""" - #: A class body in the order: Ethernet's utilities, with a registry fallback. + #: A class body in the order: ``_make_data()``, a helper pair and a registry fallback. SOUND = ''' class Sound: @property @@ -314,13 +314,15 @@ def test_exceptions_are_a_minority(self) -> 'None': self.assertLess(len(KNOWN_EXCEPTIONS) * 2, len(self.findings)) def test_the_reference_is_in_order(self) -> 'None': - name = 'pcapkit.protocols.link.ethernet.Ethernet' + name = 'pcapkit.protocols.link.arp.ARP' self.assertEqual(self.findings[name], {}) self.assertNotIn(name, KNOWN_EXCEPTIONS) - # The page's example: rule b finds the helper pair it shows. - node = dict(classes(PROTOCOLS / 'link' / 'ethernet.py'))[name] + # The page's example: its utilities are _make_data() and the helper pairs rule b finds. + node = dict(classes(PROTOCOLS / 'link' / 'arp.py'))[name] utilities = [member for member, group in members(node) if group == 3] - self.assertEqual(helpers(node, utilities), ['_read_mac_addr', '_make_mac_addr']) + pairs = ['_read_addr_resolve', '_make_addr_resolve', '_read_proto_resolve', '_make_proto_resolve'] + self.assertEqual(helpers(node, utilities), pairs) + self.assertEqual(utilities, ['_make_data', *pairs]) def test_every_exception_is_still_broken(self) -> 'None': stale = sorted(f'{name} [{kind}] ({remover})' diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index 45eaf48827..b4a2925c4d 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -209,9 +209,14 @@ #: the module, the expression (see :func:`_context`) and why. A hit here is in a #: module whose own protocol owns a constant of that value, or that reads one. NOT_A_COPY = { - ('pcapkit/protocols/application/ssl.py', '_take(32)'): + ('pcapkit/protocols/application/ssl.py', 'self._read_octets(size=32)'): "a TLS/DTLS ClientHello/ServerHello Random is 32 octets (:rfc:`9846#section-4.2.2`), " 'not an Interface Description Block minimum', + ('pcapkit/protocols/protocol.py', 'def _make_mac_addr(size=6)'): + 'a MAC address is 6 octets (IEEE 802), not the IPv6 Next Header offset', + ('pcapkit/protocols/misc/pcapng.py', 'def _make_mac_addr(size=6)'): + "the base MAC maker's 6-octet default, which the override repeats, not the IPv6 Next " + 'Header offset', ('pcapkit/protocols/schema/application/dtls.py', 'NumberField(length=6)'): "DTLS's 48-bit sequence_number (:rfc:`9147#section-4`), not the IPv6 Next " 'Header offset', @@ -402,11 +407,12 @@ 'pcapkit/protocols/schema/internet/ipv6_option.py': ((16,), 'an MPL Seed-ID of 16 octets'), 'pcapkit/protocols/schema/application/httpv2.py': ((6, 32), 'an HTTP/2 setting size and a flag bit'), - 'pcapkit/protocols/application/ospf.py': ((16, 24, 65535), - 'OSPF header length and field bounds; 16 is the ' - 'offset of the Authentication field the checksum ' - 'leaves out (RFC 2328 appendix A.3.1), not a PCAP ' - 'record header'), + 'pcapkit/protocols/application/ospf.py': ((16, 24), + 'OSPF header length; 16 is the bit width of the ' + 'packet length, and the offset of the ' + 'Authentication field the checksum leaves out ' + '(RFC 2328 appendix A.3.1), not a PCAP record ' + 'header'), 'pcapkit/protocols/schema/application/ospf.py': ((24,), 'the OSPF header length'), 'pcapkit/protocols/internet/esp.py': ((12, 16, 20, 24, 32), 'ESP ICV and key sizes'), 'pcapkit/protocols/internet/igmpv3.py': ((16,), "the floating-point Max Resp Code/QQIC " diff --git a/tests/protocols/application/test_ospf_dead_pattern_1844_unit.py b/tests/protocols/application/test_ospf_dead_pattern_1844_unit.py index e53df5e70c..fc7a09cbc8 100644 --- a/tests/protocols/application/test_ospf_dead_pattern_1844_unit.py +++ b/tests/protocols/application/test_ospf_dead_pattern_1844_unit.py @@ -1,12 +1,16 @@ # -*- coding: utf-8 -*- -"""The unused Ethernet-address pattern in the OSPF module. +"""The unused Ethernet-address pattern in the OSPF, Ethernet and PCAP-NG modules. GitHub issue #1844. Guards against :mod:`pcapkit.protocols.application.ospf` defining a ``PAT_MAC_ADDR`` that nothing reads, and importing :mod:`re` only to compile it. These tests pin that the constant is absent and that the module does not import :mod:`re` -without using it. The copies of the pattern in the Ethernet, ARP and PCAP-NG -modules have readers, and stay. +without using it. + +The Ethernet and PCAP-NG copies lost their readers when the MAC address maker +moved to ``ProtocolBase`` (#2614), so they are pinned absent too, as is the +Ethernet module's :mod:`re` import. The ARP copy, which matches an address of +any number of octets, has a reader and stays. """ from __future__ import annotations @@ -28,29 +32,36 @@ def setUp(self) -> None: def test_pat_mac_addr_is_gone(self) -> None: from pcapkit.protocols.application import ospf + from pcapkit.protocols.link import ethernet + from pcapkit.protocols.misc import pcapng - self.assertFalse(hasattr(ospf, 'PAT_MAC_ADDR')) + for module in (ospf, ethernet, pcapng): + with self.subTest(module=module.__name__): + self.assertFalse(hasattr(module, 'PAT_MAC_ADDR')) def test_re_is_not_imported_unused(self) -> None: # Read the source of the module actually imported, so the check runs # against the tree under test. ``re`` may come back if something uses # it; what must not return is the import with no reader. from pcapkit.protocols.application import ospf + from pcapkit.protocols.link import ethernet + + for module in (ospf, ethernet): + with open(module.__file__, encoding='utf-8') as file: + tree = ast.parse(file.read()) - with open(ospf.__file__, encoding='utf-8') as file: - tree = ast.parse(file.read()) - - imports_re = any( - alias.name == 're' and alias.asname is None - for node in tree.body if isinstance(node, ast.Import) - for alias in node.names - ) - reads_re = any( - isinstance(node, ast.Name) and node.id == 're' - for node in ast.walk(tree) - ) - self.assertFalse(imports_re and not reads_re, - f'{ospf.__file__} imports re but never uses it') + imports_re = any( + alias.name == 're' and alias.asname is None + for node in tree.body if isinstance(node, ast.Import) + for alias in node.names + ) + reads_re = any( + isinstance(node, ast.Name) and node.id == 're' + for node in ast.walk(tree) + ) + with self.subTest(module=module.__name__): + self.assertFalse(imports_re and not reads_re, + f'{module.__file__} imports re but never uses it') if __name__ == '__main__': diff --git a/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py b/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py index 6774cef5f6..4838932aae 100644 --- a/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py +++ b/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py @@ -98,7 +98,7 @@ def test_make_rejects_invalid_packet_length(self) -> None: with self.subTest(packet_length=packet_length): with self.assertRaises(ProtocolError) as caught: OSPF(packet_length=packet_length) - self.assertIn('invalid packet length', str(caught.exception)) + self.assertEqual(str(caught.exception), f'OSPF: invalid packet length: {packet_length!r}') self.assertEqual(bytes(OSPF(packet_length=0xFFFF))[2:4], b'\xff\xff') def test_make_crypto_auth_from_bytes(self) -> None: diff --git a/tests/protocols/application/test_tls_extensions_2514_unit.py b/tests/protocols/application/test_tls_extensions_2514_unit.py index 95c5d0fbc1..807c84f007 100644 --- a/tests/protocols/application/test_tls_extensions_2514_unit.py +++ b/tests/protocols/application/test_tls_extensions_2514_unit.py @@ -361,6 +361,125 @@ def test_a_structure_with_nothing_to_build_from_raises(self) -> None: with self.subTest(case=name), self.assertRaises(ProtocolError): self._hello(extensions, server=server) + def test_a_value_too_wide_for_its_field_raises_struct_error(self) -> None: + """Every field ``make`` packs refuses an oversize value the same way, as ``Protocol._make_pack`` does.""" + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.protocols.data.application.tls import Extension, KeyShareEntry, ServerName + from pcapkit.utilities.exceptions import StructError + + share = KeyShareEntry(group=0x10000, key_exchange=b'\x00') + for name, extensions, kwargs in ( + ('client_shares group', ((ExtensionType.key_share, {'client_shares': (share,)}),), {}), + ('server_share group', ((ExtensionType.key_share, {'server_share': share}),), {'server': True}), + ('selected_group', ((ExtensionType.key_share, {'selected_group': 0x10000}),), + {'server': True, 'random': HRR_RANDOM}), + ('selected_version', ((ExtensionType.supported_versions, {'selected_version': 0x10000}),), + {'server': True}), + ('name_type', ((ExtensionType.server_name, + {'server_name_list': (ServerName(name_type=0x100, name=b'x'),)}),), {}), + ('extension type', (Extension(type=0x10000, data=b''),), {}), + ('extension body', (Extension(type=ExtensionType.padding, data=bytes(0x10000)),), {}), + ('hello extensions', (Extension(type=ExtensionType.padding, data=bytes(0x8000)),) * 2, {}), + ): + with self.subTest(case=name), self.assertRaisesRegex(StructError, r'^TLS: pack failed$'): + self._hello(extensions, **kwargs) + + for name, alias, content_type, content in self._oversize_contents(): + with self.subTest(case=name), self.assertRaisesRegex(StructError, rf'^{alias}: pack failed$'): + self._content(alias, content_type, content) + + # The record header, which the schema would otherwise mask to its width. + for name, alias, kwargs in ( + ('record content_type', 'TLS', {'content_type': 0x100}), + ('record version', 'TLS', {'version': 0x10000}), + ('record negative version', 'TLS', {'version': -1}), + ('record length', 'TLS', {'data': bytes(0x10000)}), + ('DTLS record content_type', 'DTLS', {'content_type': 0x100}), + ('DTLS record version', 'DTLS', {'version': 0x10000}), + ('DTLS record epoch', 'DTLS', {'epoch': 0x10000}), + ('DTLS record sequence_number', 'DTLS', {'sequence_number': 1 << 48}), + ('DTLS record length', 'DTLS', {'data': bytes(0x10000)}), + ('tls12_cid record sequence_number', 'DTLS', + {'content_type': 25, 'cid': b'\x01', 'sequence_number': 1 << 48, 'data': b'x'}), + ('unified header short sequence_number', 'DTLS', + {'ciphertext': True, 'sequence_number': 0x100, 'content': b'x'}), + ('unified header long sequence_number', 'DTLS', + {'ciphertext': True, 'sequence_long': True, 'sequence_number': 0x10000, 'content': b'x'}), + ('unified header length', 'DTLS', {'ciphertext': True, 'content': bytes(0x10000)}), + ): + with self.subTest(case=name), self.assertRaisesRegex(StructError, rf'^{alias}: pack failed$'): + self._record(alias, **kwargs) + + @staticmethod + def _record(alias: str, **kwargs: object) -> bytes: + """``make`` a TLS or DTLS record from ``kwargs``.""" + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.application.tls import TLS + + maker = object.__new__(TLS if alias == 'TLS' else DTLS) + return bytes(maker.make(**kwargs)) + + @classmethod + def _content(cls, alias: str, content_type: int, content: object) -> bytes: + """``make`` a TLS or DTLS record carrying ``content``.""" + return cls._record(alias, content_type=content_type, version=0x0303 if alias == 'TLS' else 0xFEFD, + content=content) + + @staticmethod + def _oversize_contents() -> 'tuple': + """One ``(case, maker, content type, content)`` per field the hello, handshake and content makers pack.""" + from pcapkit.const.tls.content_type import ContentType + from pcapkit.const.tls.handshake_type import HandshakeType + from pcapkit.protocols.data.application.dtls import DTLSHandshake + from pcapkit.protocols.data.application.tls import (Alert, ChangeCipherSpec, ClientHello, Handshake, + ServerHello) + + def client(**fields: object) -> 'ClientHello': + values = {'version': 0x0303, 'random': bytes(32), 'session_id': b'', 'cookie': None, + 'cipher_suites': (0x1301,), 'compression_methods': (0,), 'extensions': None} + return ClientHello(**{**values, **fields}) + + def server(**fields: object) -> 'ServerHello': + values = {'version': 0x0303, 'random': bytes(32), 'session_id': b'', 'cipher_suite': 0x1301, + 'compression_method': 0, 'extensions': None} + return ServerHello(**{**values, **fields}) + + def dtls(msg_type: int = HandshakeType.client_hello, body: object = b'', **fields: object) -> 'tuple': + values = {'length': 0, 'message_seq': 0, 'fragment_offset': 0, 'fragment_length': 0, + 'is_fragmented': False} + return (DTLSHandshake(msg_type=msg_type, body=body, **{**values, **fields}),) + + handshake, alert, ccs = ContentType.handshake, ContentType.alert, ContentType.change_cipher_spec + return ( + ('client_hello.version', 'TLS', handshake, + (Handshake(msg_type=1, length=0, body=client(version=0x10000)),)), + ('client_hello.session_id', 'TLS', handshake, + (Handshake(msg_type=1, length=0, body=client(session_id=bytes(0x100))),)), + ('client_hello.cookie', 'DTLS', handshake, dtls(body=client(cookie=bytes(0x100)))), + ('client_hello.cipher_suites', 'TLS', handshake, + (Handshake(msg_type=1, length=0, body=client(cipher_suites=(0x10000,))),)), + ('client_hello.compression_methods', 'TLS', handshake, + (Handshake(msg_type=1, length=0, body=client(compression_methods=(0x100,))),)), + ('client_hello.compression_methods length', 'TLS', handshake, + (Handshake(msg_type=1, length=0, body=client(compression_methods=(0,) * 0x100)),)), + ('server_hello.version', 'TLS', handshake, + (Handshake(msg_type=2, length=0, body=server(version=0x10000)),)), + ('server_hello.session_id', 'TLS', handshake, + (Handshake(msg_type=2, length=0, body=server(session_id=bytes(0x100))),)), + ('server_hello.cipher_suite', 'TLS', handshake, + (Handshake(msg_type=2, length=0, body=server(cipher_suite=0x10000)),)), + ('server_hello.compression_method', 'TLS', handshake, + (Handshake(msg_type=2, length=0, body=server(compression_method=0x100)),)), + ('handshake msg_type', 'TLS', handshake, (Handshake(msg_type=0x100, length=0, body=b''),)), + ('handshake length', 'DTLS', handshake, + dtls(body=b'\x00', length=0x1000000, fragment_length=1, is_fragmented=True)), + ('message_seq', 'DTLS', handshake, dtls(message_seq=0x10000)), + ('fragment_offset', 'DTLS', handshake, dtls(fragment_offset=0x1000000)), + ('alert level', 'TLS', alert, Alert(level=0x100, description=0)), + ('alert description', 'TLS', alert, Alert(level=2, description=0x100)), + ('change_cipher_spec', 'TLS', ccs, ChangeCipherSpec(msg=0x100)), + ) + class ExtensionRegistryTests(unittest.TestCase): """``register_tls_extension`` writes ``SSL.__extension__``, which TLS and DTLS read.""" diff --git a/tests/protocols/test_maker_invalid_format_messages_2296_unit.py b/tests/protocols/test_maker_invalid_format_messages_2296_unit.py index be490ed75a..449f42ddad 100644 --- a/tests/protocols/test_maker_invalid_format_messages_2296_unit.py +++ b/tests/protocols/test_maker_invalid_format_messages_2296_unit.py @@ -85,10 +85,9 @@ def test_wrong_width_flags_name_the_argument(self) -> None: with self.subTest(maker=maker, flags=flags): with self.assertRaises(ProtocolError) as context: getattr(sctp, maker)(code, flags=flags) - self.assertEqual( - str(context.exception), - f'SCTP: [Chunk {code}] invalid flags: ' - f'must be one octet, not {flags!r}') + # ``ProtocolBase._make_octets`` words it, the chunk named with the field (#2614). + self.assertEqual(str(context.exception), + f'SCTP: invalid [Chunk {code}] flags: must be 1 octet, not {flags!r}') def test_one_octet_flags_still_build(self) -> None: from pcapkit.const.sctp.chunk import Chunk @@ -169,10 +168,9 @@ def test_wrong_width_reserved_names_the_argument(self) -> None: with self.subTest(reserved=reserved): with self.assertRaises(ProtocolError) as context: sctp._make_cause_invalid_stream(code, reserved=reserved) - self.assertEqual( - str(context.exception), - f'SCTP: [Cause {code}] invalid reserved: ' - f'must be two octets, not {reserved!r}') + # ``ProtocolBase._make_octets`` words it, the cause named with the field (#2614). + self.assertEqual(str(context.exception), + f'SCTP: invalid [Cause {code}] reserved: must be 2 octets, not {reserved!r}') def test_two_octet_reserved_still_builds(self) -> None: from pcapkit.const.sctp.cause_code import CauseCode diff --git a/tests/protocols/test_protocol_value_helpers_2514_unit.py b/tests/protocols/test_protocol_value_helpers_2514_unit.py index 7e228873f8..019d5a5119 100644 --- a/tests/protocols/test_protocol_value_helpers_2514_unit.py +++ b/tests/protocols/test_protocol_value_helpers_2514_unit.py @@ -17,9 +17,21 @@ fit the width; * ``_make_uint`` -- the check that a value given for a field fits its width. -The errors the readers and ``_make_uint`` raise name the instance's ``alias``, -which is per instance -- a TLS record's is its tier, such as ``TLS 1.3`` -- so -that is pinned with an alias that differs from the class name. +The rest of ``ProtocolBase``'s value helpers, from the audit on #2614: + +* ``_read_octets`` and ``_read_uint`` -- exactly so many octets, or an unsigned + integer of so many, out of a buffer, refusing a short read; ``_make_octets`` + checks the octets a field is given; +* ``_read_sint`` and ``_make_sint`` -- a two's-complement field, read from its + unsigned value and checked on the way back; +* ``_uint_size`` -- the octets an unsigned integer needs; +* ``_read_mac_addr`` and ``_make_mac_addr`` -- a MAC address, an EUI-64 one, or + one of any length; +* ``_read_text`` -- text decoded so that it encodes back to its octets. + +The errors the readers and makers raise name the instance's ``alias``, which is +per instance -- a TLS record's is its tier, such as ``TLS 1.3`` -- so that is +pinned with an alias that differs from the class name. """ @@ -35,6 +47,7 @@ if TYPE_CHECKING: from typing import Callable + from pcapkit.protocols.application.ospf import OSPF from pcapkit.protocols.misc.raw import Raw #: Test tier (#1958): the per-helper cases of the protocol base, beside the @@ -63,6 +76,15 @@ def assertRefused(self, pattern: 'str', func: 'Callable[..., object]', *args: 'o with self.assertRaisesRegex(ProtocolError, f'^{ALIAS}: {pattern}'): func(*args, **kwargs) + def unread(self) -> 'OSPF': + """An OSPF instance ``read()`` has not run on, so its ``alias`` raises.""" + from pcapkit.protocols.application.ospf import OSPF + + proto = object.__new__(OSPF) + with self.assertRaises(AttributeError): + getattr(proto, 'alias') + return proto + class ReadVectorTests(_HelperTestCase): """``_read_vector`` returns the vector's octets and the offset just past them.""" @@ -208,6 +230,235 @@ def test_a_non_int_is_refused(self) -> None: with self.subTest(value=value): self.assertRefused(f'invalid field: {shown}$', self.proto._make_uint, value, bits=4, name='field') + def test_an_alias_given_prefixes_the_message_before_read(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + proto = self.unread() + self.assertEqual(proto._make_uint(0xFFFF, bits=16, name='packet length', alias='OSPF'), 0xFFFF) + with self.assertRaisesRegex(ProtocolError, '^OSPF: invalid packet length: 65536$'): + proto._make_uint(0x10000, bits=16, name='packet length', alias='OSPF') + + +class ReadOctetsTests(_HelperTestCase): + """``_read_octets`` returns exactly the octets asked for, or refuses.""" + + def test_each_size_and_offset(self) -> None: + data = bytes(range(10)) + for pos, size in ((0, 1), (0, 10), (3, 4), (9, 1), (10, 0)): + with self.subTest(pos=pos, size=size): + self.assertEqual(self.proto._read_octets(data, pos, size=size, what='f'), data[pos:pos + size]) + + def test_the_default_offset(self) -> None: + self.assertEqual(self.proto._read_octets(b'abc', size=2, what='f'), b'ab') + + def test_a_short_read_is_refused(self) -> None: + for pos, size in ((0, 4), (2, 2), (3, 1), (4, 1)): + with self.subTest(pos=pos, size=size): + self.assertRefused('invalid format: f runs past the end of the buffer$', + self.proto._read_octets, b'abc', pos, size=size, what='f') + + +class ReadUintTests(_HelperTestCase): + """``_read_uint`` reads one unsigned integer, the inverse of ``_make_pack``.""" + + def test_each_width_and_byte_order(self) -> None: + data = b'\xee' + bytes(range(1, 9)) + for size in (1, 2, 3, 4, 8): + for lilendian in (False, True): + with self.subTest(size=size, lilendian=lilendian): + expected = int.from_bytes(data[1:1 + size], 'little' if lilendian else 'big') + self.assertEqual(self.proto._read_uint(data, 1, size=size, what='f', lilendian=lilendian), + expected) + + def test_it_undoes_make_pack_at_the_boundaries(self) -> None: + for size in (1, 2, 3, 4, 8): + for value in (0, 1, (1 << (8 * size)) - 1): + for lilendian in (False, True): + with self.subTest(size=size, value=value, lilendian=lilendian): + packed = type(self.proto)._make_pack(value, size=size, lilendian=lilendian) + self.assertEqual(self.proto._read_uint(packed, size=size, what='f', lilendian=lilendian), + value) + + def test_a_short_read_is_refused(self) -> None: + for size in (1, 2, 3): + with self.subTest(size=size): + self.assertRefused('invalid format: f runs past the end of the buffer$', + self.proto._read_uint, b'\x00' * size, 1, size=size, what='f') + + +class ReadSintTests(_HelperTestCase): + """``_read_sint`` reads an unsigned field's bits as two's complement.""" + + def test_the_boundaries_of_each_width(self) -> None: + for bits in (1, 8, 24): + top = 1 << bits + for unsigned, signed in ((0, 0), (top // 2 - 1, top // 2 - 1), (top // 2, -(top // 2)), (top - 1, -1)): + with self.subTest(bits=bits, unsigned=unsigned): + self.assertEqual(self.proto._read_sint(unsigned, bits), signed) + + +class MakeOctetsTests(_HelperTestCase): + """``_make_octets`` passes exactly so many octets, as :class:`bytes`.""" + + def test_octets_of_the_size_are_returned_as_bytes(self) -> None: + for size, value in ((1, b'\x01'), (2, bytearray(b'ab')), (8, bytes(8))): + with self.subTest(size=size): + made = self.proto._make_octets(value, size=size, name='field') + self.assertEqual(made, bytes(value)) + self.assertIs(type(made), bytes) + + def test_a_wrong_size_is_refused(self) -> None: + for size, value, unit in ((1, b'', '1 octet'), (1, b'ab', '1 octet'), (2, b'a', '2 octets')): + with self.subTest(size=size, value=value): + self.assertRefused(f'invalid field: must be {unit}, not {value!r}$', + self.proto._make_octets, value, size=size, name='field') + + def test_a_non_octet_string_is_refused(self) -> None: + for value in ('a', 1, True, None): + with self.subTest(value=value): + self.assertRefused(f'invalid field: must be 1 octet, not {value!r}$', + self.proto._make_octets, value, size=1, name='field') + + def test_an_alias_given_prefixes_the_message_before_read(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + proto = self.unread() + with self.assertRaisesRegex(ProtocolError, "^OSPF: invalid field: must be 1 octet, not b''$"): + proto._make_octets(b'', size=1, name='field', alias='OSPF') + + +class MakeSintTests(_HelperTestCase): + """``_make_sint`` checks a signed value and returns the bits ``_read_sint`` reads.""" + + def test_the_boundaries_round_trip(self) -> None: + for bits in (2, 8, 24): + half = 1 << (bits - 1) + for value in (-half, -1, 0, 1, half - 1): + with self.subTest(bits=bits, value=value): + made = self.proto._make_sint(value, bits=bits, name='field') + self.assertTrue(0 <= made < 1 << bits) + self.assertEqual(self.proto._read_sint(made, bits), value) + + def test_a_negative_value_is_twos_complement(self) -> None: + self.assertEqual(self.proto._make_sint(-1, bits=24, name='field'), 0xFFFFFF) + self.assertEqual(self.proto._make_sint(-(1 << 23), bits=24, name='field'), 0x800000) + + def test_one_past_either_end_is_refused(self) -> None: + for bits in (2, 8, 24): + half = 1 << (bits - 1) + for value in (-half - 1, half): + with self.subTest(bits=bits, value=value): + self.assertRefused(rf'invalid field: {value}$', self.proto._make_sint, value, bits=bits, + name='field') + + def test_a_bool_or_a_non_int_is_refused(self) -> None: + for value, shown in ((True, 'True'), (False, 'False'), ('1', "'1'"), (1.0, r'1\.0'), (None, 'None')): + with self.subTest(value=value): + self.assertRefused(f'invalid field: {shown}$', self.proto._make_sint, value, bits=8, name='field') + + +class UintSizeTests(_HelperTestCase): + """``_uint_size`` counts whole octets, rounding a partial one up.""" + + def test_each_boundary(self) -> None: + for value, size in ((0, 0), (1, 1), (0xFF, 1), (0x100, 2), (0xFFFF, 2), (0x10000, 3), + ((1 << 64) - 1, 8), (1 << 64, 9)): + with self.subTest(value=value): + self.assertEqual(type(self.proto)._uint_size(value), size) + + def test_a_minimum(self) -> None: + self.assertEqual(type(self.proto)._uint_size(0, minimum=1), 1) + self.assertEqual(type(self.proto)._uint_size(0x100, minimum=1), 2) + self.assertEqual(type(self.proto)._uint_size(1, minimum=4), 4) + + +class ReadMacAddrTests(_HelperTestCase): + """``_read_mac_addr`` writes the octets as colon-separated *hex*.""" + + def test_a_mac_and_an_eui_64_address(self) -> None: + self.assertEqual(self.proto._read_mac_addr(bytes.fromhex('00005e0053af')), '00:00:5e:00:53:af') + self.assertEqual(self.proto._read_mac_addr(bytes.fromhex('02005efffe0053af')), '02:00:5e:ff:fe:00:53:af') + + +class MakeMacAddrTests(_HelperTestCase): + """``_make_mac_addr`` reads a ``:``- or ``-``-separated address of ``size`` octets.""" + + MAC = bytes.fromhex('00005e0053af') + EUI = bytes.fromhex('02005efffe0053af') + + def test_each_spelling_of_a_mac_address(self) -> None: + for addr in ('00:00:5e:00:53:af', '00-00-5E-00-53-AF', '00:00-5e:00-53:aF', + b'00:00:5e:00:53:af', bytearray(b'00-00-5e-00-53-af')): + with self.subTest(addr=addr): + made = self.proto._make_mac_addr(addr) + self.assertEqual(made, self.MAC) + self.assertIs(type(made), bytes) + self.assertEqual(self.proto._make_mac_addr(self.proto._read_mac_addr(self.MAC)), self.MAC) + + def test_an_eui_64_address(self) -> None: + self.assertEqual(self.proto._make_mac_addr('02:00:5e:ff:fe:00:53:af', size=8, name='EUI address'), + self.EUI) + for addr in ('00:00:5e:00:53:af', '02:00:5e:ff:fe:00:53:af:00'): + with self.subTest(addr=addr): + self.assertRefused(f'invalid EUI address: {addr!r}$', self.proto._make_mac_addr, addr, + size=8, name='EUI address') + + def test_any_number_of_octets_without_a_size(self) -> None: + for addr, made in (('', b''), (b'', b''), ('ab', b'\xab'), ('00:00:5e:00:53:af', self.MAC), + ('02-00-5e-ff-fe-00-53-af', self.EUI), ('01:02:03', b'\x01\x02\x03')): + with self.subTest(addr=addr): + self.assertEqual(self.proto._make_mac_addr(addr, size=None), made) + + def test_a_malformed_address_is_refused(self) -> None: + for addr in ('', '00:00:5e:00:53', '00:00:5e:00:53:af:01', '00:00:5e:00:53:a', '000:00:5e:00:53:af', + '00:00:5e:00:53:af:', '00::00:5e:00:53:af', '00.00.5e.00.53.af', '00:00:5e:00:53:ag', + '00 00 5e 00 53 af', '00005e0053af', 'é0:00:5e:00:53:af'): + with self.subTest(addr=addr): + self.assertRefused(f'invalid MAC address: {addr!r}$', self.proto._make_mac_addr, addr) + for addr in (':', 'a', 'abc', 'ab:', ':ab', 'ab::cd'): + with self.subTest(addr=addr, size=None): + self.assertRefused(f'invalid MAC address: {addr!r}$', self.proto._make_mac_addr, addr, size=None) + + def test_a_value_of_another_type_is_refused(self) -> None: + for addr in (True, 0, None, 1.0): + with self.subTest(addr=addr): + self.assertRefused(f'invalid MAC address: must be a str, bytes or bytearray, not {addr!r}$', + self.proto._make_mac_addr, addr) + + def test_an_alias_given_prefixes_the_message(self) -> None: + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaisesRegex(ProtocolError, "^PCAP-NG: invalid MAC address: 'x'$"): + self.proto._make_mac_addr('x', alias='PCAP-NG') + proto = self.unread() + with self.assertRaisesRegex(ProtocolError, '^ARP: invalid MAC address: must be a str, bytes or ' + 'bytearray, not None$'): + proto._make_mac_addr(None, alias='ARP') + + +class ReadTextTests(_HelperTestCase): + """``_read_text`` returns text and the charset that encodes it back to its octets.""" + + def test_ascii_needs_no_charset(self) -> None: + for data in (b'', b'/index.html', b'200 OK'): + with self.subTest(data=data): + self.assertEqual(self.proto._read_text(data), (data.decode(), None)) + + def test_a_detected_charset_that_cannot_encode_falls_back_to_latin_1(self) -> None: + """A charset Python does not know, or one the text does not fit, is passed over.""" + for charset in ('no-such-codec', 'ascii'): + with self.subTest(charset=charset), \ + mock.patch.object(type(self.proto), 'decode', return_value='café'), \ + mock.patch('pcapkit.protocols.protocol.detect', return_value=charset): + self.assertEqual(self.proto._read_text(b'caf\xe9'), ('café', 'latin-1')) + + def test_other_octets_encode_back_with_the_charset_returned(self) -> None: + for data in (b'caf\xe9', b'\xff\xfe\xfd', b'\x80abc', 'café'.encode('cp1252') * 4): + with self.subTest(data=data): + text, charset = self.proto._read_text(data) + self.assertIsNotNone(charset) + self.assertEqual(text.encode(charset), data) + if __name__ == '__main__': unittest.main()