import ipaddress import socket import pytest from constants import protocols from cli import CloudflaredCli from util import get_tunnel_connector_id, LOGGER, wait_tunnel_ready, write_config class TestEdgeDiscovery: def _extra_config(self, protocol, edge_ip_version): config = { "protocol": protocol, } if edge_ip_version: config["edge-ip-version"] = edge_ip_version return config @pytest.mark.parametrize("protocol", protocols()) def test_default_only(self, tmp_path, component_tests_config, protocol): """ This test runs a tunnel to connect via IPv4-only edge addresses (default is unset "--edge-ip-version 4") """ if self.has_ipv6_only(): pytest.skip("Host has IPv6 only support and current default is IPv4 only") self.expect_address_connections( tmp_path, component_tests_config, protocol, None, self.expect_ipv4_address) @pytest.mark.parametrize("protocol", protocols()) def test_ipv4_only(self, tmp_path, component_tests_config, protocol): """ This test runs a tunnel to connect via IPv4-only edge addresses """ if self.has_ipv6_only(): pytest.skip("Host has IPv6 only support") self.expect_address_connections( tmp_path, component_tests_config, protocol, "4", self.expect_ipv4_address) @pytest.mark.parametrize("protocol", protocols()) def test_ipv6_only(self, tmp_path, component_tests_config, protocol): """ This test runs a tunnel to connect via IPv6-only edge addresses """ if self.has_ipv4_only(): pytest.skip("Host has IPv4 only support") self.expect_address_connections( tmp_path, component_tests_config, protocol, "6", self.expect_ipv6_address) @pytest.mark.parametrize("protocol", protocols()) def test_auto_ip64(self, tmp_path, component_tests_config, protocol): """ This test runs a tunnel to connect via auto with a preference of IPv6 then IPv4 addresses for a dual stack host This test also assumes that the host has IPv6 preference. """ if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET6): pytest.skip("Host does not support dual stack with IPv6 preference") self.expect_address_connections( tmp_path, component_tests_config, protocol, "auto", self.expect_ipv6_address) @pytest.mark.parametrize("protocol", protocols()) def test_auto_ip46(self, tmp_path, component_tests_config, protocol): """ This test runs a tunnel to connect via auto with a preference of IPv4 then IPv6 addresses for a dual stack host This test also assumes that the host has IPv4 preference. """ if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET): pytest.skip("Host does not support dual stack with IPv4 preference") self.expect_address_connections( tmp_path, component_tests_config, protocol, "auto", self.expect_ipv4_address) def expect_address_connections(self, tmp_path, component_tests_config, protocol, edge_ip_version, assert_address_type): config = component_tests_config( self._extra_config(protocol, edge_ip_version)) config_path = write_config(tmp_path, config.full_config) LOGGER.debug(config) with CloudflaredCli(config, config_path, LOGGER): wait_tunnel_ready(tunnel_url=config.get_url(), require_min_connections=4) cfd_cli = CloudflaredCli(config, config_path, LOGGER) tunnel_id = config.get_tunnel_id() info = cfd_cli.get_tunnel_info(tunnel_id) connector_id = get_tunnel_connector_id() connector = next( (c for c in info["conns"] if c["id"] == connector_id), None) assert connector, f"Expected connection info from get tunnel info for the connected instance: {info}" conns = connector["conns"] assert conns == None or len( conns) == 4, f"There should be 4 connections registered: {conns}" for conn in conns: origin_ip = conn["origin_ip"] assert origin_ip, f"No available origin_ip for this connection: {conn}" assert_address_type(origin_ip) def expect_ipv4_address(self, address): assert type(ipaddress.ip_address( address)) is ipaddress.IPv4Address, f"Expected connection from origin to be a valid IPv4 address: {address}" def expect_ipv6_address(self, address): assert type(ipaddress.ip_address( address)) is ipaddress.IPv6Address, f"Expected connection from origin to be a valid IPv6 address: {address}" def get_addresses(self): """ Returns a list of addresses for the host. """ host_addresses = socket.getaddrinfo( "region1.v2.argotunnel.com", 7844, socket.AF_UNSPEC, socket.SOCK_STREAM) assert len( host_addresses) > 0, "No addresses returned from getaddrinfo" return host_addresses def has_dual_stack(self, address_family_preference=None): """ Returns true if the host has dual stack support and can optionally check the provided IP family preference. """ dual_stack = not self.has_ipv6_only() and not self.has_ipv4_only() if address_family_preference: address = self.get_addresses()[0] return dual_stack and address[0] == address_family_preference return dual_stack def has_ipv6_only(self): """ Returns True if the host has only IPv6 address support. """ return self.attempt_connection(socket.AddressFamily.AF_INET6) and not self.attempt_connection(socket.AddressFamily.AF_INET) def has_ipv4_only(self): """ Returns True if the host has only IPv4 address support. """ return self.attempt_connection(socket.AddressFamily.AF_INET) and not self.attempt_connection(socket.AddressFamily.AF_INET6) def attempt_connection(self, address_family): """ Returns True if a successful socket connection can be made to the remote host with the provided address family to validate host support for the provided address family. """ address = None for a in self.get_addresses(): if a[0] == address_family: address = a break if address is None: # Couldn't even lookup the address family so we can't connect return False af, socktype, proto, canonname, sockaddr = address s = None try: s = socket.socket(af, socktype, proto) except OSError: return False try: s.connect(sockaddr) except OSError: s.close() return False s.close() return True