1#!/usr/bin/python
2#
3# Copyright 2017 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16
17# pylint: disable=g-bad-todo,g-bad-file-header,wildcard-import
18from errno import *  # pylint: disable=wildcard-import
19import os
20import random
21import re
22from scapy import all as scapy
23from socket import *  # pylint: disable=wildcard-import
24import struct
25import subprocess
26import time
27import unittest
28
29import multinetwork_base
30import net_test
31import netlink
32import packets
33import xfrm
34
35XFRM_ADDR_ANY = 16 * "\x00"
36LOOPBACK = 15 * "\x00" + "\x01"
37ENCRYPTED_PAYLOAD = ("b1c74998efd6326faebe2061f00f2c750e90e76001664a80c287b150"
38                     "59e74bf949769cc6af71e51b539e7de3a2a14cb05a231b969e035174"
39                     "d98c5aa0cef1937db98889ec0d08fa408fecf616")
40ENCRYPTION_KEY = ("308146eb3bd84b044573d60f5a5fd159"
41                  "57c7d4fe567a2120f35bae0f9869ec22".decode("hex"))
42AUTH_TRUNC_KEY = "af442892cdcd0ef650e9c299f9a8436a".decode("hex")
43
44TEST_ADDR1 = "2001:4860:4860::8888"
45TEST_ADDR2 = "2001:4860:4860::8844"
46
47TEST_SPI = 0x1234
48
49ALL_ALGORITHMS = 0xffffffff
50ALGO_CBC_AES_256 = xfrm.XfrmAlgo(("cbc(aes)", 256))
51ALGO_HMAC_SHA1 = xfrm.XfrmAlgoAuth(("hmac(sha1)", 128, 96))
52
53class XfrmTest(multinetwork_base.MultiNetworkBaseTest):
54
55  @classmethod
56  def setUpClass(cls):
57    super(XfrmTest, cls).setUpClass()
58    cls.xfrm = xfrm.Xfrm()
59
60  def setUp(self):
61    # TODO: delete this when we're more diligent about deleting our SAs.
62    super(XfrmTest, self).setUp()
63    subprocess.call("ip xfrm state flush".split())
64
65  def expectIPv6EspPacketOn(self, netid, spi, seq, length):
66    packets = self.ReadAllPacketsOn(netid)
67    self.assertEquals(1, len(packets))
68    packet = packets[0]
69    self.assertEquals(IPPROTO_ESP, packet.nh)
70    spi_seq = struct.pack("!II", spi, seq)
71    self.assertEquals(spi_seq, str(packet.payload)[:len(spi_seq)])
72    self.assertEquals(length, len(packet.payload))
73
74  def assertIsUdpEncapEsp(self, packet, spi, seq, length):
75    self.assertEquals(IPPROTO_UDP, packet.proto)
76    self.assertEquals(4500, packet.dport)
77    # Skip UDP header. TODO: isn't there a better way to do this?
78    payload = str(packet.payload)[8:]
79    self.assertEquals(length, len(payload))
80    spi_seq = struct.pack("!II", ntohl(spi), seq)
81    self.assertEquals(spi_seq, str(payload)[:len(spi_seq)])
82
83  def testAddSa(self):
84    self.xfrm.AddMinimalSaInfo("::", TEST_ADDR1, htonl(TEST_SPI), IPPROTO_ESP,
85                               xfrm.XFRM_MODE_TRANSPORT, 3320,
86                               ALGO_CBC_AES_256, ENCRYPTION_KEY,
87                               ALGO_HMAC_SHA1, AUTH_TRUNC_KEY, None)
88    expected = (
89        "src :: dst 2001:4860:4860::8888\n"
90        "\tproto esp spi 0x00001234 reqid 3320 mode transport\n"
91        "\treplay-window 4 \n"
92        "\tauth-trunc hmac(sha1) 0x%s 96\n"
93        "\tenc cbc(aes) 0x%s\n"
94        "\tsel src ::/0 dst ::/0 \n" % (
95            AUTH_TRUNC_KEY.encode("hex"), ENCRYPTION_KEY.encode("hex")))
96
97    actual = subprocess.check_output("ip xfrm state".split())
98    try:
99      self.assertMultiLineEqual(expected, actual)
100    finally:
101      self.xfrm.DeleteSaInfo(TEST_ADDR1, htonl(TEST_SPI), IPPROTO_ESP)
102
103
104  @unittest.skipUnless(net_test.LINUX_VERSION < (4, 4, 0), "regression")
105  def testSocketPolicy(self):
106    # Open an IPv6 UDP socket and connect it.
107    s = socket(AF_INET6, SOCK_DGRAM, 0)
108    netid = random.choice(self.NETIDS)
109    self.SelectInterface(s, netid, "mark")
110    s.connect((TEST_ADDR1, 53))
111    saddr, sport = s.getsockname()[:2]
112    daddr, dport = s.getpeername()[:2]
113
114    # Create a selector that matches all UDP packets. It's not actually used to
115    # select traffic, that will be done by the socket policy, which selects the
116    # SA entry (i.e., xfrm state) via the SPI and reqid.
117    sel = xfrm.XfrmSelector((XFRM_ADDR_ANY, XFRM_ADDR_ANY, 0, 0, 0, 0,
118                             AF_INET6, 0, 0, IPPROTO_UDP, 0, 0))
119
120    # Create a user policy that specifies that all outbound packets matching the
121    # (essentially no-op) selector should be encrypted.
122    info = xfrm.XfrmUserpolicyInfo((sel,
123                                    xfrm.NO_LIFETIME_CFG, xfrm.NO_LIFETIME_CUR,
124                                    100, 0,
125                                    xfrm.XFRM_POLICY_OUT,
126                                    xfrm.XFRM_POLICY_ALLOW,
127                                    xfrm.XFRM_POLICY_LOCALOK,
128                                    xfrm.XFRM_SHARE_UNIQUE))
129
130    # Create a template that specifies the SPI and the protocol.
131    xfrmid = xfrm.XfrmId((XFRM_ADDR_ANY, htonl(TEST_SPI), IPPROTO_ESP))
132    tmpl = xfrm.XfrmUserTmpl((xfrmid, AF_INET6, XFRM_ADDR_ANY, 0,
133                              xfrm.XFRM_MODE_TRANSPORT, xfrm.XFRM_SHARE_UNIQUE,
134                              0,                # require
135                              ALL_ALGORITHMS,   # auth algos
136                              ALL_ALGORITHMS,   # encryption algos
137                              ALL_ALGORITHMS))  # compression algos
138
139    # Set the policy and template on our socket.
140    data = info.Pack() + tmpl.Pack()
141    s.setsockopt(IPPROTO_IPV6, xfrm.IPV6_XFRM_POLICY, data)
142
143    # Because the policy has level set to "require" (the default), attempting
144    # to send a packet results in an error, because there is no SA that
145    # matches the socket policy we set.
146    self.assertRaisesErrno(
147        EAGAIN,
148        s.sendto, net_test.UDP_PAYLOAD, (TEST_ADDR1, 53))
149
150    # Adding a matching SA causes the packet to go out encrypted. The SA's
151    # SPI must match the one in our template, and the destination address must
152    # match the packet's destination address (in tunnel mode, it has to match
153    # the tunnel destination).
154    reqid = 0
155    self.xfrm.AddMinimalSaInfo("::", TEST_ADDR1, htonl(TEST_SPI), IPPROTO_ESP,
156                               xfrm.XFRM_MODE_TRANSPORT, reqid,
157                               ALGO_CBC_AES_256, ENCRYPTION_KEY,
158                               ALGO_HMAC_SHA1, AUTH_TRUNC_KEY, None)
159
160    s.sendto(net_test.UDP_PAYLOAD, (TEST_ADDR1, 53))
161    self.expectIPv6EspPacketOn(netid, TEST_SPI, 1, 84)
162
163    # Sending to another destination doesn't work: again, no matching SA.
164    self.assertRaisesErrno(
165        EAGAIN,
166        s.sendto, net_test.UDP_PAYLOAD, (TEST_ADDR2, 53))
167
168    # Sending on another socket without the policy applied results in an
169    # unencrypted packet going out.
170    s2 = socket(AF_INET6, SOCK_DGRAM, 0)
171    self.SelectInterface(s2, netid, "mark")
172    s2.sendto(net_test.UDP_PAYLOAD, (TEST_ADDR1, 53))
173    packets = self.ReadAllPacketsOn(netid)
174    self.assertEquals(1, len(packets))
175    packet = packets[0]
176    self.assertEquals(IPPROTO_UDP, packet.nh)
177
178    # Deleting the SA causes the first socket to return errors again.
179    self.xfrm.DeleteSaInfo(TEST_ADDR1, htonl(TEST_SPI), IPPROTO_ESP)
180    self.assertRaisesErrno(
181        EAGAIN,
182        s.sendto, net_test.UDP_PAYLOAD, (TEST_ADDR1, 53))
183
184
185  def testUdpEncapWithSocketPolicy(self):
186    # TODO: test IPv6 instead of IPv4.
187    netid = random.choice(self.NETIDS)
188    myaddr = self.MyAddress(4, netid)
189    remoteaddr = self.GetRemoteAddress(4)
190
191    # Reserve a port on which to receive UDP encapsulated packets. Sending
192    # packets works without this (and potentially can send packets with a source
193    # port belonging to another application), but receiving requires the port to
194    # be bound and the encapsulation socket option enabled.
195    encap_socket = net_test.Socket(AF_INET, SOCK_DGRAM, 0)
196    encap_socket.bind((myaddr, 0))
197    encap_port = encap_socket.getsockname()[1]
198    encap_socket.setsockopt(IPPROTO_UDP, xfrm.UDP_ENCAP,
199                               xfrm.UDP_ENCAP_ESPINUDP)
200
201    # Open a socket to send traffic.
202    s = socket(AF_INET, SOCK_DGRAM, 0)
203    self.SelectInterface(s, netid, "mark")
204    s.connect((remoteaddr, 53))
205
206    # Create a UDP encap policy and template inbound and outbound and apply
207    # them to s.
208    sel = xfrm.XfrmSelector((XFRM_ADDR_ANY, XFRM_ADDR_ANY, 0, 0, 0, 0,
209                             AF_INET, 0, 0, IPPROTO_UDP, 0, 0))
210
211    # Use the same SPI both inbound and outbound because this lets us receive
212    # encrypted packets by simply replaying the packets the kernel sends.
213    in_reqid = 123
214    in_spi = htonl(TEST_SPI)
215    out_reqid = 456
216    out_spi = htonl(TEST_SPI)
217
218    # Start with the outbound policy.
219    # TODO: what happens without XFRM_SHARE_UNIQUE?
220    info = xfrm.XfrmUserpolicyInfo((sel,
221                                    xfrm.NO_LIFETIME_CFG, xfrm.NO_LIFETIME_CUR,
222                                    100, 0,
223                                    xfrm.XFRM_POLICY_OUT,
224                                    xfrm.XFRM_POLICY_ALLOW,
225                                    xfrm.XFRM_POLICY_LOCALOK,
226                                    xfrm.XFRM_SHARE_UNIQUE))
227    xfrmid = xfrm.XfrmId((XFRM_ADDR_ANY, out_spi, IPPROTO_ESP))
228    usertmpl = xfrm.XfrmUserTmpl((xfrmid, AF_INET, XFRM_ADDR_ANY, out_reqid,
229                              xfrm.XFRM_MODE_TRANSPORT, xfrm.XFRM_SHARE_UNIQUE,
230                              0,                # require
231                              ALL_ALGORITHMS,   # auth algos
232                              ALL_ALGORITHMS,   # encryption algos
233                              ALL_ALGORITHMS))  # compression algos
234
235    data = info.Pack() + usertmpl.Pack()
236    s.setsockopt(IPPROTO_IP, xfrm.IP_XFRM_POLICY, data)
237
238    # Uncomment for debugging.
239    # subprocess.call("ip xfrm policy".split())
240
241    # Create inbound and outbound SAs that specify UDP encapsulation.
242    encaptmpl = xfrm.XfrmEncapTmpl((xfrm.UDP_ENCAP_ESPINUDP, htons(encap_port),
243                                    htons(4500), 16 * "\x00"))
244    self.xfrm.AddMinimalSaInfo(myaddr, remoteaddr, out_spi, IPPROTO_ESP,
245                               xfrm.XFRM_MODE_TRANSPORT, out_reqid,
246                               ALGO_CBC_AES_256, ENCRYPTION_KEY,
247                               ALGO_HMAC_SHA1, AUTH_TRUNC_KEY, encaptmpl)
248
249    # Add an encap template that's the mirror of the outbound one.
250    encaptmpl.sport, encaptmpl.dport = encaptmpl.dport, encaptmpl.sport
251    self.xfrm.AddMinimalSaInfo(remoteaddr, myaddr, in_spi, IPPROTO_ESP,
252                               xfrm.XFRM_MODE_TRANSPORT, in_reqid,
253                               ALGO_CBC_AES_256, ENCRYPTION_KEY,
254                               ALGO_HMAC_SHA1, AUTH_TRUNC_KEY, encaptmpl)
255
256    # Uncomment for debugging.
257    # subprocess.call("ip xfrm state".split())
258
259    # Now send a packet.
260    s.sendto("foo", (remoteaddr, 53))
261    srcport = s.getsockname()[1]
262    # s.send("foo")  # TODO: WHY DOES THIS NOT WORK?
263
264    # Expect to see an UDP encapsulated packet.
265    packets = self.ReadAllPacketsOn(netid)
266    self.assertEquals(1, len(packets))
267    packet = packets[0]
268    self.assertIsUdpEncapEsp(packet, out_spi, 1, 52)
269
270    # Now test the receive path. Because we don't know how to decrypt packets,
271    # we just play back the encrypted packet that kernel sent earlier. We swap
272    # the addresses in the IP header to make the packet look like it's bound for
273    # us, but we can't do that for the port numbers because the UDP header is
274    # part of the integrity protected payload, which we can only replay as is.
275    # So the source and destination ports are swapped and the packet appears to
276    # be sent from srcport to port 53. Open another socket on that port, and
277    # apply the inbound policy to it.
278    twisted_socket = socket(AF_INET, SOCK_DGRAM, 0)
279    net_test.SetSocketTimeout(twisted_socket, 100)
280    twisted_socket.bind(("0.0.0.0", 53))
281
282    # TODO: why does this work even without the per-socket policy applied? The
283    # received packet obviously matches an SA, but don't inbound packets need to
284    # match a policy as well?
285    info.dir = xfrm.XFRM_POLICY_IN
286    xfrmid.spi = in_spi
287    usertmpl.reqid = in_reqid
288    data = info.Pack() + usertmpl.Pack()
289    twisted_socket.setsockopt(IPPROTO_IP, xfrm.IP_XFRM_POLICY, data)
290
291    # Save the payload of the packet so we can replay it back to ourselves.
292    payload = str(packet.payload)[8:]
293    spi_seq = struct.pack("!II", ntohl(in_spi), 1)
294    payload = spi_seq + payload[len(spi_seq):]
295
296    # Tamper with the packet and check that it's dropped and counted as invalid.
297    sainfo = self.xfrm.FindSaInfo(in_spi)
298    self.assertEquals(0, sainfo.stats.integrity_failed)
299    broken = payload[:25] + chr((ord(payload[25]) + 1) % 256) + payload[26:]
300    incoming = (scapy.IP(src=remoteaddr, dst=myaddr) /
301                scapy.UDP(sport=4500, dport=encap_port) / broken)
302    self.ReceivePacketOn(netid, incoming)
303    sainfo = self.xfrm.FindSaInfo(in_spi)
304    self.assertEquals(1, sainfo.stats.integrity_failed)
305
306    # Now play back the valid packet and check that we receive it.
307    incoming = (scapy.IP(src=remoteaddr, dst=myaddr) /
308                scapy.UDP(sport=4500, dport=encap_port) / payload)
309    self.ReceivePacketOn(netid, incoming)
310    data, src = twisted_socket.recvfrom(4096)
311    self.assertEquals("foo", data)
312    self.assertEquals((remoteaddr, srcport), src)
313
314    # Check that unencrypted packets are not received.
315    unencrypted = (scapy.IP(src=remoteaddr, dst=myaddr) /
316                   scapy.UDP(sport=srcport, dport=53) / "foo")
317    self.assertRaisesErrno(EAGAIN, twisted_socket.recv, 4096)
318
319
320if __name__ == "__main__":
321  unittest.main()
322