From d77427d5913535bbe7c8dfa38a43fed3a105f364 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 28 Oct 2026 10:22:36 -0400 Subject: [PATCH] fix(internet): the ESP read path and integrity-key warning name the protocol (#2411) #2366 (fixing #2297) put the house `ESP: ` prefix on `ESP.make`'s ProtocolError messages and on `SecurityAssociation.unavailable`, but eight sites were outside its checklist. The same condition therefore read two ways depending on the direction it was hit from -- `make` said `ESP: no security association for SPI 0x...`, `read` said `no security association for SPI 0x...`. - esp.py `ESP._read_trailer` (a staticmethod, so the prefix is spelled literally, as every `SecurityAssociation` site does): the plaintext-too-short diagnostic, the pad-length-beyond-the-plaintext one, and the padding-pattern one, which is both returned and warned with. - esp.py `ESP.read` (an instance method, so `{self.alias}:`, as `ESP.make`): the NO_SA, TRUNCATED, separate-integrity AUTH_FAILED and AEAD-tag AUTH_FAILED messages, surfaced as `esp.error`. - esp.py `SecurityAssociation.__init__`: the short-integrity-key warning. `warn` reports the caller's frame rather than the protocol, so the message text is the only thing naming ESP -- it needs the prefix more than the exceptions do, not less. - test_tier_guard.py `DependencyGateFalsifiabilityTests .test_removing_crypto_from_the_test_job_is_caught`: the new module is the second to carry a `HAS_CRYPTO` gate, so it joins that test's expected set of gated modules and its `describe_gap` location assertions. #2324 dropped this test's `len(gap.gates)` pin and kept the set equality as the structural claim, so the fix is to extend the set rather than to loosen it. Two messages were reworded rather than only prefixed, to avoid `ESP: ESP payload ...` and `ESP: decrypted ESP payload ...`, and the three that read as ` for SPI x does/declares y` were turned into the `description: detail` house form. Every substring the existing tests match on is preserved. Test: tests/protocols/internet/test_esp_message_prefix_2411_unit.py (new, 13 tests, every message pinned whole rather than by substring -- matching fragments is why the omission survived #2366). Confirmed 10 failed / 3 passed against unpatched esp.py, each failure a missing prefix; the 3 that passed are the controls that assert the guards did not widen. All but one of the new tests runs without `cryptography`: an ENCR_NULL SA reaches the trailer and separate-integrity paths on hmac and hashlib alone. tests/test_tier_guard.py in full: 135 passed / 2497 subtests / 0 failed. tests/ in full, after `examples/generators/make_samples.py`: 6776 passed / 118 skipped / 101828 subtests / 0 failed. Fixes #2411. --- pcapkit/protocols/internet/esp.py | 37 ++- .../test_esp_message_prefix_2411_unit.py | 292 ++++++++++++++++++ tests/test_tier_guard.py | 9 +- 3 files changed, 320 insertions(+), 18 deletions(-) create mode 100644 tests/protocols/internet/test_esp_message_prefix_2411_unit.py diff --git a/pcapkit/protocols/internet/esp.py b/pcapkit/protocols/internet/esp.py index cf06386799..e7eb6a2637 100644 --- a/pcapkit/protocols/internet/esp.py +++ b/pcapkit/protocols/internet/esp.py @@ -597,9 +597,12 @@ def __init__(self, spi: 'Optional[int]' = None, *, if self.integrity is not Integrity.NONE: expected = self.integrity_suite.key_size if len(self.__integrity_key__) != expected: - warn(f'{self.integrity.name} expects a {expected}-octet key, got ' - f'{len(self.__integrity_key__)} octets; the ICV will very likely ' - f'fail to verify', ProtocolWarning) + # NOTE: #2411 -- the ``ESP:`` prefix matters more here than on + # the exceptions above: ``warn`` reports the caller's frame, so + # the message text is the only thing that names the protocol. + warn(f'ESP: invalid integrity key: {self.integrity.name} expects a ' + f'{expected}-octet key, got {len(self.__integrity_key__)} octets; ' + f'the ICV will very likely fail to verify', ProtocolWarning) if icv_length is not None and icv_length < 0: raise ProtocolError(f'ESP: invalid ICV length: {icv_length!r}') @@ -1126,7 +1129,7 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, if association is None: return self._make_opaque( spi, seq, total, data, ESPStatus.NO_SA, - f'no security association for SPI {spi:#010x}', + f'{self.alias}: no security association for SPI {spi:#010x}', version=version, packet=packet, warning=False, ) @@ -1141,8 +1144,8 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, if icv_length > len(data): return self._make_opaque( spi, seq, total, data, ESPStatus.TRUNCATED, - f'ESP payload is {len(data)} octets, shorter than the {icv_length}-octet ' - f'ICV the security association declares', + f'{self.alias}: payload too short: {len(data)} octets, shorter than the ' + f'{icv_length}-octet ICV the security association declares', version=version, packet=packet, ) body, icv = (data[:len(data) - icv_length], data[len(data) - icv_length:]) \ @@ -1161,8 +1164,8 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, if not hmac.compare_digest(association.compute_icv(spi, seq, body), icv): return self._make_opaque( spi, seq, total, body, ESPStatus.AUTH_FAILED, - f'integrity check value does not verify for SPI {spi:#010x} ' - f'sequence {seq}', icv=icv, version=version, packet=packet, + f'{self.alias}: integrity check value does not verify for SPI ' + f'{spi:#010x} sequence {seq}', icv=icv, version=version, packet=packet, ) crypto = load_cryptography() @@ -1177,8 +1180,8 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, except invalid_tag: return self._make_opaque( spi, seq, total, body, ESPStatus.AUTH_FAILED, - f'authentication tag does not verify for SPI {spi:#010x} sequence {seq}; ' - f'the encryption key is most likely wrong', + f'{self.alias}: authentication tag does not verify for SPI {spi:#010x} ' + f'sequence {seq}; the encryption key is most likely wrong', icv=icv, version=version, packet=packet, ) @@ -1435,21 +1438,21 @@ def _read_trailer(plaintext: 'bytes', association: 'SecurityAssociation', """ if len(plaintext) < 2: - return (f'decrypted ESP payload for SPI {spi:#010x} is {len(plaintext)} octets, ' - f'too short to hold a pad length and next header') + return (f'ESP: decrypted payload too short: SPI {spi:#010x} decrypts to ' + f'{len(plaintext)} octets, too short to hold a pad length and next header') pad_len, next_ = plaintext[-2], plaintext[-1] if pad_len + 2 > len(plaintext): - return (f'decrypted ESP payload for SPI {spi:#010x} declares {pad_len} octets of ' - f'padding but holds only {len(plaintext) - 2}; the encryption key is most ' - f'likely wrong') + return (f'ESP: invalid pad length: the decrypted payload for SPI {spi:#010x} ' + f'declares {pad_len} octets of padding but holds only ' + f'{len(plaintext) - 2}; the encryption key is most likely wrong') padding = plaintext[len(plaintext) - 2 - pad_len:len(plaintext) - 2] inner = plaintext[:len(plaintext) - 2 - pad_len] if padding != bytes(range(1, pad_len + 1)): - message = (f'padding of the decrypted ESP payload for SPI {spi:#010x} does not ' - f'follow the monotonically increasing sequence of RFC 4303 s2.4') + message = (f'ESP: invalid padding: the decrypted payload for SPI {spi:#010x} does ' + f'not follow the monotonically increasing sequence of RFC 4303 s2.4') if association.strict and not association.authenticated: # Nothing authenticated this packet, so the padding pattern # is the only wrong-key signal available. diff --git a/tests/protocols/internet/test_esp_message_prefix_2411_unit.py b/tests/protocols/internet/test_esp_message_prefix_2411_unit.py new file mode 100644 index 0000000000..431f002804 --- /dev/null +++ b/tests/protocols/internet/test_esp_message_prefix_2411_unit.py @@ -0,0 +1,292 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2411: the ESP read path and one warning missed the ``ESP:`` prefix. + +#2366 (fixing #2297) put the house ``ESP: `` prefix on +:meth:`ESP.make `'s +:exc:`~pcapkit.utilities.exceptions.ProtocolError` messages and on +:meth:`SecurityAssociation.unavailable +`, but eight +sites were outside its checklist and kept the bare wording -- so the *same* +condition read two different ways depending on which direction it was hit from +(``ESP: no security association for SPI 0x...`` from ``make``, and +``no security association for SPI 0x...`` from ``read``). + +The eight, as #2411 enumerates them: + +* three diagnostics returned by :meth:`ESP._read_trailer + ` -- the one for a + plaintext too short to hold a trailer, the one for a pad length the + plaintext cannot hold, and the padding-pattern one, which is both returned + and warned with (:class:`ReadTrailerMessages`); +* four built inline in :meth:`ESP.read + ` and surfaced as + :attr:`esp.error ` + (:class:`ReadErrorMessages`); +* the integrity-key warning in :meth:`SecurityAssociation.__init__ + ` + (:class:`IntegrityKeyWarning`). :func:`~pcapkit.utilities.warnings.warn` + blames the caller's frame rather than the protocol, so the message text is + the only thing that names ESP at all -- it needs the prefix more than the + exceptions do, not less. + +Every message is pinned **whole**, not by substring: the existing tests in +``tests/protocols/internet/test_esp_unit.py`` match fragments and so pass with +or without a prefix, which is exactly why the omission survived #2366. This +follows the pinning style of +``tests/protocols/test_maker_invalid_format_messages_2296_unit.py`` and its +#2406 companion. + +Nothing here needs |cryptography|_ bar +:meth:`ReadErrorMessages.test_aead_tag_mismatch_names_the_protocol`: an +``ENCR_NULL`` SA exercises the trailer and the separate-integrity paths on +:mod:`hmac` and :mod:`hashlib` alone. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +from __future__ import annotations + +import hashlib +import hmac +import importlib.util +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +HAS_CRYPTO = importlib.util.find_spec('cryptography') is not None + +#: The SPI every message below names, and its ``{spi:#010x}`` rendering. +SPI = 0x4321 +SPI_TEXT = '0x00004321' + + +def esp_packet(payload: bytes, *, spi: int = SPI, seq: int = 1) -> bytes: + """An ESP packet: the 8-octet header, then ``payload`` verbatim.""" + return struct.pack('!II', spi, seq) + payload + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ESPMessageTestCase(unittest.TestCase): + """Re-import :mod:`pcapkit` once per class, as the suite's convention asks.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def sa(self, **kwargs: 'Any') -> 'Any': + """A Security Association for :data:`SPI`, ``ENCR_NULL`` unless told otherwise. + + ``ENCR_NULL`` returns the payload as its own plaintext and needs no + |cryptography|_, so it reaches the trailer and separate-integrity + paths that carry the messages under test without a cipher in the way. + + """ + from pcapkit.protocols.internet.esp import SecurityAssociation + + kwargs.setdefault('spi', SPI) + return SecurityAssociation(**kwargs) + + def parse(self, packet: bytes, *associations: 'Any') -> 'Any': + """Parse ``packet`` with ``associations`` in context, swallowing warnings. + + Returns: + The parsed data model, i.e. ``esp.info``. + + """ + from pcapkit.corekit.context import ContextRegistry + from pcapkit.protocols.internet.esp import ESP, ESPContext + + registry = ContextRegistry.make(ESPContext(*associations)) if associations else None + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + if registry is None: + return ESP(packet, len(packet)).info + return ESP(packet, len(packet), __context__=registry).info + + +class ReadTrailerMessages(ESPMessageTestCase): + """``ESP._read_trailer``'s three diagnostics name the protocol. + + ``_read_trailer`` is a :func:`staticmethod`, so it has no ``self.alias`` to + read and spells the prefix literally -- the same way every site in + :class:`~pcapkit.protocols.internet.esp.SecurityAssociation` does. + + """ + + def trailer(self, plaintext: bytes, **kwargs: 'Any') -> 'Any': + """``_read_trailer`` over ``plaintext``, for an SA built from ``kwargs``.""" + from pcapkit.protocols.internet.esp import ESP + + return ESP._read_trailer(plaintext, self.sa(**kwargs), SPI) + + def test_plaintext_too_short_for_a_trailer(self) -> None: + # one octet cannot hold both the pad length and the next header, which + # are the last two octets of the plaintext + self.assertEqual( + self.trailer(b'\x06'), + f'ESP: decrypted payload too short: SPI {SPI_TEXT} decrypts to 1 octets, ' + f'too short to hold a pad length and next header') + + def test_an_empty_plaintext_too(self) -> None: + self.assertEqual( + self.trailer(b''), + f'ESP: decrypted payload too short: SPI {SPI_TEXT} decrypts to 0 octets, ' + f'too short to hold a pad length and next header') + + def test_pad_length_beyond_the_plaintext(self) -> None: + # pad length 255 over a 5-octet plaintext: 3 octets remain once the + # trailer's own two are taken off + self.assertEqual( + self.trailer(b'abc' + bytes([255, 6])), + f'ESP: invalid pad length: the decrypted payload for SPI {SPI_TEXT} declares ' + f'255 octets of padding but holds only 3; the encryption key is most likely wrong') + + def test_padding_pattern_is_decisive_when_nothing_authenticated_it(self) -> None: + # zero padding where RFC 4303 s2.4 asks for 1, 2, ...; a strict SA with + # no integrity has the pattern as its only wrong-key signal, so the + # mismatch is returned rather than warned about + self.assertEqual( + self.trailer(b'payload' + bytes(2) + bytes([2, 6])), + f'ESP: invalid padding: the decrypted payload for SPI {SPI_TEXT} does not follow ' + f'the monotonically increasing sequence of RFC 4303 s2.4; the encryption key is ' + f'most likely wrong') + + def test_a_lenient_sa_warns_with_the_same_prefixed_text(self) -> None: + from pcapkit.utilities.warnings import ProtocolWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + trailer = self.trailer(b'payload' + bytes(2) + bytes([2, 6]), strict=False) + + # accepted, so the trailer comes back parsed rather than as a reason + self.assertEqual(trailer, (b'payload', bytes(2), 2, 6)) + messages = [str(w.message) for w in caught + if issubclass(w.category, ProtocolWarning)] + self.assertEqual( + messages, + [f'ESP: invalid padding: the decrypted payload for SPI {SPI_TEXT} does not follow ' + f'the monotonically increasing sequence of RFC 4303 s2.4']) + + def test_a_consistent_trailer_is_still_read(self) -> None: + # the guards have not widened: padding 01 02, pad length 2, next 6 (TCP) + self.assertEqual(self.trailer(b'payload' + bytes([1, 2]) + bytes([2, 6])), + (b'payload', bytes([1, 2]), 2, 6)) + + +class ReadErrorMessages(ESPMessageTestCase): + """``ESP.read``'s four inline diagnostics name the protocol. + + ``read`` is an instance method, so it spells the prefix ``{self.alias}:``, + as ``ESP.make``'s messages do. + + """ + + def test_no_security_association_names_the_protocol(self) -> None: + from pcapkit.protocols.internet.esp import ESPStatus + + info = self.parse(esp_packet(b'opaque-ciphertext')) + self.assertIs(info.status, ESPStatus.NO_SA) + self.assertEqual(info.error, f'ESP: no security association for SPI {SPI_TEXT}') + + def test_a_payload_shorter_than_the_declared_icv(self) -> None: + from pcapkit.protocols.internet.esp import ESPStatus + + # the ICV length comes from the SA, so declaring 16 octets over a + # 4-octet payload is reported as TRUNCATED with no cipher involved + info = self.parse(esp_packet(b'tiny'), self.sa(icv_length=16)) + self.assertIs(info.status, ESPStatus.TRUNCATED) + self.assertEqual( + info.error, + 'ESP: payload too short: 4 octets, shorter than the 16-octet ICV the ' + 'security association declares') + + def test_an_icv_that_does_not_verify(self) -> None: + from pcapkit.protocols.internet.esp import ESPStatus, Integrity + + sa = self.sa(integrity=Integrity.HMAC_SHA2_256_128, integrity_key=bytes(32)) + # a 16-octet ICV that is not the one the key computes + packet = esp_packet(b'payload' + bytes([1, 2]) + bytes([2, 6]) + bytes(16), seq=7) + info = self.parse(packet, sa) + + self.assertIs(info.status, ESPStatus.AUTH_FAILED) + self.assertEqual( + info.error, + f'ESP: integrity check value does not verify for SPI {SPI_TEXT} sequence 7') + + def test_the_right_icv_still_verifies(self) -> None: + from pcapkit.protocols.internet.esp import ESPStatus, Integrity + + key = bytes(range(32)) + sa = self.sa(integrity=Integrity.HMAC_SHA2_256_128, integrity_key=key) + body = b'payload' + bytes([1, 2]) + bytes([2, 6]) + icv = hmac.new(key, struct.pack('!II', SPI, 7) + body, hashlib.sha256).digest()[:16] + info = self.parse(esp_packet(body + icv, seq=7), sa) + + self.assertIs(info.status, ESPStatus.DECRYPTED) + self.assertIsNone(info.error) + + @unittest.skipUnless(HAS_CRYPTO, 'cryptography not installed') + def test_aead_tag_mismatch_names_the_protocol(self) -> None: + from pcapkit.protocols.internet.esp import Cipher, ESPStatus + + # a combined-mode SA verifies its own tag inside decryption, so the + # wrong key surfaces as the library's InvalidTag rather than as a + # ProtocolError -- the one message here that needs |cryptography|_ + sa = self.sa(encryption=Cipher.AES_GCM_16, encryption_key=bytes(16) + b'salt') + packet = esp_packet(bytes(8) + bytes(16) + bytes(16), seq=3) + info = self.parse(packet, sa) + + self.assertIs(info.status, ESPStatus.AUTH_FAILED) + self.assertEqual( + info.error, + f'ESP: authentication tag does not verify for SPI {SPI_TEXT} sequence 3; ' + f'the encryption key is most likely wrong') + + +class IntegrityKeyWarning(ESPMessageTestCase): + """``SecurityAssociation.__init__``'s integrity-key warning names the protocol. + + :func:`~pcapkit.utilities.warnings.warn` reports the *caller's* frame, so + without the prefix the text is all a reader has to go on -- there is no + ``pcapkit/protocols/internet/esp.py`` in the location it prints. + + """ + + def test_a_short_integrity_key_warns_by_name(self) -> None: + from pcapkit.protocols.internet.esp import Integrity + from pcapkit.utilities.warnings import ProtocolWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + self.sa(integrity=Integrity.HMAC_SHA2_256_128, integrity_key=bytes(16)) + + messages = [str(w.message) for w in caught + if issubclass(w.category, ProtocolWarning)] + self.assertEqual( + messages, + ['ESP: invalid integrity key: AUTH_HMAC_SHA2_256_128 expects a 32-octet key, ' + 'got 16 octets; the ICV will very likely fail to verify']) + + def test_a_key_of_the_right_size_does_not_warn(self) -> None: + from pcapkit.protocols.internet.esp import Integrity + from pcapkit.utilities.warnings import ProtocolWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + self.sa(integrity=Integrity.HMAC_SHA2_256_128, integrity_key=bytes(32)) + + self.assertEqual([str(w.message) for w in caught + if issubclass(w.category, ProtocolWarning)], []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_tier_guard.py b/tests/test_tier_guard.py index bca81d6de6..6373f0e553 100644 --- a/tests/test_tier_guard.py +++ b/tests/test_tier_guard.py @@ -2600,14 +2600,21 @@ def test_removing_crypto_from_the_test_job_is_caught(self) -> None: gap = gaps[('HAS_CRYPTO', 'test')] self.assertEqual(gap.missing, ('cryptography',)) + # #2324 dropped this test's `len(gap.gates)` pin and kept the set equality + # as the structural claim, so a module that legitimately gains a crypto gate + # is added here rather than the assertion being loosened. #2411's module is + # the second: one AEAD-tag case needs `cryptography`, the rest reach the same + # read paths through an ENCR_NULL SA on hmac and hashlib alone. self.assertEqual({gate.module for gate in gap.gates}, - {'tests/protocols/internet/test_esp_unit.py'}) + {'tests/protocols/internet/test_esp_unit.py', + 'tests/protocols/internet/test_esp_message_prefix_2411_unit.py'}) message = _dependency_gates.describe_gap(gap) self.assertIn("the 'test' job", message) self.assertIn('cryptography', message) self.assertIn('crypto', message) self.assertIn('test_esp_unit.py', message) + self.assertIn('test_esp_message_prefix_2411_unit.py', message) self.assertNotIn( 'HAS_CRYPTO', _dependency_gates.DEPENDENCY_GATE_EXCLUSIONS,