| #!/usr/bin/python3 |
| # |
| # 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. |
| |
| from errno import * # pylint: disable=wildcard-import,g-importing-member |
| import itertools |
| import os |
| from socket import * # pylint: disable=wildcard-import,g-importing-member |
| import threading |
| import time |
| import unittest |
| |
| import net_test |
| from scapy import all as scapy |
| from tun_twister import TapTwister |
| import util |
| import xfrm |
| import xfrm_base |
| import xfrm_test |
| |
| ANY_KVER = net_test.LINUX_ANY_VERSION |
| |
| # List of encryption algorithms for use in ParamTests. |
| CRYPT_ALGOS = [ |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CBC_AES, 128)), ANY_KVER), |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CBC_AES, 192)), ANY_KVER), |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CBC_AES, 256)), ANY_KVER), |
| # RFC 3686 specifies that key length must be 128, 192 or 256 bits, with |
| # an additional 4 bytes (32 bits) of nonce. A fresh nonce value MUST be |
| # assigned for each SA. |
| # CTR-AES is enforced since kernel version 5.8 |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CTR_AES, 128+32)), (5, 8)), |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CTR_AES, 192+32)), (5, 8)), |
| (xfrm.XfrmAlgo((xfrm.XFRM_EALG_CTR_AES, 256+32)), (5, 8)), |
| ] |
| |
| # List of auth algorithms for use in ParamTests. |
| AUTH_ALGOS = [ |
| # RFC 4868 specifies that the only supported truncation length is half the |
| # hash size. |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_MD5, 128, 96)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA1, 160, 96)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA256, 256, 128)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA384, 384, 192)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA512, 512, 256)), ANY_KVER), |
| # Test larger truncation lengths for good measure. |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_MD5, 128, 128)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA1, 160, 160)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA256, 256, 256)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA384, 384, 384)), ANY_KVER), |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_HMAC_SHA512, 512, 512)), ANY_KVER), |
| # RFC 3566 specifies that the only supported truncation length |
| # is 96 bits. |
| # XCBC-AES is enforced since kernel version 5.8 |
| (xfrm.XfrmAlgoAuth((xfrm.XFRM_AALG_AUTH_XCBC_AES, 128, 96)), (5, 8)), |
| ] |
| |
| # List of aead algorithms for use in ParamTests. |
| AEAD_ALGOS = [ |
| # RFC 4106 specifies that key length must be 128, 192 or 256 bits, |
| # with an additional 4 bytes (32 bits) of salt. The salt must be unique |
| # for each new SA using the same key. |
| # RFC 4106 specifies that ICV length must be 8, 12, or 16 bytes |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 128+32, 8*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 128+32, 12*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 128+32, 16*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 192+32, 8*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 192+32, 12*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 192+32, 16*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 256+32, 8*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 256+32, 12*8)), ANY_KVER), |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_GCM_AES, 256+32, 16*8)), ANY_KVER), |
| # RFC 7634 specifies that key length must be 256 bits, with an additional |
| # 4 bytes (32 bits) of nonce. A fresh nonce value MUST be assigned for |
| # each SA. RFC 7634 also specifies that ICV length must be 16 bytes. |
| # ChaCha20-Poly1305 is enforced since kernel version 5.8 |
| (xfrm.XfrmAlgoAead((xfrm.XFRM_AEAD_CHACHA20_POLY1305, 256+32, 16*8)), |
| (5, 8)), |
| ] |
| |
| |
| def GenerateKey(key_len): |
| if key_len % 8 != 0: |
| raise ValueError("Invalid key length in bits: " + str(key_len)) |
| return os.urandom(key_len // 8) |
| |
| |
| # Does the kernel support this algorithm? |
| def HaveAlgo(crypt_algo, auth_algo, aead_algo): |
| try: |
| test_xfrm = xfrm.Xfrm() |
| test_xfrm.FlushSaInfo() |
| test_xfrm.FlushPolicyInfo() |
| |
| test_xfrm.AddSaInfo( |
| src=xfrm_test.TEST_ADDR1, |
| dst=xfrm_test.TEST_ADDR2, |
| spi=xfrm_test.TEST_SPI, |
| mode=xfrm.XFRM_MODE_TRANSPORT, |
| reqid=100, |
| encryption=(crypt_algo, |
| GenerateKey(crypt_algo.key_len)) if crypt_algo else None, |
| auth_trunc=(auth_algo, |
| GenerateKey(auth_algo.key_len)) if auth_algo else None, |
| aead=(aead_algo, GenerateKey(aead_algo.key_len)) if aead_algo else None, |
| encap=None, |
| mark=None, |
| output_mark=None) |
| |
| test_xfrm.FlushSaInfo() |
| test_xfrm.FlushPolicyInfo() |
| |
| return True |
| except IOError as err: |
| if err.errno == ENOSYS: |
| return False |
| else: |
| print("Unexpected error:", err.errno) |
| return True |
| |
| # Dictionary to record the algorithm state. Mark the state True if this |
| # algorithm is enforced or enabled on this kernel. Otherwise, mark it |
| # False. |
| algoState = {} |
| |
| |
| def AlgoEnforcedOrEnabled(crypt, auth, aead, target_algo, target_kernel): |
| if algoState.get(target_algo) is None: |
| algoState[target_algo] = (net_test.LINUX_VERSION >= target_kernel |
| or HaveAlgo(crypt, auth, aead)) |
| return algoState.get(target_algo) |
| |
| |
| # Return true if this algorithm should be enforced or is enabled on this kernel |
| def AuthEnforcedOrEnabled(auth_case): |
| auth = auth_case[0] |
| crypt = xfrm.XfrmAlgo((b"ecb(cipher_null)", 0)) |
| return AlgoEnforcedOrEnabled(crypt, auth, None, auth.name, auth_case[1]) |
| |
| |
| # Return true if this algorithm should be enforced or is enabled on this kernel |
| def CryptEnforcedOrEnabled(crypt_case): |
| crypt = crypt_case[0] |
| auth = xfrm.XfrmAlgoAuth((b"digest_null", 0, 0)) |
| return AlgoEnforcedOrEnabled(crypt, auth, None, crypt.name, crypt_case[1]) |
| |
| |
| # Return true if this algorithm should be enforced or is enabled on this kernel |
| def AeadEnforcedOrEnabled(aead_case): |
| aead = aead_case[0] |
| return AlgoEnforcedOrEnabled(None, None, aead, aead.name, aead_case[1]) |
| |
| |
| def InjectTests(): |
| XfrmAlgorithmTest.InjectTests() |
| |
| |
| class XfrmAlgorithmTest(xfrm_base.XfrmLazyTest): |
| @classmethod |
| def InjectTests(cls): |
| versions = (4, 6) |
| types = (SOCK_DGRAM, SOCK_STREAM) |
| |
| # Tests all combinations of auth & crypt. Mutually exclusive with aead. |
| param_list = itertools.product(versions, types, AUTH_ALGOS, CRYPT_ALGOS, |
| [None]) |
| util.InjectParameterizedTest(cls, param_list, cls.TestNameGenerator) |
| |
| # Tests all combinations of aead. Mutually exclusive with auth/crypt. |
| param_list = itertools.product(versions, types, [None], [None], AEAD_ALGOS) |
| util.InjectParameterizedTest(cls, param_list, cls.TestNameGenerator) |
| |
| @staticmethod |
| def TestNameGenerator(version, proto, auth_case, crypt_case, aead_case): |
| # Produce a unique and readable name for each test. e.g. |
| # testSocketPolicySimple_cbc-aes_256_hmac-sha512_512_256_IPv6_UDP |
| param_string = "" |
| if crypt_case is not None: |
| crypt = crypt_case[0] |
| param_string += "%s_%d_" % (crypt.name.decode(), crypt.key_len) |
| |
| if auth_case is not None: |
| auth = auth_case[0] |
| param_string += "%s_%d_%d_" % (auth.name.decode(), auth.key_len, |
| auth.trunc_len) |
| |
| if aead_case is not None: |
| aead = aead_case[0] |
| param_string += "%s_%d_%d_" % (aead.name.decode(), aead.key_len, |
| aead.icv_len) |
| |
| param_string += "%s_%s" % ("IPv4" if version == 4 else "IPv6", |
| "UDP" if proto == SOCK_DGRAM else "TCP") |
| return param_string |
| |
| def ParamTestSocketPolicySimple(self, version, proto, auth_case, crypt_case, |
| aead_case): |
| """Test two-way traffic using transport mode and socket policies.""" |
| |
| # Bypass the test if any algorithm going to be tested is not enforced |
| # or enabled on this kernel |
| if auth_case is not None and not AuthEnforcedOrEnabled(auth_case): |
| return |
| if crypt_case is not None and not CryptEnforcedOrEnabled(crypt_case): |
| return |
| if aead_case is not None and not AeadEnforcedOrEnabled(aead_case): |
| return |
| |
| auth = auth_case[0] if auth_case else None |
| crypt = crypt_case[0] if crypt_case else None |
| aead = aead_case[0] if aead_case else None |
| |
| def AssertEncrypted(packet): |
| # This gives a free pass to ICMP and ICMPv6 packets, which show up |
| # nondeterministically in tests. |
| self.assertEqual(None, |
| packet.getlayer(scapy.UDP), |
| "UDP packet sent in the clear") |
| self.assertEqual(None, |
| packet.getlayer(scapy.TCP), |
| "TCP packet sent in the clear") |
| |
| # We create a pair of sockets, "left" and "right", that will talk to each |
| # other using transport mode ESP. Because of TapTwister, both sockets |
| # perceive each other as owning "remote_addr". |
| netid = self.RandomNetid() |
| family = net_test.GetAddressFamily(version) |
| local_addr = self.MyAddress(version, netid) |
| remote_addr = self.GetRemoteSocketAddress(version) |
| auth_left = (xfrm.XfrmAlgoAuth((auth.name, auth.key_len, auth.trunc_len)), |
| os.urandom(auth.key_len // 8)) if auth else None |
| auth_right = (xfrm.XfrmAlgoAuth((auth.name, auth.key_len, auth.trunc_len)), |
| os.urandom(auth.key_len // 8)) if auth else None |
| crypt_left = (xfrm.XfrmAlgo((crypt.name, crypt.key_len)), |
| os.urandom(crypt.key_len // 8)) if crypt else None |
| crypt_right = (xfrm.XfrmAlgo((crypt.name, crypt.key_len)), |
| os.urandom(crypt.key_len // 8)) if crypt else None |
| aead_left = (xfrm.XfrmAlgoAead((aead.name, aead.key_len, aead.icv_len)), |
| os.urandom(aead.key_len // 8)) if aead else None |
| aead_right = (xfrm.XfrmAlgoAead((aead.name, aead.key_len, aead.icv_len)), |
| os.urandom(aead.key_len // 8)) if aead else None |
| spi_left = 0xbeefface |
| spi_right = 0xcafed00d |
| req_ids = [100, 200, 300, 400] # Used to match templates and SAs. |
| |
| # Left outbound SA |
| self.xfrm.AddSaInfo( |
| src=local_addr, |
| dst=remote_addr, |
| spi=spi_right, |
| mode=xfrm.XFRM_MODE_TRANSPORT, |
| reqid=req_ids[0], |
| encryption=crypt_right, |
| auth_trunc=auth_right, |
| aead=aead_right, |
| encap=None, |
| mark=None, |
| output_mark=None) |
| # Right inbound SA |
| self.xfrm.AddSaInfo( |
| src=remote_addr, |
| dst=local_addr, |
| spi=spi_right, |
| mode=xfrm.XFRM_MODE_TRANSPORT, |
| reqid=req_ids[1], |
| encryption=crypt_right, |
| auth_trunc=auth_right, |
| aead=aead_right, |
| encap=None, |
| mark=None, |
| output_mark=None) |
| # Right outbound SA |
| self.xfrm.AddSaInfo( |
| src=local_addr, |
| dst=remote_addr, |
| spi=spi_left, |
| mode=xfrm.XFRM_MODE_TRANSPORT, |
| reqid=req_ids[2], |
| encryption=crypt_left, |
| auth_trunc=auth_left, |
| aead=aead_left, |
| encap=None, |
| mark=None, |
| output_mark=None) |
| # Left inbound SA |
| self.xfrm.AddSaInfo( |
| src=remote_addr, |
| dst=local_addr, |
| spi=spi_left, |
| mode=xfrm.XFRM_MODE_TRANSPORT, |
| reqid=req_ids[3], |
| encryption=crypt_left, |
| auth_trunc=auth_left, |
| aead=aead_left, |
| encap=None, |
| mark=None, |
| output_mark=None) |
| |
| # Make two sockets. |
| sock_left = socket(family, proto, 0) |
| sock_left.settimeout(2.0) |
| sock_left.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) |
| self.SelectInterface(sock_left, netid, "mark") |
| sock_right = socket(family, proto, 0) |
| sock_right.settimeout(2.0) |
| sock_right.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) |
| self.SelectInterface(sock_right, netid, "mark") |
| |
| # For UDP, set SO_LINGER to 0, to prevent TCP sockets from hanging around |
| # in a TIME_WAIT state. |
| if proto == SOCK_STREAM: |
| net_test.DisableFinWait(sock_left) |
| net_test.DisableFinWait(sock_right) |
| |
| # Apply the left outbound socket policy. |
| xfrm_base.ApplySocketPolicy(sock_left, family, xfrm.XFRM_POLICY_OUT, |
| spi_right, req_ids[0], None) |
| # Apply right inbound socket policy. |
| xfrm_base.ApplySocketPolicy(sock_right, family, xfrm.XFRM_POLICY_IN, |
| spi_right, req_ids[1], None) |
| # Apply right outbound socket policy. |
| xfrm_base.ApplySocketPolicy(sock_right, family, xfrm.XFRM_POLICY_OUT, |
| spi_left, req_ids[2], None) |
| # Apply left inbound socket policy. |
| xfrm_base.ApplySocketPolicy(sock_left, family, xfrm.XFRM_POLICY_IN, |
| spi_left, req_ids[3], None) |
| |
| server_ready = threading.Event() |
| server_error = None # Save exceptions thrown by the server. |
| |
| def TcpServer(sock, client_port): |
| try: |
| sock.listen(1) |
| server_ready.set() |
| accepted, peer = sock.accept() |
| self.assertEqual(remote_addr, peer[0]) |
| self.assertEqual(client_port, peer[1]) |
| data = accepted.recv(2048) |
| self.assertEqual(b"hello request", data) |
| accepted.send(b"hello response") |
| time.sleep(0.1) |
| accepted.close() |
| except Exception as e: # pylint: disable=broad-exception-caught |
| nonlocal server_error |
| server_error = e |
| finally: |
| sock.close() |
| |
| def UdpServer(sock, client_port): |
| try: |
| server_ready.set() |
| data, peer = sock.recvfrom(2048) |
| self.assertEqual(remote_addr, peer[0]) |
| self.assertEqual(client_port, peer[1]) |
| self.assertEqual(b"hello request", data) |
| sock.sendto(b"hello response", peer) |
| except Exception as e: # pylint: disable=broad-exception-caught |
| nonlocal server_error |
| server_error = e |
| finally: |
| sock.close() |
| |
| # Server and client need to know each other's port numbers in advance. |
| wildcard_addr = net_test.GetWildcardAddress(version) |
| sock_left.bind((wildcard_addr, 0)) |
| sock_right.bind((wildcard_addr, 0)) |
| left_port = sock_left.getsockname()[1] |
| right_port = sock_right.getsockname()[1] |
| |
| # Start the appropriate server type on sock_right. |
| target = TcpServer if proto == SOCK_STREAM else UdpServer |
| server = threading.Thread( |
| target=target, |
| args=(sock_right, left_port), |
| name="SocketServer") |
| server.start() |
| # Wait for server to be ready before attempting to connect. TCP retries |
| # hide this problem, but UDP will fail outright if the server socket has |
| # not bound when we send. |
| self.assertTrue(server_ready.wait(3.0), |
| "Timed out waiting for server thread") |
| |
| with TapTwister(fd=self.tuns[netid].fileno(), validator=AssertEncrypted): |
| sock_left.connect((remote_addr, right_port)) |
| sock_left.send(b"hello request") |
| data = sock_left.recv(2048) |
| self.assertEqual(b"hello response", data) |
| sock_left.close() |
| server.join(timeout=3.0) |
| self.assertFalse(server.is_alive(), "Timed out waiting for server exit") |
| if server_error: |
| raise server_error |
| |
| |
| if __name__ == "__main__": |
| XfrmAlgorithmTest.InjectTests() |
| unittest.main() |
