From b32b8e583654116351a14550bc2ed367a92b5046 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 31 Oct 2026 00:37:11 -0400 Subject: [PATCH 1/2] refactor(protocols): order application protocol members Move public members ahead of dunders and read helpers ahead of make helpers in Application, HTTP v1/v2, OSPF, SSL and the HTTP v2 Settings data model. Pure reorder: no bodies, names or behaviour changed. Refs #2490 --- pcapkit/protocols/application/application.py | 36 +-- pcapkit/protocols/application/httpv1.py | 164 ++++++------ pcapkit/protocols/application/httpv2.py | 86 +++---- pcapkit/protocols/application/ospf.py | 62 ++--- pcapkit/protocols/application/ssl.py | 248 +++++++++---------- pcapkit/protocols/data/application/httpv2.py | 10 +- 6 files changed, 305 insertions(+), 301 deletions(-) diff --git a/pcapkit/protocols/application/application.py b/pcapkit/protocols/application/application.py index 22475bde88..7005c9298c 100644 --- a/pcapkit/protocols/application/application.py +++ b/pcapkit/protocols/application/application.py @@ -66,6 +66,26 @@ def layer(self) -> 'Literal["Application"]': """Protocol layer.""" return self.__layer__ + ########################################################################## + # Methods. + ########################################################################## + + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. + **kwargs: Construction keywords, as for :meth:`Protocol.from_data + `. + + Returns: + Protocol instance, cut back to what was captured when the header + it was parsed from was cut short (:issue:`1458`). + + """ + return replay_short_read(super().from_data(data, **kwargs), data) + ########################################################################## # Data models. ########################################################################## @@ -115,22 +135,6 @@ def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned """ raise IntError(f'{cls.__name__!r} object cannot be interpreted as an integer') - @classmethod - def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': - """Create protocol instance from data. - - Args: - data: Protocol data. - **kwargs: Construction keywords, as for :meth:`Protocol.from_data - `. - - Returns: - Protocol instance, cut back to what was captured when the header - it was parsed from was cut short (:issue:`1458`). - - """ - return replay_short_read(super().from_data(data, **kwargs), data) - ########################################################################## # Utilities. ########################################################################## diff --git a/pcapkit/protocols/application/httpv1.py b/pcapkit/protocols/application/httpv1.py index ad0336555c..5c3922991a 100644 --- a/pcapkit/protocols/application/httpv1.py +++ b/pcapkit/protocols/application/httpv1.py @@ -277,6 +277,16 @@ def make(self, # type: ignore[override] data=header_line + b''.join(header_fields) + b'\r\n' + body, ) + @classmethod + def id(cls) -> 'tuple[Literal["HTTP"], Literal["HTTPv1"]]': # type: ignore[override] + """Index ID of the protocol. + + Returns: + Index ID of the protocol. + + """ + return (cls.__name__, 'HTTPv1') # type: ignore[return-value] + @staticmethod def _check_raw(raw: 'bytes', *values: 'str', fold: 'bool') -> 'None': """Refuse a raw line that would write a line break its values lack. @@ -301,88 +311,6 @@ def _check_raw(raw: 'bytes', *values: 'str', fold: 'bool') -> 'None': if raw.count(char.encode()) != sum(value.count(char) for value in values): raise ProtocolError('HTTP: raw line carries a CR or LF its values do not') - def _make_start_line(self, http_version: 'str | bytes', *, - method: 'Optional[Enum_Method | str | bytes]' = None, - uri: 'Optional[str | bytes]' = None, - status: 'Optional[Enum_StatusCode | str | bytes | int]' = None, - status_default: 'Optional[int]' = None, - status_namespace: 'Optional[dict[str, int] | dict[int, str] | _Type[StdlibEnum] | _Type[AenumEnum]]' = None, # pylint: disable=line-too-long - status_reversed: 'bool' = False, - message: 'Optional[str | bytes]' = None, - charset: 'Optional[str]' = None) -> 'bytes': - """Make an HTTP/1.* start line from its values. - - Args: - http_version: HTTP version. - method: HTTP method. - uri: HTTP request URI. - status: HTTP status code. - status_default: Default HTTP status code. - status_namespace: Namespace of HTTP status code. - status_reversed: Whether to reverse the namespace. - message: HTTP status message. - charset: Encoding for a :obj:`str` request URI or status message, - **UTF-8** if not given. - - Returns: - The start line, CRLF included. - - Raises: - ProtocolError: If the values are neither a request's nor a - response's, or if they are a request's and ``uri`` is - :obj:`None`. - - """ - version = http_version.encode() if isinstance(http_version, str) else http_version - if method is not None and status is None: - if uri is None: - raise ProtocolError('HTTP: invalid start line: a request needs a URI') - - if isinstance(method, Enum_Method): - meth = method.value.encode() - elif isinstance(method, bytes): - meth = method - elif isinstance(method, str): - meth = method.encode() - else: - meth = method.value.encode() - uri_val = uri.encode(charset or 'utf-8') if isinstance(uri, str) else uri - - header_line = b'%s %s HTTP/%s\r\n' % (meth, uri_val, version) - elif method is None and status is not None: - # NOTE: ``status`` defaults to :data:`None`, so ``_make_index`` has no - # enum default to infer the namespace of a :obj:`str` member name from; - # name it here rather than leave the name unresolvable (#2356). - status_code = self._make_index( - status, status_default, - namespace=status_namespace if status_namespace is not None else Enum_StatusCode, - reversed=status_reversed, pack=False) - status_code_val = int(status_code) - - if message is None: - msg = getattr(status, 'message', None) or getattr(status_code, 'message', b'') or b'' - else: - msg = message.encode(charset or 'utf-8') if isinstance(message, str) else message - if isinstance(msg, str): - msg = msg.encode() - - # status-code is 3DIGIT (RFC 9112, section 4), so keep a leading zero - header_line = b'HTTP/%s %03d %s\r\n' % (version, status_code_val, msg) - else: - raise ProtocolError('HTTP: invalid start line: must have either a method or a status, ' - f'not method={method!r} and status={status!r}') - return header_line - - @classmethod - def id(cls) -> 'tuple[Literal["HTTP"], Literal["HTTPv1"]]': # type: ignore[override] - """Index ID of the protocol. - - Returns: - Index ID of the protocol. - - """ - return (cls.__name__, 'HTTPv1') # type: ignore[return-value] - ########################################################################## # Utilities. ########################################################################## @@ -620,3 +548,75 @@ def _read_http_body(self, body: 'bytes', *, """ return body + + def _make_start_line(self, http_version: 'str | bytes', *, + method: 'Optional[Enum_Method | str | bytes]' = None, + uri: 'Optional[str | bytes]' = None, + status: 'Optional[Enum_StatusCode | str | bytes | int]' = None, + status_default: 'Optional[int]' = None, + status_namespace: 'Optional[dict[str, int] | dict[int, str] | _Type[StdlibEnum] | _Type[AenumEnum]]' = None, # pylint: disable=line-too-long + status_reversed: 'bool' = False, + message: 'Optional[str | bytes]' = None, + charset: 'Optional[str]' = None) -> 'bytes': + """Make an HTTP/1.* start line from its values. + + Args: + http_version: HTTP version. + method: HTTP method. + uri: HTTP request URI. + status: HTTP status code. + status_default: Default HTTP status code. + status_namespace: Namespace of HTTP status code. + status_reversed: Whether to reverse the namespace. + message: HTTP status message. + charset: Encoding for a :obj:`str` request URI or status message, + **UTF-8** if not given. + + Returns: + The start line, CRLF included. + + Raises: + ProtocolError: If the values are neither a request's nor a + response's, or if they are a request's and ``uri`` is + :obj:`None`. + + """ + version = http_version.encode() if isinstance(http_version, str) else http_version + if method is not None and status is None: + if uri is None: + raise ProtocolError('HTTP: invalid start line: a request needs a URI') + + if isinstance(method, Enum_Method): + meth = method.value.encode() + elif isinstance(method, bytes): + meth = method + elif isinstance(method, str): + meth = method.encode() + else: + meth = method.value.encode() + uri_val = uri.encode(charset or 'utf-8') if isinstance(uri, str) else uri + + header_line = b'%s %s HTTP/%s\r\n' % (meth, uri_val, version) + elif method is None and status is not None: + # NOTE: ``status`` defaults to :data:`None`, so ``_make_index`` has no + # enum default to infer the namespace of a :obj:`str` member name from; + # name it here rather than leave the name unresolvable (#2356). + status_code = self._make_index( + status, status_default, + namespace=status_namespace if status_namespace is not None else Enum_StatusCode, + reversed=status_reversed, pack=False) + status_code_val = int(status_code) + + if message is None: + msg = getattr(status, 'message', None) or getattr(status_code, 'message', b'') or b'' + else: + msg = message.encode(charset or 'utf-8') if isinstance(message, str) else message + if isinstance(msg, str): + msg = msg.encode() + + # status-code is 3DIGIT (RFC 9112, section 4), so keep a leading zero + header_line = b'HTTP/%s %03d %s\r\n' % (version, status_code_val, msg) + else: + raise ProtocolError('HTTP: invalid start line: must have either a method or a status, ' + f'not method={method!r} and status={status!r}') + return header_line diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index b22bc0a192..5101e7078b 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -452,49 +452,6 @@ def pack(self, **kwargs: 'Any') -> 'bytes': self.__header__ = None # type: ignore[assignment] return data - @staticmethod - def _make_http_length(frame: 'bytes | Schema_FrameType', flags: 'Flags') -> 'int': - """Calculate HTTP/2 frame payload length for construction. - - Several frame payload schemas use conditional fields keyed by the - containing HTTP/2 header flags, so packing them without that context can - produce incorrect lengths or fail outright. - """ - if isinstance(frame, bytes): - return len(frame) - - flags_int = int(flags) - if isinstance(frame, Schema_UnassignedFrame): - return len(frame.data) - if isinstance(frame, Schema_DataFrame): - padded = flags_int & int(Schema_DataFrame.Flags.PADDED) - return len(frame.data) + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_HeadersFrame): - padded = flags_int & int(Schema_HeadersFrame.Flags.PADDED) - priority_len = 5 if flags_int & int(Schema_HeadersFrame.Flags.PRIORITY) else 0 - return len(frame.fragment) + priority_len + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_PriorityFrame): - return 5 - if isinstance(frame, Schema_RSTStreamFrame): - return 4 - if isinstance(frame, Schema_SettingsFrame): - settings = cast('object', frame.settings) - if isinstance(settings, bytes): - return len(settings) - return len(frame.settings) * 6 - if isinstance(frame, Schema_PushPromiseFrame): - padded = flags_int & int(Schema_PushPromiseFrame.Flags.PADDED) - return len(frame.fragment) + 4 + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_PingFrame): - return 8 - if isinstance(frame, Schema_GoawayFrame): - return len(frame.debug) + 8 - if isinstance(frame, Schema_WindowUpdateFrame): - return 4 - if isinstance(frame, Schema_ContinuationFrame): - return len(frame.fragment) - return len(frame.pack()) - @classmethod def id(cls) -> 'tuple[Literal["HTTP"], Literal["HTTPv2"]]': # type: ignore[override] """Index ID of the protocol. @@ -1106,6 +1063,49 @@ def _read_http_continuation(self, schema: 'Schema_ContinuationFrame', *, ) return data + @staticmethod + def _make_http_length(frame: 'bytes | Schema_FrameType', flags: 'Flags') -> 'int': + """Calculate HTTP/2 frame payload length for construction. + + Several frame payload schemas use conditional fields keyed by the + containing HTTP/2 header flags, so packing them without that context can + produce incorrect lengths or fail outright. + """ + if isinstance(frame, bytes): + return len(frame) + + flags_int = int(flags) + if isinstance(frame, Schema_UnassignedFrame): + return len(frame.data) + if isinstance(frame, Schema_DataFrame): + padded = flags_int & int(Schema_DataFrame.Flags.PADDED) + return len(frame.data) + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_HeadersFrame): + padded = flags_int & int(Schema_HeadersFrame.Flags.PADDED) + priority_len = 5 if flags_int & int(Schema_HeadersFrame.Flags.PRIORITY) else 0 + return len(frame.fragment) + priority_len + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_PriorityFrame): + return 5 + if isinstance(frame, Schema_RSTStreamFrame): + return 4 + if isinstance(frame, Schema_SettingsFrame): + settings = cast('object', frame.settings) + if isinstance(settings, bytes): + return len(settings) + return len(frame.settings) * 6 + if isinstance(frame, Schema_PushPromiseFrame): + padded = flags_int & int(Schema_PushPromiseFrame.Flags.PADDED) + return len(frame.fragment) + 4 + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_PingFrame): + return 8 + if isinstance(frame, Schema_GoawayFrame): + return len(frame.debug) + 8 + if isinstance(frame, Schema_WindowUpdateFrame): + return 4 + if isinstance(frame, Schema_ContinuationFrame): + return len(frame.fragment) + return len(frame.pack()) + def _make_http_none(self, frame: 'Optional[Data_UnassignedFrame]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'tuple[Schema_UnassignedFrame, Flags]': diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index f1cda117b7..ef89a4ee6c 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -379,6 +379,37 @@ def _read_id_numbers(self, id: 'bytes') -> 'IPv4Address': #_addr = '.'.join(str(_) for _ in _byte) return ipaddress.ip_address(id) # type: ignore[return-value] + def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'Data_CryptographicAuthentication': + """Read Authentication field when Cryptographic Authentication is employed, + i.e. :attr:`~pcapkit.protocols.data.application.ospf.OSPF.autype` is ``2``. + + Structure of Cryptographic Authentication [:rfc:`2328`]: + + .. code-block:: text + + 0 1 2 3 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | 0 | Key ID | Auth Data Len | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Cryptographic sequence number | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: parsed authentication data + + Returns: + Parsed packet data. + + """ + auth = Data_CryptographicAuthentication( + reserved=schema.reserved, + key_id=schema.key_id, + len=schema.len, + seq=schema.seq, + ) + return auth + def _make_id_numbers(self, id: 'IPv4Address | str | bytes | bytearray') -> 'bytes': """Make router and area IDs. @@ -412,37 +443,6 @@ def _make_id_numbers(self, id: 'IPv4Address | str | bytes | bytearray') -> 'byte return cast('IPv4Address', parse_ip_address( id, f'{self.__class__.__name__}: invalid ID', version=4)).packed - def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'Data_CryptographicAuthentication': - """Read Authentication field when Cryptographic Authentication is employed, - i.e. :attr:`~pcapkit.protocols.data.application.ospf.OSPF.autype` is ``2``. - - Structure of Cryptographic Authentication [:rfc:`2328`]: - - .. code-block:: text - - 0 1 2 3 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | 0 | Key ID | Auth Data Len | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Cryptographic sequence number | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - - Args: - schema: parsed authentication data - - Returns: - Parsed packet data. - - """ - auth = Data_CryptographicAuthentication( - reserved=schema.reserved, - key_id=schema.key_id, - len=schema.len, - seq=schema.seq, - ) - return auth - def _make_encrypt_auth(self, auth_data: 'bytes | Schema_CryptographicAuthentication | Data_CryptographicAuthentication | dict[str, Any]' # pylint: disable=line-too-long ) -> 'Schema_CryptographicAuthentication': diff --git a/pcapkit/protocols/application/ssl.py b/pcapkit/protocols/application/ssl.py index bf82524d3e..17bb980ab8 100644 --- a/pcapkit/protocols/application/ssl.py +++ b/pcapkit/protocols/application/ssl.py @@ -372,30 +372,6 @@ def _read_tls_content(self, content_type: 'Enum_ContentType', fragment: 'bytes', return Data_ChangeCipherSpec(msg=fragment[0]) return fragment - def _make_tls_content(self, content_type: 'Enum_ContentType', content: 'Any', *, - dtls: 'bool') -> 'bytes': - """Encode a record fragment by its content type -- the inverse of - :meth:`_read_tls_content`. - - Args: - content_type: Record content type. - content: Decoded record content, or already-encoded :obj:`bytes`. - dtls: Whether this is a DTLS record. - - Returns: - Encoded record fragment. - - """ - if isinstance(content, bytes): - return content - 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)]) - if content_type == Enum_ContentType.change_cipher_spec: - return bytes([int(content.msg)]) - return bytes(content) - def _read_tls_handshake_messages(self, fragment: 'bytes', *, dtls: 'bool') -> 'tuple[Any, ...]': """Decode a ``handshake`` record fragment into its handshake messages. @@ -470,50 +446,6 @@ def _read_tls_handshake_messages(self, fragment: 'bytes', *, messages.append(Data_Handshake(msg_type=msg_type, length=length, body=body)) return tuple(messages) - def _make_tls_handshake_messages(self, messages: 'tuple[Any, ...]', *, - dtls: 'bool') -> 'bytes': - """Encode handshake messages back into a record fragment -- the - inverse of :meth:`_read_tls_handshake_messages`. - - Args: - messages: Handshake messages, as returned by - :meth:`_read_tls_handshake_messages`, or already-encoded - :obj:`bytes` per message. - dtls: Whether to write DTLS's fragment header after each - message's ``msg_type``/``length`` header. - - Returns: - Encoded record fragment. - - """ - out = bytearray() - for message in messages: - if isinstance(message, bytes): - out += message - continue - - body = message.body - if isinstance(body, Data_ClientHello): - body_bytes = self._make_tls_client_hello(body, dtls=dtls) - elif isinstance(body, Data_ServerHello): - body_bytes = self._make_tls_server_hello(body) - else: - body_bytes = bytes(body) - - # NOTE: the message's ``length`` is trusted only for a fragmented DTLS message; - # otherwise it is recomputed from ``body_bytes``, so a rebuild stays exact. - is_fragmented = dtls and message.is_fragmented - length = message.length if is_fragmented else len(body_bytes) - - out += bytes([int(message.msg_type)]) + length.to_bytes(3, 'big') - if dtls: - fragment_length = message.fragment_length if is_fragmented else len(body_bytes) - out += (message.message_seq.to_bytes(2, 'big') - + message.fragment_offset.to_bytes(3, 'big') - + fragment_length.to_bytes(3, 'big')) - out += body_bytes - return bytes(out) - def _read_tls_client_hello(self, body: 'bytes', *, dtls: 'bool') -> 'Data_ClientHello': """Decode a ``ClientHello`` handshake message body. @@ -618,39 +550,6 @@ def _read_tls_client_hello(self, body: 'bytes', *, dtls: 'bool') -> 'Data_Client extensions=extensions, ) - def _make_tls_client_hello(self, data: 'Data_ClientHello', *, dtls: 'bool') -> 'bytes': - """Encode a ``ClientHello`` handshake message body -- the inverse of - :meth:`_read_tls_client_hello`. - - Args: - data: Decoded ``ClientHello``. - dtls: Whether this is a DTLS ``ClientHello``, i.e. whether to - write the ``cookie`` field. - - Returns: - Encoded message body. - - """ - out = bytearray() - out += int(data.version).to_bytes(2, 'big') - out += data.random - out += bytes([len(data.session_id)]) + data.session_id - - 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 - - compression_methods = bytes(int(method) for method in data.compression_methods) - out += bytes([len(compression_methods)]) + compression_methods - - if data.extensions is not None: - extensions_raw = self._make_tls_extensions(data.extensions) - out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw - return bytes(out) - def _read_tls_server_hello(self, body: 'bytes') -> 'Data_ServerHello': """Decode a ``ServerHello`` handshake message body. @@ -709,29 +608,6 @@ def _read_tls_server_hello(self, body: 'bytes') -> 'Data_ServerHello': compression_method=compression_method, extensions=extensions, ) - def _make_tls_server_hello(self, data: 'Data_ServerHello') -> 'bytes': - """Encode a ``ServerHello`` handshake message body -- the inverse of - :meth:`_read_tls_server_hello`. - - Args: - data: Decoded ``ServerHello``. - - Returns: - Encoded message body. - - """ - out = bytearray() - out += int(data.version).to_bytes(2, 'big') - 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)]) - - if data.extensions is not None: - extensions_raw = self._make_tls_extensions(data.extensions) - out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw - return bytes(out) - def _read_tls_extensions(self, data: 'bytes') -> 'tuple[Data_Extension, ...]': """Decode an extension list -- TYPE and LENGTH framing only. @@ -771,6 +647,130 @@ def _read_tls_extensions(self, data: 'bytes') -> 'tuple[Data_Extension, ...]': extensions.append(Data_Extension(type=ext_type, data=ext_data)) return tuple(extensions) + def _make_tls_content(self, content_type: 'Enum_ContentType', content: 'Any', *, + dtls: 'bool') -> 'bytes': + """Encode a record fragment by its content type -- the inverse of + :meth:`_read_tls_content`. + + Args: + content_type: Record content type. + content: Decoded record content, or already-encoded :obj:`bytes`. + dtls: Whether this is a DTLS record. + + Returns: + Encoded record fragment. + + """ + if isinstance(content, bytes): + return content + 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)]) + if content_type == Enum_ContentType.change_cipher_spec: + return bytes([int(content.msg)]) + return bytes(content) + + def _make_tls_handshake_messages(self, messages: 'tuple[Any, ...]', *, + dtls: 'bool') -> 'bytes': + """Encode handshake messages back into a record fragment -- the + inverse of :meth:`_read_tls_handshake_messages`. + + Args: + messages: Handshake messages, as returned by + :meth:`_read_tls_handshake_messages`, or already-encoded + :obj:`bytes` per message. + dtls: Whether to write DTLS's fragment header after each + message's ``msg_type``/``length`` header. + + Returns: + Encoded record fragment. + + """ + out = bytearray() + for message in messages: + if isinstance(message, bytes): + out += message + continue + + body = message.body + if isinstance(body, Data_ClientHello): + body_bytes = self._make_tls_client_hello(body, dtls=dtls) + elif isinstance(body, Data_ServerHello): + body_bytes = self._make_tls_server_hello(body) + else: + body_bytes = bytes(body) + + # NOTE: the message's ``length`` is trusted only for a fragmented DTLS message; + # otherwise it is recomputed from ``body_bytes``, so a rebuild stays exact. + is_fragmented = dtls and message.is_fragmented + length = message.length if is_fragmented else len(body_bytes) + + out += bytes([int(message.msg_type)]) + length.to_bytes(3, 'big') + if dtls: + fragment_length = message.fragment_length if is_fragmented else len(body_bytes) + out += (message.message_seq.to_bytes(2, 'big') + + message.fragment_offset.to_bytes(3, 'big') + + fragment_length.to_bytes(3, 'big')) + out += body_bytes + return bytes(out) + + def _make_tls_client_hello(self, data: 'Data_ClientHello', *, dtls: 'bool') -> 'bytes': + """Encode a ``ClientHello`` handshake message body -- the inverse of + :meth:`_read_tls_client_hello`. + + Args: + data: Decoded ``ClientHello``. + dtls: Whether this is a DTLS ``ClientHello``, i.e. whether to + write the ``cookie`` field. + + Returns: + Encoded message body. + + """ + out = bytearray() + out += int(data.version).to_bytes(2, 'big') + out += data.random + out += bytes([len(data.session_id)]) + data.session_id + + 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 + + compression_methods = bytes(int(method) for method in data.compression_methods) + out += bytes([len(compression_methods)]) + compression_methods + + if data.extensions is not None: + extensions_raw = self._make_tls_extensions(data.extensions) + out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw + return bytes(out) + + def _make_tls_server_hello(self, data: 'Data_ServerHello') -> 'bytes': + """Encode a ``ServerHello`` handshake message body -- the inverse of + :meth:`_read_tls_server_hello`. + + Args: + data: Decoded ``ServerHello``. + + Returns: + Encoded message body. + + """ + out = bytearray() + out += int(data.version).to_bytes(2, 'big') + 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)]) + + if data.extensions is not None: + extensions_raw = self._make_tls_extensions(data.extensions) + out += len(extensions_raw).to_bytes(2, 'big') + extensions_raw + return bytes(out) + def _make_tls_extensions(self, extensions: 'tuple[Data_Extension, ...]') -> 'bytes': """Encode an extension list -- the inverse of :meth:`_read_tls_extensions`. diff --git a/pcapkit/protocols/data/application/httpv2.py b/pcapkit/protocols/data/application/httpv2.py index 8bcbd2ddfe..06a56d7da7 100644 --- a/pcapkit/protocols/data/application/httpv2.py +++ b/pcapkit/protocols/data/application/httpv2.py @@ -187,11 +187,6 @@ class Settings(OrderedMultiInfo['Setting', int]): """ - def __getitem__(self, key: 'Setting') -> 'int': # type: ignore[override] - if key in self: - return self.getlist(key)[-1] - return super().__getitem__(key) # pylint: disable=no-value-for-parameter - def items(self, multi: 'bool' = False) -> 'Iterator[tuple[Setting, int]]': # type: ignore[override] if multi: yield from super().items(multi=True) # pylint: disable=no-value-for-parameter @@ -199,6 +194,11 @@ def items(self, multi: 'bool' = False) -> 'Iterator[tuple[Setting, int]]': # ty for key, _ in super().items(): # pylint: disable=no-value-for-parameter yield key, self[key] + def __getitem__(self, key: 'Setting') -> 'int': # type: ignore[override] + if key in self: + return self.getlist(key)[-1] + return super().__getitem__(key) # pylint: disable=no-value-for-parameter + @info_final class SettingsFrameFlags(Flags): From fe6ea7f7d9cc9bc859082bbc5ed8cc3e772ec4f8 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 31 Oct 2026 03:19:44 -0400 Subject: [PATCH 2/2] refactor(protocols): move HTTP private helpers into Utilities - httpv1: move _check_raw out of Methods into Utilities, before _read_field_line, since make() calls it. - httpv1: read _read_http_body before _split_field_lines, the order read() first uses them. - httpv1: keep _read_start_line and _make_start_line together, reader first. - httpv2: move _make_http_length, called directly by make(), ahead of the frame readers in Utilities. Pure reorder: no bodies, names or behaviour changed. Refs #2490 --- pcapkit/protocols/application/httpv1.py | 148 ++++++++++++------------ pcapkit/protocols/application/httpv2.py | 86 +++++++------- 2 files changed, 117 insertions(+), 117 deletions(-) diff --git a/pcapkit/protocols/application/httpv1.py b/pcapkit/protocols/application/httpv1.py index 5c3922991a..1f9e895d24 100644 --- a/pcapkit/protocols/application/httpv1.py +++ b/pcapkit/protocols/application/httpv1.py @@ -287,30 +287,6 @@ def id(cls) -> 'tuple[Literal["HTTP"], Literal["HTTPv1"]]': # type: ignore[over """ return (cls.__name__, 'HTTPv1') # type: ignore[return-value] - @staticmethod - def _check_raw(raw: 'bytes', *values: 'str', fold: 'bool') -> 'None': - """Refuse a raw line that would write a line break its values lack. - - Reading a line back strips the whitespace around its tokens, and that - includes a lone CR or LF, so a raw line can read back as values that - carry no line break while writing one to the wire. - - Args: - raw: Raw start line or field line, CRLF excluded. - *values: The values ``raw`` reads back as. - fold: Whether the CRLF of an ``obs-fold``, followed by SP or HTAB, - is allowed in ``raw``. - - Raises: - ProtocolError: If ``raw`` carries a CR or LF that ``values`` do not. - - """ - if fold: - raw = re.sub(rb'\r\n(?=[ \t])', b'', raw) - for char in ('\r', '\n'): - if raw.count(char.encode()) != sum(value.count(char) for value in values): - raise ProtocolError('HTTP: raw line carries a CR or LF its values do not') - ########################################################################## # Utilities. ########################################################################## @@ -409,6 +385,20 @@ def _read_http_header(self, header: 'bytes') -> 'tuple[Data_Header, OrderedMulti return header_line, header_fields + def _read_http_body(self, body: 'bytes', *, + headers: 'OrderedMultiDict[str, str]') -> 'Any': + """Read HTTP/1.* body. + + Args: + body: HTTP body data. + headers: HTTP header fields. + + Returns: + Raw HTTP body. + + """ + return body + @staticmethod def _split_field_lines(headerfield: 'bytes') -> 'tuple[bytes, ...]': """Split the field lines of an HTTP/1.* header. @@ -445,42 +435,6 @@ def _split_field_lines(headerfield: 'bytes') -> 'tuple[bytes, ...]': raw_fields.append(line) return tuple(raw_fields) - def _read_field_line(self, line: 'bytes') -> 'tuple[str, str]': - """Read one HTTP/1.* field line. - - Args: - line: Field line as received, CRLF excluded, with any ``obs-fold`` - continuations still in it. - - Returns: - The field name and the field value, the latter unfolded and without - the optional whitespace around it. - - Raises: - ProtocolError: If the line is not a field line. - - """ - # NOTE: The accumulator is right-stripped as well as the continuation, - # because the production is ``obs-fold = OWS CRLF RWS`` and it is the - # *whole* obs-fold that is replaced by a single space -- the RFC's own - # remedy (:rfc:`9112#section-5.2`). The OWS before the CRLF belongs to - # the fold, not to the value, so ``X: a \t\r\n\tb`` unfolds to ``'a b'``. - field, *continuations = line.split(b'\r\n') - for continuation in continuations: - if not continuation.startswith((b' ', b'\t')): - raise ProtocolError('HTTP: invalid format') - field = field.rstrip() + b' ' + continuation.strip() - - # NOTE: Checked rather than left to ``item[1]``, and refused rather than - # skipped: a field line with no colon is not a header field, and dropping - # it would hand back a message whose fields are quietly not the ones on - # the wire. ``ProtocolError`` for the reason ``read`` gives, and to the - # same message. - item = re.split(rb'\s*:\s*', field, maxsplit=1) - if len(item) != 2: - raise ProtocolError('HTTP: invalid format') - return self.decode(item[0].strip()), self.decode(item[1].strip()) - def _read_start_line(self, startline: 'bytes') -> 'Data_Header': """Read an HTTP/1.* start line. @@ -535,20 +489,6 @@ def _read_start_line(self, startline: 'bytes') -> 'Data_Header': raise ProtocolError('HTTP: invalid format') return header_line - def _read_http_body(self, body: 'bytes', *, - headers: 'OrderedMultiDict[str, str]') -> 'Any': - """Read HTTP/1.* body. - - Args: - body: HTTP body data. - headers: HTTP header fields. - - Returns: - Raw HTTP body. - - """ - return body - def _make_start_line(self, http_version: 'str | bytes', *, method: 'Optional[Enum_Method | str | bytes]' = None, uri: 'Optional[str | bytes]' = None, @@ -620,3 +560,63 @@ def _make_start_line(self, http_version: 'str | bytes', *, raise ProtocolError('HTTP: invalid start line: must have either a method or a status, ' f'not method={method!r} and status={status!r}') return header_line + + @staticmethod + def _check_raw(raw: 'bytes', *values: 'str', fold: 'bool') -> 'None': + """Refuse a raw line that would write a line break its values lack. + + Reading a line back strips the whitespace around its tokens, and that + includes a lone CR or LF, so a raw line can read back as values that + carry no line break while writing one to the wire. + + Args: + raw: Raw start line or field line, CRLF excluded. + *values: The values ``raw`` reads back as. + fold: Whether the CRLF of an ``obs-fold``, followed by SP or HTAB, + is allowed in ``raw``. + + Raises: + ProtocolError: If ``raw`` carries a CR or LF that ``values`` do not. + + """ + if fold: + raw = re.sub(rb'\r\n(?=[ \t])', b'', raw) + for char in ('\r', '\n'): + if raw.count(char.encode()) != sum(value.count(char) for value in values): + raise ProtocolError('HTTP: raw line carries a CR or LF its values do not') + + def _read_field_line(self, line: 'bytes') -> 'tuple[str, str]': + """Read one HTTP/1.* field line. + + Args: + line: Field line as received, CRLF excluded, with any ``obs-fold`` + continuations still in it. + + Returns: + The field name and the field value, the latter unfolded and without + the optional whitespace around it. + + Raises: + ProtocolError: If the line is not a field line. + + """ + # NOTE: The accumulator is right-stripped as well as the continuation, + # because the production is ``obs-fold = OWS CRLF RWS`` and it is the + # *whole* obs-fold that is replaced by a single space -- the RFC's own + # remedy (:rfc:`9112#section-5.2`). The OWS before the CRLF belongs to + # the fold, not to the value, so ``X: a \t\r\n\tb`` unfolds to ``'a b'``. + field, *continuations = line.split(b'\r\n') + for continuation in continuations: + if not continuation.startswith((b' ', b'\t')): + raise ProtocolError('HTTP: invalid format') + field = field.rstrip() + b' ' + continuation.strip() + + # NOTE: Checked rather than left to ``item[1]``, and refused rather than + # skipped: a field line with no colon is not a header field, and dropping + # it would hand back a message whose fields are quietly not the ones on + # the wire. ``ProtocolError`` for the reason ``read`` gives, and to the + # same message. + item = re.split(rb'\s*:\s*', field, maxsplit=1) + if len(item) != 2: + raise ProtocolError('HTTP: invalid format') + return self.decode(item[0].strip()), self.decode(item[1].strip()) diff --git a/pcapkit/protocols/application/httpv2.py b/pcapkit/protocols/application/httpv2.py index 5101e7078b..d9045cadc4 100644 --- a/pcapkit/protocols/application/httpv2.py +++ b/pcapkit/protocols/application/httpv2.py @@ -523,6 +523,49 @@ def _make_data(cls, data: 'Data_HTTP') -> 'dict[str, Any]': # type: ignore[over 'payload': cls._make_payload(data), } + @staticmethod + def _make_http_length(frame: 'bytes | Schema_FrameType', flags: 'Flags') -> 'int': + """Calculate HTTP/2 frame payload length for construction. + + Several frame payload schemas use conditional fields keyed by the + containing HTTP/2 header flags, so packing them without that context can + produce incorrect lengths or fail outright. + """ + if isinstance(frame, bytes): + return len(frame) + + flags_int = int(flags) + if isinstance(frame, Schema_UnassignedFrame): + return len(frame.data) + if isinstance(frame, Schema_DataFrame): + padded = flags_int & int(Schema_DataFrame.Flags.PADDED) + return len(frame.data) + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_HeadersFrame): + padded = flags_int & int(Schema_HeadersFrame.Flags.PADDED) + priority_len = 5 if flags_int & int(Schema_HeadersFrame.Flags.PRIORITY) else 0 + return len(frame.fragment) + priority_len + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_PriorityFrame): + return 5 + if isinstance(frame, Schema_RSTStreamFrame): + return 4 + if isinstance(frame, Schema_SettingsFrame): + settings = cast('object', frame.settings) + if isinstance(settings, bytes): + return len(settings) + return len(frame.settings) * 6 + if isinstance(frame, Schema_PushPromiseFrame): + padded = flags_int & int(Schema_PushPromiseFrame.Flags.PADDED) + return len(frame.fragment) + 4 + ((frame.pad_len or 0) + 1 if padded else 0) + if isinstance(frame, Schema_PingFrame): + return 8 + if isinstance(frame, Schema_GoawayFrame): + return len(frame.debug) + 8 + if isinstance(frame, Schema_WindowUpdateFrame): + return 4 + if isinstance(frame, Schema_ContinuationFrame): + return len(frame.fragment) + return len(frame.pack()) + @staticmethod def _read_http_flags(schema: 'Schema_FrameType') -> 'Data_Flags': """Read the flags of an HTTP/2 frame type that defines none. @@ -1063,49 +1106,6 @@ def _read_http_continuation(self, schema: 'Schema_ContinuationFrame', *, ) return data - @staticmethod - def _make_http_length(frame: 'bytes | Schema_FrameType', flags: 'Flags') -> 'int': - """Calculate HTTP/2 frame payload length for construction. - - Several frame payload schemas use conditional fields keyed by the - containing HTTP/2 header flags, so packing them without that context can - produce incorrect lengths or fail outright. - """ - if isinstance(frame, bytes): - return len(frame) - - flags_int = int(flags) - if isinstance(frame, Schema_UnassignedFrame): - return len(frame.data) - if isinstance(frame, Schema_DataFrame): - padded = flags_int & int(Schema_DataFrame.Flags.PADDED) - return len(frame.data) + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_HeadersFrame): - padded = flags_int & int(Schema_HeadersFrame.Flags.PADDED) - priority_len = 5 if flags_int & int(Schema_HeadersFrame.Flags.PRIORITY) else 0 - return len(frame.fragment) + priority_len + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_PriorityFrame): - return 5 - if isinstance(frame, Schema_RSTStreamFrame): - return 4 - if isinstance(frame, Schema_SettingsFrame): - settings = cast('object', frame.settings) - if isinstance(settings, bytes): - return len(settings) - return len(frame.settings) * 6 - if isinstance(frame, Schema_PushPromiseFrame): - padded = flags_int & int(Schema_PushPromiseFrame.Flags.PADDED) - return len(frame.fragment) + 4 + ((frame.pad_len or 0) + 1 if padded else 0) - if isinstance(frame, Schema_PingFrame): - return 8 - if isinstance(frame, Schema_GoawayFrame): - return len(frame.debug) + 8 - if isinstance(frame, Schema_WindowUpdateFrame): - return 4 - if isinstance(frame, Schema_ContinuationFrame): - return len(frame.fragment) - return len(frame.pack()) - def _make_http_none(self, frame: 'Optional[Data_UnassignedFrame]' = None, *, data: 'bytes' = b'', **kwargs: 'Any') -> 'tuple[Schema_UnassignedFrame, Flags]':