Skip to content

fix(esp): give ESP's errors the ESP: prefix and !r values (#2297) #2366

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Oct 27, 2026
Merged

fix(esp): give ESP's errors the ESP: prefix and !r values (#2297) #2366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions docs/source/pcapkit/protocols/internet/esp.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
77 changes: 42 additions & 35 deletions pcapkit/protocols/internet/esp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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


Expand Down Expand Up @@ -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
Expand All @@ -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.
Expand All @@ -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)
Expand All @@ -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()
Expand Down Expand Up @@ -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':
Expand All @@ -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,
Expand Down Expand Up @@ -751,30 +755,31 @@ 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()

# AEAD, i.e. AES-GCM [RFC 4106]: the nonce is the salt from the
# 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(
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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

##########################################################################
Expand Down Expand Up @@ -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',
Expand Down Expand Up @@ -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)
Expand All @@ -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.
Expand Down
110 changes: 110 additions & 0 deletions tests/protocols/internet/test_esp_unit.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: <description>``.

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()
Loading