mirror of
https://github.com/cloudflare/cloudflared.git
synced 2026-06-23 04:10:20 +00:00
Miguel da Costa Martins Marcelino
b0b898c235
Check / check (1.22.x, macos-latest) (push) Has been cancelled
Check / check (1.22.x, ubuntu-latest) (push) Has been cancelled
Check / check (1.22.x, windows-latest) (push) Has been cancelled
Semgrep config / semgrep/ci (push) Has been cancelled
To allow pre-checks to test both IPv6 and IPv4, we must change the default value of edge-ip-version's from 4 to auto. This will allows the tunnel (and pre-check) to probe both IPv4 and IPv6 addresses by default, respecting the system's DNS preference. Instead of always preferring IPv4, cloudflared will now use whichever address family the system resolver returns first.
175 lines
7.4 KiB
Python
175 lines
7.4 KiB
Python
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 with the default edge-ip-version (auto), which will use
|
|
whichever address family the system resolver returns first.
|
|
"""
|
|
if self.has_ipv6_only():
|
|
self.expect_address_connections(
|
|
tmp_path, component_tests_config, protocol, None, self.expect_ipv6_address)
|
|
elif self.has_ipv4_only():
|
|
self.expect_address_connections(
|
|
tmp_path, component_tests_config, protocol, None, self.expect_ipv4_address)
|
|
elif self.has_dual_stack(address_family_preference=socket.AddressFamily.AF_INET6):
|
|
self.expect_address_connections(
|
|
tmp_path, component_tests_config, protocol, None, self.expect_ipv6_address)
|
|
else:
|
|
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
|