| ############################################################################# |
| ## 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(raw(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=b'\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 |
| """ |
| |
| from __future__ import absolute_import |
| from fractions import gcd |
| import os |
| import socket |
| import struct |
| |
| from scapy.config import conf, crypto_validator |
| from scapy.compat import orb, raw |
| from scapy.data import IP_PROTOS |
| from scapy.compat import * |
| from scapy.error import log_loading |
| from scapy.fields import ByteEnumField, ByteField, IntField, PacketField, \ |
| ShortField, StrField, XIntField, XStrField, XStrLenField |
| from scapy.packet import Packet, bind_layers, Raw |
| from scapy.layers.inet import IP, UDP |
| import scapy.modules.six as six |
| from scapy.modules.six.moves import range |
| from scapy.layers.inet6 import IPv6, IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, \ |
| IPv6ExtHdrRouting |
| |
| |
| #------------------------------------------------------------------------------ |
| class AH(Packet): |
| """ |
| Authentication Header |
| |
| See https://tools.ietf.org/rfc/rfc4302.txt |
| """ |
| |
| name = 'AH' |
| |
| def __get_icv_len(self): |
| """ |
| Compute the size of the ICV based on the payloadlen field. |
| Padding size is included as it can only be known from the authentication |
| algorithm provided by the Security Association. |
| """ |
| # payloadlen = length of AH in 32-bit words (4-byte units), minus "2" |
| # payloadlen = 3 32-bit word fixed fields + ICV + padding - 2 |
| # ICV = (payloadlen + 2 - 3 - padding) in 32-bit words |
| return (self.payloadlen - 1) * 4 |
| |
| fields_desc = [ |
| ByteEnumField('nh', None, IP_PROTOS), |
| ByteField('payloadlen', None), |
| ShortField('reserved', None), |
| XIntField('spi', 0x0), |
| IntField('seq', 0), |
| XStrLenField('icv', None, length_from=__get_icv_len), |
| # Padding len can only be known with the SecurityAssociation.auth_algo |
| XStrLenField('padding', None, length_from=lambda x: 0), |
| ] |
| |
| 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) |
| bind_layers(AH, IP, nh=socket.IPPROTO_IP) |
| bind_layers(AH, IPv6, nh=socket.IPPROTO_IPV6) |
| |
| #------------------------------------------------------------------------------ |
| class ESP(Packet): |
| """ |
| Encapsulated Security Payload |
| |
| See https://tools.ietf.org/rfc/rfc4303.txt |
| """ |
| name = 'ESP' |
| |
| fields_desc = [ |
| XIntField('spi', 0x0), |
| IntField('seq', 0), |
| XStrField('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 raw(self.data) + self.padding + struct.pack("BB", self.padlen, self.nh) |
| |
| #------------------------------------------------------------------------------ |
| if conf.crypto_valid: |
| from cryptography.exceptions import InvalidTag |
| from cryptography.hazmat.backends import default_backend |
| from cryptography.hazmat.primitives.ciphers import ( |
| Cipher, |
| algorithms, |
| modes, |
| ) |
| else: |
| log_loading.info("Can't import python-cryptography v1.7+. " |
| "Disabled IPsec encryption/authentication.") |
| InvalidTag = default_backend = None |
| Cipher = algorithms = modes = None |
| |
| #------------------------------------------------------------------------------ |
| def _lcm(a, b): |
| """ |
| Least Common Multiple between 2 integers. |
| """ |
| if a == 0 or b == 0: |
| return 0 |
| else: |
| return abs(a * b) // gcd(a, b) |
| |
| class CryptAlgo(object): |
| """ |
| IPsec encryption algorithm |
| """ |
| |
| def __init__(self, name, cipher, mode, block_size=None, iv_size=None, |
| key_size=None, icv_size=None, salt_size=None, format_mode_iv=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. |
| @param icv_size: the length of the Integrity Check Value of this algo. |
| Used by Combined Mode Algorithms e.g. GCM |
| @param salt_size: the length of the salt to use as the IV prefix. |
| Usually used by Counter modes e.g. CTR |
| @param format_mode_iv: function to format the Initialization Vector |
| e.g. handle the salt value |
| Default is the random buffer from `generate_iv` |
| """ |
| self.name = name |
| self.cipher = cipher |
| self.mode = mode |
| self.icv_size = icv_size |
| |
| if modes and self.mode is not None: |
| self.is_aead = issubclass(self.mode, |
| modes.ModeWithAuthenticationTag) |
| else: |
| self.is_aead = False |
| |
| if block_size is not None: |
| self.block_size = block_size |
| elif cipher is not None: |
| self.block_size = cipher.block_size // 8 |
| 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 = tuple(i // 8 for i in cipher.key_sizes) |
| else: |
| self.key_size = None |
| |
| if salt_size is None: |
| self.salt_size = 0 |
| else: |
| self.salt_size = salt_size |
| |
| if format_mode_iv is None: |
| self._format_mode_iv = lambda iv, **kw: iv |
| else: |
| self._format_mode_iv = format_mode_iv |
| |
| 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. |
| """ |
| # XXX: Handle counter modes with real counters? RFCs allow the use of |
| # XXX: random bytes for counters, so it is not wrong to do it that way |
| return os.urandom(self.iv_size) |
| |
| @crypto_validator |
| def new_cipher(self, key, mode_iv, digest=None): |
| """ |
| @param key: the secret key, a byte string |
| @param mode_iv: the initialization vector or nonce, a byte string. |
| Formatted by `format_mode_iv`. |
| @param digest: also known as tag or icv. A byte string containing the |
| digest of the encrypted data. Only use this during |
| decryption! |
| |
| @return: an initialized cipher object for this algo |
| """ |
| if self.is_aead and digest is not None: |
| # With AEAD, the mode needs the digest during decryption. |
| return Cipher( |
| self.cipher(key), |
| self.mode(mode_iv, digest, len(digest)), |
| default_backend(), |
| ) |
| else: |
| return Cipher( |
| self.cipher(key), |
| self.mode(mode_iv), |
| default_backend(), |
| ) |
| |
| 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 |
| bytes. |
| |
| @param esp: an unencrypted _ESPPlain packet |
| |
| @return: an unencrypted _ESPPlain packet with valid padding |
| """ |
| # 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 |
| |
| # Still according to the RFC, the default value for padding *MUST* be an |
| # array of bytes starting from 1 to padlen |
| # TODO: Handle padding function according to the encryption algo |
| esp.padding = struct.pack("B" * esp.padlen, *range(1, esp.padlen + 1)) |
| |
| # 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, sa, esp, key): |
| """ |
| Encrypt an ESP packet |
| |
| @param sa: the SecurityAssociation associated with the 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: |
| mode_iv = self._format_mode_iv(algo=self, sa=sa, iv=esp.iv) |
| cipher = self.new_cipher(key, mode_iv) |
| encryptor = cipher.encryptor() |
| |
| if self.is_aead: |
| aad = struct.pack('!LL', esp.spi, esp.seq) |
| encryptor.authenticate_additional_data(aad) |
| data = encryptor.update(data) + encryptor.finalize() |
| data += encryptor.tag[:self.icv_size] |
| else: |
| data = encryptor.update(data) + encryptor.finalize() |
| |
| return ESP(spi=esp.spi, seq=esp.seq, data=esp.iv + data) |
| |
| def decrypt(self, sa, esp, key, icv_size=None): |
| """ |
| Decrypt an ESP packet |
| |
| @param sa: the SecurityAssociation associated with the 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 |
| @raise IPSecIntegrityError: if the integrity check fails with an AEAD |
| algorithm |
| """ |
| if icv_size is None: |
| icv_size = self.icv_size if self.is_aead else 0 |
| |
| 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: |
| mode_iv = self._format_mode_iv(sa=sa, iv=iv) |
| cipher = self.new_cipher(key, mode_iv, icv) |
| decryptor = cipher.decryptor() |
| |
| if self.is_aead: |
| # Tag value check is done during the finalize method |
| decryptor.authenticate_additional_data( |
| struct.pack('!LL', esp.spi, esp.seq) |
| ) |
| |
| try: |
| data = decryptor.update(data) + decryptor.finalize() |
| except InvalidTag as err: |
| raise IPSecIntegrityError(err) |
| |
| # extract padlen and nh |
| padlen = orb(data[-2]) |
| nh = orb(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 algorithms: |
| CRYPT_ALGOS['AES-CBC'] = CryptAlgo('AES-CBC', |
| cipher=algorithms.AES, |
| mode=modes.CBC) |
| _aes_ctr_format_mode_iv = lambda sa, iv, **kw: sa.crypt_salt + iv + b'\x00\x00\x00\x01' |
| CRYPT_ALGOS['AES-CTR'] = CryptAlgo('AES-CTR', |
| cipher=algorithms.AES, |
| mode=modes.CTR, |
| iv_size=8, |
| salt_size=4, |
| format_mode_iv=_aes_ctr_format_mode_iv) |
| _salt_format_mode_iv = lambda sa, iv, **kw: sa.crypt_salt + iv |
| CRYPT_ALGOS['AES-GCM'] = CryptAlgo('AES-GCM', |
| cipher=algorithms.AES, |
| mode=modes.GCM, |
| salt_size=4, |
| iv_size=8, |
| icv_size=16, |
| format_mode_iv=_salt_format_mode_iv) |
| if hasattr(modes, 'CCM'): |
| CRYPT_ALGOS['AES-CCM'] = CryptAlgo('AES-CCM', |
| cipher=algorithms.AES, |
| mode=modes.CCM, |
| iv_size=8, |
| salt_size=3, |
| icv_size=16, |
| format_mode_iv=_salt_format_mode_iv) |
| # XXX: Flagged as weak by 'cryptography'. Kept for backward compatibility |
| CRYPT_ALGOS['Blowfish'] = CryptAlgo('Blowfish', |
| cipher=algorithms.Blowfish, |
| mode=modes.CBC) |
| # XXX: RFC7321 states that DES *MUST NOT* be implemented. |
| # XXX: Keep for backward compatibility? |
| # Using a TripleDES cipher algorithm for DES is done by using the same 64 |
| # bits key 3 times (done by cryptography when given a 64 bits key) |
| CRYPT_ALGOS['DES'] = CryptAlgo('DES', |
| cipher=algorithms.TripleDES, |
| mode=modes.CBC, |
| key_size=(8,)) |
| CRYPT_ALGOS['3DES'] = CryptAlgo('3DES', |
| cipher=algorithms.TripleDES, |
| mode=modes.CBC) |
| CRYPT_ALGOS['CAST'] = CryptAlgo('CAST', |
| cipher=algorithms.CAST5, |
| mode=modes.CBC) |
| |
| #------------------------------------------------------------------------------ |
| if conf.crypto_valid: |
| from cryptography.hazmat.primitives.hmac import HMAC |
| from cryptography.hazmat.primitives.cmac import CMAC |
| from cryptography.hazmat.primitives import hashes |
| else: |
| # no error if cryptography is not available but authentication won't be supported |
| HMAC = CMAC = hashes = 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)) |
| |
| @crypto_validator |
| def new_mac(self, key): |
| """ |
| @param key: a byte string |
| @return: an initialized mac object for this algo |
| """ |
| if self.mac is CMAC: |
| return self.mac(self.digestmod(key), default_backend()) |
| else: |
| return self.mac(key, self.digestmod(), default_backend()) |
| |
| 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 |
| |
| mac = self.new_mac(key) |
| |
| if pkt.haslayer(ESP): |
| mac.update(raw(pkt[ESP])) |
| pkt[ESP].data += mac.finalize()[:self.icv_size] |
| |
| elif pkt.haslayer(AH): |
| clone = zero_mutable_fields(pkt.copy(), sending=True) |
| mac.update(raw(clone)) |
| pkt[AH].icv = mac.finalize()[: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 |
| |
| 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:] |
| clone = pkt.copy() |
| clone.data = clone.data[:len(clone.data) - self.icv_size] |
| |
| elif pkt.haslayer(AH): |
| if len(pkt[AH].icv) != self.icv_size: |
| # Fill padding since we know the actual icv_size |
| pkt[AH].padding = pkt[AH].icv[self.icv_size:] |
| pkt[AH].icv = pkt[AH].icv[:self.icv_size] |
| pkt_icv = pkt[AH].icv |
| clone = zero_mutable_fields(pkt.copy(), sending=False) |
| |
| mac.update(raw(clone)) |
| computed_icv = mac.finalize()[:self.icv_size] |
| |
| # XXX: Cannot use mac.verify because the ICV can be truncated |
| 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 and hashes: |
| # XXX: NIST has deprecated SHA1 but is required by RFC7321 |
| AUTH_ALGOS['HMAC-SHA1-96'] = AuthAlgo('HMAC-SHA1-96', |
| mac=HMAC, |
| digestmod=hashes.SHA1, |
| icv_size=12) |
| AUTH_ALGOS['SHA2-256-128'] = AuthAlgo('SHA2-256-128', |
| mac=HMAC, |
| digestmod=hashes.SHA256, |
| icv_size=16) |
| AUTH_ALGOS['SHA2-384-192'] = AuthAlgo('SHA2-384-192', |
| mac=HMAC, |
| digestmod=hashes.SHA384, |
| icv_size=24) |
| AUTH_ALGOS['SHA2-512-256'] = AuthAlgo('SHA2-512-256', |
| mac=HMAC, |
| digestmod=hashes.SHA512, |
| icv_size=32) |
| # XXX:Flagged as deprecated by 'cryptography'. Kept for backward compat |
| AUTH_ALGOS['HMAC-MD5-96'] = AuthAlgo('HMAC-MD5-96', |
| mac=HMAC, |
| digestmod=hashes.MD5, |
| icv_size=12) |
| if CMAC and algorithms: |
| AUTH_ALGOS['AES-CMAC-96'] = AuthAlgo('AES-CMAC-96', |
| mac=CMAC, |
| digestmod=algorithms.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. |
| """ |
| # force resolution of default fields to avoid padding errors |
| header = orig_pkt.__class__(raw(orig_pkt)) |
| 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 = b"\x00" * 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(b"\x00" * 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 = b"\x00" * 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, six.string_types): |
| 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, list(CRYPT_ALGOS))) |
| self.crypt_algo = CRYPT_ALGOS[crypt_algo] |
| |
| if crypt_key: |
| salt_size = self.crypt_algo.salt_size |
| self.crypt_key = crypt_key[:len(crypt_key) - salt_size] |
| self.crypt_salt = crypt_key[len(crypt_key) - salt_size:] |
| else: |
| self.crypt_key = None |
| self.crypt_salt = None |
| |
| 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, list(AUTH_ALGOS))) |
| self.auth_algo = AUTH_ALGOS[auth_algo] |
| 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__(raw(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(self, 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__(raw(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 = b"\x00" * 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__(raw(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 = b"\x00" * (-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 = b"\x00" * (-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__(raw(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(self, encrypted, self.crypt_key, |
| self.crypt_algo.icv_size or |
| 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__(raw(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__(raw(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)) |
