1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
|
diff --git a/scapy/layers/ipsec.py b/scapy/layers/ipsec.py
index 8251dc14..a8390fc5 100644
--- a/scapy/layers/ipsec.py
+++ b/scapy/layers/ipsec.py
@@ -60,7 +60,7 @@ import scapy.modules.six as six
from scapy.modules.six.moves import range
from scapy.layers.inet6 import IPv6, IPv6ExtHdrHopByHop, IPv6ExtHdrDestOpt, \
IPv6ExtHdrRouting
-
+from scapy.contrib.mpls import MPLS
###############################################################################
class AH(Packet):
@@ -360,13 +360,16 @@ class CryptAlgo(object):
encryptor = cipher.encryptor()
if self.is_aead:
- if esn_en:
- aad = struct.pack('!LLL', esp.spi, esn, esp.seq)
- else:
- aad = struct.pack('!LL', esp.spi, esp.seq)
+ aad = sa.build_aead(esp)
+ if self.name == 'AES-NULL-GMAC':
+ aad = aad + esp.iv + data
+ aes_null_gmac_data = data
+ data = b''
encryptor.authenticate_additional_data(aad)
data = encryptor.update(data) + encryptor.finalize()
data += encryptor.tag[:self.icv_size]
+ if self.name == 'AES-NULL-GMAC':
+ data = aes_null_gmac_data + data
else:
data = encryptor.update(data) + encryptor.finalize()
@@ -402,16 +405,18 @@ class CryptAlgo(object):
if self.is_aead:
# Tag value check is done during the finalize method
- if esn_en:
- decryptor.authenticate_additional_data(
- struct.pack('!LLL', esp.spi, esn, esp.seq))
- else:
- decryptor.authenticate_additional_data(
- struct.pack('!LL', esp.spi, esp.seq))
+ aad = sa.build_aead(esp)
+ if self.name == 'AES-NULL-GMAC':
+ aad = aad + iv + data
+ aes_null_gmac_data = data
+ data = b''
+ decryptor.authenticate_additional_data(aad)
try:
data = decryptor.update(data) + decryptor.finalize()
except InvalidTag as err:
raise IPSecIntegrityError(err)
+ if self.name == 'AES-NULL-GMAC':
+ data = aes_null_gmac_data + data
# extract padlen and nh
padlen = orb(data[-2])
@@ -458,6 +463,13 @@ if algorithms:
iv_size=8,
icv_size=16,
format_mode_iv=_salt_format_mode_iv)
+ CRYPT_ALGOS['AES-NULL-GMAC'] = CryptAlgo('AES-NULL-GMAC',
+ 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,
@@ -546,7 +558,7 @@ class AuthAlgo(object):
else:
return self.mac(key, self.digestmod(), default_backend())
- def sign(self, pkt, key, esn_en=False, esn=0):
+ def sign(self, pkt, key, trailer=None, esn_en=False, esn=0):
"""
Sign an IPsec (ESP or AH) packet with this algo.
@@ -565,6 +577,8 @@ class AuthAlgo(object):
if pkt.haslayer(ESP):
mac.update(raw(pkt[ESP]))
+ if trailer:
+ mac.update(trailer)
pkt[ESP].data += mac.finalize()[:self.icv_size]
elif pkt.haslayer(AH):
@@ -574,11 +588,13 @@ class AuthAlgo(object):
else:
temp = raw(clone)
mac.update(temp)
+ if trailer:
+ mac.update(trailer)
pkt[AH].icv = mac.finalize()[:self.icv_size]
return pkt
- def verify(self, pkt, key, esn_en=False, esn=0):
+ def verify(self, pkt, key, trailer, esn_en=False, esn=0):
"""
Check that the integrity check value (icv) of a packet is valid.
@@ -617,6 +633,8 @@ class AuthAlgo(object):
temp = raw(clone)
mac.update(temp)
+ if trailer:
+ mac.update(trailer) # bytearray(4)) #raw(trailer))
computed_icv = mac.finalize()[:self.icv_size]
# XXX: Cannot use mac.verify because the ICV can be truncated
@@ -805,7 +823,7 @@ class SecurityAssociation(object):
This class is responsible of "encryption" and "decryption" of IPsec packets. # noqa: E501
"""
- SUPPORTED_PROTOS = (IP, IPv6)
+ SUPPORTED_PROTOS = (IP, IPv6, MPLS)
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, esn_en=False, esn=0): # noqa: E501
@@ -880,6 +898,23 @@ class SecurityAssociation(object):
raise TypeError('nat_t_header must be %s' % UDP.name)
self.nat_t_header = nat_t_header
+ def build_aead(self, esp):
+ if self.esn_en:
+ return (struct.pack('!LLL', esp.spi, self.seq_num >> 32, esp.seq))
+ else:
+ return (struct.pack('!LL', esp.spi, esp.seq))
+
+ def build_seq_num(self, num):
+ # only lower order bits are transmitted
+ # higher order bits are used in the ICV
+ lower = num & 0xffffffff
+ upper = num >> 32
+
+ if self.esn_en:
+ return lower, struct.pack("!I", upper)
+ else:
+ return lower, None
+
def check_spi(self, pkt):
if pkt.spi != self.spi:
raise TypeError('packet spi=0x%x does not match the SA spi=0x%x' %
@@ -893,7 +928,8 @@ class SecurityAssociation(object):
if len(iv) != self.crypt_algo.iv_size:
raise TypeError('iv length must be %s' % self.crypt_algo.iv_size) # noqa: E501
- esp = _ESPPlain(spi=self.spi, seq=seq_num or self.seq_num, iv=iv)
+ low_seq_num, high_seq_num = self.build_seq_num(seq_num or self.seq_num)
+ esp = _ESPPlain(spi=self.spi, seq=low_seq_num, iv=iv)
if self.tunnel_header:
tunnel = self.tunnel_header.copy()
@@ -917,7 +953,7 @@ class SecurityAssociation(object):
esn_en=esn_en or self.esn_en,
esn=esn or self.esn)
- self.auth_algo.sign(esp, self.auth_key)
+ self.auth_algo.sign(esp, self.auth_key, high_seq_num)
if self.nat_t_header:
nat_t_header = self.nat_t_header.copy()
@@ -944,7 +980,8 @@ class SecurityAssociation(object):
def _encrypt_ah(self, pkt, seq_num=None, esn_en=False, esn=0):
- ah = AH(spi=self.spi, seq=seq_num or self.seq_num,
+ low_seq_num, high_seq_num = self.build_seq_num(seq_num or self.seq_num)
+ ah = AH(spi=self.spi, seq=low_seq_num,
icv=b"\x00" * self.auth_algo.icv_size)
if self.tunnel_header:
@@ -985,7 +1022,7 @@ class SecurityAssociation(object):
ip_header.plen = len(ip_header.payload) + len(ah) + len(payload)
signed_pkt = self.auth_algo.sign(ip_header / ah / payload,
- self.auth_key,
+ self.auth_key, high_seq_num,
esn_en=esn_en or self.esn_en,
esn=esn or self.esn)
@@ -1025,11 +1062,12 @@ class SecurityAssociation(object):
def _decrypt_esp(self, pkt, verify=True, esn_en=None, esn=None):
+ low_seq_num, high_seq_num = self.build_seq_num(self.seq_num)
encrypted = pkt[ESP]
if verify:
self.check_spi(pkt)
- self.auth_algo.verify(encrypted, self.auth_key)
+ self.auth_algo.verify(encrypted, self.auth_key, high_seq_num)
esp = self.crypt_algo.decrypt(self, encrypted, self.crypt_key,
self.crypt_algo.icv_size or
@@ -1070,9 +1108,10 @@ class SecurityAssociation(object):
def _decrypt_ah(self, pkt, verify=True, esn_en=None, esn=None):
+ low_seq_num, high_seq_num = self.build_seq_num(self.seq_num)
if verify:
self.check_spi(pkt)
- self.auth_algo.verify(pkt, self.auth_key,
+ self.auth_algo.verify(pkt, self.auth_key, high_seq_num,
esn_en=esn_en or self.esn_en,
esn=esn or self.esn)
|
