diff --git a/scapy/config.py b/scapy/config.py
index 836de46831e7b8c5c4d2bf0d525aa704af0a77c1..119022d5a91c769278f65a8d17b9d35db7912199 100644
--- a/scapy/config.py
+++ b/scapy/config.py
@@ -372,7 +372,8 @@ extensions_paths: path or list of paths where extensions are to be looked for
netcache = NetCache()
load_layers = ["l2", "inet", "dhcp", "dns", "dot11", "gprs", "hsrp", "inet6", "ir", "isakmp", "l2tp",
"mgcp", "mobileip", "netbios", "netflow", "ntp", "ppp", "radius", "rip", "rtp",
- "sebek", "skinny", "smb", "snmp", "tftp", "x509", "bluetooth", "dhcp6", "llmnr", "sctp", "vrrp" ]
+ "sebek", "skinny", "smb", "snmp", "tftp", "x509", "bluetooth", "dhcp6", "llmnr", "sctp", "vrrp",
+ "ipsec" ]
if not Conf.ipv6_enabled:
diff --git a/scapy/layers/ipsec.py b/scapy/layers/ipsec.py
new file mode 100644
index 0000000000000000000000000000000000000000..692a6e1879a1f2cfe4955fb38ecdf6b98711f4c4
--- /dev/null
+++ b/scapy/layers/ipsec.py
@@ -0,0 +1,981 @@
+#############################################################################
+## ipsec.py --- IPSec support for Scapy ##
+## ##
+## Copyright (C) 2014 6WIND ##
+## ##
+## This program is free software; you can redistribute it and/or modify it ##
+## under the terms of the GNU General Public License version 2 as ##
+## published by the Free Software Foundation. ##
+## ##
+## This program is distributed in the hope that it will be useful, but ##
+## WITHOUT ANY WARRANTY; without even the implied warranty of ##
+## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU ##
+## General Public License for more details. ##
+#############################################################################
+"""
+IPSec layer
+===========
+
+Example of use:
+
+>>> sa = SecurityAssociation(ESP, spi=0xdeadbeef, crypt_algo='AES-CBC',
+... crypt_key='sixteenbytes key')
+>>> p = IP(src='1.1.1.1', dst='2.2.2.2')
+>>> p /= TCP(sport=45012, dport=80)
+>>> p /= Raw('testdata')
+>>> p = IP(str(p))
+>>> p
+<IP version=4L ihl=5L tos=0x0 len=48 id=1 flags= frag=0L ttl=64 proto=tcp chksum=0x74c2 src=1.1.1.1 dst=2.2.2.2 options=[] |<TCP sport=45012 dport=http seq=0 ack=0 dataofs=5L reserved=0L flags=S window=8192 chksum=0x1914 urgptr=0 options=[] |<Raw load='testdata' |>>>
+>>>
+>>> e = sa.encrypt(p)
+>>> e
+<IP version=4L ihl=5L tos=0x0 len=76 id=1 flags= frag=0L ttl=64 proto=esp chksum=0x747a src=1.1.1.1 dst=2.2.2.2 |<ESP spi=0xdeadbeef seq=1 data='\xf8\xdb\x1e\x83[T\xab\\\xd2\x1b\xed\xd1\xe5\xc8Y\xc2\xa5d\x92\xc1\x05\x17\xa6\x92\x831\xe6\xc1]\x9a\xd6K}W\x8bFfd\xa5B*+\xde\xc8\x89\xbf{\xa9' |>>
+>>>
+>>> d = sa.decrypt(e)
+>>> d
+<IP version=4L ihl=5L tos=0x0 len=48 id=1 flags= frag=0L ttl=64 proto=tcp chksum=0x74c2 src=1.1.1.1 dst=2.2.2.2 |<TCP sport=45012 dport=http seq=0 ack=0 dataofs=5L reserved=0L flags=S window=8192 chksum=0x1914 urgptr=0 options=[] |<Raw load='testdata' |>>>
+>>>
+>>> d == p
+True
+"""
+
+import socket
+import fractions
+
+from scapy.data import IP_PROTOS
+
+from scapy.fields import ByteEnumField, ByteField, StrField, XIntField, IntField, \
+ ShortField, PacketField
+
+from scapy.packet import Packet, bind_layers, Raw
+
+from scapy.layers.inet import IP, UDP
+from scapy.layers.inet6 import IPv6, IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, \
+ IPv6ExtHdrRouting
+
+
+#------------------------------------------------------------------------------
+class AH(Packet):
+ """
+ Authentication Header
+
+ See https://tools.ietf.org/rfc/rfc4302.txt
+ """
+
+ name = 'AH'
+
+ fields_desc = [
+ ByteEnumField('nh', None, IP_PROTOS),
+ ByteField('payloadlen', None),
+ ShortField('reserved', None),
+ XIntField('spi', 0x0),
+ IntField('seq', 0),
+ StrField('icv', None),
+ StrField('padding', None),
+ ]
+
+ overload_fields = {
+ IP: {'proto': socket.IPPROTO_AH},
+ IPv6: {'nh': socket.IPPROTO_AH},
+ IPv6ExtHdrHopByHop: {'nh': socket.IPPROTO_AH},
+ IPv6ExtHdrDestOpt: {'nh': socket.IPPROTO_AH},
+ IPv6ExtHdrRouting: {'nh': socket.IPPROTO_AH},
+ }
+
+bind_layers(IP, AH, proto=socket.IPPROTO_AH)
+bind_layers(IPv6, AH, nh=socket.IPPROTO_AH)
+
+#------------------------------------------------------------------------------
+class ESP(Packet):
+ """
+ Encapsulated Security Payload
+
+ See https://tools.ietf.org/rfc/rfc4303.txt
+ """
+ name = 'ESP'
+
+ fields_desc = [
+ XIntField('spi', 0x0),
+ IntField('seq', 0),
+ StrField('data', None),
+ ]
+
+ overload_fields = {
+ IP: {'proto': socket.IPPROTO_ESP},
+ IPv6: {'nh': socket.IPPROTO_ESP},
+ IPv6ExtHdrHopByHop: {'nh': socket.IPPROTO_ESP},
+ IPv6ExtHdrDestOpt: {'nh': socket.IPPROTO_ESP},
+ IPv6ExtHdrRouting: {'nh': socket.IPPROTO_ESP},
+ }
+
+bind_layers(IP, ESP, proto=socket.IPPROTO_ESP)
+bind_layers(IPv6, ESP, nh=socket.IPPROTO_ESP)
+bind_layers(UDP, ESP, dport=4500) # NAT-Traversal encapsulation
+bind_layers(UDP, ESP, sport=4500) # NAT-Traversal encapsulation
+
+#------------------------------------------------------------------------------
+class _ESPPlain(Packet):
+ """
+ Internal class to represent unencrypted ESP packets.
+ """
+ name = 'ESP'
+
+ fields_desc = [
+ XIntField('spi', 0x0),
+ IntField('seq', 0),
+
+ StrField('iv', ''),
+ PacketField('data', '', Raw),
+ StrField('padding', ''),
+
+ ByteField('padlen', 0),
+ ByteEnumField('nh', 0, IP_PROTOS),
+ StrField('icv', ''),
+ ]
+
+ def data_for_encryption(self):
+ return str(self.data) + self.padding + chr(self.padlen) + chr(self.nh)
+
+#------------------------------------------------------------------------------
+try:
+ from Crypto.Cipher import AES
+ from Crypto.Cipher import DES
+ from Crypto.Cipher import DES3
+ from Crypto.Cipher import CAST
+ from Crypto.Cipher import Blowfish
+ from Crypto.Util import Counter
+ from Crypto import Random
+except ImportError:
+ # no error if pycrypto is not available but encryption won't be supported
+ AES = None
+ DES = None
+ DES3 = None
+ CAST = None
+ Blowfish = None
+ Random = None
+
+#------------------------------------------------------------------------------
+def _lcm(a, b):
+ """
+ Least Common Multiple between 2 integers.
+ """
+ if a == 0 or b == 0:
+ return 0
+ else:
+ return abs(a * b) // fractions.gcd(a, b)
+
+class CryptAlgo(object):
+ """
+ IPSec encryption algorithm
+ """
+
+ def __init__(self, name, cipher, mode, block_size=None, iv_size=None, key_size=None):
+ """
+ @param name: the name of this encryption algorithm
+ @param cipher: a Cipher module
+ @param mode: the mode used with the cipher module
+ @param block_size: the length a block for this algo. Defaults to the
+ `block_size` of the cipher.
+ @param iv_size: the length of the initialization vector of this algo.
+ Defaults to the `block_size` of the cipher.
+ @param key_size: an integer or list/tuple of integers. If specified,
+ force the secret keys length to one of the values.
+ Defaults to the `key_size` of the cipher.
+ """
+ self.name = name
+ self.cipher = cipher
+ self.mode = mode
+
+ if block_size is not None:
+ self.block_size = block_size
+ elif cipher is not None:
+ self.block_size = cipher.block_size
+ else:
+ self.block_size = 1
+
+ if iv_size is None:
+ self.iv_size = self.block_size
+ else:
+ self.iv_size = iv_size
+
+ if key_size is not None:
+ self.key_size = key_size
+ elif cipher is not None:
+ self.key_size = cipher.key_size
+ else:
+ self.key_size = None
+
+ def check_key(self, key):
+ """
+ Check that the key length is valid.
+
+ @param key: a byte string
+ """
+ if self.key_size and not (len(key) == self.key_size or len(key) in self.key_size):
+ raise TypeError('invalid key size %s, must be %s' %
+ (len(key), self.key_size))
+
+ def generate_iv(self):
+ """
+ Generate a random initialization vector. If pycrypto is not available,
+ return a buffer of the correct length filled with only '\x00'.
+ """
+ if Random:
+ return Random.get_random_bytes(self.iv_size)
+ else:
+ return chr(0) * self.iv_size
+
+ def new_cipher(self, key, iv):
+ """
+ @param key: the secret key, a byte string
+ @param iv: the initialization vector, a byte string
+ @return: an initialized cipher object for this algo
+ """
+ if (hasattr(self.cipher, 'MODE_CTR') and self.mode == self.cipher.MODE_CTR
+ or hasattr(self.cipher, 'MODE_GCM') and self.mode == self.cipher.MODE_GCM):
+ # in counter mode, the "iv" must be incremented for each block
+ # it is calculated like this:
+ # +---------+------------------+---------+
+ # | nonce | IV | counter |
+ # +---------+------------------+---------+
+ # m bytes n bytes 4 bytes
+ # <-------------------------------------->
+ # block_size
+ nonce_size = self.cipher.block_size - self.iv_size - 4
+
+ # instead of asking for an extra parameter, we extract the last
+ # nonce_size bytes of the key and use them as the nonce.
+ # +----------------------------+---------+
+ # | cipher key | nonce |
+ # +----------------------------+---------+
+ # <--------->
+ # nonce_size
+ cipher_key, nonce = key[:-nonce_size], key[-nonce_size:]
+
+ return self.cipher.new(cipher_key, self.mode,
+ counter=Counter.new(4 * 8, prefix=nonce + iv))
+ else:
+ return self.cipher.new(key, self.mode, iv)
+
+ def pad(self, esp):
+ """
+ Add the correct amount of padding so that the data to encrypt is
+ exactly a multiple of the algorithm's block size.
+
+ Also, make sure that the total ESP packet length is a multiple of 4 or
+ 8 bytes with IP or IPv6 respectively.
+
+ @param esp: an unencrypted _ESPPlain packet
+ """
+ # 2 extra bytes for padlen and nh
+ data_len = len(esp.data) + 2
+
+ # according to the RFC4303, section 2.4. Padding (for Encryption)
+ # the size of the ESP payload must be a multiple of 32 bits
+ align = _lcm(self.block_size, 4)
+
+ # pad for block size
+ esp.padlen = -data_len % align
+
+ # padding must be an array of bytes starting from 1 to padlen
+ esp.padding = ''
+ for b in range(1, esp.padlen + 1):
+ esp.padding += chr(b)
+
+ # If the following test fails, it means that this algo does not comply
+ # with the RFC
+ payload_len = len(esp.iv) + len(esp.data) + len(esp.padding) + 2
+ if payload_len % 4 != 0:
+ raise ValueError('The size of the ESP data is not aligned to 32 bits after padding.')
+
+ return esp
+
+ def encrypt(self, esp, key):
+ """
+ Encrypt an ESP packet
+
+ @param esp: an unencrypted _ESPPlain packet with valid padding
+ @param key: the secret key used for encryption
+
+ @return: a valid ESP packet encrypted with this algorithm
+ """
+ data = esp.data_for_encryption()
+
+ if self.cipher:
+ self.check_key(key)
+ cipher = self.new_cipher(key, esp.iv)
+ data = cipher.encrypt(data)
+
+ return ESP(spi=esp.spi, seq=esp.seq, data=esp.iv + data)
+
+ def decrypt(self, esp, key, icv_size=0):
+ """
+ Decrypt an ESP packet
+
+ @param esp: an encrypted ESP packet
+ @param key: the secret key used for encryption
+ @param icv_size: the length of the icv used for integrity check
+
+ @return: a valid ESP packet encrypted with this algorithm
+ """
+ self.check_key(key)
+
+ iv = esp.data[:self.iv_size]
+ data = esp.data[self.iv_size:len(esp.data) - icv_size]
+ icv = esp.data[len(esp.data) - icv_size:]
+
+ if self.cipher:
+ cipher = self.new_cipher(key, iv)
+ data = cipher.decrypt(data)
+
+ # extract padlen and nh
+ padlen = ord(data[-2])
+ nh = ord(data[-1])
+
+ # then use padlen to determine data and padding
+ data = data[:len(data) - padlen - 2]
+ padding = data[len(data) - padlen - 2: len(data) - 2]
+
+ return _ESPPlain(spi=esp.spi,
+ seq=esp.seq,
+ iv=iv,
+ data=data,
+ padding=padding,
+ padlen=padlen,
+ nh=nh,
+ icv=icv)
+
+#------------------------------------------------------------------------------
+# The names of the encryption algorithms are the same than in scapy.contrib.ikev2
+# see http://www.iana.org/assignments/ikev2-parameters/ikev2-parameters.xhtml
+
+CRYPT_ALGOS = {
+ 'NULL': CryptAlgo('NULL', cipher=None, mode=None, iv_size=0),
+}
+
+if AES:
+ CRYPT_ALGOS['AES-CBC'] = CryptAlgo('AES-CBC',
+ cipher=AES,
+ mode=AES.MODE_CBC)
+ # specific case for counter mode:
+ # the last 4 bytes of the key are used to carry the nonce of the counter
+ CRYPT_ALGOS['AES-CTR'] = CryptAlgo('AES-CTR',
+ cipher=AES,
+ mode=AES.MODE_CTR,
+ block_size=1,
+ iv_size=8,
+ key_size=(16 + 4, 24 + 4, 32 + 4))
+if DES:
+ CRYPT_ALGOS['DES'] = CryptAlgo('DES',
+ cipher=DES,
+ mode=DES.MODE_CBC)
+if Blowfish:
+ CRYPT_ALGOS['Blowfish'] = CryptAlgo('Blowfish',
+ cipher=Blowfish,
+ mode=Blowfish.MODE_CBC)
+if DES3:
+ CRYPT_ALGOS['3DES'] = CryptAlgo('3DES',
+ cipher=DES3,
+ mode=DES3.MODE_CBC)
+if CAST:
+ CRYPT_ALGOS['CAST'] = CryptAlgo('CAST',
+ cipher=CAST,
+ mode=CAST.MODE_CBC)
+
+#------------------------------------------------------------------------------
+try:
+ from Crypto.Hash import HMAC
+ from Crypto.Hash import SHA
+ from Crypto.Hash import MD5
+ from Crypto.Hash import SHA256
+ from Crypto.Hash import SHA384
+ from Crypto.Hash import SHA512
+except ImportError:
+ # no error if pycrypto is not available but authentication won't be supported
+ HMAC = None
+ SHA = None
+ MD5 = None
+ SHA256 = None
+ SHA384 = None
+try:
+ from Crypto.Hash import XCBCMAC
+except ImportError:
+ XCBCMAC = None
+
+#------------------------------------------------------------------------------
+class IPSecIntegrityError(Exception):
+ """
+ Error risen when the integrity check fails.
+ """
+ pass
+
+class AuthAlgo(object):
+ """
+ IPSec integrity algorithm
+ """
+
+ def __init__(self, name, mac, digestmod, icv_size, key_size=None):
+ """
+ @param name: the name of this integrity algorithm
+ @param mac: a Message Authentication Code module
+ @param digestmod: a Hash or Cipher module
+ @param icv_size: the length of the integrity check value of this algo
+ @param key_size: an integer or list/tuple of integers. If specified,
+ force the secret keys length to one of the values.
+ Defaults to the `key_size` of the cipher.
+ """
+ self.name = name
+ self.mac = mac
+ self.digestmod = digestmod
+ self.icv_size = icv_size
+ self.key_size = key_size
+
+ def check_key(self, key):
+ """
+ Check that the key length is valid.
+
+ @param key: a byte string
+ """
+ if self.key_size and len(key) not in self.key_size:
+ raise TypeError('invalid key size %s, must be one of %s' %
+ (len(key), self.key_size))
+
+ def new_mac(self, key):
+ """
+ @param key: a byte string
+ @return: an initialized mac object for this algo
+ """
+ if self.mac is XCBCMAC:
+ # specific case here, ciphermod instead of digestmod
+ return self.mac.new(key, ciphermod=self.digestmod)
+ else:
+ return self.mac.new(key, digestmod=self.digestmod)
+
+ def sign(self, pkt, key):
+ """
+ Sign an IPSec (ESP or AH) packet with this algo.
+
+ @param pkt: a packet that contains a valid encrypted ESP or AH layer
+ @param key: the authentication key, a byte string
+
+ @return: the signed packet
+ """
+ if not self.mac:
+ return pkt
+
+ self.check_key(key)
+
+ mac = self.new_mac(key)
+
+ if pkt.haslayer(ESP):
+ mac.update(str(pkt[ESP]))
+ pkt[ESP].data += mac.digest()[:self.icv_size]
+
+ elif pkt.haslayer(AH):
+ clone = zero_mutable_fields(pkt.copy(), sending=True)
+ mac.update(str(clone))
+ pkt[AH].icv = mac.digest()[:self.icv_size]
+
+ return pkt
+
+ def verify(self, pkt, key):
+ """
+ Check that the integrity check value (icv) of a packet is valid.
+
+ @param pkt: a packet that contains a valid encrypted ESP or AH layer
+ @param key: the authentication key, a byte string
+
+ @raise IPSecIntegrityError: if the integrity check fails
+ """
+ if not self.mac or self.icv_size == 0:
+ return
+
+ self.check_key(key)
+
+ mac = self.new_mac(key)
+
+ pkt_icv = 'not found'
+ computed_icv = 'not computed'
+
+ if isinstance(pkt, ESP):
+ pkt_icv = pkt.data[len(pkt.data) - self.icv_size:]
+
+ pkt = pkt.copy()
+ pkt.data = pkt.data[:len(pkt.data) - self.icv_size]
+ mac.update(str(pkt))
+ computed_icv = mac.digest()[:self.icv_size]
+
+ elif pkt.haslayer(AH):
+ pkt_icv = pkt[AH].icv[:self.icv_size]
+
+ clone = zero_mutable_fields(pkt.copy(), sending=False)
+ mac.update(str(clone))
+ computed_icv = mac.digest()[:self.icv_size]
+
+ if pkt_icv != computed_icv:
+ raise IPSecIntegrityError('pkt_icv=%r, computed_icv=%r' %
+ (pkt_icv, computed_icv))
+
+#------------------------------------------------------------------------------
+# The names of the integrity algorithms are the same than in scapy.contrib.ikev2
+# see http://www.iana.org/assignments/ikev2-parameters/ikev2-parameters.xhtml
+
+AUTH_ALGOS = {
+ 'NULL': AuthAlgo('NULL', mac=None, digestmod=None, icv_size=0),
+}
+
+if HMAC:
+ if SHA:
+ AUTH_ALGOS['HMAC-SHA1-96'] = AuthAlgo('HMAC-SHA1-96',
+ mac=HMAC,
+ digestmod=SHA,
+ icv_size=12)
+ if SHA256:
+ AUTH_ALGOS['SHA2-256-128'] = AuthAlgo('SHA2-256-128',
+ mac=HMAC,
+ digestmod=SHA256,
+ icv_size=16)
+ if SHA384:
+ AUTH_ALGOS['SHA2-384-192'] = AuthAlgo('SHA2-384-192',
+ mac=HMAC,
+ digestmod=SHA384,
+ icv_size=24)
+ if SHA512:
+ AUTH_ALGOS['SHA2-512-256'] = AuthAlgo('SHA2-512-256',
+ mac=HMAC,
+ digestmod=SHA512,
+ icv_size=32)
+ if MD5:
+ AUTH_ALGOS['HMAC-MD5-96'] = AuthAlgo('HMAC-MD5-96',
+ mac=HMAC,
+ digestmod=MD5,
+ icv_size=12)
+if AES and XCBCMAC:
+ AUTH_ALGOS['AES-XCBC-96'] = AuthAlgo('AES-XCBC-96',
+ mac=XCBCMAC,
+ digestmod=AES,
+ icv_size=12,
+ key_size=(16,))
+
+#------------------------------------------------------------------------------
+
+
+#------------------------------------------------------------------------------
+def split_for_transport(orig_pkt, transport_proto):
+ """
+ Split an IP(v6) packet in the correct location to insert an ESP or AH
+ header.
+
+ @param orig_pkt: the packet to split. Must be an IP or IPv6 packet
+ @param transport_proto: the IPSec protocol number that will be inserted
+ at the split position.
+ @return: a tuple (header, nh, payload) where nh is the protocol number of
+ payload.
+ """
+ header = orig_pkt.copy()
+ next_hdr = header.payload
+ nh = None
+
+ if header.version == 4:
+ nh = header.proto
+ header.proto = transport_proto
+ header.remove_payload()
+ del header.chksum
+ del header.len
+
+ return header, nh, next_hdr
+ else:
+ found_rt_hdr = False
+ prev = header
+
+ # Since the RFC 4302 is vague about where the ESP/AH headers should be
+ # inserted in IPv6, I chose to follow the linux implementation.
+ while isinstance(next_hdr, (IPv6ExtHdrHopByHop, IPv6ExtHdrRouting, IPv6ExtHdrDestOpt)):
+ if isinstance(next_hdr, IPv6ExtHdrHopByHop):
+ pass
+ if isinstance(next_hdr, IPv6ExtHdrRouting):
+ found_rt_hdr = True
+ elif isinstance(next_hdr, IPv6ExtHdrDestOpt) and found_rt_hdr:
+ break
+
+ prev = next_hdr
+ next_hdr = next_hdr.payload
+
+ nh = prev.nh
+ prev.nh = transport_proto
+ prev.remove_payload()
+ del header.plen
+
+ return header, nh, next_hdr
+
+#------------------------------------------------------------------------------
+# see RFC 4302 - Appendix A. Mutability of IP Options/Extension Headers
+IMMUTABLE_IPV4_OPTIONS = (
+ 0, # End Of List
+ 1, # No OPeration
+ 2, # Security
+ 5, # Extended Security
+ 6, # Commercial Security
+ 20, # Router Alert
+ 21, # Sender Directed Multi-Destination Delivery
+)
+def zero_mutable_fields(pkt, sending=False):
+ """
+ When using AH, all "mutable" fields must be "zeroed" before calculating
+ the ICV. See RFC 4302, Section 3.3.3.1. Handling Mutable Fields.
+
+ @param pkt: an IP(v6) packet containing an AH layer.
+ NOTE: The packet will be modified
+ @param sending: if true, ipv6 routing headers will not be reordered
+ """
+
+ if pkt.haslayer(AH):
+ pkt[AH].icv = chr(0) * len(pkt[AH].icv)
+ else:
+ raise TypeError('no AH layer found')
+
+ if pkt.version == 4:
+ # the tos field has been replaced by DSCP and ECN
+ # Routers may rewrite the DS field as needed to provide a
+ # desired local or end-to-end service
+ pkt.tos = 0
+ # an intermediate router might set the DF bit, even if the source
+ # did not select it.
+ pkt.flags = 0
+ # changed en route as a normal course of processing by routers
+ pkt.ttl = 0
+ # will change if any of these other fields change
+ pkt.chksum = 0
+
+ immutable_opts = []
+ for opt in pkt.options:
+ if opt.option in IMMUTABLE_IPV4_OPTIONS:
+ immutable_opts.append(opt)
+ else:
+ immutable_opts.append(Raw(chr(0) * len(opt)))
+ pkt.options = immutable_opts
+
+ else:
+ # holds DSCP and ECN
+ pkt.tc = 0
+ # The flow label described in AHv1 was mutable, and in RFC 2460 [DH98]
+ # was potentially mutable. To retain compatibility with existing AH
+ # implementations, the flow label is not included in the ICV in AHv2.
+ pkt.fl = 0
+ # same as ttl
+ pkt.hlim = 0
+
+ next_hdr = pkt.payload
+
+ while isinstance(next_hdr, (IPv6ExtHdrHopByHop, IPv6ExtHdrRouting, IPv6ExtHdrDestOpt)):
+ if isinstance(next_hdr, (IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt)):
+ for opt in next_hdr.options:
+ if opt.otype & 0x20:
+ # option data can change en-route and must be zeroed
+ opt.optdata = chr(0) * opt.optlen
+ elif isinstance(next_hdr, IPv6ExtHdrRouting) and sending:
+ # The sender must order the field so that it appears as it
+ # will at the receiver, prior to performing the ICV computation.
+ next_hdr.segleft = 0
+ if next_hdr.addresses:
+ final = next_hdr.addresses.pop()
+ next_hdr.addresses.insert(0, pkt.dst)
+ pkt.dst = final
+ else:
+ break
+
+ next_hdr = next_hdr.payload
+
+ return pkt
+
+#------------------------------------------------------------------------------
+class SecurityAssociation(object):
+ """
+ This class is responsible of "encryption" and "decryption" of IPSec packets.
+ """
+
+ SUPPORTED_PROTOS = (IP, IPv6)
+
+ def __init__(self, proto, spi, seq_num=1, crypt_algo=None, crypt_key=None,
+ auth_algo=None, auth_key=None, tunnel_header=None, nat_t_header=None):
+ """
+ @param proto: the IPSec proto to use (ESP or AH)
+ @param spi: the Security Parameters Index of this SA
+ @param seq_num: the initial value for the sequence number on encrypted
+ packets
+ @param crypt_algo: the encryption algorithm name (only used with ESP)
+ @param crypt_key: the encryption key (only used with ESP)
+ @param auth_algo: the integrity algorithm name
+ @param auth_key: the integrity key
+ @param tunnel_header: an instance of a IP(v6) header that will be used
+ to encapsulate the encrypted packets.
+ @param nat_t_header: an instance of a UDP header that will be used
+ for NAT-Traversal.
+ """
+
+ if proto not in (ESP, AH, ESP.name, AH.name):
+ raise ValueError("proto must be either ESP or AH")
+ if isinstance(proto, basestring):
+ self.proto = eval(proto)
+ else:
+ self.proto = proto
+
+ self.spi = spi
+ self.seq_num = seq_num
+
+ if crypt_algo:
+ if crypt_algo not in CRYPT_ALGOS:
+ raise TypeError('unsupported encryption algo %r, try %r' %
+ (crypt_algo, CRYPT_ALGOS.keys()))
+ self.crypt_algo = CRYPT_ALGOS[crypt_algo]
+ self.crypt_algo.check_key(crypt_key)
+ self.crypt_key = crypt_key
+ else:
+ self.crypt_algo = CRYPT_ALGOS['NULL']
+ self.crypt_key = None
+
+ if auth_algo:
+ if auth_algo not in AUTH_ALGOS:
+ raise TypeError('unsupported integrity algo %r, try %r' %
+ (auth_algo, AUTH_ALGOS.keys()))
+ self.auth_algo = AUTH_ALGOS[auth_algo]
+ self.auth_algo.check_key(auth_key)
+ self.auth_key = auth_key
+ else:
+ self.auth_algo = AUTH_ALGOS['NULL']
+ self.auth_key = None
+
+ if tunnel_header and not isinstance(tunnel_header, (IP, IPv6)):
+ raise TypeError('tunnel_header must be %s or %s' % (IP.name, IPv6.name))
+ self.tunnel_header = tunnel_header
+
+ if nat_t_header:
+ if proto is not ESP:
+ raise TypeError('nat_t_header is only allowed with ESP')
+ if not isinstance(nat_t_header, UDP):
+ raise TypeError('nat_t_header must be %s' % UDP.name)
+ self.nat_t_header = nat_t_header
+
+ def check_spi(self, pkt):
+ if pkt.spi != self.spi:
+ raise TypeError('packet spi=0x%x does not match the SA spi=0x%x' %
+ (pkt.spi, self.spi))
+
+ def _encrypt_esp(self, pkt, seq_num=None, iv=None):
+
+ if iv is None:
+ iv = self.crypt_algo.generate_iv()
+ else:
+ if len(iv) != self.crypt_algo.iv_size:
+ raise TypeError('iv length must be %s' % self.crypt_algo.iv_size)
+
+ esp = _ESPPlain(spi=self.spi, seq=seq_num or self.seq_num, iv=iv)
+
+ if self.tunnel_header:
+ tunnel = self.tunnel_header.copy()
+
+ if tunnel.version == 4:
+ del tunnel.proto
+ del tunnel.len
+ del tunnel.chksum
+ else:
+ del tunnel.nh
+ del tunnel.plen
+
+ pkt = tunnel.__class__(str(tunnel / pkt))
+
+ ip_header, nh, payload = split_for_transport(pkt, socket.IPPROTO_ESP)
+ esp.data = payload
+ esp.nh = nh
+
+ esp = self.crypt_algo.pad(esp)
+ esp = self.crypt_algo.encrypt(esp, self.crypt_key)
+
+ self.auth_algo.sign(esp, self.auth_key)
+
+ if self.nat_t_header:
+ nat_t_header = self.nat_t_header.copy()
+ nat_t_header.chksum = 0
+ del nat_t_header.len
+ if ip_header.version == 4:
+ del ip_header.proto
+ else:
+ del ip_header.nh
+ ip_header /= nat_t_header
+
+ if ip_header.version == 4:
+ ip_header.len = len(ip_header) + len(esp)
+ del ip_header.chksum
+ ip_header = ip_header.__class__(str(ip_header))
+ else:
+ ip_header.plen = len(ip_header.payload) + len(esp)
+
+ # sequence number must always change, unless specified by the user
+ if seq_num is None:
+ self.seq_num += 1
+
+ return ip_header / esp
+
+ def _encrypt_ah(self, pkt, seq_num=None):
+
+ ah = AH(spi=self.spi, seq=seq_num or self.seq_num,
+ icv=chr(0) * self.auth_algo.icv_size)
+
+ if self.tunnel_header:
+ tunnel = self.tunnel_header.copy()
+
+ if tunnel.version == 4:
+ del tunnel.proto
+ del tunnel.len
+ del tunnel.chksum
+ else:
+ del tunnel.nh
+ del tunnel.plen
+
+ pkt = tunnel.__class__(str(tunnel / pkt))
+
+ ip_header, nh, payload = split_for_transport(pkt, socket.IPPROTO_AH)
+ ah.nh = nh
+
+ if ip_header.version == 6 and len(ah) % 8 != 0:
+ # For IPv6, the total length of the header must be a multiple of
+ # 8-octet units.
+ ah.padding = chr(0) * (-len(ah) % 8)
+ elif len(ah) % 4 != 0:
+ # For IPv4, the total length of the header must be a multiple of
+ # 4-octet units.
+ ah.padding = chr(0) * (-len(ah) % 4)
+
+ # RFC 4302 - Section 2.2. Payload Length
+ # This 8-bit field specifies the length of AH in 32-bit words (4-byte
+ # units), minus "2".
+ ah.payloadlen = len(ah) / 4 - 2
+
+ if ip_header.version == 4:
+ ip_header.len = len(ip_header) + len(ah) + len(payload)
+ del ip_header.chksum
+ ip_header = ip_header.__class__(str(ip_header))
+ else:
+ ip_header.plen = len(ip_header.payload) + len(ah) + len(payload)
+
+ signed_pkt = self.auth_algo.sign(ip_header / ah / payload, self.auth_key)
+
+ # sequence number must always change, unless specified by the user
+ if seq_num is None:
+ self.seq_num += 1
+
+ return signed_pkt
+
+ def encrypt(self, pkt, seq_num=None, iv=None):
+ """
+ Encrypt (and encapsulate) an IP(v6) packet with ESP or AH according
+ to this SecurityAssociation.
+
+ @param pkt: the packet to encrypt
+ @param seq_num: if specified, use this sequence number instead of the
+ generated one
+ @param iv: if specified, use this initialization vector for
+ encryption instead of a random one.
+
+ @return: the encrypted/encapsulated packet
+ """
+ if not isinstance(pkt, self.SUPPORTED_PROTOS):
+ raise TypeError('cannot encrypt %s, supported protos are %s'
+ % (pkt.__class__, self.SUPPORTED_PROTOS))
+ if self.proto is ESP:
+ return self._encrypt_esp(pkt, seq_num=seq_num, iv=iv)
+ else:
+ return self._encrypt_ah(pkt, seq_num=seq_num)
+
+ def _decrypt_esp(self, pkt, verify=True):
+
+ encrypted = pkt[ESP]
+
+ if verify:
+ self.check_spi(pkt)
+ self.auth_algo.verify(encrypted, self.auth_key)
+
+ esp = self.crypt_algo.decrypt(encrypted, self.crypt_key,
+ self.auth_algo.icv_size)
+
+ if self.tunnel_header:
+ # drop the tunnel header and return the payload untouched
+
+ pkt.remove_payload()
+ if pkt.version == 4:
+ pkt.proto = esp.nh
+ else:
+ pkt.nh = esp.nh
+ cls = pkt.guess_payload_class(esp.data)
+
+ return cls(esp.data)
+ else:
+ ip_header = pkt
+
+ if ip_header.version == 4:
+ ip_header.proto = esp.nh
+ del ip_header.chksum
+ ip_header.remove_payload()
+ ip_header.len = len(ip_header) + len(esp.data)
+ # recompute checksum
+ ip_header = ip_header.__class__(str(ip_header))
+ else:
+ encrypted.underlayer.nh = esp.nh
+ encrypted.underlayer.remove_payload()
+ ip_header.plen = len(ip_header.payload) + len(esp.data)
+
+ cls = ip_header.guess_payload_class(esp.data)
+
+ # reassemble the ip_header with the ESP payload
+ return ip_header / cls(esp.data)
+
+ def _decrypt_ah(self, pkt, verify=True):
+
+ if verify:
+ self.check_spi(pkt)
+ self.auth_algo.verify(pkt, self.auth_key)
+
+ ah = pkt[AH]
+ payload = ah.payload
+ payload.remove_underlayer(None) # useless argument...
+
+ if self.tunnel_header:
+ return payload
+ else:
+ ip_header = pkt
+
+ if ip_header.version == 4:
+ ip_header.proto = ah.nh
+ del ip_header.chksum
+ ip_header.remove_payload()
+ ip_header.len = len(ip_header) + len(payload)
+ # recompute checksum
+ ip_header = ip_header.__class__(str(ip_header))
+ else:
+ ah.underlayer.nh = ah.nh
+ ah.underlayer.remove_payload()
+ ip_header.plen = len(ip_header.payload) + len(payload)
+
+ # reassemble the ip_header with the AH payload
+ return ip_header / payload
+
+ def decrypt(self, pkt, verify=True):
+ """
+ Decrypt (and decapsulate) an IP(v6) packet containing ESP or AH.
+
+ @param pkt: the packet to decrypt
+ @param verify: if False, do not perform the integrity check
+
+ @return: the decrypted/decapsulated packet
+ @raise IPSecIntegrityError: if the integrity check fails
+ """
+ if not isinstance(pkt, self.SUPPORTED_PROTOS):
+ raise TypeError('cannot decrypt %s, supported protos are %s'
+ % (pkt.__class__, self.SUPPORTED_PROTOS))
+
+ if self.proto is ESP and pkt.haslayer(ESP):
+ return self._decrypt_esp(pkt, verify=verify)
+ elif self.proto is AH and pkt.haslayer(AH):
+ return self._decrypt_ah(pkt, verify=verify)
+ else:
+ raise TypeError('%s has no %s layer' % (pkt, self.proto.name))
diff --git a/test/ipsec.uts b/test/ipsec.uts
new file mode 100644
index 0000000000000000000000000000000000000000..01f13571417ecfc6aef0a2bcd0d306ca4a2de6d2
--- /dev/null
+++ b/test/ipsec.uts
@@ -0,0 +1,752 @@
+##############################
+% IPSec layer regression tests
+##############################
+
+###############################################################################
++ IPv4 / ESP
+
+#######################################
+= IPv4 / ESP - Transport - AES-CBC - NULL
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='AES-CBC', crypt_key='sixteenbytes key',
+ auth_algo='NULL', auth_key=None)
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '1.1.1.1' and e.dst == '2.2.2.2')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should NOT be readable
+assert('testdata' not in e[ESP].data)
+
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv4 / ESP - Transport - NULL - HMAC-SHA1-96
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '1.1.1.1' and e.dst == '2.2.2.2')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+assert('testdata' in e[ESP].data)
+
+* integrity verification should pass
+d = sa.decrypt(e)
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv4 / ESP - Transport - NULL - HMAC-SHA1-96 - altered packet
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '1.1.1.1' and e.dst == '2.2.2.2')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+assert('testdata' in e[ESP].data)
+
+* simulate the alteration of the packet before decryption
+e[ESP].data = e[ESP].data.replace('\x01', '\x21')
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+#######################################
+= IPv4 / ESP - Tunnel - AES-CTR - NULL
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='AES-CTR', crypt_key='16bytekey+4bytenonce',
+ auth_algo='NULL', auth_key=None,
+ tunnel_header=IP(src='11.11.11.11', dst='22.22.22.22'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == '11.11.11.11' and e.dst == '22.22.22.22')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should NOT be readable
+assert('testdata' not in e[ESP].data)
+
+d = sa.decrypt(e)
+d
+
+* after decryption original packet should be preserved
+assert(d == p)
+
+#######################################
+= IPv4 / ESP - Tunnel - NULL - SHA2-256-128
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='SHA2-256-128', auth_key='secret key',
+ tunnel_header=IP(src='11.11.11.11', dst='22.22.22.22'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == '11.11.11.11' and e.dst == '22.22.22.22')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* integrity verification should pass
+d = sa.decrypt(e)
+
+* after decryption the original packet should be preserved
+assert(d == p)
+
+#######################################
+= IPv4 / ESP - Tunnel - NULL - SHA2-256-128 - altered packet
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='SHA2-256-128', auth_key='secret key',
+ tunnel_header=IP(src='11.11.11.11', dst='22.22.22.22'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == '11.11.11.11' and e.dst == '22.22.22.22')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* simulate the alteration of the packet before decryption
+e[ESP].data = e[ESP].data.replace('\x01', '\x21')
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+###############################################################################
++ IPv6 / ESP
+
+#######################################
+= IPv6 / ESP - Transport - DES - NULL
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='DES', crypt_key='8bytekey',
+ auth_algo='NULL', auth_key=None)
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+assert(e.src == '11::22' and e.dst == '22::11')
+* the encrypted packet should have an ESP layer
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should NOT be readable
+assert('testdata' not in e[ESP].data)
+
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv6 / ESP - Transport - NULL - HMAC-MD5-96
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='HMAC-MD5-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+assert(e.src == '11::22' and e.dst == '22::11')
+* the encrypted packet should have an ESP layer
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* integrity verification should pass
+d = sa.decrypt(e)
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv6 / ESP - Transport - NULL - HMAC-MD5-96 - altered packet
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='HMAC-MD5-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+assert(e.src == '11::22' and e.dst == '22::11')
+* the encrypted packet should have an ESP layer
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* simulate the alteration of the packet before decryption
+e[ESP].data = e[ESP].data.replace('\x01', '\x21')
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+#######################################
+= IPv6 / ESP - Tunnel - 3DES - NULL
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='AES-CBC', crypt_key='sixteenbytes key',
+ auth_algo='NULL', auth_key=None,
+ tunnel_header=IPv6(src='aa::bb', dst='bb::aa'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == 'aa::bb' and e.dst == 'bb::aa')
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should NOT be readable
+assert('testdata' not in e[ESP].data)
+
+d = sa.decrypt(e)
+d
+
+* after decryption original packet should be preserved
+assert(d == p)
+
+#######################################
+= IPv6 / ESP - Tunnel - NULL - SHA2-384-192
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='SHA2-384-192', auth_key='secret key',
+ tunnel_header=IPv6(src='aa::bb', dst='bb::aa'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == 'aa::bb' and e.dst == 'bb::aa')
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* integrity verification should pass
+d = sa.decrypt(e)
+
+* after decryption the original packet should be preserved
+assert(d == p)
+
+#######################################
+= IPv6 / ESP - Tunnel - NULL - SHA2-384-192 - altered packet
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(ESP, spi=0x222,
+ crypt_algo='NULL', crypt_key=None,
+ auth_algo='SHA2-384-192', auth_key='secret key',
+ tunnel_header=IPv6(src='aa::bb', dst='bb::aa'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == 'aa::bb' and e.dst == 'bb::aa')
+assert(e.nh == socket.IPPROTO_ESP)
+assert(e.haslayer(ESP))
+assert(not e.haslayer(TCP))
+assert(e[ESP].spi == sa.spi)
+* after encryption the original packet payload should be readable
+assert('testdata' in e[ESP].data)
+
+* simulate the alteration of the packet before decryption
+e[ESP].data = e[ESP].data.replace('\x01', '\x21')
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+###############################################################################
++ IPv4 / AH
+
+#######################################
+= IPv4 / AH - Transport - HMAC-SHA1-96
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='sixteenbytes key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '1.1.1.1' and e.dst == '2.2.2.2')
+assert(e.chksum != p.chksum)
+* the encrypted packet should have an AH layer
+assert(e.proto == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* alter mutable fields in the packet
+e.ttl = 2
+
+* integrity verification should pass
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv4 / AH - Transport - HMAC-SHA1-96 - altered packet
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='sixteenbytes key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '1.1.1.1' and e.dst == '2.2.2.2')
+assert(e.chksum != p.chksum)
+* the encrypted packet should have an AH layer
+assert(e.proto == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* simulate the alteration of the packet before decryption
+e[TCP].sport = 5
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+#######################################
+= IPv4 / AH - Tunnel - SHA2-256-128
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='SHA2-256-128', auth_key='secret key',
+ tunnel_header=IP(src='11.11.11.11', dst='22.22.22.22'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '11.11.11.11' and e.dst == '22.22.22.22')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* alter mutable fields in the packet
+e.ttl = 2
+
+* integrity verification should pass
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet should be unaltered
+assert(d == p)
+
+#######################################
+= IPv4 / AH - Tunnel - HMAC-SHA1-96 - altered packet
+
+p = IP(src='1.1.1.1', dst='2.2.2.2')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IP(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key',
+ tunnel_header=IP(src='11.11.11.11', dst='22.22.22.22'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IP))
+assert(e.src == '11.11.11.11' and e.dst == '22.22.22.22')
+assert(e.chksum != p.chksum)
+assert(e.proto == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* simulate the alteration of the packet before verification
+e.dst = '4.4.4.4'
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+###############################################################################
++ IPv6 / AH
+
+#######################################
+= IPv6 / AH - Transport - HMAC-SHA1-96
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+assert(e.src == '11::22' and e.dst == '22::11')
+* the encrypted packet should have an AH layer
+assert(e.nh == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* alter mutable fields in the packet
+e.hlim = 2
+
+* integrity verification should pass
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet payload should be unaltered
+assert(d[TCP] == p[TCP])
+
+#######################################
+= IPv6 / AH - Transport - HMAC-SHA1-96 - altered packet
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+assert(e.src == '11::22' and e.dst == '22::11')
+* the encrypted packet should have an AH layer
+assert(e.nh == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* simulate the alteration of the packet before verification
+e[TCP].dport = 46
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+#######################################
+= IPv6 / AH - Tunnel - HMAC-SHA1-96
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key',
+ tunnel_header=IPv6(src='aa::bb', dst='bb::aa'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == 'aa::bb' and e.dst == 'bb::aa')
+assert(e.nh == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* alter mutable fields in the packet
+e.hlim = 2
+
+* integrity verification should pass
+d = sa.decrypt(e)
+d
+
+* after decryption the original packet payload should be unaltered
+assert(d == p)
+
+#######################################
+= IPv6 / AH - Tunnel - HMAC-SHA1-96 - altered packet
+
+p = IPv6(src='11::22', dst='22::11')
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key',
+ tunnel_header=IPv6(src='aa::bb', dst='bb::aa'))
+
+e = sa.encrypt(p)
+e
+
+assert(isinstance(e, IPv6))
+* after encryption packet should be encapsulated with the given ip tunnel header
+assert(e.src == 'aa::bb' and e.dst == 'bb::aa')
+assert(e.nh == socket.IPPROTO_AH)
+assert(e.haslayer(AH))
+assert(e.haslayer(TCP))
+assert(e[AH].spi == sa.spi)
+
+* simulate the alteration of the packet before verification
+e.src = 'cc::ee'
+
+* integrity verification should fail
+try:
+ d = sa.decrypt(e)
+ assert(False)
+except IPSecIntegrityError, err:
+ err
+
+###############################################################################
++ IPv6 + Extensions / AH
+
+#######################################
+= IPv6 + Extensions / AH - Transport
+
+p = IPv6(src='11::22', dst='22::11')
+p /= IPv6ExtHdrHopByHop()
+p /= IPv6ExtHdrDestOpt()
+p /= IPv6ExtHdrRouting()
+p /= IPv6ExtHdrDestOpt()
+p /= IPv6ExtHdrFragment()
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(e.src == '11::22' and e.dst == '22::11')
+* AH header should be inserted between the routing header and the dest options header
+assert(isinstance(e[AH].underlayer, IPv6ExtHdrRouting))
+assert(isinstance(e[AH].payload, IPv6ExtHdrDestOpt))
+
+#######################################
+= IPv6 + Routing Header / AH - Transport
+
+p = IPv6(src='11::22', dst='22::11')
+p /= IPv6ExtHdrHopByHop()
+p /= IPv6ExtHdrRouting(addresses=['aa::bb', 'cc::dd', 'ee::ff'])
+p /= TCP(sport=45012, dport=80)
+p /= Raw('testdata')
+p = IPv6(str(p))
+p
+
+sa = SecurityAssociation(AH, spi=0x222,
+ auth_algo='HMAC-SHA1-96', auth_key='secret key')
+
+e = sa.encrypt(p)
+e
+
+assert(e.src == '11::22' and e.dst == '22::11')
+* AH header should be inserted between the routing header and TCP
+assert(isinstance(e[AH].underlayer, IPv6ExtHdrRouting))
+assert(isinstance(e[AH].payload, TCP))
+
+* reorder the routing header as the receiver will get it
+final = e[IPv6ExtHdrRouting].addresses.pop()
+e[IPv6ExtHdrRouting].addresses.insert(0, e.dst)
+e.dst = final
+e[IPv6ExtHdrRouting].segleft = 0
+
+* integrity verification should pass
+d = sa.decrypt(e)
+d