#!/usr/bin/python # # Copyright 2017 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # pylint: disable=g-bad-todo,g-bad-file-header,wildcard-import from errno import * # pylint: disable=wildcard-import from scapy import all as scapy from socket import * # pylint: disable=wildcard-import import struct import subprocess import threading import unittest import csocket import cstruct import multinetwork_base import net_test import packets import xfrm import xfrm_base ENCRYPTED_PAYLOAD = ("b1c74998efd6326faebe2061f00f2c750e90e76001664a80c287b150" "59e74bf949769cc6af71e51b539e7de3a2a14cb05a231b969e035174" "d98c5aa0cef1937db98889ec0d08fa408fecf616") TEST_ADDR1 = "2001:4860:4860::8888" TEST_ADDR2 = "2001:4860:4860::8844" XFRM_STATS_PROCFILE = "/proc/net/xfrm_stat" XFRM_STATS_OUT_NO_STATES = "XfrmOutNoStates" # IP addresses to use for tunnel endpoints. For generality, these should be # different from the addresses we send packets to. TUNNEL_ENDPOINTS = {4: "8.8.4.4", 6: TEST_ADDR2} TEST_SPI = 0x1234 TEST_SPI2 = 0x1235 class XfrmFunctionalTest(xfrm_base.XfrmLazyTest): def assertIsUdpEncapEsp(self, packet, spi, seq, length): self.assertEquals(IPPROTO_UDP, packet.proto) udp_hdr = packet[scapy.UDP] self.assertEquals(4500, udp_hdr.dport) self.assertEquals(length, len(udp_hdr)) esp_hdr, _ = cstruct.Read(str(udp_hdr.payload), xfrm.EspHdr) # FIXME: this file currently swaps SPI byte order manually, so SPI needs to # be double-swapped here. self.assertEquals(xfrm.EspHdr((spi, seq)), esp_hdr) def CreateNewSa(self, localAddr, remoteAddr, spi, reqId, encap_tmpl, null_auth=False): auth_algo = ( xfrm_base._ALGO_AUTH_NULL if null_auth else xfrm_base._ALGO_HMAC_SHA1) self.xfrm.AddSaInfo(localAddr, remoteAddr, spi, xfrm.XFRM_MODE_TRANSPORT, reqId, xfrm_base._ALGO_CBC_AES_256, auth_algo, None, encap_tmpl, None, None) def testAddSa(self): self.CreateNewSa("::", TEST_ADDR1, TEST_SPI, 3320, None) expected = ( "src :: dst 2001:4860:4860::8888\n" "\tproto esp spi 0x00001234 reqid 3320 mode transport\n" "\treplay-window 4 \n" "\tauth-trunc hmac(sha1) 0x%s 96\n" "\tenc cbc(aes) 0x%s\n" "\tsel src ::/0 dst ::/0 \n" % ( xfrm_base._AUTHENTICATION_KEY_128.encode("hex"), xfrm_base._ENCRYPTION_KEY_256.encode("hex"))) actual = subprocess.check_output("ip xfrm state".split()) # Newer versions of IP also show anti-replay context. Don't choke if it's # missing. actual = actual.replace( "\tanti-replay context: seq 0x0, oseq 0x0, bitmap 0x00000000\n", "") try: self.assertMultiLineEqual(expected, actual) finally: self.xfrm.DeleteSaInfo(TEST_ADDR1, TEST_SPI, IPPROTO_ESP) def testFlush(self): self.assertEquals(0, len(self.xfrm.DumpSaInfo())) self.CreateNewSa("::", "2000::", TEST_SPI, 1234, None) self.CreateNewSa("0.0.0.0", "192.0.2.1", TEST_SPI, 4321, None) self.assertEquals(2, len(self.xfrm.DumpSaInfo())) self.xfrm.FlushSaInfo() self.assertEquals(0, len(self.xfrm.DumpSaInfo())) def _TestSocketPolicy(self, version): # Open a UDP socket and connect it. family = net_test.GetAddressFamily(version) s = socket(family, SOCK_DGRAM, 0) netid = self.RandomNetid() self.SelectInterface(s, netid, "mark") remotesockaddr = self.GetRemoteSocketAddress(version) s.connect((remotesockaddr, 53)) saddr, sport = s.getsockname()[:2] daddr, dport = s.getpeername()[:2] if version == 5: saddr = saddr.replace("::ffff:", "") daddr = daddr.replace("::ffff:", "") reqid = 0 desc, pkt = packets.UDP(version, saddr, daddr, sport=sport) s.sendto(net_test.UDP_PAYLOAD, (remotesockaddr, 53)) self.ExpectPacketOn(netid, "Send after socket, expected %s" % desc, pkt) # Using IPv4 XFRM on a dual-stack socket requires setting an AF_INET policy # that's written in terms of IPv4 addresses. xfrm_version = 4 if version == 5 else version xfrm_family = net_test.GetAddressFamily(xfrm_version) xfrm_base.ApplySocketPolicy(s, xfrm_family, xfrm.XFRM_POLICY_OUT, TEST_SPI, reqid, None) # Because the policy has level set to "require" (the default), attempting # to send a packet results in an error, because there is no SA that # matches the socket policy we set. self.assertRaisesErrno( EAGAIN, s.sendto, net_test.UDP_PAYLOAD, (remotesockaddr, 53)) # If there is a user space key manager, calling sendto() after applying the socket policy # creates an SA whose state is XFRM_STATE_ACQ. So this just deletes it. # If there is no user space key manager, deleting SA returns ESRCH as the error code. try: self.xfrm.DeleteSaInfo(self.GetRemoteAddress(xfrm_version), TEST_SPI, IPPROTO_ESP) except IOError as e: self.assertEquals(ESRCH, e.errno, "Unexpected error when deleting ACQ SA") # Adding a matching SA causes the packet to go out encrypted. The SA's # SPI must match the one in our template, and the destination address must # match the packet's destination address (in tunnel mode, it has to match # the tunnel destination). self.CreateNewSa( net_test.GetWildcardAddress(xfrm_version), self.GetRemoteAddress(xfrm_version), TEST_SPI, reqid, None) s.sendto(net_test.UDP_PAYLOAD, (remotesockaddr, 53)) expected_length = xfrm_base.GetEspPacketLength(xfrm.XFRM_MODE_TRANSPORT, version, False, net_test.UDP_PAYLOAD, xfrm_base._ALGO_HMAC_SHA1, xfrm_base._ALGO_CBC_AES_256) self._ExpectEspPacketOn(netid, TEST_SPI, 1, expected_length, None, None) # Sending to another destination doesn't work: again, no matching SA. remoteaddr2 = self.GetOtherRemoteSocketAddress(version) self.assertRaisesErrno( EAGAIN, s.sendto, net_test.UDP_PAYLOAD, (remoteaddr2, 53)) # Sending on another socket without the policy applied results in an # unencrypted packet going out. s2 = socket(family, SOCK_DGRAM, 0) self.SelectInterface(s2, netid, "mark") s2.sendto(net_test.UDP_PAYLOAD, (remotesockaddr, 53)) pkts = self.ReadAllPacketsOn(netid) self.assertEquals(1, len(pkts)) packet = pkts[0] protocol = packet.nh if version == 6 else packet.proto self.assertEquals(IPPROTO_UDP, protocol) # Deleting the SA causes the first socket to return errors again. self.xfrm.DeleteSaInfo(self.GetRemoteAddress(xfrm_version), TEST_SPI, IPPROTO_ESP) self.assertRaisesErrno( EAGAIN, s.sendto, net_test.UDP_PAYLOAD, (remotesockaddr, 53)) # Clear the socket policy and expect a cleartext packet. xfrm_base.SetPolicySockopt(s, family, None) s.sendto(net_test.UDP_PAYLOAD, (remotesockaddr, 53)) self.ExpectPacketOn(netid, "Send after clear, expected %s" % desc, pkt) # Clearing the policy twice is safe. xfrm_base.SetPolicySockopt(s, family, None) s.sendto(net_test.UDP_PAYLOAD, (remotesockaddr, 53)) self.ExpectPacketOn(netid, "Send after clear 2, expected %s" % desc, pkt) # Clearing if a policy was never set is safe. s = socket(AF_INET6, SOCK_DGRAM, 0) xfrm_base.SetPolicySockopt(s, family, None) def testSocketPolicyIPv4(self): self._TestSocketPolicy(4) def testSocketPolicyIPv6(self): self._TestSocketPolicy(6) def testSocketPolicyMapped(self): self._TestSocketPolicy(5) # Sets up sockets and marks to correct netid def _SetupUdpEncapSockets(self): netid = self.RandomNetid() myaddr = self.MyAddress(4, netid) remoteaddr = self.GetRemoteAddress(4) # Reserve a port on which to receive UDP encapsulated packets. Sending # packets works without this (and potentially can send packets with a source # port belonging to another application), but receiving requires the port to # be bound and the encapsulation socket option enabled. encap_sock = net_test.Socket(AF_INET, SOCK_DGRAM, 0) encap_sock.bind((myaddr, 0)) encap_port = encap_sock.getsockname()[1] encap_sock.setsockopt(IPPROTO_UDP, xfrm.UDP_ENCAP, xfrm.UDP_ENCAP_ESPINUDP) # Open a socket to send traffic. s = socket(AF_INET, SOCK_DGRAM, 0) self.SelectInterface(s, netid, "mark") s.connect((remoteaddr, 53)) return netid, myaddr, remoteaddr, encap_sock, encap_port, s # Sets up SAs and applies socket policy to given socket def _SetupUdpEncapSaPair(self, myaddr, remoteaddr, in_spi, out_spi, encap_port, s, use_null_auth): in_reqid = 123 out_reqid = 456 # Create inbound and outbound SAs that specify UDP encapsulation. encaptmpl = xfrm.XfrmEncapTmpl((xfrm.UDP_ENCAP_ESPINUDP, htons(encap_port), htons(4500), 16 * "\x00")) self.CreateNewSa(myaddr, remoteaddr, out_spi, out_reqid, encaptmpl, use_null_auth) # Add an encap template that's the mirror of the outbound one. encaptmpl.sport, encaptmpl.dport = encaptmpl.dport, encaptmpl.sport self.CreateNewSa(remoteaddr, myaddr, in_spi, in_reqid, encaptmpl, use_null_auth) # Apply socket policies to s. xfrm_base.ApplySocketPolicy(s, AF_INET, xfrm.XFRM_POLICY_OUT, out_spi, out_reqid, None) # TODO: why does this work without a per-socket policy applied? # The received packet obviously matches an SA, but don't inbound packets # need to match a policy as well? (b/71541609) xfrm_base.ApplySocketPolicy(s, AF_INET, xfrm.XFRM_POLICY_IN, in_spi, in_reqid, None) # Uncomment for debugging. # subprocess.call("ip xfrm state".split()) # Check that packets can be sent and received. def _VerifyUdpEncapSocket(self, netid, remoteaddr, myaddr, encap_port, sock, in_spi, out_spi, null_auth, seq_num): # Now send a packet. sock.sendto(net_test.UDP_PAYLOAD, (remoteaddr, 53)) srcport = sock.getsockname()[1] # Expect to see an UDP encapsulated packet. pkts = self.ReadAllPacketsOn(netid) self.assertEquals(1, len(pkts)) packet = pkts[0] auth_algo = ( xfrm_base._ALGO_AUTH_NULL if null_auth else xfrm_base._ALGO_HMAC_SHA1) expected_len = xfrm_base.GetEspPacketLength( xfrm.XFRM_MODE_TRANSPORT, 4, True, net_test.UDP_PAYLOAD, auth_algo, xfrm_base._ALGO_CBC_AES_256) self.assertIsUdpEncapEsp(packet, out_spi, seq_num, expected_len) # Now test the receive path. Because we don't know how to decrypt packets, # we just play back the encrypted packet that kernel sent earlier. We swap # the addresses in the IP header to make the packet look like it's bound for # us, but we can't do that for the port numbers because the UDP header is # part of the integrity protected payload, which we can only replay as is. # So the source and destination ports are swapped and the packet appears to # be sent from srcport to port 53. Open another socket on that port, and # apply the inbound policy to it. twisted_socket = socket(AF_INET, SOCK_DGRAM, 0) csocket.SetSocketTimeout(twisted_socket, 100) twisted_socket.bind(("0.0.0.0", 53)) # Save the payload of the packet so we can replay it back to ourselves, and # replace the SPI with our inbound SPI. payload = str(packet.payload)[8:] spi_seq = xfrm.EspHdr((in_spi, seq_num)).Pack() payload = spi_seq + payload[len(spi_seq):] sainfo = self.xfrm.FindSaInfo(in_spi) start_integrity_failures = sainfo.stats.integrity_failed # Now play back the valid packet and check that we receive it. incoming = (scapy.IP(src=remoteaddr, dst=myaddr) / scapy.UDP(sport=4500, dport=encap_port) / payload) incoming = scapy.IP(str(incoming)) self.ReceivePacketOn(netid, incoming) sainfo = self.xfrm.FindSaInfo(in_spi) # TODO: break this out into a separate test # If our SPIs are different, and we aren't using null authentication, # we expect the packet to be dropped. We also expect that the integrity # failure counter to increase, as SPIs are part of the authenticated or # integrity-verified portion of the packet. if not null_auth and in_spi != out_spi: self.assertRaisesErrno(EAGAIN, twisted_socket.recv, 4096) self.assertEquals(start_integrity_failures + 1, sainfo.stats.integrity_failed) else: data, src = twisted_socket.recvfrom(4096) self.assertEquals(net_test.UDP_PAYLOAD, data) self.assertEquals((remoteaddr, srcport), src) self.assertEquals(start_integrity_failures, sainfo.stats.integrity_failed) # Check that unencrypted packets on twisted_socket are not received. unencrypted = ( scapy.IP(src=remoteaddr, dst=myaddr) / scapy.UDP( sport=srcport, dport=53) / net_test.UDP_PAYLOAD) self.assertRaisesErrno(EAGAIN, twisted_socket.recv, 4096) def _RunEncapSocketPolicyTest(self, in_spi, out_spi, use_null_auth): netid, myaddr, remoteaddr, encap_sock, encap_port, s = \ self._SetupUdpEncapSockets() self._SetupUdpEncapSaPair(myaddr, remoteaddr, in_spi, out_spi, encap_port, s, use_null_auth) # Check that UDP encap sockets work with socket policy and given SAs self._VerifyUdpEncapSocket(netid, remoteaddr, myaddr, encap_port, s, in_spi, out_spi, use_null_auth, 1) # TODO: Add tests for ESP (non-encap) sockets. def testUdpEncapSameSpisNullAuth(self): # Use the same SPI both inbound and outbound because this lets us receive # encrypted packets by simply replaying the packets the kernel sends # without having to disable authentication self._RunEncapSocketPolicyTest(TEST_SPI, TEST_SPI, True) def testUdpEncapSameSpis(self): self._RunEncapSocketPolicyTest(TEST_SPI, TEST_SPI, False) def testUdpEncapDifferentSpisNullAuth(self): self._RunEncapSocketPolicyTest(TEST_SPI, TEST_SPI2, True) def testUdpEncapDifferentSpis(self): self._RunEncapSocketPolicyTest(TEST_SPI, TEST_SPI2, False) def testUdpEncapRekey(self): # Select the two SPIs that will be used start_spi = TEST_SPI rekey_spi = TEST_SPI2 # Setup sockets netid, myaddr, remoteaddr, encap_sock, encap_port, s = \ self._SetupUdpEncapSockets() # The SAs must use null authentication, since we change SPIs on the fly # Without null authentication, this would result in an ESP authentication # error since the SPI is part of the authenticated section. The packet # would then be dropped self._SetupUdpEncapSaPair(myaddr, remoteaddr, start_spi, start_spi, encap_port, s, True) # Check that UDP encap sockets work with socket policy and given SAs self._VerifyUdpEncapSocket(netid, remoteaddr, myaddr, encap_port, s, start_spi, start_spi, True, 1) # Rekey this socket using the make-before-break paradigm. First we create # new SAs, update the per-socket policies, and only then remove the old SAs # # This allows us to switch to the new SA without breaking the outbound path. self._SetupUdpEncapSaPair(myaddr, remoteaddr, rekey_spi, rekey_spi, encap_port, s, True) # Check that UDP encap socket works with updated socket policy, sending # using new SA, but receiving on both old and new SAs self._VerifyUdpEncapSocket(netid, remoteaddr, myaddr, encap_port, s, rekey_spi, rekey_spi, True, 1) self._VerifyUdpEncapSocket(netid, remoteaddr, myaddr, encap_port, s, start_spi, rekey_spi, True, 2) # Delete old SAs self.xfrm.DeleteSaInfo(remoteaddr, start_spi, IPPROTO_ESP) self.xfrm.DeleteSaInfo(myaddr, start_spi, IPPROTO_ESP) # Check that UDP encap socket works with updated socket policy and new SAs self._VerifyUdpEncapSocket(netid, remoteaddr, myaddr, encap_port, s, rekey_spi, rekey_spi, True, 3) def testAllocSpecificSpi(self): spi = 0xABCD new_sa = self.xfrm.AllocSpi("::", IPPROTO_ESP, spi, spi) self.assertEquals(spi, new_sa.id.spi) def testAllocSpecificSpiUnavailable(self): """Attempt to allocate the same SPI twice.""" spi = 0xABCD new_sa = self.xfrm.AllocSpi("::", IPPROTO_ESP, spi, spi) self.assertEquals(spi, new_sa.id.spi) with self.assertRaisesErrno(ENOENT): new_sa = self.xfrm.AllocSpi("::", IPPROTO_ESP, spi, spi) def testAllocRangeSpi(self): start, end = 0xABCD0, 0xABCDF new_sa = self.xfrm.AllocSpi("::", IPPROTO_ESP, start, end) spi = new_sa.id.spi self.assertGreaterEqual(spi, start) self.assertLessEqual(spi, end) def testAllocRangeSpiUnavailable(self): """Attempt to allocate N+1 SPIs from a range of size N.""" start, end = 0xABCD0, 0xABCDF range_size = end - start + 1 spis = set() # Assert that allocating SPI fails when none are available. with self.assertRaisesErrno(ENOENT): # Allocating range_size + 1 SPIs is guaranteed to fail. Due to the way # kernel picks random SPIs, this has a high probability of failing before # reaching that limit. for i in xrange(range_size + 1): new_sa = self.xfrm.AllocSpi("::", IPPROTO_ESP, start, end) spi = new_sa.id.spi self.assertNotIn(spi, spis) spis.add(spi) def testSocketPolicyDstCacheV6(self): self._TestSocketPolicyDstCache(6) def testSocketPolicyDstCacheV4(self): self._TestSocketPolicyDstCache(4) def _TestSocketPolicyDstCache(self, version): """Test that destination cache is cleared with socket policy. This relies on the fact that connect() on a UDP socket populates the destination cache. """ # Create UDP socket. family = net_test.GetAddressFamily(version) netid = self.RandomNetid() s = socket(family, SOCK_DGRAM, 0) self.SelectInterface(s, netid, "mark") # Populate the socket's destination cache. remote = self.GetRemoteAddress(version) s.connect((remote, 53)) # Apply a policy to the socket. Should clear dst cache. reqid = 123 xfrm_base.ApplySocketPolicy(s, family, xfrm.XFRM_POLICY_OUT, TEST_SPI, reqid, None) # Policy with no matching SA should result in EAGAIN. If destination cache # failed to clear, then the UDP packet will be sent normally. with self.assertRaisesErrno(EAGAIN): s.send(net_test.UDP_PAYLOAD) self.ExpectNoPacketsOn(netid, "Packet not blocked by policy") def _CheckNullEncryptionTunnelMode(self, version): family = net_test.GetAddressFamily(version) netid = self.RandomNetid() local_addr = self.MyAddress(version, netid) remote_addr = self.GetRemoteAddress(version) # Borrow the address of another netId as the source address of the tunnel tun_local = self.MyAddress(version, self.RandomNetid(netid)) # For generality, pick a tunnel endpoint that's not the address we # connect the socket to. tun_remote = TUNNEL_ENDPOINTS[version] # Output self.xfrm.AddSaInfo( tun_local, tun_remote, 0xABCD, xfrm.XFRM_MODE_TUNNEL, 123, xfrm_base._ALGO_CRYPT_NULL, xfrm_base._ALGO_AUTH_NULL, None, None, None, netid) # Input self.xfrm.AddSaInfo( tun_remote, tun_local, 0x9876, xfrm.XFRM_MODE_TUNNEL, 456, xfrm_base._ALGO_CRYPT_NULL, xfrm_base._ALGO_AUTH_NULL, None, None, None, None) sock = net_test.UDPSocket(family) self.SelectInterface(sock, netid, "mark") sock.bind((local_addr, 0)) local_port = sock.getsockname()[1] remote_port = 5555 xfrm_base.ApplySocketPolicy( sock, family, xfrm.XFRM_POLICY_OUT, 0xABCD, 123, (tun_local, tun_remote)) xfrm_base.ApplySocketPolicy( sock, family, xfrm.XFRM_POLICY_IN, 0x9876, 456, (tun_remote, tun_local)) # Create and receive an ESP packet. IpType = {4: scapy.IP, 6: scapy.IPv6}[version] input_pkt = (IpType(src=remote_addr, dst=local_addr) / scapy.UDP(sport=remote_port, dport=local_port) / "input hello") input_pkt = IpType(str(input_pkt)) # Compute length, checksum. input_pkt = xfrm_base.EncryptPacketWithNull(input_pkt, 0x9876, 1, (tun_remote, tun_local)) self.ReceivePacketOn(netid, input_pkt) msg, addr = sock.recvfrom(1024) self.assertEquals("input hello", msg) self.assertEquals((remote_addr, remote_port), addr[:2]) # Send and capture a packet. sock.sendto("output hello", (remote_addr, remote_port)) packets = self.ReadAllPacketsOn(netid) self.assertEquals(1, len(packets)) output_pkt = packets[0] output_pkt, esp_hdr = xfrm_base.DecryptPacketWithNull(output_pkt) self.assertEquals(output_pkt[scapy.UDP].len, len("output_hello") + 8) self.assertEquals(remote_addr, output_pkt.dst) self.assertEquals(remote_port, output_pkt[scapy.UDP].dport) # length of the payload plus the UDP header self.assertEquals("output hello", str(output_pkt[scapy.UDP].payload)) self.assertEquals(0xABCD, esp_hdr.spi) def testNullEncryptionTunnelMode(self): """Verify null encryption in tunnel mode. This test verifies both manual assembly and disassembly of UDP packets with ESP in IPsec tunnel mode. """ for version in [4, 6]: self._CheckNullEncryptionTunnelMode(version) def _CheckNullEncryptionTransportMode(self, version): family = net_test.GetAddressFamily(version) netid = self.RandomNetid() local_addr = self.MyAddress(version, netid) remote_addr = self.GetRemoteAddress(version) # Output self.xfrm.AddSaInfo( local_addr, remote_addr, 0xABCD, xfrm.XFRM_MODE_TRANSPORT, 123, xfrm_base._ALGO_CRYPT_NULL, xfrm_base._ALGO_AUTH_NULL, None, None, None, None) # Input self.xfrm.AddSaInfo( remote_addr, local_addr, 0x9876, xfrm.XFRM_MODE_TRANSPORT, 456, xfrm_base._ALGO_CRYPT_NULL, xfrm_base._ALGO_AUTH_NULL, None, None, None, None) sock = net_test.UDPSocket(family) self.SelectInterface(sock, netid, "mark") sock.bind((local_addr, 0)) local_port = sock.getsockname()[1] remote_port = 5555 xfrm_base.ApplySocketPolicy( sock, family, xfrm.XFRM_POLICY_OUT, 0xABCD, 123, None) xfrm_base.ApplySocketPolicy( sock, family, xfrm.XFRM_POLICY_IN, 0x9876, 456, None) # Create and receive an ESP packet. IpType = {4: scapy.IP, 6: scapy.IPv6}[version] input_pkt = (IpType(src=remote_addr, dst=local_addr) / scapy.UDP(sport=remote_port, dport=local_port) / "input hello") input_pkt = IpType(str(input_pkt)) # Compute length, checksum. input_pkt = xfrm_base.EncryptPacketWithNull(input_pkt, 0x9876, 1, None) self.ReceivePacketOn(netid, input_pkt) msg, addr = sock.recvfrom(1024) self.assertEquals("input hello", msg) self.assertEquals((remote_addr, remote_port), addr[:2]) # Send and capture a packet. sock.sendto("output hello", (remote_addr, remote_port)) packets = self.ReadAllPacketsOn(netid) self.assertEquals(1, len(packets)) output_pkt = packets[0] output_pkt, esp_hdr = xfrm_base.DecryptPacketWithNull(output_pkt) # length of the payload plus the UDP header self.assertEquals(output_pkt[scapy.UDP].len, len("output_hello") + 8) self.assertEquals(remote_addr, output_pkt.dst) self.assertEquals(remote_port, output_pkt[scapy.UDP].dport) self.assertEquals("output hello", str(output_pkt[scapy.UDP].payload)) self.assertEquals(0xABCD, esp_hdr.spi) def testNullEncryptionTransportMode(self): """Verify null encryption in transport mode. This test verifies both manual assembly and disassembly of UDP packets with ESP in IPsec transport mode. """ for version in [4, 6]: self._CheckNullEncryptionTransportMode(version) def _CheckGlobalPoliciesByMark(self, version): """Tests that global policies may differ by only the mark.""" family = net_test.GetAddressFamily(version) sel = xfrm.EmptySelector(family) # Pick 2 arbitrary mark values. mark1 = xfrm.XfrmMark(mark=0xf00, mask=xfrm_base.MARK_MASK_ALL) mark2 = xfrm.XfrmMark(mark=0xf00d, mask=xfrm_base.MARK_MASK_ALL) # Create a global policy. policy = xfrm.UserPolicy(xfrm.XFRM_POLICY_OUT, sel) tmpl = xfrm.UserTemplate(AF_UNSPEC, 0xfeed, 0, None) # Create the policy with the first mark. self.xfrm.AddPolicyInfo(policy, tmpl, mark1) # Create the same policy but with the second (different) mark. self.xfrm.AddPolicyInfo(policy, tmpl, mark2) # Delete the policies individually self.xfrm.DeletePolicyInfo(sel, xfrm.XFRM_POLICY_OUT, mark1) self.xfrm.DeletePolicyInfo(sel, xfrm.XFRM_POLICY_OUT, mark2) def testGlobalPoliciesByMarkV4(self): self._CheckGlobalPoliciesByMark(4) def testGlobalPoliciesByMarkV6(self): self._CheckGlobalPoliciesByMark(6) def _CheckUpdatePolicy(self, version): """Tests that we can can update the template on a policy.""" family = net_test.GetAddressFamily(version) tmpl1 = xfrm.UserTemplate(family, 0xdead, 0, None) tmpl2 = xfrm.UserTemplate(family, 0xbeef, 0, None) sel = xfrm.EmptySelector(family) policy = xfrm.UserPolicy(xfrm.XFRM_POLICY_OUT, sel) mark = xfrm.XfrmMark(mark=0xf00, mask=xfrm_base.MARK_MASK_ALL) def _CheckTemplateMatch(tmpl): """Dump the SPD and match a single template on a single policy.""" dump = self.xfrm.DumpPolicyInfo() self.assertEquals(1, len(dump)) _, attributes = dump[0] self.assertEquals(attributes['XFRMA_TMPL'], tmpl) # Create a new policy using update. self.xfrm.UpdatePolicyInfo(policy, tmpl1, mark, None) # NEWPOLICY will not update the existing policy. This checks both that # UPDPOLICY created a policy and that NEWPOLICY will not perform updates. _CheckTemplateMatch(tmpl1) with self.assertRaisesErrno(EEXIST): self.xfrm.AddPolicyInfo(policy, tmpl2, mark, None) # Update the policy using UPDPOLICY. self.xfrm.UpdatePolicyInfo(policy, tmpl2, mark, None) # There should only be one policy after update, and it should have the # updated template. _CheckTemplateMatch(tmpl2) def testUpdatePolicyV4(self): self._CheckUpdatePolicy(4) def testUpdatePolicyV6(self): self._CheckUpdatePolicy(6) def _CheckPolicyDifferByDirection(self,version): """Tests that policies can differ only by direction.""" family = net_test.GetAddressFamily(version) tmpl = xfrm.UserTemplate(family, 0xdead, 0, None) sel = xfrm.EmptySelector(family) mark = xfrm.XfrmMark(mark=0xf00, mask=xfrm_base.MARK_MASK_ALL) policy = xfrm.UserPolicy(xfrm.XFRM_POLICY_OUT, sel) self.xfrm.AddPolicyInfo(policy, tmpl, mark) policy = xfrm.UserPolicy(xfrm.XFRM_POLICY_IN, sel) self.xfrm.AddPolicyInfo(policy, tmpl, mark) def testPolicyDifferByDirectionV4(self): self._CheckPolicyDifferByDirection(4) def testPolicyDifferByDirectionV6(self): self._CheckPolicyDifferByDirection(6) class XfrmOutputMarkTest(xfrm_base.XfrmLazyTest): def _CheckTunnelModeOutputMark(self, version, tunsrc, mark, expected_netid): """Tests sending UDP packets to tunnel mode SAs with output marks. Opens a UDP socket and binds it to a random netid, then sets up tunnel mode SAs with an output_mark of mark and sets a socket policy to use the SA. Then checks that sending on those SAs sends a packet on expected_netid, or, if expected_netid is zero, checks that sending returns ENETUNREACH. Args: version: 4 or 6. tunsrc: A string, the source address of the tunnel. mark: An integer, the output_mark to set in the SA. expected_netid: An integer, the netid to expect the kernel to send the packet on. If None, expect that sendto will fail with ENETUNREACH. """ # Open a UDP socket and bind it to a random netid. family = net_test.GetAddressFamily(version) s = socket(family, SOCK_DGRAM, 0) self.SelectInterface(s, self.RandomNetid(), "mark") # For generality, pick a tunnel endpoint that's not the address we # connect the socket to. tundst = TUNNEL_ENDPOINTS[version] tun_addrs = (tunsrc, tundst) # Create a tunnel mode SA and use XFRM_OUTPUT_MARK to bind it to netid. spi = TEST_SPI * mark reqid = 100 + spi self.xfrm.AddSaInfo(tunsrc, tundst, spi, xfrm.XFRM_MODE_TUNNEL, reqid, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, None, mark) # Set a socket policy to use it. xfrm_base.ApplySocketPolicy(s, family, xfrm.XFRM_POLICY_OUT, spi, reqid, tun_addrs) # Send a packet and check that we see it on the wire. remoteaddr = self.GetRemoteAddress(version) packetlen = xfrm_base.GetEspPacketLength(xfrm.XFRM_MODE_TUNNEL, version, False, net_test.UDP_PAYLOAD, xfrm_base._ALGO_HMAC_SHA1, xfrm_base._ALGO_CBC_AES_256) if expected_netid is not None: s.sendto(net_test.UDP_PAYLOAD, (remoteaddr, 53)) self._ExpectEspPacketOn(expected_netid, spi, 1, packetlen, tunsrc, tundst) else: with self.assertRaisesErrno(ENETUNREACH): s.sendto(net_test.UDP_PAYLOAD, (remoteaddr, 53)) def testTunnelModeOutputMarkIPv4(self): for netid in self.NETIDS: tunsrc = self.MyAddress(4, netid) self._CheckTunnelModeOutputMark(4, tunsrc, netid, netid) def testTunnelModeOutputMarkIPv6(self): for netid in self.NETIDS: tunsrc = self.MyAddress(6, netid) self._CheckTunnelModeOutputMark(6, tunsrc, netid, netid) def testTunnelModeOutputNoMarkIPv4(self): tunsrc = self.MyAddress(4, self.RandomNetid()) self._CheckTunnelModeOutputMark(4, tunsrc, 0, None) def testTunnelModeOutputNoMarkIPv6(self): tunsrc = self.MyAddress(6, self.RandomNetid()) self._CheckTunnelModeOutputMark(6, tunsrc, 0, None) def testTunnelModeOutputInvalidMarkIPv4(self): tunsrc = self.MyAddress(4, self.RandomNetid()) self._CheckTunnelModeOutputMark(4, tunsrc, 9999, None) def testTunnelModeOutputInvalidMarkIPv6(self): tunsrc = self.MyAddress(6, self.RandomNetid()) self._CheckTunnelModeOutputMark(6, tunsrc, 9999, None) def testTunnelModeOutputMarkAttributes(self): mark = 1234567 self.xfrm.AddSaInfo(TEST_ADDR1, TUNNEL_ENDPOINTS[6], 0x1234, xfrm.XFRM_MODE_TUNNEL, 100, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, None, mark) dump = self.xfrm.DumpSaInfo() self.assertEquals(1, len(dump)) sainfo, attributes = dump[0] self.assertEquals(mark, attributes["XFRMA_OUTPUT_MARK"]) def testInvalidAlgorithms(self): key = "af442892cdcd0ef650e9c299f9a8436a".decode("hex") invalid_auth = (xfrm.XfrmAlgoAuth(("invalid(algo)", 128, 96)), key) invalid_crypt = (xfrm.XfrmAlgo(("invalid(algo)", 128)), key) with self.assertRaisesErrno(ENOSYS): self.xfrm.AddSaInfo(TEST_ADDR1, TEST_ADDR2, 0x1234, xfrm.XFRM_MODE_TRANSPORT, 0, xfrm_base._ALGO_CBC_AES_256, invalid_auth, None, None, None, 0) with self.assertRaisesErrno(ENOSYS): self.xfrm.AddSaInfo(TEST_ADDR1, TEST_ADDR2, 0x1234, xfrm.XFRM_MODE_TRANSPORT, 0, invalid_crypt, xfrm_base._ALGO_HMAC_SHA1, None, None, None, 0) def testUpdateSaAddMark(self): """Test that an embryonic SA can be updated to add a mark.""" for version in [4, 6]: spi = 0xABCD # Test that an SA created with ALLOCSPI can be updated with the mark. new_sa = self.xfrm.AllocSpi(net_test.GetWildcardAddress(version), IPPROTO_ESP, spi, spi) mark = xfrm.ExactMatchMark(0xf00d) self.xfrm.AddSaInfo(net_test.GetWildcardAddress(version), net_test.GetWildcardAddress(version), spi, xfrm.XFRM_MODE_TUNNEL, 0, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, mark, 0, is_update=True) dump = self.xfrm.DumpSaInfo() self.assertEquals(1, len(dump)) # check that update updated sainfo, attributes = dump[0] self.assertEquals(mark, attributes["XFRMA_MARK"]) self.xfrm.DeleteSaInfo(net_test.GetWildcardAddress(version), spi, IPPROTO_ESP, mark) def getXfrmStat(self, statName): stateVal = 0 with open(XFRM_STATS_PROCFILE, 'r') as f: for line in f: if statName in line: stateVal = int(line.split()[1]) break f.close() return stateVal def testUpdateActiveSaMarks(self): """Test that the OUTPUT_MARK can be updated on an ACTIVE SA.""" for version in [4, 6]: family = net_test.GetAddressFamily(version) netid = self.RandomNetid() remote = self.GetRemoteAddress(version) local = self.MyAddress(version, netid) s = socket(family, SOCK_DGRAM, 0) self.SelectInterface(s, netid, "mark") # Create a mark that we will apply to the policy and later the SA mark = xfrm.ExactMatchMark(netid) # Create a global policy that selects using the mark. sel = xfrm.EmptySelector(family) policy = xfrm.UserPolicy(xfrm.XFRM_POLICY_OUT, sel) tmpl = xfrm.UserTemplate(family, 0, 0, (local, remote)) self.xfrm.AddPolicyInfo(policy, tmpl, mark) # Pull /proc/net/xfrm_stats for baseline outNoStateCount = self.getXfrmStat(XFRM_STATS_OUT_NO_STATES); # should increment XfrmOutNoStates s.sendto(net_test.UDP_PAYLOAD, (remote, 53)) # Check to make sure XfrmOutNoStates is incremented by exactly 1 self.assertEquals(outNoStateCount + 1, self.getXfrmStat(XFRM_STATS_OUT_NO_STATES)) length = xfrm_base.GetEspPacketLength(xfrm.XFRM_MODE_TUNNEL, version, False, net_test.UDP_PAYLOAD, xfrm_base._ALGO_HMAC_SHA1, xfrm_base._ALGO_CBC_AES_256) # Add a default SA with no mark that routes to nowhere. self.xfrm.AddSaInfo(local, remote, TEST_SPI, xfrm.XFRM_MODE_TUNNEL, 0, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, None, 0, is_update=False) self.assertRaisesErrno( ENETUNREACH, s.sendto, net_test.UDP_PAYLOAD, (remote, 53)) # Update the SA to route to a valid netid. self.xfrm.AddSaInfo(local, remote, TEST_SPI, xfrm.XFRM_MODE_TUNNEL, 0, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, None, netid, is_update=True) # Now the payload routes to the updated netid. s.sendto(net_test.UDP_PAYLOAD, (remote, 53)) self._ExpectEspPacketOn(netid, TEST_SPI, 1, length, None, None) # Get a new netid and reroute the packets to the new netid. reroute_netid = self.RandomNetid(netid) # Update the SA to change the output mark. self.xfrm.AddSaInfo(local, remote, TEST_SPI, xfrm.XFRM_MODE_TUNNEL, 0, xfrm_base._ALGO_CBC_AES_256, xfrm_base._ALGO_HMAC_SHA1, None, None, None, reroute_netid, is_update=True) s.sendto(net_test.UDP_PAYLOAD, (remote, 53)) self._ExpectEspPacketOn(reroute_netid, TEST_SPI, 2, length, None, None) dump = self.xfrm.DumpSaInfo() self.assertEquals(1, len(dump)) # check that update updated sainfo, attributes = dump[0] self.assertEquals(reroute_netid, attributes["XFRMA_OUTPUT_MARK"]) self.xfrm.DeleteSaInfo(remote, TEST_SPI, IPPROTO_ESP, None) self.xfrm.DeletePolicyInfo(sel, xfrm.XFRM_POLICY_OUT, mark) if __name__ == "__main__": unittest.main()