|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""OSPF and NGAP rebuild what they parsed, or refuse it. |
| 3 | +
|
| 4 | +GitHub issues: |
| 5 | +
|
| 6 | +* #1249: :meth:`OSPF.make <pcapkit.protocols.application.ospf.OSPF.make>` |
| 7 | + recomputed Packet Length from the payload and ``_make_data`` dropped the |
| 8 | + parsed one, so the 16-octet digest that cryptographic authentication appends |
| 9 | + outside the packet (:rfc:`2328#appendix-D.4.3`) was counted in it. |
| 10 | +* #1250: ``OSPF(auth_type=2, auth_data=<bytes>)`` raised |
| 11 | + :exc:`AttributeError`. |
| 12 | +* #1251 (OSPF half): an ``auth_data`` that is not 8 octets was padded or cut. |
| 13 | +* #1252: octets after the aligned PER ``NGAP-PDU`` were dropped on rebuild. |
| 14 | +
|
| 15 | +Every case builds its own octets in memory and reads no capture. Classes are |
| 16 | +imported inside each test, after :func:`~tests._support.reimport_once_per_class`, |
| 17 | +c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``. |
| 18 | +
|
| 19 | +""" |
| 20 | + |
| 21 | +import importlib.util |
| 22 | +import unittest |
| 23 | + |
| 24 | +from tests._support import reimport_once_per_class |
| 25 | + |
| 26 | +HAS_PYCRATE = importlib.util.find_spec('pycrate_asn1dir') is not None |
| 27 | + |
| 28 | +#: Cryptographic authentication: Key ID 1, Auth Data Len 16, sequence 7. |
| 29 | +CRYPTO_AUTH = bytes.fromhex('0000011000000007') |
| 30 | +#: A 20-octet Hello body. |
| 31 | +HELLO_BODY = bytes.fromhex('ffffff00000a020100000028c0a8000100000000') |
| 32 | + |
| 33 | + |
| 34 | +def ospf_header(length: int, auth_type: int, auth: bytes) -> bytes: |
| 35 | + """OSPFv2 Hello header with the given Packet Length and authentication.""" |
| 36 | + return (bytes.fromhex('0201') + length.to_bytes(2, 'big') |
| 37 | + + bytes.fromhex('c0a8000100000000' '0000') |
| 38 | + + auth_type.to_bytes(2, 'big') + auth) |
| 39 | + |
| 40 | + |
| 41 | +#: Packets whose Packet Length is not ``24 + len(payload)``. |
| 42 | +OSPF_PACKETS = { |
| 43 | + 'crypto auth, digest outside Length': ospf_header(0x2c, 2, CRYPTO_AUTH) + HELLO_BODY + bytes(range(16)), |
| 44 | + 'Length zero': ospf_header(0x00, 0, bytes(8)) + HELLO_BODY, |
| 45 | + 'Length short of the payload': ospf_header(0x1e, 0, bytes(8)) + HELLO_BODY, |
| 46 | + 'Length matching the payload': ospf_header(0x2c, 0, bytes(8)) + HELLO_BODY, |
| 47 | +} |
| 48 | + |
| 49 | +#: ``NGSetupRequest`` in aligned PER, c.f. ``test_ngap_unit.NGSETUP_REQUEST``. |
| 50 | +NGSETUP_REQUEST = bytes.fromhex( |
| 51 | + '00150036000004001b00080002f839100001020052400d0500706361706b69742d676e62' |
| 52 | + '0066000d00000000010002f839000000080015400140' |
| 53 | +) |
| 54 | + |
| 55 | + |
| 56 | +class TestOSPFRoundTrip(unittest.TestCase): |
| 57 | + """Pin Packet Length and the 8-octet Authentication field.""" |
| 58 | + |
| 59 | + def setUp(self) -> None: |
| 60 | + reimport_once_per_class(self) |
| 61 | + |
| 62 | + def test_from_data_keeps_parsed_packet_length(self) -> None: |
| 63 | + from pcapkit.protocols.application.ospf import OSPF |
| 64 | + |
| 65 | + for name, packet in OSPF_PACKETS.items(): |
| 66 | + with self.subTest(packet=name): |
| 67 | + parsed = OSPF(packet, len(packet)) |
| 68 | + self.assertEqual(bytes(OSPF.from_data(parsed.info)), packet) |
| 69 | + |
| 70 | + def test_make_packet_length_default_and_explicit(self) -> None: |
| 71 | + from pcapkit.protocols.application.ospf import OSPF |
| 72 | + |
| 73 | + self.assertEqual(bytes(OSPF(payload=HELLO_BODY))[2:4], b'\x00\x2c') |
| 74 | + self.assertEqual(bytes(OSPF(packet_length=0x1e, payload=HELLO_BODY))[2:4], b'\x00\x1e') |
| 75 | + |
| 76 | + def test_from_data_without_len_computes_packet_length(self) -> None: |
| 77 | + from ipaddress import ip_address |
| 78 | + |
| 79 | + from pcapkit.protocols.application.ospf import OSPF |
| 80 | + from pcapkit.protocols.data.application.ospf import OSPF as Data_OSPF |
| 81 | + |
| 82 | + data = Data_OSPF.from_dict({ |
| 83 | + 'version': 2, 'type': 1, 'router_id': ip_address('192.168.0.1'), |
| 84 | + 'area_id': ip_address('0.0.0.0'), 'chksum': b'\x00\x00', 'autype': 0, |
| 85 | + 'auth': bytes(8), |
| 86 | + }) |
| 87 | + self.assertNotIn('len', data) |
| 88 | + self.assertEqual(bytes(OSPF.from_data(data)), ospf_header(24, 0, bytes(8))) |
| 89 | + |
| 90 | + def test_make_rejects_invalid_packet_length(self) -> None: |
| 91 | + from pcapkit.protocols.application.ospf import OSPF |
| 92 | + from pcapkit.utilities.exceptions import ProtocolError |
| 93 | + |
| 94 | + for packet_length in (-1, 0x10000, True, 24.0, '24'): |
| 95 | + with self.subTest(packet_length=packet_length): |
| 96 | + with self.assertRaises(ProtocolError) as caught: |
| 97 | + OSPF(packet_length=packet_length) |
| 98 | + self.assertIn('invalid packet length', str(caught.exception)) |
| 99 | + self.assertEqual(bytes(OSPF(packet_length=0xFFFF))[2:4], b'\xff\xff') |
| 100 | + |
| 101 | + def test_make_crypto_auth_from_bytes(self) -> None: |
| 102 | + from pcapkit.protocols.application.ospf import OSPF |
| 103 | + |
| 104 | + packet = bytes(OSPF(auth_type=2, auth_data=CRYPTO_AUTH)) |
| 105 | + self.assertEqual(packet[16:24], CRYPTO_AUTH) |
| 106 | + |
| 107 | + info = OSPF(packet, len(packet)).info |
| 108 | + self.assertEqual((info.auth.key_id, info.auth.len, info.auth.seq), (1, 16, 7)) |
| 109 | + self.assertEqual(bytes(OSPF.from_data(info)), packet) |
| 110 | + |
| 111 | + def test_make_rejects_wrong_width_auth_data(self) -> None: |
| 112 | + from pcapkit.protocols.application.ospf import OSPF |
| 113 | + from pcapkit.utilities.exceptions import ProtocolError |
| 114 | + |
| 115 | + for auth_type in (0, 1, 2): |
| 116 | + for auth_data in (b'abc', b'0123456789', b''): |
| 117 | + with self.subTest(auth_type=auth_type, auth_data=auth_data): |
| 118 | + with self.assertRaises(ProtocolError) as caught: |
| 119 | + OSPF(auth_type=auth_type, auth_data=auth_data) |
| 120 | + self.assertIn('8 octets', str(caught.exception)) |
| 121 | + |
| 122 | + |
| 123 | +@unittest.skipUnless(HAS_PYCRATE, 'pycrate not installed') |
| 124 | +class TestNGAPTrailingOctets(unittest.TestCase): |
| 125 | + """Octets after the ``NGAP-PDU`` are refused rather than dropped.""" |
| 126 | + |
| 127 | + def setUp(self) -> None: |
| 128 | + reimport_once_per_class(self) |
| 129 | + |
| 130 | + def test_exact_pdu_rebuilds_byte_for_byte(self) -> None: |
| 131 | + from pcapkit.protocols.application.ngap import NGAP |
| 132 | + |
| 133 | + parsed = NGAP(NGSETUP_REQUEST, len(NGSETUP_REQUEST)) |
| 134 | + self.assertEqual(bytes(NGAP.from_data(parsed.info)), NGSETUP_REQUEST) |
| 135 | + |
| 136 | + def test_trailing_octets_raise_protocol_error(self) -> None: |
| 137 | + from pcapkit.protocols.application.ngap import NGAP |
| 138 | + from pcapkit.utilities.exceptions import ProtocolError |
| 139 | + |
| 140 | + for suffix in (b'\x00', b'\xff\xff', b'\xde\xad\xbe\xef'): |
| 141 | + with self.subTest(suffix=suffix): |
| 142 | + packet = NGSETUP_REQUEST + suffix |
| 143 | + with self.assertRaises(ProtocolError) as caught: |
| 144 | + NGAP(packet, len(packet)) |
| 145 | + self.assertIn(f'{len(suffix)} octet(s) after the NGAP-PDU', str(caught.exception)) |
| 146 | + |
| 147 | + |
| 148 | +if __name__ == '__main__': |
| 149 | + unittest.main() |
0 commit comments