From e26ee2352d5c7f929a934a7f58e4bd08eff71fa6 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 26 Oct 2026 17:04:14 -0400 Subject: [PATCH] fix(esp): give ESP's errors the ESP: prefix and !r values (#2297) No ProtocolError in pcapkit.protocols.internet.esp carried the ": : ... {value!r}" form #2232 and #2257 gave Ethernet, PCAP-NG and ARP: none had an "ESP:" prefix, some put "ESP" mid-sentence, and most printed the value bare. Every message raised by _resolve, CipherSuite.get, IntegritySuite.get, SecurityAssociation (SPI, destination, AEAD plus integrity, ICV length, compute_icv, decrypt, encrypt, _split_key), ESPContext.register and ESP.make now reads "ESP: : ...", with the offending value in !r form. SecurityAssociation.unavailable gives the same prefixed reason. The traceback quoted in the module docstring and esp.rst follows. Test: ESPErrorMessageTests pins each message whole; the HAS_CRYPTO gate count in test_tier_guard rises by its one gated test. --- .../source/pcapkit/protocols/internet/esp.rst | 4 +- pcapkit/protocols/internet/esp.py | 77 ++++++------ tests/protocols/internet/test_esp_unit.py | 110 ++++++++++++++++++ 3 files changed, 154 insertions(+), 37 deletions(-) diff --git a/docs/source/pcapkit/protocols/internet/esp.rst b/docs/source/pcapkit/protocols/internet/esp.rst index eb356537ea..dc673f14bf 100644 --- a/docs/source/pcapkit/protocols/internet/esp.rst +++ b/docs/source/pcapkit/protocols/internet/esp.rst @@ -156,8 +156,8 @@ refuse anything outside them rather than half-working: >>> CipherSuite.get('ENCR_3DES') # but ESP cannot apply it Traceback (most recent call last): ... - pcapkit.utilities.exceptions.ProtocolError: unsupported ESP encryption - algorithm: ENCR_3DES; pcapkit implements ENCR_NULL, ENCR_AES_CBC, + pcapkit.utilities.exceptions.ProtocolError: ESP: unsupported encryption + algorithm: 'ENCR_3DES'; pcapkit implements ENCR_NULL, ENCR_AES_CBC, ENCR_AES_GCM_8, ENCR_AES_GCM_12, ENCR_AES_GCM_16 Decryption requires the optional |cryptography|_ dependency diff --git a/pcapkit/protocols/internet/esp.py b/pcapkit/protocols/internet/esp.py index 0e91f8a6a6..cf06386799 100644 --- a/pcapkit/protocols/internet/esp.py +++ b/pcapkit/protocols/internet/esp.py @@ -79,8 +79,8 @@ >>> CipherSuite.get('ENCR_3DES') # but ESP cannot apply it Traceback (most recent call last): ... - pcapkit.utilities.exceptions.ProtocolError: unsupported ESP encryption - algorithm: ENCR_3DES; pcapkit implements ENCR_NULL, ENCR_AES_CBC, + pcapkit.utilities.exceptions.ProtocolError: ESP: unsupported encryption + algorithm: 'ENCR_3DES'; pcapkit implements ENCR_NULL, ENCR_AES_CBC, ENCR_AES_GCM_8, ENCR_AES_GCM_12, ENCR_AES_GCM_16 Decryption requires the optional |cryptography|_ dependency @@ -278,13 +278,13 @@ def _resolve(registry: 'Type[Cipher] | Type[Integrity]', value: 'Any', try: return registry(value) except ValueError: - raise ProtocolError(f'unknown ESP {kind} algorithm: {value}') from None + raise ProtocolError(f'ESP: unknown {kind} algorithm: {value!r}') from None name = str(value).upper().replace('-', '_') for candidate in (name, f'{prefix}{name}'): if candidate in registry.__members__: return registry[candidate] - raise ProtocolError(f'unknown ESP {kind} algorithm: {value!r}') + raise ProtocolError(f'ESP: unknown {kind} algorithm: {value!r}') @info_final @@ -341,7 +341,7 @@ def get(cls, value: 'Cipher | str | int') -> 'CipherSuite': cipher = _resolve(Cipher, value, 'ENCR_', 'encryption') suite = CIPHER_SUITES.get(cipher) if suite is None: - raise ProtocolError(f'unsupported ESP encryption algorithm: {cipher.name}; pcapkit ' + raise ProtocolError(f'ESP: unsupported encryption algorithm: {cipher.name!r}; pcapkit ' f'implements {", ".join(key.name for key in CIPHER_SUITES)}') return suite @@ -395,8 +395,9 @@ def get(cls, value: 'Integrity | str | int') -> 'IntegritySuite': integrity = _resolve(Integrity, value, 'AUTH_', 'integrity') suite = INTEGRITY_SUITES.get(integrity) if suite is None: - raise ProtocolError(f'unsupported ESP integrity algorithm: {integrity.name}; pcapkit ' - f'implements {", ".join(key.name for key in INTEGRITY_SUITES)}') + raise ProtocolError(f'ESP: unsupported integrity algorithm: {integrity.name!r}; ' + f'pcapkit implements ' + f'{", ".join(key.name for key in INTEGRITY_SUITES)}') return suite @@ -551,7 +552,7 @@ def __init__(self, spi: 'Optional[int]' = None, *, destination: 'Optional[IPv4Address | IPv6Address | str | int | bytes]' = None, strict: 'bool' = True) -> 'None': if spi is not None and not 0 <= spi <= 0xFFFFFFFF: - raise ProtocolError(f'invalid SPI: {spi}') + raise ProtocolError(f'ESP: invalid SPI: {spi!r}') if isinstance(destination, bool): # NOTE: checked before the ``ipaddress.ip_address`` call below, for @@ -564,7 +565,7 @@ def __init__(self, spi: 'Optional[int]' = None, *, # would silently become ``IPv4Address('0.0.0.1')``, with no # exception and no warning. raise ProtocolError( - f'invalid destination: must not be a bool, not {destination!r} -- ' + f'ESP: invalid destination: must not be a bool, not {destination!r} -- ' f'pass int({destination!r}) if the numeric value is what is wanted') #: Optional[int]: Security Parameters Index, or :data:`None` for any. @@ -584,8 +585,9 @@ def __init__(self, spi: 'Optional[int]' = None, *, if self.cipher_suite.is_aead and self.integrity is not Integrity.NONE: raise ProtocolError( - f'{self.encryption.name} is a combined mode algorithm and provides its own ' - f'integrity; {self.integrity.name} must not be configured alongside it' + f'ESP: invalid integrity algorithm: {self.encryption.name} is a combined mode ' + f'algorithm and provides its own integrity, so {self.integrity.name!r} must ' + f'not be configured alongside it' ) key, self.__salt__ = self._split_key(self.cipher_suite, encryption_key, salt) @@ -600,7 +602,7 @@ def __init__(self, spi: 'Optional[int]' = None, *, f'fail to verify', ProtocolWarning) if icv_length is not None and icv_length < 0: - raise ProtocolError(f'invalid ICV length: {icv_length}') + raise ProtocolError(f'ESP: invalid ICV length: {icv_length!r}') self.__icv_length__ = icv_length unavailable = self.unavailable() @@ -689,8 +691,9 @@ def unavailable(self) -> 'Optional[str]': """ if self.cipher_suite.requires_cryptography and load_cryptography() is None: - return (f'{self.encryption.name} needs the optional "cryptography" dependency, ' - f'which is not installed (pip install pypcapkit[crypto])') + return (f'ESP: unavailable encryption algorithm: {self.encryption.name!r} needs the ' + f'optional "cryptography" dependency, which is not installed ' + f'(pip install pypcapkit[crypto])') return None def compute_icv(self, spi: 'int', seq: 'int', body: 'bytes') -> 'bytes': @@ -715,7 +718,8 @@ def compute_icv(self, spi: 'int', seq: 'int', body: 'bytes') -> 'bytes': """ digest = self.integrity_suite.digest if digest is None: - raise ProtocolError(f'{self.integrity.name} computes no ICV') + raise ProtocolError(f'ESP: integrity algorithm computes no ICV: ' + f'{self.integrity.name!r}') mac = hmac.new(self.__integrity_key__, spi.to_bytes(4, 'big') + seq.to_bytes(4, 'big') + body, @@ -751,21 +755,22 @@ def decrypt(self, spi: 'int', seq: 'int', body: 'bytes', icv: 'bytes') -> 'bytes crypto = load_cryptography() if crypto is None: - raise ProtocolError(f'{cipher.name} needs the optional "cryptography" dependency, ' - f'which is not installed') + raise ProtocolError(f'ESP: unavailable encryption algorithm: {cipher.name!r} needs ' + f'the optional "cryptography" dependency, which is not ' + f'installed') crypto_cipher, algorithms, modes, _ = crypto iv_length = suite.iv_length if len(body) < iv_length: - raise ProtocolError(f'ESP payload is {len(body)} octets, too short for the ' + raise ProtocolError(f'ESP: payload too short: {len(body)!r} octets, less than the ' f'{iv_length}-octet {cipher.name} IV') iv, ciphertext = body[:iv_length], body[iv_length:] if cipher is Cipher.ENCR_AES_CBC: if not ciphertext or len(ciphertext) % suite.block_size: - raise ProtocolError(f'ESP ciphertext of {len(ciphertext)} octets is not a ' - f'positive multiple of the {suite.block_size}-octet ' - f'{cipher.name} block size') + raise ProtocolError(f'ESP: invalid ciphertext length: {len(ciphertext)!r} ' + f'octets is not a positive multiple of the ' + f'{suite.block_size}-octet {cipher.name} block size') decryptor = crypto_cipher(algorithms.AES(self.__key__), modes.CBC(iv)).decryptor() return decryptor.update(ciphertext) + decryptor.finalize() @@ -773,8 +778,8 @@ def decrypt(self, spi: 'int', seq: 'int', body: 'bytes', icv: 'bytes') -> 'bytes # keying material followed by the explicit IV, and the associated # data is the SPI and the sequence number. if not icv: - raise ProtocolError(f'{cipher.name} requires an authentication tag, but the ' - f'packet carries no ICV') + raise ProtocolError(f'ESP: missing authentication tag: {cipher.name} requires ' + f'one, but the packet carries no ICV') nonce = self.__salt__ + iv aad = spi.to_bytes(4, 'big') + seq.to_bytes(4, 'big') decryptor = crypto_cipher( @@ -813,15 +818,17 @@ def encrypt(self, spi: 'int', seq: 'int', plaintext: 'bytes', crypto = load_cryptography() if crypto is None: - raise ProtocolError(f'{cipher.name} needs the optional "cryptography" dependency, ' - f'which is not installed') + raise ProtocolError(f'ESP: unavailable encryption algorithm: {cipher.name!r} needs ' + f'the optional "cryptography" dependency, which is not ' + f'installed') crypto_cipher, algorithms, modes, _ = crypto iv_length = suite.iv_length if iv is None: iv = os.urandom(iv_length) elif len(iv) != iv_length: - raise ProtocolError(f'{cipher.name} needs a {iv_length}-octet IV, got {len(iv)}') + raise ProtocolError(f'ESP: invalid IV: {cipher.name} needs a {iv_length}-octet IV, ' + f'got {len(iv)!r} octets') if cipher is Cipher.ENCR_AES_CBC: encryptor = crypto_cipher(algorithms.AES(self.__key__), modes.CBC(iv)).encryptor() @@ -869,12 +876,12 @@ def _split_key(suite: 'CipherSuite', material: 'bytes', salt = bytes(salt) if len(salt) != salt_length: - raise ProtocolError(f'{suite.cipher.name} needs a {salt_length}-octet salt, ' - f'got {len(salt)}') + raise ProtocolError(f'ESP: invalid salt: {suite.cipher.name} needs a ' + f'{salt_length}-octet salt, got {len(salt)!r} octets') if len(material) not in suite.key_sizes: - raise ProtocolError(f'{suite.cipher.name} needs a key of ' + raise ProtocolError(f'ESP: invalid key: {suite.cipher.name} needs a key of ' f'{" or ".join(map(str, suite.key_sizes))} octets, ' - f'got {len(material)}') + f'got {len(material)!r}') return material, salt ########################################################################## @@ -933,7 +940,7 @@ def register(self, association: 'SecurityAssociation') -> 'None': """ if not isinstance(association, SecurityAssociation): - raise ProtocolError(f'not a security association: {association!r}') + raise ProtocolError(f'ESP: not a security association: {association!r}') self.__associations__.append(association) def match(self, spi: 'int', @@ -1270,8 +1277,8 @@ def make(self, context = self._get_context(ESPContext) association = context.match(spi) if context is not None else None if association is None: - raise ProtocolError(f'no security association for SPI {spi:#010x}; ESP cannot be ' - f'constructed with encrypt=True without one') + raise ProtocolError(f'{self.alias}: no security association for SPI {spi:#010x}; ' + f'encrypt=True needs one') next_value = self._make_index(next, next_default, namespace=next_namespace, reversed=next_reversed, pack=False) @@ -1281,11 +1288,11 @@ def make(self, if pad_len is None: pad_len = -(len(plain) + 2) % align elif (len(plain) + pad_len + 2) % align: - raise ProtocolError(f'pad length {pad_len} leaves ' + raise ProtocolError(f'{self.alias}: invalid pad length: {pad_len!r} leaves ' f'{len(plain) + pad_len + 2} octets, which is not a multiple ' f'of the required {align}-octet alignment') if not 0 <= pad_len <= 255: - raise ProtocolError(f'invalid pad length: {pad_len}') + raise ProtocolError(f'{self.alias}: invalid pad length: {pad_len!r}') # RFC 4303 s2.4: padding bytes are a monotonically increasing # sequence starting at 1. diff --git a/tests/protocols/internet/test_esp_unit.py b/tests/protocols/internet/test_esp_unit.py index d76cac2605..9bc2c85364 100644 --- a/tests/protocols/internet/test_esp_unit.py +++ b/tests/protocols/internet/test_esp_unit.py @@ -1112,5 +1112,115 @@ def test_docstring_cites_the_extension_header_registry(self) -> None: self.assertIn('qualifies it for a base besides', doc) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class ESPErrorMessageTests(unittest.TestCase): + """Every :exc:`ProtocolError` the module raises reads ``ESP: ``. + + The other tests here match substrings, so they pass whatever the prefix + is. These pin each message whole: the ``ESP:`` prefix and the ``!r`` form + of the offending value, as #2232 and #2257 gave Ethernet, PCAP-NG and ARP + (#2297). + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def assertMessage(self, expected: str, func, *args, **kwargs) -> None: # noqa: N802 + from pcapkit.utilities.exceptions import ProtocolError + + with self.assertRaises(ProtocolError) as ctx: + func(*args, **kwargs) + self.assertEqual(str(ctx.exception), expected) + + def test_algorithm_lookup(self) -> None: + from pcapkit.protocols.internet.esp import (CIPHER_SUITES, INTEGRITY_SUITES, CipherSuite, + IntegritySuite) + + ciphers = ', '.join(key.name for key in CIPHER_SUITES) + integrities = ', '.join(key.name for key in INTEGRITY_SUITES) + self.assertMessage(f"ESP: unsupported encryption algorithm: 'ENCR_3DES'; " + f"pcapkit implements {ciphers}", CipherSuite.get, 'ENCR_3DES') + self.assertMessage(f"ESP: unsupported integrity algorithm: 'AUTH_HMAC_MD5_96'; " + f"pcapkit implements {integrities}", IntegritySuite.get, 'HMAC_MD5_96') + self.assertMessage("ESP: unknown encryption algorithm: 'ROT13'", CipherSuite.get, 'ROT13') + self.assertMessage('ESP: unknown encryption algorithm: -1', CipherSuite.get, -1) + self.assertMessage("ESP: unknown integrity algorithm: 'HMAC_SHA3_256'", + IntegritySuite.get, 'HMAC_SHA3_256') + + def test_security_association(self) -> None: + from pcapkit.protocols.internet.esp import (Cipher, ESPContext, Integrity, + SecurityAssociation) + + self.assertMessage('ESP: invalid SPI: -1', SecurityAssociation, spi=-1) + self.assertMessage('ESP: invalid destination: must not be a bool, not True -- ' + 'pass int(True) if the numeric value is what is wanted', + SecurityAssociation, spi=1, destination=True) + self.assertMessage('ESP: invalid integrity algorithm: ENCR_AES_GCM_16 is a combined ' + 'mode algorithm and provides its own integrity, so ' + "'AUTH_HMAC_SHA2_256_128' must not be configured alongside it", + SecurityAssociation, spi=1, encryption=Cipher.AES_GCM_16, + encryption_key=GCM128_KEYMAT, integrity=Integrity.HMAC_SHA2_256_128, + integrity_key=bytes(32)) + self.assertMessage('ESP: invalid ICV length: -1', SecurityAssociation, spi=1, + icv_length=-1) + self.assertMessage('ESP: invalid key: ENCR_AES_CBC needs a key of 16 or 24 or 32 ' + 'octets, got 15', SecurityAssociation, spi=1, + encryption=Cipher.AES_CBC, encryption_key=bytes(15)) + self.assertMessage('ESP: invalid salt: ENCR_AES_GCM_16 needs a 4-octet salt, ' + 'got 2 octets', SecurityAssociation, spi=1, + encryption=Cipher.AES_GCM_16, encryption_key=bytes(16), + salt=b'ab') + self.assertMessage("ESP: integrity algorithm computes no ICV: 'NONE'", + SecurityAssociation(spi=1).compute_icv, 1, 1, b'body') + self.assertMessage('ESP: not a security association: 1', ESPContext().register, 1) + + def test_missing_cryptography(self) -> None: + from pcapkit.protocols.internet import esp as esp_module + from pcapkit.protocols.internet.esp import Cipher, SecurityAssociation + + expected = ("ESP: unavailable encryption algorithm: 'ENCR_AES_CBC' needs the optional " + '"cryptography" dependency, which is not installed') + with mock.patch.object(esp_module, 'load_cryptography', return_value=None), \ + mock.patch.object(esp_module, 'warn'): + sa = SecurityAssociation(spi=1, encryption=Cipher.AES_CBC, + encryption_key=CASE5_KEY) + self.assertMessage(expected, sa.decrypt, 1, 1, bytes(32), b'') + self.assertMessage(expected, sa.encrypt, 1, 1, bytes(16)) + # the same reason, as the packet's ``error`` and the SA's warning carry it + self.assertEqual(sa.unavailable(), f'{expected} (pip install pypcapkit[crypto])') + + @unittest.skipUnless(HAS_CRYPTO, 'cryptography not installed') + def test_malformed_payload(self) -> None: + from pcapkit.protocols.internet.esp import Cipher, SecurityAssociation + + cbc = SecurityAssociation(spi=1, encryption=Cipher.AES_CBC, encryption_key=CASE5_KEY) + gcm = SecurityAssociation(spi=1, encryption=Cipher.AES_GCM_16, + encryption_key=GCM128_KEYMAT) + self.assertMessage('ESP: payload too short: 5 octets, less than the 16-octet ' + 'ENCR_AES_CBC IV', cbc.decrypt, 1, 1, bytes(5), b'') + self.assertMessage('ESP: invalid ciphertext length: 5 octets is not a positive ' + 'multiple of the 16-octet ENCR_AES_CBC block size', + cbc.decrypt, 1, 1, bytes(21), b'') + self.assertMessage('ESP: missing authentication tag: ENCR_AES_GCM_16 requires one, ' + 'but the packet carries no ICV', gcm.decrypt, 1, 1, bytes(9), b'') + self.assertMessage('ESP: invalid IV: ENCR_AES_CBC needs a 16-octet IV, got 2 octets', + cbc.encrypt, 1, 1, bytes(16), b'ab') + + def test_make(self) -> None: + from pcapkit.corekit.context import ContextRegistry + from pcapkit.protocols.internet.esp import ESP, ESPContext, SecurityAssociation + + registry = ContextRegistry.make(ESPContext(SecurityAssociation(spi=0x77))) + self.assertMessage('ESP: no security association for SPI 0x00000099; ' + 'encrypt=True needs one', ESP, spi=0x99, seq=1, encrypt=True, + payload=b'x') + self.assertMessage('ESP: invalid pad length: 1 leaves 5 octets, which is not a ' + 'multiple of the required 4-octet alignment', ESP, spi=0x77, seq=1, + encrypt=True, pad_len=1, payload=bytes(2), __context__=registry) + self.assertMessage('ESP: invalid pad length: 256', ESP, spi=0x77, seq=1, encrypt=True, + pad_len=256, payload=bytes(2), __context__=registry) + + if __name__ == '__main__': unittest.main()