romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-08-25 21:53:32 +03:00
Code Issues Projects Releases Packages Wiki Activity

HTTP/3: add proxy CONNECT and MASQUE CONNECT-UDP support (ngtcp2 QUIC)

Browse source 
This patch adds two major proxy capabilities to curl (ngtcp2 QUIC):
- HTTP/3 Proxy CONNECT: Tunnel HTTP/1.1 or HTTP/2 traffic through an
  HTTPS proxy that speaks HTTP/3 (QUIC) using the standard CONNECT
  method over an HTTP/3 connection.
- MASQUE CONNECT-UDP: Tunnel HTTP/3 (QUIC) traffic through an HTTP
  proxy (speaking HTTP/1.1, HTTP/2, or HTTP/3) using the extended
  CONNECT method with the CONNECT-UDP protocol (RFC9297 & RFC9298).

Public API additions:
- `CURLPROXY_HTTPS3`: new proxy type constant for HTTP/3 proxy
- `--proxy-http3`: new CLI flag to negotiate HTTP/3 with HTTPS proxy

The implementation adds two new filters:
- `H3-PROXY` - enables negotiating HTTP/3 (QUIC) to the proxy and
  running CONNECT/CONNECT-UDP through that proxy transport.
- `CAPSULE` - dedicated filter inserted between QUIC transport and
  HTTP-PROXY to handle datagram capsule encapsulation/decapsulation.

Here is how the curl filter chaining looks in different scenarios:
- HTTP/3 Proxy CONNECT (tunneling TCP protocols over QUIC proxy):
  conn -> HTTP/1.1 or HTTP/2  -> SSL -> HTTP-PROXY ->
                                 H3-PROXY -> HAPPY-EYEBALLS -> UDP
- MASQUE CONNECT-UDP (tunneling QUIC over any proxy):
  conn -> HTTP/3 -> CAPSULE -> HTTP-PROXY -> H3-PROXY ->
                               HAPPY-EYEBALLS -> UDP
  conn -> HTTP/3 -> CAPSULE -> HTTP-PROXY -> H1-PROXY or H2-PROXY ->
                               SSL -> HAPPY-EYEBALLS -> TCP

- Both features currently require the ngtcp2 QUIC backend.
- Both features are experimental (disabled by default). Enable with
  `--enable-proxy-http3`(autotools) or `-DUSE_PROXY_HTTP3=ON`(CMake).

Tests:
- tests/unit/unit3400.c: Unit tests for capsule protocol encode/decode
- tests/http/test_60_h3_proxy.py: Comprehensive pytest integration suite
- tests/http/testenv/h2o.py: Managing h2o instances with HTTP/1.1, HTTP/2,
  and HTTP/3 (QUIC) listeners, proxy.connect and proxy.connect-udp enabled.

References:
  RFC 9297 - HTTP Datagrams and the Capsule Protocol
  RFC 9298 - Proxying UDP in HTTP
  RFC 9000 §16 — Variable-Length Integer Encoding

Signed-off-by: Aritra Basu <aritrbas+gh@cisco.com>

Closes #21153
This commit is contained in:
Aritra Basu 2026-04-27 19:35:38 -04:00 • committed by Daniel Stenberg
parent efc3f2309e
commit e78b1b3ecc
No known key found for this signature in database
GPG key ID: 5CC908FDB71E12C2
66 changed files with 7401 additions and 473 deletions
Download patch file Download diff file Expand all files Collapse all files

491
tests/http/testenv/env.py
View file

