cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
component-tests/test_edge_discovery.py
165lines · modecode
| 1 | import ipaddress |
| 2 | import socket |
| 3 | |
| 4 | import pytest |
| 5 | |
| 6 | from constants import protocols |
| 7 | from cli import CloudflaredCli |
| 8 | from util import get_tunnel_connector_id, LOGGER, wait_tunnel_ready, write_config |
| 9 | |
| 10 | |
| 11 | class TestEdgeDiscovery: |
| 12 | def _extra_config(self, protocol, edge_ip_version): |
| 13 | config = { |
| 14 | "protocol": protocol, |
| 15 | } |
| 16 | if edge_ip_version: |
| 17 | config["edge-ip-version"] = edge_ip_version |
| 18 | return config |
| 19 | |
| 20 | @pytest.mark.parametrize("protocol", protocols()) |
| 21 | def test_default_only(self, tmp_path, component_tests_config, protocol): |
| 22 | """ |
| 23 | This test runs a tunnel to connect via IPv4-only edge addresses (default is unset "--edge-ip-version 4") |
| 24 | """ |
| 25 | if self.has_ipv6_only(): |
| 26 | pytest.skip("Host has IPv6 only support and current default is IPv4 only") |
| 27 | self.expect_address_connections( |
| 28 | tmp_path, component_tests_config, protocol, None, self.expect_ipv4_address) |
| 29 | |
| 30 | @pytest.mark.parametrize("protocol", protocols()) |
| 31 | def test_ipv4_only(self, tmp_path, component_tests_config, protocol): |
| 32 | """ |
| 33 | This test runs a tunnel to connect via IPv4-only edge addresses |
| 34 | """ |
| 35 | if self.has_ipv6_only(): |
| 36 | pytest.skip("Host has IPv6 only support") |
| 37 | self.expect_address_connections( |
| 38 | tmp_path, component_tests_config, protocol, "4", self.expect_ipv4_address) |
| 39 | |
| 40 | @pytest.mark.parametrize("protocol", protocols()) |
| 41 | def test_ipv6_only(self, tmp_path, component_tests_config, protocol): |
| 42 | """ |
| 43 | This test runs a tunnel to connect via IPv6-only edge addresses |
| 44 | """ |
| 45 | if self.has_ipv4_only(): |
| 46 | pytest.skip("Host has IPv4 only support") |
| 47 | self.expect_address_connections( |
| 48 | tmp_path, component_tests_config, protocol, "6", self.expect_ipv6_address) |
| 49 | |
| 50 | @pytest.mark.parametrize("protocol", protocols()) |
| 51 | def test_auto_ip64(self, tmp_path, component_tests_config, protocol): |
| 52 | """ |
| 53 | This test runs a tunnel to connect via auto with a preference of IPv6 then IPv4 addresses for a dual stack host |
| 54 | |
| 55 | This test also assumes that the host has IPv6 preference. |
| 56 | """ |
| 57 | if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET6): |
| 58 | pytest.skip("Host does not support dual stack with IPv6 preference") |
| 59 | self.expect_address_connections( |
| 60 | tmp_path, component_tests_config, protocol, "auto", self.expect_ipv6_address) |
| 61 | |
| 62 | @pytest.mark.parametrize("protocol", protocols()) |
| 63 | def test_auto_ip46(self, tmp_path, component_tests_config, protocol): |
| 64 | """ |
| 65 | This test runs a tunnel to connect via auto with a preference of IPv4 then IPv6 addresses for a dual stack host |
| 66 | |
| 67 | This test also assumes that the host has IPv4 preference. |
| 68 | """ |
| 69 | if not self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET): |
| 70 | pytest.skip("Host does not support dual stack with IPv4 preference") |
| 71 | self.expect_address_connections( |
| 72 | tmp_path, component_tests_config, protocol, "auto", self.expect_ipv4_address) |
| 73 | |
| 74 | def expect_address_connections(self, tmp_path, component_tests_config, protocol, edge_ip_version, assert_address_type): |
| 75 | config = component_tests_config( |
| 76 | self._extra_config(protocol, edge_ip_version)) |
| 77 | config_path = write_config(tmp_path, config.full_config) |
| 78 | LOGGER.debug(config) |
| 79 | with CloudflaredCli(config, config_path, LOGGER): |
| 80 | wait_tunnel_ready(tunnel_url=config.get_url(), |
| 81 | require_min_connections=4) |
| 82 | cfd_cli = CloudflaredCli(config, config_path, LOGGER) |
| 83 | tunnel_id = config.get_tunnel_id() |
| 84 | info = cfd_cli.get_tunnel_info(tunnel_id) |
| 85 | connector_id = get_tunnel_connector_id() |
| 86 | connector = next( |
| 87 | (c for c in info["conns"] if c["id"] == connector_id), None) |
| 88 | assert connector, f"Expected connection info from get tunnel info for the connected instance: {info}" |
| 89 | conns = connector["conns"] |
| 90 | assert conns == None or len( |
| 91 | conns) == 4, f"There should be 4 connections registered: {conns}" |
| 92 | for conn in conns: |
| 93 | origin_ip = conn["origin_ip"] |
| 94 | assert origin_ip, f"No available origin_ip for this connection: {conn}" |
| 95 | assert_address_type(origin_ip) |
| 96 | |
| 97 | def expect_ipv4_address(self, address): |
| 98 | assert type(ipaddress.ip_address( |
| 99 | address)) is ipaddress.IPv4Address, f"Expected connection from origin to be a valid IPv4 address: {address}" |
| 100 | |
| 101 | def expect_ipv6_address(self, address): |
| 102 | assert type(ipaddress.ip_address( |
| 103 | address)) is ipaddress.IPv6Address, f"Expected connection from origin to be a valid IPv6 address: {address}" |
| 104 | |
| 105 | def get_addresses(self): |
| 106 | """ |
| 107 | Returns a list of addresses for the host. |
| 108 | """ |
| 109 | host_addresses = socket.getaddrinfo( |
| 110 | "region1.v2.argotunnel.com", 7844, socket.AF_UNSPEC, socket.SOCK_STREAM) |
| 111 | assert len( |
| 112 | host_addresses) > 0, "No addresses returned from getaddrinfo" |
| 113 | return host_addresses |
| 114 | |
| 115 | def has_dual_stack(self, address_family_preference=None): |
| 116 | """ |
| 117 | Returns true if the host has dual stack support and can optionally check |
| 118 | the provided IP family preference. |
| 119 | """ |
| 120 | dual_stack = not self.has_ipv6_only() and not self.has_ipv4_only() |
| 121 | if address_family_preference: |
| 122 | address = self.get_addresses()[0] |
| 123 | return dual_stack and address[0] == address_family_preference |
| 124 | |
| 125 | return dual_stack |
| 126 | |
| 127 | def has_ipv6_only(self): |
| 128 | """ |
| 129 | Returns True if the host has only IPv6 address support. |
| 130 | """ |
| 131 | return self.attempt_connection(socket.AddressFamily.AF_INET6) and not self.attempt_connection(socket.AddressFamily.AF_INET) |
| 132 | |
| 133 | def has_ipv4_only(self): |
| 134 | """ |
| 135 | Returns True if the host has only IPv4 address support. |
| 136 | """ |
| 137 | return self.attempt_connection(socket.AddressFamily.AF_INET) and not self.attempt_connection(socket.AddressFamily.AF_INET6) |
| 138 | |
| 139 | def attempt_connection(self, address_family): |
| 140 | """ |
| 141 | Returns True if a successful socket connection can be made to the |
| 142 | remote host with the provided address family to validate host support |
| 143 | for the provided address family. |
| 144 | """ |
| 145 | address = None |
| 146 | for a in self.get_addresses(): |
| 147 | if a[0] == address_family: |
| 148 | address = a |
| 149 | break |
| 150 | if address is None: |
| 151 | # Couldn't even lookup the address family so we can't connect |
| 152 | return False |
| 153 | af, socktype, proto, canonname, sockaddr = address |
| 154 | s = None |
| 155 | try: |
| 156 | s = socket.socket(af, socktype, proto) |
| 157 | except OSError: |
| 158 | return False |
| 159 | try: |
| 160 | s.connect(sockaddr) |
| 161 | except OSError: |
| 162 | s.close() |
| 163 | return False |
| 164 | s.close() |
| 165 | return True |
| 166 | |