From 17bf03c4343a75b17438a5fde35b7d77dcea8f4b Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 14:22:34 -0400 Subject: [PATCH] fix(ospf,ngap): keep the parsed OSPF Packet Length and refuse octets after the NGAP-PDU - OSPF: make() takes packet_length, and _make_data passes the parsed len, so the RFC 2328 D.4.3 digest outside Packet Length is no longer counted and a declared Length is not rewritten (#1249). Data without len falls back to 24 + len(payload); a packet_length that is not an int in 0..65535 raises ProtocolError. - OSPF: 8 raw octets are accepted as cryptographic auth_data and unpacked into the schema (#1250); auth_data of any other width raises ProtocolError instead of being padded or cut (#1251, OSPF half). - NGAP: octets after the aligned PER NGAP-PDU raise ProtocolError instead of being dropped (#1252). Closes #1249 Closes #1250 Closes #1252 Refs #1251 --- pcapkit/protocols/application/ngap.py | 29 +++- pcapkit/protocols/application/ospf.py | 38 ++++- tests/protocols/application/test_ngap_unit.py | 8 +- .../test_ospf_ngap_roundtrip_unit.py | 149 ++++++++++++++++++ tests/protocols/application/test_ospf_unit.py | 5 +- 5 files changed, 215 insertions(+), 14 deletions(-) create mode 100644 tests/protocols/application/test_ospf_ngap_roundtrip_unit.py diff --git a/pcapkit/protocols/application/ngap.py b/pcapkit/protocols/application/ngap.py index 823cbc6d1a..5153c37bce 100644 --- a/pcapkit/protocols/application/ngap.py +++ b/pcapkit/protocols/application/ngap.py @@ -396,8 +396,9 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_NGAP': Parsed packet data. Raises: - ProtocolError: If |pycrate|_ is not installed, or if the payload is - not a well formed aligned PER ``NGAP-PDU``. + ProtocolError: If |pycrate|_ is not installed, if the payload is + not a well formed aligned PER ``NGAP-PDU``, or if octets follow + the ``NGAP-PDU``. """ if length is None: @@ -417,7 +418,23 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_NGAP': # some 700x the ~0.15 ms of the decode itself; and `from_aper()` # overwrites the stored value outright, so it would buy nothing # even if it were free. - pdu.from_aper(schema.data) + from pycrate_core.charpy import Charpy # pylint: disable=import-outside-toplevel + + # NOTE: decoded from a `Charpy` rather than from the bytes, so + # that what the decoder did not consume can be measured: a bare + # `from_aper(bytes)` ignores a suffix. SCTP preserves message + # boundaries: each user message occupies its own DATA chunk + # (RFC 9260 §1.3), and one with both B and E set is a complete, + # unfragmented message (§3.3.1). The outermost APER encoding is + # padded with zero to seven zero bits up to the next octet + # (X.691 (02/2021) §11.1.4), and the result is the complete + # encoding (§11.1.5). So a whole octet left over is not part of + # the NGAP-PDU. + char = Charpy(schema.data) + pdu.from_aper(char) + trailing = char.len_byte() + if trailing: + raise ProtocolError(f'NGAP: {trailing} octet(s) after the NGAP-PDU') kind, body = pdu.get_val() message, message_body = body['value'] @@ -447,9 +464,9 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_NGAP': ) except ProtocolError: # NOTE: a ProtocolError raised inside this block is already - # specific about what went wrong, so it must not be caught below - # and relabelled "malformed NGAP-PDU". Nothing here raises one - # now; this keeps that true of the next edit. + # specific about what went wrong, such as the trailing-octet + # check above, so it must not be caught below and relabelled + # "malformed NGAP-PDU". raise except Exception as exc: # NOTE: `Exception` rather than something narrower on purpose. diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index 290d347de7..3344e0ec7f 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/pcapkit/protocols/application/ospf.py @@ -197,6 +197,7 @@ def make(self, type_default: 'Optional[int]' = None, type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long type_reversed: 'bool' = False, + packet_length: 'Optional[int]' = None, router_id: 'IPv4Address | str | bytes | bytearray' = '0.0.0.0', # nosec: B104 area_id: 'IPv4Address | str | bytes | bytearray' = '0.0.0.0', # nosec: B104 checksum: 'bytes' = b'\x00\x00', @@ -215,6 +216,12 @@ def make(self, type_default: Default value for ``type`` if not specified. type_namespace: Namespace for ``type``. type_reversed: Reverse namespace for ``type``. + packet_length: Packet length (header included). If not given, it is + computed as 24 plus the length of ``payload``. Give it + explicitly when ``payload`` carries octets outside the OSPF + packet, such as the message digest appended under + cryptographic authentication (:rfc:`2328#appendix-D.4.3`), + which Packet Length does not count. router_id: Router ID. area_id: Area ID. checksum: Checksum. @@ -229,23 +236,37 @@ def make(self, Returns: Constructed packet data. + Raises: + ProtocolError: If ``auth_data`` is of an invalid type, or is + :obj:`bytes` that are not exactly 8 octets long; or if + ``packet_length`` is not an :obj:`int` in ``0..65535``. + """ type_ = self._make_index(type, type_default, namespace=type_namespace, reversed=type_reversed, pack=False) auth_type_ = self._make_index(auth_type, auth_type_default, namespace=auth_type_namespace, reversed=auth_type_reversed, pack=False) + data: 'bytes | Schema_CryptographicAuthentication' if auth_type_ == Enum_Authentication.Cryptographic_authentication: data = self._make_encrypt_auth(auth_data) else: if not isinstance(auth_data, bytes): raise ProtocolError(f'OSPF: invalid type for authentication data: {auth_data!r}') + if len(auth_data) != 8: + raise ProtocolError(f'OSPF: authentication data must be 8 octets: {auth_data!r}') data = auth_data + 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}') + return Schema_OSPF( version=version, type=type_, # type: ignore[arg-type] - length=24 + len(payload), + length=packet_length, router_id=router_id, area_id=area_id, checksum=checksum, @@ -292,6 +313,7 @@ def _make_data(cls, data: 'Data_OSPF') -> 'dict[str, Any]': # type: ignore[over return { 'version': data.version, 'type': data.type, + 'packet_length': data.get('len'), 'router_id': data.router_id, 'area_id': data.area_id, 'checksum': data.chksum, @@ -378,18 +400,26 @@ def _read_encrypt_auth(self, schema: 'Schema_CryptographicAuthentication') -> 'D def _make_encrypt_auth(self, auth_data: 'bytes | Schema_CryptographicAuthentication | Data_CryptographicAuthentication' # pylint: disable=line-too-long - ) -> 'bytes | Schema_CryptographicAuthentication': + ) -> 'Schema_CryptographicAuthentication': """Make Authentication field when Cryptographic Authentication is employed. Args: auth_data: Authentication data. Returns: - Authentication bytes. + Authentication schema. + + Raises: + ProtocolError: If ``auth_data`` is of an invalid type, or is + :obj:`bytes` that are not exactly 8 octets long. """ - if isinstance(auth_data, (Schema_CryptographicAuthentication, bytes)): + if isinstance(auth_data, Schema_CryptographicAuthentication): return auth_data + if isinstance(auth_data, bytes): + if len(auth_data) != 8: + raise ProtocolError(f'OSPF: authentication data must be 8 octets: {auth_data!r}') + return Schema_CryptographicAuthentication.unpack(auth_data) # type: ignore[call-arg,misc] if isinstance(auth_data, Data_CryptographicAuthentication): return Schema_CryptographicAuthentication( key_id=auth_data.key_id, diff --git a/tests/protocols/application/test_ngap_unit.py b/tests/protocols/application/test_ngap_unit.py index 4301f7c264..c9dd508d08 100644 --- a/tests/protocols/application/test_ngap_unit.py +++ b/tests/protocols/application/test_ngap_unit.py @@ -475,9 +475,11 @@ def test_malformed_payloads_raise_protocol_error_not_pycrate_errors(self) -> Non except ProtocolError as exc: self.assertIn('NGAP:', str(exc)) else: - # Decoding a prefix or ignoring a suffix is legitimate PER - # behaviour; what matters is that it did not raise something - # raw. Assert it produced a usable model. + # Decoding a prefix is legitimate PER behaviour; what + # matters is that it did not raise something raw. A suffix + # is not: the chunk carries exactly one NGAP-PDU, so octets + # after it are refused rather than dropped (#1252). + self.assertNotEqual(label, 'trailing garbage') self.assertIsNotNone(info.message) @unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') diff --git a/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py b/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py new file mode 100644 index 0000000000..ee16b6e468 --- /dev/null +++ b/tests/protocols/application/test_ospf_ngap_roundtrip_unit.py @@ -0,0 +1,149 @@ +# -*- coding: utf-8 -*- +"""OSPF and NGAP rebuild what they parsed, or refuse it. + +GitHub issues: + +* #1249: :meth:`OSPF.make ` + recomputed Packet Length from the payload and ``_make_data`` dropped the + parsed one, so the 16-octet digest that cryptographic authentication appends + outside the packet (:rfc:`2328#appendix-D.4.3`) was counted in it. +* #1250: ``OSPF(auth_type=2, auth_data=)`` raised + :exc:`AttributeError`. +* #1251 (OSPF half): an ``auth_data`` that is not 8 octets was padded or cut. +* #1252: octets after the aligned PER ``NGAP-PDU`` were dropped on rebuild. + +Every case builds its own octets in memory and reads no capture. Classes are +imported inside each test, after :func:`~tests._support.reimport_once_per_class`, +c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``. + +""" + +import importlib.util +import unittest + +from tests._support import reimport_once_per_class + +HAS_PYCRATE = importlib.util.find_spec('pycrate_asn1dir') is not None + +#: Cryptographic authentication: Key ID 1, Auth Data Len 16, sequence 7. +CRYPTO_AUTH = bytes.fromhex('0000011000000007') +#: A 20-octet Hello body. +HELLO_BODY = bytes.fromhex('ffffff00000a020100000028c0a8000100000000') + + +def ospf_header(length: int, auth_type: int, auth: bytes) -> bytes: + """OSPFv2 Hello header with the given Packet Length and authentication.""" + return (bytes.fromhex('0201') + length.to_bytes(2, 'big') + + bytes.fromhex('c0a8000100000000' '0000') + + auth_type.to_bytes(2, 'big') + auth) + + +#: Packets whose Packet Length is not ``24 + len(payload)``. +OSPF_PACKETS = { + 'crypto auth, digest outside Length': ospf_header(0x2c, 2, CRYPTO_AUTH) + HELLO_BODY + bytes(range(16)), + 'Length zero': ospf_header(0x00, 0, bytes(8)) + HELLO_BODY, + 'Length short of the payload': ospf_header(0x1e, 0, bytes(8)) + HELLO_BODY, + 'Length matching the payload': ospf_header(0x2c, 0, bytes(8)) + HELLO_BODY, +} + +#: ``NGSetupRequest`` in aligned PER, c.f. ``test_ngap_unit.NGSETUP_REQUEST``. +NGSETUP_REQUEST = bytes.fromhex( + '00150036000004001b00080002f839100001020052400d0500706361706b69742d676e62' + '0066000d00000000010002f839000000080015400140' +) + + +class TestOSPFRoundTrip(unittest.TestCase): + """Pin Packet Length and the 8-octet Authentication field.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_from_data_keeps_parsed_packet_length(self) -> None: + from pcapkit.protocols.application.ospf import OSPF + + for name, packet in OSPF_PACKETS.items(): + with self.subTest(packet=name): + parsed = OSPF(packet, len(packet)) + self.assertEqual(bytes(OSPF.from_data(parsed.info)), packet) + + def test_make_packet_length_default_and_explicit(self) -> None: + from pcapkit.protocols.application.ospf import OSPF + + self.assertEqual(bytes(OSPF(payload=HELLO_BODY))[2:4], b'\x00\x2c') + self.assertEqual(bytes(OSPF(packet_length=0x1e, payload=HELLO_BODY))[2:4], b'\x00\x1e') + + def test_from_data_without_len_computes_packet_length(self) -> None: + from ipaddress import ip_address + + from pcapkit.protocols.application.ospf import OSPF + from pcapkit.protocols.data.application.ospf import OSPF as Data_OSPF + + data = Data_OSPF.from_dict({ + 'version': 2, 'type': 1, 'router_id': ip_address('192.168.0.1'), + 'area_id': ip_address('0.0.0.0'), 'chksum': b'\x00\x00', 'autype': 0, + 'auth': bytes(8), + }) + self.assertNotIn('len', data) + self.assertEqual(bytes(OSPF.from_data(data)), ospf_header(24, 0, bytes(8))) + + def test_make_rejects_invalid_packet_length(self) -> None: + from pcapkit.protocols.application.ospf import OSPF + from pcapkit.utilities.exceptions import ProtocolError + + for packet_length in (-1, 0x10000, True, 24.0, '24'): + 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(bytes(OSPF(packet_length=0xFFFF))[2:4], b'\xff\xff') + + def test_make_crypto_auth_from_bytes(self) -> None: + from pcapkit.protocols.application.ospf import OSPF + + packet = bytes(OSPF(auth_type=2, auth_data=CRYPTO_AUTH)) + self.assertEqual(packet[16:24], CRYPTO_AUTH) + + info = OSPF(packet, len(packet)).info + self.assertEqual((info.auth.key_id, info.auth.len, info.auth.seq), (1, 16, 7)) + self.assertEqual(bytes(OSPF.from_data(info)), packet) + + def test_make_rejects_wrong_width_auth_data(self) -> None: + from pcapkit.protocols.application.ospf import OSPF + from pcapkit.utilities.exceptions import ProtocolError + + for auth_type in (0, 1, 2): + for auth_data in (b'abc', b'0123456789', b''): + with self.subTest(auth_type=auth_type, auth_data=auth_data): + with self.assertRaises(ProtocolError) as caught: + OSPF(auth_type=auth_type, auth_data=auth_data) + self.assertIn('8 octets', str(caught.exception)) + + +@unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') +class TestNGAPTrailingOctets(unittest.TestCase): + """Octets after the ``NGAP-PDU`` are refused rather than dropped.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_exact_pdu_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + + parsed = NGAP(NGSETUP_REQUEST, len(NGSETUP_REQUEST)) + self.assertEqual(bytes(NGAP.from_data(parsed.info)), NGSETUP_REQUEST) + + def test_trailing_octets_raise_protocol_error(self) -> None: + from pcapkit.protocols.application.ngap import NGAP + from pcapkit.utilities.exceptions import ProtocolError + + for suffix in (b'\x00', b'\xff\xff', b'\xde\xad\xbe\xef'): + with self.subTest(suffix=suffix): + packet = NGSETUP_REQUEST + suffix + with self.assertRaises(ProtocolError) as caught: + NGAP(packet, len(packet)) + self.assertIn(f'{len(suffix)} octet(s) after the NGAP-PDU', str(caught.exception)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/application/test_ospf_unit.py b/tests/protocols/application/test_ospf_unit.py index 7ba668809e..111dcdac5c 100644 --- a/tests/protocols/application/test_ospf_unit.py +++ b/tests/protocols/application/test_ospf_unit.py @@ -141,7 +141,10 @@ def test_ospf_properties_read_make_and_auth_helpers(self) -> None: ) self.assertEqual(crypto_schema_from_data.auth_data.key_id, 2) self.assertIs(maker._make_encrypt_auth(crypto_schema), crypto_schema) - self.assertEqual(maker._make_encrypt_auth(b'\x02' * 8), b'\x02' * 8) + crypto_schema_from_bytes = maker._make_encrypt_auth(b'\x00\x00\x02\x14' + b'\x00\x00\x00\x64') + self.assertIsInstance(crypto_schema_from_bytes, SchemaCryptoAuth) + self.assertEqual((crypto_schema_from_bytes.key_id, crypto_schema_from_bytes.len, + crypto_schema_from_bytes.seq), (2, 20, 100)) self.assertEqual(maker._read_encrypt_auth(crypto_schema).len, 16) self.assertEqual(maker._read_id_numbers(b'\xc0\x00\x02\x04'), ip_address('192.0.2.4')) self.assertEqual(maker._make_id_numbers('192.0.2.5'), b'\xc0\x00\x02\x05')