@ -1,6 +1,6 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
# ***************************************************************************
# _ _ ____ _
# Project ___| | | | _ \| |
# / __| | | | |_) | |
@ -54,20 +54,21 @@ def init_config_from(conf_path):
TESTS_HTTPD_PATH = os.path.dirname(os.path.dirname(__file__))
PROJ_PATH = os.path.dirname(os.path.dirname(TESTS_HTTPD_PATH))
TOP_PATH = os.path.join(os.getcwd(), os.path.pardir)
CONFIG_PATH = os.path.join(TOP_PATH, 'tests', 'http', 'config.ini')
CONFIG_PATH = os.path.join(TOP_PATH, "tests", "http", "config.ini")
if not os.path.exists(CONFIG_PATH):
ALT_CONFIG_PATH = os.path.join(PROJ_PATH, 'tests', 'http', 'config.ini')
ALT_CONFIG_PATH = os.path.join(PROJ_PATH, "tests", "http", "config.ini")
if not os.path.exists(ALT_CONFIG_PATH):
raise Exception(f'unable to find config.ini in {CONFIG_PATH} nor {ALT_CONFIG_PATH}')
raise Exception(
f"unable to find config.ini in {CONFIG_PATH} nor {ALT_CONFIG_PATH}"
)
TOP_PATH = PROJ_PATH
CONFIG_PATH = ALT_CONFIG_PATH
DEF_CONFIG = init_config_from(CONFIG_PATH)
CURL = os.path.join(TOP_PATH, 'src', 'curl')
CURLINFO = os.path.join(TOP_PATH, 'src', 'curlinfo')
CURL = os.path.join(TOP_PATH, "src", "curl")
CURLINFO = os.path.join(TOP_PATH, "src", "curlinfo")
class NghttpxUtil:
CMD = None
VERSION_FULL = None
@ -76,34 +77,37 @@ class NghttpxUtil:
if cmd is None:
return None
if cls.VERSION_FULL is None or cmd != cls.CMD:
p = subprocess.run(args=[cmd, '--version'],
capture_output=True, text=True)
p = subprocess.run(args=[cmd, "--version"], capture_output=True, text=True)
if p.returncode != 0:
raise RuntimeError(f'{cmd} --version failed with exit code: {p.returncode}')
raise RuntimeError(
f"{cmd} --version failed with exit code: {p.returncode}"
)
cls.CMD = cmd
for line in p.stdout.splitlines(keepends=False):
if line.startswith('nghttpx '):
if line.startswith("nghttpx "):
cls.VERSION_FULL = line
if cls.VERSION_FULL is None:
raise RuntimeError(f'{cmd}: unable to determine version')
raise RuntimeError(f"{cmd}: unable to determine version")
return cls.VERSION_FULL
@staticmethod
def version_with_h3(version):
return re.match(r'.* ngtcp2/\d+\.\d+\.\d+.*', version) is not None
return re.match(r".* ngtcp2/\d+\.\d+\.\d+.*", version) is not None
class EnvConfig:
def __init__(self, pytestconfig: Optional[pytest.Config] = None,
testrun_uid=None,
worker_id=None):
def __init__(
self,
pytestconfig: Optional[pytest.Config] = None,
testrun_uid=None,
worker_id=None,
):
self.pytestconfig = pytestconfig
self.testrun_uid = testrun_uid
self.worker_id = worker_id if worker_id is not None else 'master'
self.worker_id = worker_id if worker_id is not None else "master"
self.tests_dir = TESTS_HTTPD_PATH
self.gen_root = self.gen_dir = os.path.join(self.tests_dir, 'gen')
if self.worker_id != 'master':
self.gen_root = self.gen_dir = os.path.join(self.tests_dir, "gen")
if self.worker_id != "master":
self.gen_dir = os.path.join(self.gen_dir, self.worker_id)
self.project_dir = os.path.dirname(os.path.dirname(self.tests_dir))
self.build_dir = TOP_PATH
@ -111,57 +115,56 @@ class EnvConfig:
# check cur and its features
self.curl = CURL
self.curlinfo = CURLINFO
if 'CURL' in os.environ:
self.curl = os.environ['CURL']
if "CURL" in os.environ:
self.curl = os.environ["CURL"]
self.curl_props = {
'version_string': '',
'version': '',
'os': '',
'fullname': '',
'features_string': '',
'features': set(),
'protocols_string': '',
'protocols': set(),
'libs': set(),
'lib_versions': set(),
"version_string": "",
"version": "",
"os": "",
"fullname": "",
"features_string": "",
"features": set(),
"protocols_string": "",
"protocols": set(),
"libs": set(),
"lib_versions": set(),
}
self.curl_is_debug = False
self.curl_protos = []
p = subprocess.run(args=[self.curl, '-V'],
capture_output=True, text=True)
p = subprocess.run(args=[self.curl, "-V"], capture_output=True, text=True)
if p.returncode != 0:
raise RuntimeError(f'{self.curl} -V failed with exit code: {p.returncode}')
if p.stderr.startswith('WARNING:'):
raise RuntimeError(f"{self.curl} -V failed with exit code: {p.returncode}")
if p.stderr.startswith("WARNING:"):
self.curl_is_debug = True
for line in p.stdout.splitlines(keepends=False):
if line.startswith('curl '):
self.curl_props['version_string'] = line
m = re.match(r'^curl (?P<version>\S+) (?P<os>\S+) (?P<libs>.*)$', line)
if line.startswith("curl "):
self.curl_props["version_string"] = line
m = re.match(r"^curl (?P<version>\S+) (?P<os>\S+) (?P<libs>.*)$", line)
if m:
self.curl_props['fullname'] = m.group(0)
self.curl_props['version'] = m.group('version')
self.curl_props['os'] = m.group('os')
self.curl_props['lib_versions'] = {
lib.lower() for lib in m.group('libs').split(' ')
self.curl_props["fullname"] = m.group(0)
self.curl_props["version"] = m.group("version")
self.curl_props["os"] = m.group("os")
self.curl_props["lib_versions"] = {
lib.lower() for lib in m.group("libs").split(" ")
}
self.curl_props['libs'] = {
re.sub(r'/[a-z0-9.-]*', '', lib) for lib in self.curl_props['lib_versions']
self.curl_props["libs"] = {
re.sub(r"/[a-z0-9.-]*", "", lib)
for lib in self.curl_props["lib_versions"]
}
if line.startswith('Features: '):
self.curl_props['features_string'] = line[10:]
self.curl_props['features'] = {
feat.lower() for feat in line[10:].split(' ')
if line.startswith("Features: "):
self.curl_props["features_string"] = line[10:]
self.curl_props["features"] = {
feat.lower() for feat in line[10:].split(" ")
}
if line.startswith('Protocols: '):
self.curl_props['protocols_string'] = line[11:]
self.curl_props['protocols'] = {
prot.lower() for prot in line[11:].split(' ')
if line.startswith("Protocols: "):
self.curl_props["protocols_string"] = line[11:]
self.curl_props["protocols"] = {
prot.lower() for prot in line[11:].split(" ")
}
p = subprocess.run(args=[self.curlinfo],
capture_output=True, text=True)
p = subprocess.run(args=[self.curlinfo], capture_output=True, text=True)
if p.returncode != 0:
raise RuntimeError(f'{self.curlinfo} failed with exit code: {p.returncode}')
raise RuntimeError(f"{self.curlinfo} failed with exit code: {p.returncode}")
self.curl_is_verbose = 'verbose-strings: ON' in p.stdout
self.curl_can_cert_status = 'cert-status: ON' in p.stdout
self.curl_override_dns = 'override-dns: ON' in p.stdout
@ -169,18 +172,18 @@ class EnvConfig:
self.ports = {}
self.httpd = self.config['httpd']['httpd']
self.apxs = self.config['httpd']['apxs']
self.httpd = self.config["httpd"]["httpd"]
self.apxs = self.config["httpd"]["apxs"]
if len(self.apxs) == 0:
self.apxs = None
self._httpd_version = None
self.examples_pem = {
'key': 'xxx',
'cert': 'xxx',
"key": "xxx",
"cert": "xxx",
}
self.htdocs_dir = os.path.join(self.gen_dir, 'htdocs')
self.tld = 'http.curl.se'
self.htdocs_dir = os.path.join(self.gen_dir, "htdocs")
self.tld = "http.curl.se"
self.domain1 = f"one.{self.tld}"
self.domain1brotli = f"brotli.one.{self.tld}"
self.domain2 = f"two.{self.tld}"
@ -188,22 +191,43 @@ class EnvConfig:
self.proxy_domain = f"proxy.{self.tld}"
self.expired_domain = f"expired.{self.tld}"
self.cert_specs = [
CertificateSpec(domains=[self.domain1, self.domain1brotli, 'localhost', '127.0.0.1'], key_type='rsa2048'),
CertificateSpec(name='domain1-no-ip', domains=[self.domain1, self.domain1brotli], key_type='rsa2048'),
CertificateSpec(name='domain1-very-bad', domains=[self.domain1, 'dns:127.0.0.1'], key_type='rsa2048'),
CertificateSpec(domains=[self.domain2], key_type='rsa2048'),
CertificateSpec(domains=[self.ftp_domain], key_type='rsa2048'),
CertificateSpec(domains=[self.proxy_domain, '127.0.0.1'], key_type='rsa2048'),
CertificateSpec(domains=[self.expired_domain], key_type='rsa2048',
valid_from=timedelta(days=-100), valid_to=timedelta(days=-10)),
CertificateSpec(name="clientsX", sub_specs=[
CertificateSpec(name="user1", client=True),
]),
CertificateSpec(
domains=[self.domain1, self.domain1brotli, "localhost", "127.0.0.1"],
key_type="rsa2048",
),
CertificateSpec(
name="domain1-no-ip",
domains=[self.domain1, self.domain1brotli],
key_type="rsa2048",
),
CertificateSpec(
name="domain1-very-bad",
domains=[self.domain1, "dns:127.0.0.1"],
key_type="rsa2048",
),
CertificateSpec(domains=[self.domain2], key_type="rsa2048"),
CertificateSpec(domains=[self.ftp_domain], key_type="rsa2048"),
CertificateSpec(
domains=[self.proxy_domain, "127.0.0.1"], key_type="rsa2048"
),
CertificateSpec(
domains=[self.expired_domain],
key_type="rsa2048",
valid_from=timedelta(days=-100),
valid_to=timedelta(days=-10),
),
CertificateSpec(
name="clientsX",
sub_specs=[
CertificateSpec(name="user1", client=True),
],
),
]
self.openssl = 'openssl'
p = subprocess.run(args=[self.openssl, 'version'],
capture_output=True, text=True)
self.openssl = "openssl"
p = subprocess.run(
args=[self.openssl, "version"], capture_output=True, text=True
)
if p.returncode != 0:
# no openssl in path
self.openssl = None
@ -211,7 +235,7 @@ class EnvConfig:
else:
self.openssl_version = p.stdout.strip()
self.nghttpx = self.config['nghttpx']['nghttpx']
self.nghttpx = self.config["nghttpx"]["nghttpx"]
if len(self.nghttpx.strip()) == 0:
self.nghttpx = None
self._nghttpx_version = None
@ -220,30 +244,58 @@ class EnvConfig:
self._nghttpx_version = NghttpxUtil.version(self.nghttpx)
self.nghttpx_with_h3 = NghttpxUtil.version_with_h3(self._nghttpx_version)
self.caddy = self.config['caddy']['caddy']
self.caddy = self.config["caddy"]["caddy"]
self._caddy_version = None
if len(self.caddy.strip()) == 0:
self.caddy = None
self.h2o = self.config["h2o"]["h2o"]
if len(self.h2o.strip()) == 0:
self.h2o = None
self._h2o_version = None
if self.h2o is not None:
try:
p = subprocess.run(
args=[self.h2o, "--version"], capture_output=True, text=True
)
if p.returncode != 0:
# not a working h2o
self.h2o = None
else:
# h2o --version output format: "h2o version 2.3.0"
m = re.search(r"h2o version (\S+)", p.stdout)
if m:
self._h2o_version = m.group(1)
else:
self.h2o = None
except Exception:
log.exception("checking h2o version")
self.h2o = None
if self.caddy is not None:
p = subprocess.run(args=[self.caddy, 'version'],
capture_output=True, text=True)
p = subprocess.run(
args=[self.caddy, "version"], capture_output=True, text=True
)
if p.returncode != 0:
# not a working caddy
self.caddy = None
m = re.match(r'v?(\d+\.\d+\.\d+).*', p.stdout)
m = re.match(r"v?(\d+\.\d+\.\d+).*", p.stdout)
if m:
self._caddy_version = m.group(1)
else:
raise RuntimeError(f'Unable to determine caddy version from: {p.stdout}')
raise RuntimeError(
f"Unable to determine caddy version from: {p.stdout}"
)
self.vsftpd = self.config['vsftpd']['vsftpd']
if self.vsftpd == '':
self.vsftpd = self.config["vsftpd"]["vsftpd"]
if self.vsftpd == "":
self.vsftpd = None
self._vsftpd_version = None
if self.vsftpd is not None:
with tempfile.TemporaryFile('w+') as tmp:
p = subprocess.run(args=[self.vsftpd, '-v'],
capture_output=True, text=True, stdin=tmp)
with tempfile.TemporaryFile("w+") as tmp:
p = subprocess.run(
args=[self.vsftpd, "-v"], capture_output=True, text=True, stdin=tmp
)
if p.returncode != 0:
# not a working vsftpd
self.vsftpd = None
@ -256,80 +308,83 @@ class EnvConfig:
# any data there instead.
tmp.seek(0)
ver_text = tmp.read()
m = re.match(r'vsftpd: version (\d+\.\d+\.\d+)', ver_text)
m = re.match(r"vsftpd: version (\d+\.\d+\.\d+)", ver_text)
if m:
self._vsftpd_version = m.group(1)
elif len(p.stderr) == 0:
# vsftp does not use stdout or stderr for printing its version... -.-
self._vsftpd_version = 'unknown'
self._vsftpd_version = "unknown"
else:
raise Exception(f'Unable to determine VsFTPD version from: {p.stderr}')
raise Exception(f"Unable to determine VsFTPD version from: {p.stderr}")
self.danted = self.config['danted']['danted']
if self.danted == '':
self.danted = self.config["danted"]["danted"]
if self.danted == "":
self.danted = None
self._danted_version = None
if self.danted is not None:
p = subprocess.run(args=[self.danted, '-v'],
capture_output=True, text=True)
p = subprocess.run(args=[self.danted, "-v"], capture_output=True, text=True)
assert p.returncode == 0
if p.returncode != 0:
# not a working vsftpd
self.danted = None
m = re.match(r'^Dante v(\d+\.\d+\.\d+).*', p.stdout)
m = re.match(r"^Dante v(\d+\.\d+\.\d+).*", p.stdout)
if not m:
m = re.match(r'^Dante v(\d+\.\d+\.\d+).*', p.stderr)
m = re.match(r"^Dante v(\d+\.\d+\.\d+).*", p.stderr)
if m:
self._danted_version = m.group(1)
else:
self.danted = None
raise Exception(f'Unable to determine danted version from: {p.stderr}')
raise Exception(f"Unable to determine danted version from: {p.stderr}")
self.sshd = self.config['sshd']['sshd']
if self.sshd == '':
self.sshd = self.config["sshd"]["sshd"]
if self.sshd == "":
self.sshd = None
self._sshd_version = None
if self.sshd is not None:
p = subprocess.run(args=[self.sshd, '-V'],
capture_output=True, text=True)
p = subprocess.run(args=[self.sshd, "-V"], capture_output=True, text=True)
assert p.returncode == 0
if p.returncode != 0:
self.sshd = None
else:
m = re.match(r'^OpenSSH_(\d+\.\d+.*),.*', p.stderr)
assert m, f'version: {p.stderr}'
m = re.match(r"^OpenSSH_(\d+\.\d+.*),.*", p.stderr)
assert m, f"version: {p.stderr}"
if m:
self._sshd_version = m.group(1)
else:
self.sshd = None
raise Exception(f'Unable to determine sshd version from: {p.stderr}')
raise Exception(
f"Unable to determine sshd version from: {p.stderr}"
)
if self.sshd:
self.sftpd = self.config['sshd']['sftpd']
if self.sftpd == '':
self.sftpd = self.config["sshd"]["sftpd"]
if self.sftpd == "":
self.sftpd = None
else:
self.sftpd = None
self._tcpdump = shutil.which('tcpdump')
self._tcpdump = shutil.which("tcpdump")
@property
def httpd_version(self):
if self._httpd_version is None and self.apxs is not None:
try:
p = subprocess.run(args=[self.apxs, '-q', 'HTTPD_VERSION'],
capture_output=True, text=True)
p = subprocess.run(
args=[self.apxs, "-q", "HTTPD_VERSION"],
capture_output=True,
text=True,
)
if p.returncode != 0:
log.error(f'{self.apxs} failed to query HTTPD_VERSION: {p}')
log.error(f"{self.apxs} failed to query HTTPD_VERSION: {p}")
else:
self._httpd_version = p.stdout.strip()
except Exception:
log.exception(f'{self.apxs} failed to run')
log.exception(f"{self.apxs} failed to run")
return self._httpd_version
def versiontuple(self, v):
v = re.sub(r'(\d+\.\d+(\.\d+)?)(-\S+)?', r'\1', v)
return tuple(map(int, v.split('.')))
v = re.sub(r"(\d+\.\d+(\.\d+)?)(-\S+)?", r"\1", v)
return tuple(map(int, v.split(".")))
def httpd_is_at_least(self, minv):
if self.httpd_version is None:
@ -344,15 +399,17 @@ class EnvConfig:
return hv >= self.versiontuple(minv)
def is_complete(self) -> bool:
return os.path.isfile(self.httpd) and \
self.apxs is not None and \
os.path.isfile(self.apxs)
return (
os.path.isfile(self.httpd)
and self.apxs is not None
and os.path.isfile(self.apxs)
)
def get_incomplete_reason(self) -> Optional[str]:
if self.httpd is None or len(self.httpd.strip()) == 0:
return 'httpd not configured, see `--with-test-httpd=<path>`'
return "httpd not configured, see `--with-test-httpd=<path>`"
if not os.path.isfile(self.httpd):
return f'httpd ({self.httpd}) not found'
return f"httpd ({self.httpd}) not found"
if self.apxs is None:
return "command apxs not found (commonly provided in apache2-dev)"
if not os.path.isfile(self.apxs):
@ -371,18 +428,21 @@ class EnvConfig:
def vsftpd_version(self):
return self._vsftpd_version
@property
def h2o_version(self):
return self._h2o_version
@property
def tcpdmp(self) -> Optional[str]:
return self._tcpdump
def clear_locks(self):
ca_lock = os.path.join(self.gen_root, 'ca/ca.lock')
ca_lock = os.path.join(self.gen_root, "ca/ca.lock")
if os.path.exists(ca_lock):
os.remove(ca_lock)
class Env:
SERVER_TIMEOUT = 30 # seconds to wait for server to come up/reload
CONFIG = EnvConfig()
@ -407,98 +467,106 @@ class Env:
def have_h3_server() -> bool:
return Env.CONFIG.nghttpx_with_h3
@staticmethod
def have_h2o() -> bool:
return Env.CONFIG.h2o is not None
@staticmethod
def have_ssl_curl() -> bool:
return Env.curl_has_feature('ssl') or Env.curl_has_feature('multissl')
return Env.curl_has_feature("ssl") or Env.curl_has_feature("multissl")
@staticmethod
def have_h2_curl() -> bool:
return 'http2' in Env.CONFIG.curl_props['features']
return "http2" in Env.CONFIG.curl_props["features"]
@staticmethod
def have_h3_curl() -> bool:
return 'http3' in Env.CONFIG.curl_props['features']
return "http3" in Env.CONFIG.curl_props["features"]
@staticmethod
def have_compressed_curl() -> bool:
return 'brotli' in Env.CONFIG.curl_props['libs'] or \
'zlib' in Env.CONFIG.curl_props['libs'] or \
'zstd' in Env.CONFIG.curl_props['libs']
return (
"brotli" in Env.CONFIG.curl_props["libs"]
or "zlib" in Env.CONFIG.curl_props["libs"]
or "zstd" in Env.CONFIG.curl_props["libs"]
)
@staticmethod
def curl_uses_lib(libname: str) -> bool:
return libname.lower() in Env.CONFIG.curl_props['libs']
return libname.lower() in Env.CONFIG.curl_props["libs"]
@staticmethod
def curl_uses_any_libs(libs: List[str]) -> bool:
for libname in libs:
if libname.lower() in Env.CONFIG.curl_props['libs']:
if libname.lower() in Env.CONFIG.curl_props["libs"]:
return True
return False
@staticmethod
def curl_uses_ossl_quic() -> bool:
if Env.have_h3_curl():
return not Env.curl_uses_lib('ngtcp2') and Env.curl_uses_lib('nghttp3')
return not Env.curl_uses_lib("ngtcp2") and Env.curl_uses_lib("nghttp3")
return False
@staticmethod
def curl_version_string() -> str:
return Env.CONFIG.curl_props['version_string']
return Env.CONFIG.curl_props["version_string"]
@staticmethod
def curl_features_string() -> str:
return Env.CONFIG.curl_props['features_string']
return Env.CONFIG.curl_props["features_string"]
@staticmethod
def curl_has_feature(feature: str) -> bool:
return feature.lower() in Env.CONFIG.curl_props['features']
return feature.lower() in Env.CONFIG.curl_props["features"]
@staticmethod
def curl_protocols_string() -> str:
return Env.CONFIG.curl_props['protocols_string']
return Env.CONFIG.curl_props["protocols_string"]
@staticmethod
def curl_has_protocol(protocol: str) -> bool:
return protocol.lower() in Env.CONFIG.curl_props['protocols']
return protocol.lower() in Env.CONFIG.curl_props["protocols"]
@staticmethod
def curl_lib_version(libname: str) -> str:
prefix = f'{libname.lower()}/'
for lversion in Env.CONFIG.curl_props['lib_versions']:
prefix = f"{libname.lower()}/"
for lversion in Env.CONFIG.curl_props["lib_versions"]:
if lversion.startswith(prefix):
return lversion[len(prefix):]
return 'unknown'
return lversion[len(prefix) :]
return "unknown"
@staticmethod
def curl_lib_version_at_least(libname: str, min_version) -> bool:
lversion = Env.curl_lib_version(libname)
if lversion != 'unknown':
return Env.CONFIG.versiontuple(min_version) <= \
Env.CONFIG.versiontuple(lversion)
if lversion != "unknown":
return Env.CONFIG.versiontuple(min_version) <= Env.CONFIG.versiontuple(
lversion
)
return False
@staticmethod
def curl_lib_version_before(libname: str, lib_version) -> bool:
lversion = Env.curl_lib_version(libname)
if lversion != 'unknown':
if m := re.match(r'(\d+\.\d+\.\d+).*', lversion):
if lversion != "unknown":
if m := re.match(r"(\d+\.\d+\.\d+).*", lversion):
lversion = m.group(1)
return Env.CONFIG.versiontuple(lib_version) > \
Env.CONFIG.versiontuple(lversion)
return Env.CONFIG.versiontuple(lib_version) > Env.CONFIG.versiontuple(
lversion
)
return False
@staticmethod
def curl_os() -> str:
return Env.CONFIG.curl_props['os']
return Env.CONFIG.curl_props["os"]
@staticmethod
def curl_fullname() -> str:
return Env.CONFIG.curl_props['fullname']
return Env.CONFIG.curl_props["fullname"]
@staticmethod
def curl_version() -> str:
return Env.CONFIG.curl_props['version']
return Env.CONFIG.curl_props["version"]
@staticmethod
def curl_is_debug() -> bool:
@ -528,32 +596,31 @@ class Env:
@staticmethod
def curl_can_h3_early_data() -> bool:
return Env.curl_can_early_data() and \
Env.curl_uses_lib('ngtcp2')
return Env.curl_can_early_data() and Env.curl_uses_lib("ngtcp2")
@staticmethod
def http_protos() -> List[str]:
# http protocols we can test
if Env.have_h2_curl():
if Env.have_h3():
return ['http/1.1', 'h2', 'h3']
return ['http/1.1', 'h2']
return ['http/1.1']
return ["http/1.1", "h2", "h3"]
return ["http/1.1", "h2"]
return ["http/1.1"]
@staticmethod
def http_h1_h2_protos() -> List[str]:
# http 1+2 protocols we can test
if Env.have_h2_curl():
return ['http/1.1', 'h2']
return ['http/1.1']
return ["http/1.1", "h2"]
return ["http/1.1"]
@staticmethod
def http_mplx_protos() -> List[str]:
# http multiplexing protocols we can test
if Env.have_h2_curl():
if Env.have_h3():
return ['h2', 'h3']
return ['h2']
return ["h2", "h3"]
return ["h2"]
return []
@staticmethod
@ -572,6 +639,10 @@ class Env:
def caddy_version() -> str:
return Env.CONFIG.caddy_version
@staticmethod
def h2o_version() -> str:
return Env.CONFIG.h2o_version
@staticmethod
def caddy_is_at_least(minv) -> bool:
return Env.CONFIG.caddy_is_at_least(minv)
@ -611,21 +682,20 @@ class Env:
def __init__(self, pytestconfig=None, env_config=None):
if env_config:
Env.CONFIG = env_config
self._verbose = pytestconfig.option.verbose \
if pytestconfig is not None else 0
self._verbose = pytestconfig.option.verbose if pytestconfig is not None else 0
self._ca = None
self._test_timeout = 300.0 if self._verbose > 1 else 60.0 # seconds
def issue_certs(self):
if self._ca is None:
# ca_dir = os.path.join(self.CONFIG.gen_root, 'ca')
ca_dir = os.path.join(self.gen_dir, 'ca')
ca_dir = os.path.join(self.gen_dir, "ca")
os.makedirs(ca_dir, exist_ok=True)
lock_file = os.path.join(ca_dir, 'ca.lock')
lock_file = os.path.join(ca_dir, "ca.lock")
with FileLock(lock_file):
self._ca = TestCA.create_root(name=self.CONFIG.tld,
store_dir=ca_dir,
key_type="rsa2048")
self._ca = TestCA.create_root(
name=self.CONFIG.tld, store_dir=ca_dir, key_type="rsa2048"
)
self._ca.issue_certs(self.CONFIG.cert_specs)
if self.have_openssl():
self._ca.create_hashdir(self.openssl)
@ -714,19 +784,19 @@ class Env:
@property
def http_port(self) -> int:
return self.CONFIG.ports.get('http', 0)
return self.CONFIG.ports.get("http", 0)
@property
def https_port(self) -> int:
return self.CONFIG.ports['https']
return self.CONFIG.ports["https"]
@property
def https_only_tcp_port(self) -> int:
return self.CONFIG.ports['https-tcp-only']
return self.CONFIG.ports["https-tcp-only"]
@property
def nghttpx_https_port(self) -> int:
return self.CONFIG.ports['nghttpx_https']
return self.CONFIG.ports["nghttpx_https"]
@property
def h3_port(self) -> int:
@ -734,27 +804,35 @@ class Env:
@property
def proxy_port(self) -> int:
return self.CONFIG.ports['proxy']
return self.CONFIG.ports["proxy"]
@property
def proxys_port(self) -> int:
return self.CONFIG.ports['proxys']
return self.CONFIG.ports["proxys"]
@property
def ftp_port(self) -> int:
return self.CONFIG.ports['ftp']
return self.CONFIG.ports["ftp"]
@property
def ftps_port(self) -> int:
return self.CONFIG.ports['ftps']
return self.CONFIG.ports["ftps"]
@property
def h2proxys_port(self) -> int:
return self.CONFIG.ports['h2proxys']
return self.CONFIG.ports["h2proxys"]
def pts_port(self, proto: str = 'http/1.1') -> int:
@property
def h3proxys_port(self) -> int:
return self.CONFIG.ports["h3proxys"]
def pts_port(self, proto: str = "http/1.1") -> int:
# proxy tunnel port
return self.CONFIG.ports['h2proxys' if proto == 'h2' else 'proxys']
if proto == "h3":
return self.CONFIG.ports["h3proxys"]
if proto == "h2":
return self.CONFIG.ports["h2proxys"]
return self.CONFIG.ports["proxys"]
@property
def caddy(self) -> str:
@ -762,11 +840,11 @@ class Env:
@property
def caddy_https_port(self) -> int:
return self.CONFIG.ports['caddys']
return self.CONFIG.ports["caddys"]
@property
def caddy_http_port(self) -> int:
return self.CONFIG.ports['caddy']
return self.CONFIG.ports["caddy"]
@property
def danted(self) -> str:
@ -778,7 +856,7 @@ class Env:
@property
def ws_port(self) -> int:
return self.CONFIG.ports['ws']
return self.CONFIG.ports["ws"]
@property
def curl(self) -> str:
@ -802,58 +880,65 @@ class Env:
@property
def slow_network(self) -> bool:
return "CURL_DBG_SOCK_WBLOCK" in os.environ or \
"CURL_DBG_SOCK_WPARTIAL" in os.environ
return (
"CURL_DBG_SOCK_WBLOCK" in os.environ
or "CURL_DBG_SOCK_WPARTIAL" in os.environ
)
@property
def ci_run(self) -> bool:
return "CURL_CI" in os.environ
def port_for(self, alpn_proto: Optional[str] = None):
if alpn_proto is None or \
alpn_proto in ['h2', 'http/1.1', 'http/1.0', 'http/0.9']:
if alpn_proto is None or alpn_proto in [
"h2",
"http/1.1",
"http/1.0",
"http/0.9",
]:
return self.https_port
if alpn_proto in ['h3']:
if alpn_proto in ["h3"]:
return self.h3_port
return self.http_port
def authority_for(self, domain: str, alpn_proto: Optional[str] = None):
return f'{domain}:{self.port_for(alpn_proto=alpn_proto)}'
return f"{domain}:{self.port_for(alpn_proto=alpn_proto)}"
def make_data_file(self, indir: str, fname: str, fsize: int,
line_length: int = 1024) -> str:
def make_data_file(
self, indir: str, fname: str, fsize: int, line_length: int = 1024
) -> str:
if line_length < 11:
raise RuntimeError('line_length less than 11 not supported')
raise RuntimeError("line_length less than 11 not supported")
fpath = os.path.join(indir, fname)
s10 = "0123456789"
s = round((line_length / 10) + 1) * s10
s = s[0:line_length-11]
with open(fpath, 'w') as fd:
s = s[0 : line_length - 11]
with open(fpath, "w") as fd:
for i in range(int(fsize / line_length)):
fd.write(f"{i:09d}-{s}\n")
remain = int(fsize % line_length)
if remain != 0:
i = int(fsize / line_length) + 1
fd.write(f"{i:09d}-{s}"[0:remain-1] + "\n")
fd.write(f"{i:09d}-{s}"[0 : remain - 1] + "\n")
return fpath
def make_data_gzipbomb(self, indir: str, fname: str, fsize: int) -> str:
fpath = os.path.join(indir, fname)
gzpath = f'{fpath}.gz'
varpath = f'{fpath}.var'
gzpath = f"{fpath}.gz"
varpath = f"{fpath}.var"
with open(fpath, 'w') as fd:
fd.write('not what we are looking for!\n')
with open(fpath, "w") as fd:
fd.write("not what we are looking for!\n")
count = int(fsize / 1024)
zero1k = bytearray(1024)
with gzip.open(gzpath, 'wb') as fd:
with gzip.open(gzpath, "wb") as fd:
for _ in range(count):
fd.write(zero1k)
with open(varpath, 'w') as fd:
fd.write(f'URI: {fname}\n')
fd.write('\n')
fd.write(f'URI: {fname}.gz\n')
fd.write('Content-Type: text/plain\n')
fd.write('Content-Encoding: x-gzip\n')
fd.write('\n')
with open(varpath, "w") as fd:
fd.write(f"URI: {fname}\n")
fd.write("\n")
fd.write(f"URI: {fname}.gz\n")
fd.write("Content-Type: text/plain\n")
fd.write("Content-Encoding: x-gzip\n")
fd.write("\n")
return fpath