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
|
diff --git a/scapy/layers/ipsec.py b/scapy/layers/ipsec.py
index f8c601fa..f566d288 100644
--- a/scapy/layers/ipsec.py
+++ b/scapy/layers/ipsec.py
@@ -359,11 +359,8 @@ 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)
- encryptor.authenticate_additional_data(aad)
+ encryptor.authenticate_additional_data(sa.build_aead(esp))
+
data = encryptor.update(data) + encryptor.finalize()
data += encryptor.tag[:self.icv_size]
else:
@@ -401,12 +398,7 @@ 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))
+ decryptor.authenticate_additional_data(sa.build_aead(esp))
try:
data = decryptor.update(data) + decryptor.finalize()
except InvalidTag as err:
@@ -545,7 +537,7 @@ class AuthAlgo(object):
else:
return self.mac(key, self.digestmod(), default_backend())
- def sign(self, pkt, key):
+ def sign(self, pkt, key, trailer=None):
"""
Sign an IPsec (ESP or AH) packet with this algo.
@@ -561,16 +553,20 @@ 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):
clone = zero_mutable_fields(pkt.copy(), sending=True)
mac.update(raw(clone))
+ if trailer:
+ mac.update(trailer)
pkt[AH].icv = mac.finalize()[:self.icv_size]
return pkt
- def verify(self, pkt, key):
+ def verify(self, pkt, key, trailer):
"""
Check that the integrity check value (icv) of a packet is valid.
@@ -602,6 +598,8 @@ class AuthAlgo(object):
clone = zero_mutable_fields(pkt.copy(), sending=False)
mac.update(raw(clone))
+ 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
@@ -864,6 +862,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' %
@@ -877,7 +892,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()
@@ -901,7 +917,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()
@@ -928,7 +944,8 @@ class SecurityAssociation(object):
def _encrypt_ah(self, pkt, seq_num=None):
- 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:
@@ -968,7 +985,8 @@ class SecurityAssociation(object):
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) # noqa: E501
+ signed_pkt = self.auth_algo.sign(ip_header / ah / payload,
+ self.auth_key, high_seq_num) # noqa: E501
# sequence number must always change, unless specified by the user
if seq_num is None:
@@ -1005,11 +1023,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
@@ -1050,9 +1069,10 @@ class SecurityAssociation(object):
def _decrypt_ah(self, pkt, verify=True):
+ 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)
ah = pkt[AH]
payload = ah.payload
|
