fix(ospf,ngap): keep the parsed OSPF Packet Length and refuse octets after the NGAP-PDU

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.

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,

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')

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 <pcapkit.protocols.application.ospf.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=<bytes>)`` 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()

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')