From c73cdbc5f2e860b796627c8014f1b9e151a7bbe3 Mon Sep 17 00:00:00 2001
From: Thomas Faivre <thomas.faivre@6wind.com>
Date: Thu, 23 Mar 2017 17:43:12 +0100
Subject: [PATCH] layers/ipsec: fix AH dissection
The ICV size is defined by the authentication algorithm in the SA.
it can be deduced from the payloadlen field but the padding len is
unknown aswell so there is no way to find both without the SA.
Assume everything in payloadlen is the ICV until the verification called
from the SA. Fill padding when possible.
Signed-off-by: Thomas Faivre <thomas.faivre@6wind.com>
---
scapy/layers/ipsec.py | 28 +++++++--
test/ipsec.uts | 134 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 157 insertions(+), 5 deletions(-)
diff --git a/scapy/layers/ipsec.py b/scapy/layers/ipsec.py
index fee198c3..26eea06c 100644
--- a/scapy/layers/ipsec.py
+++ b/scapy/layers/ipsec.py
@@ -47,8 +47,8 @@ import struct
from scapy.config import conf, crypto_validator
from scapy.data import IP_PROTOS
from scapy.error import log_loading
-from scapy.fields import (ByteEnumField, ByteField, StrField, XIntField,
- IntField, ShortField, PacketField)
+from scapy.fields import (ByteEnumField, ByteField, StrField, StrLenField,
+ 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,
@@ -65,14 +65,26 @@ class AH(Packet):
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),
- StrField('icv', None),
- StrField('padding', None),
+ StrLenField('icv', None, length_from=__get_icv_len),
+ # Padding len can only be known with the SecurityAssociation.auth_algo
+ StrLenField('padding', None, length_from=lambda x: 0),
]
overload_fields = {
@@ -85,6 +97,8 @@ class AH(Packet):
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):
@@ -527,7 +541,11 @@ class AuthAlgo(object):
clone.data = clone.data[:len(clone.data) - self.icv_size]
elif pkt.haslayer(AH):
- pkt_icv = pkt[AH].icv[:self.icv_size]
+ 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(str(clone))
diff --git a/test/ipsec.uts b/test/ipsec.uts
index 4e9667c9..d53d867c 100644
--- a/test/ipsec.uts
+++ b/test/ipsec.uts
@@ -3090,6 +3090,72 @@ 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 - Transport - SHA2-256-128
+
+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='SHA2-256-128', 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 - SHA2-256-128 - 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='SHA2-256-128', 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)
@@ -3158,6 +3224,74 @@ 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 / AH - Tunnel - SHA2-256-128
+
+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='SHA2-256-128', 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 - SHA2-256-128 - 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='SHA2-256-128', 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)
--
GitLab