1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3#*************************************************************************** 4# _ _ ____ _ 5# Project ___| | | | _ \| | 6# / __| | | | |_) | | 7# | (__| |_| | _ <| |___ 8# \___|\___/|_| \_\_____| 9# 10# Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al. 11# 12# This software is licensed as described in the file COPYING, which 13# you should have received as part of this distribution. The terms 14# are also available at https://curl.se/docs/copyright.html. 15# 16# You may opt to use, copy, modify, merge, publish, distribute and/or sell 17# copies of the Software, and permit persons to whom the Software is 18# furnished to do so, under the terms of the COPYING file. 19# 20# This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY 21# KIND, either express or implied. 22# 23# SPDX-License-Identifier: curl 24# 25########################################################################### 26# 27import filecmp 28import logging 29import os 30import re 31import pytest 32 33from testenv import Env, CurlClient, ExecResult 34 35 36log = logging.getLogger(__name__) 37 38 39class TestProxy: 40 41 @pytest.fixture(autouse=True, scope='class') 42 def _class_scope(self, env, httpd, nghttpx_fwd): 43 push_dir = os.path.join(httpd.docs_dir, 'push') 44 if not os.path.exists(push_dir): 45 os.makedirs(push_dir) 46 if env.have_nghttpx(): 47 nghttpx_fwd.start_if_needed() 48 env.make_data_file(indir=env.gen_dir, fname="data-100k", fsize=100*1024) 49 env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024) 50 httpd.clear_extra_configs() 51 httpd.reload() 52 53 def get_tunnel_proto_used(self, r: ExecResult): 54 for l in r.trace_lines: 55 m = re.match(r'.* CONNECT tunnel: (\S+) negotiated$', l) 56 if m: 57 return m.group(1) 58 assert False, f'tunnel protocol not found in:\n{"".join(r.trace_lines)}' 59 return None 60 61 # download via http: proxy (no tunnel) 62 def test_10_01_proxy_http(self, env: Env, httpd, repeat): 63 curl = CurlClient(env=env) 64 url = f'http://localhost:{env.http_port}/data.json' 65 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 66 extra_args=curl.get_proxy_args(proxys=False)) 67 r.check_response(count=1, http_status=200) 68 69 # download via https: proxy (no tunnel) 70 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 71 reason='curl lacks HTTPS-proxy support') 72 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 73 def test_10_02_proxys_down(self, env: Env, httpd, proto, repeat): 74 if proto == 'h2' and not env.curl_uses_lib('nghttp2'): 75 pytest.skip('only supported with nghttp2') 76 curl = CurlClient(env=env) 77 url = f'http://localhost:{env.http_port}/data.json' 78 xargs = curl.get_proxy_args(proto=proto) 79 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 80 extra_args=xargs) 81 r.check_response(count=1, http_status=200, 82 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 83 84 # upload via https: with proto (no tunnel) 85 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 86 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 87 @pytest.mark.parametrize("fname, fcount", [ 88 ['data.json', 5], 89 ['data-100k', 5], 90 ['data-1m', 2] 91 ]) 92 @pytest.mark.skipif(condition=not Env.have_nghttpx(), 93 reason="no nghttpx available") 94 def test_10_02_proxys_up(self, env: Env, httpd, nghttpx, proto, 95 fname, fcount, repeat): 96 if proto == 'h2' and not env.curl_uses_lib('nghttp2'): 97 pytest.skip('only supported with nghttp2') 98 count = fcount 99 srcfile = os.path.join(httpd.docs_dir, fname) 100 curl = CurlClient(env=env) 101 url = f'http://localhost:{env.http_port}/curltest/echo?id=[0-{count-1}]' 102 xargs = curl.get_proxy_args(proto=proto) 103 r = curl.http_upload(urls=[url], data=f'@{srcfile}', alpn_proto=proto, 104 extra_args=xargs) 105 r.check_response(count=count, http_status=200, 106 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 107 indata = open(srcfile).readlines() 108 for i in range(count): 109 respdata = open(curl.response_file(i)).readlines() 110 assert respdata == indata 111 112 # download http: via http: proxytunnel 113 def test_10_03_proxytunnel_http(self, env: Env, httpd, repeat): 114 curl = CurlClient(env=env) 115 url = f'http://localhost:{env.http_port}/data.json' 116 xargs = curl.get_proxy_args(proxys=False, tunnel=True) 117 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 118 extra_args=xargs) 119 r.check_response(count=1, http_status=200) 120 121 # download http: via https: proxytunnel 122 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 123 reason='curl lacks HTTPS-proxy support') 124 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 125 def test_10_04_proxy_https(self, env: Env, httpd, nghttpx_fwd, repeat): 126 curl = CurlClient(env=env) 127 url = f'http://localhost:{env.http_port}/data.json' 128 xargs = curl.get_proxy_args(tunnel=True) 129 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 130 extra_args=xargs) 131 r.check_response(count=1, http_status=200) 132 133 # download https: with proto via http: proxytunnel 134 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 135 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 136 def test_10_05_proxytunnel_http(self, env: Env, httpd, proto, repeat): 137 curl = CurlClient(env=env) 138 url = f'https://localhost:{env.https_port}/data.json' 139 xargs = curl.get_proxy_args(proxys=False, tunnel=True) 140 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, 141 extra_args=xargs) 142 r.check_response(count=1, http_status=200, 143 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 144 145 # download https: with proto via https: proxytunnel 146 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 147 reason='curl lacks HTTPS-proxy support') 148 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 149 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 150 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 151 def test_10_06_proxytunnel_https(self, env: Env, httpd, nghttpx_fwd, proto, tunnel, repeat): 152 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 153 pytest.skip('only supported with nghttp2') 154 curl = CurlClient(env=env) 155 url = f'https://localhost:{env.https_port}/data.json?[0-0]' 156 xargs = curl.get_proxy_args(tunnel=True, proto=tunnel) 157 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, 158 extra_args=xargs) 159 r.check_response(count=1, http_status=200, 160 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 161 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 162 if tunnel == 'h2' else 'HTTP/1.1' 163 srcfile = os.path.join(httpd.docs_dir, 'data.json') 164 dfile = curl.download_file(0) 165 assert filecmp.cmp(srcfile, dfile, shallow=False) 166 167 # download many https: with proto via https: proxytunnel 168 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 169 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 170 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 171 @pytest.mark.parametrize("fname, fcount", [ 172 ['data.json', 100], 173 ['data-100k', 20], 174 ['data-1m', 5] 175 ]) 176 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 177 def test_10_07_pts_down_small(self, env: Env, httpd, nghttpx_fwd, proto, 178 tunnel, fname, fcount, repeat): 179 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 180 pytest.skip('only supported with nghttp2') 181 count = fcount 182 curl = CurlClient(env=env) 183 url = f'https://localhost:{env.https_port}/{fname}?[0-{count-1}]' 184 xargs = curl.get_proxy_args(tunnel=True, proto=tunnel) 185 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, 186 extra_args=xargs) 187 r.check_response(count=count, http_status=200, 188 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 189 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 190 if tunnel == 'h2' else 'HTTP/1.1' 191 srcfile = os.path.join(httpd.docs_dir, fname) 192 for i in range(count): 193 dfile = curl.download_file(i) 194 assert filecmp.cmp(srcfile, dfile, shallow=False) 195 assert r.total_connects == 1, r.dump_logs() 196 197 # upload many https: with proto via https: proxytunnel 198 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 199 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 200 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 201 @pytest.mark.parametrize("fname, fcount", [ 202 ['data.json', 50], 203 ['data-100k', 20], 204 ['data-1m', 5] 205 ]) 206 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 207 def test_10_08_upload_seq_large(self, env: Env, httpd, nghttpx, proto, 208 tunnel, fname, fcount, repeat): 209 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 210 pytest.skip('only supported with nghttp2') 211 count = fcount 212 srcfile = os.path.join(httpd.docs_dir, fname) 213 curl = CurlClient(env=env) 214 url = f'https://localhost:{env.https_port}/curltest/echo?id=[0-{count-1}]' 215 xargs = curl.get_proxy_args(tunnel=True, proto=tunnel) 216 r = curl.http_upload(urls=[url], data=f'@{srcfile}', alpn_proto=proto, 217 extra_args=xargs) 218 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 219 if tunnel == 'h2' else 'HTTP/1.1' 220 r.check_response(count=count, http_status=200) 221 indata = open(srcfile).readlines() 222 for i in range(count): 223 respdata = open(curl.response_file(i)).readlines() 224 assert respdata == indata, f'resonse {i} differs' 225 assert r.total_connects == 1, r.dump_logs() 226 227 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 228 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 229 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 230 def test_10_09_reuse_ser(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat): 231 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 232 pytest.skip('only supported with nghttp2') 233 curl = CurlClient(env=env) 234 url1 = f'https://localhost:{env.https_port}/data.json' 235 url2 = f'http://localhost:{env.http_port}/data.json' 236 xargs = curl.get_proxy_args(tunnel=True, proto=tunnel) 237 r = curl.http_download(urls=[url1, url2], alpn_proto='http/1.1', with_stats=True, 238 extra_args=xargs) 239 r.check_response(count=2, http_status=200) 240 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 241 if tunnel == 'h2' else 'HTTP/1.1' 242 if tunnel == 'h2': 243 # TODO: we would like to reuse the first connection for the 244 # second URL, but this is currently not possible 245 # assert r.total_connects == 1 246 assert r.total_connects == 2 247 else: 248 assert r.total_connects == 2 249 250 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 251 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 252 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 253 def test_10_10_reuse_proxy(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat): 254 # url twice via https: proxy separated with '--next', will reuse 255 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 256 pytest.skip('only supported with nghttp2') 257 curl = CurlClient(env=env) 258 url = f'https://localhost:{env.https_port}/data.json' 259 proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel) 260 r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 261 extra_args=proxy_args) 262 r1.check_response(count=1, http_status=200) 263 assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \ 264 if tunnel == 'h2' else 'HTTP/1.1' 265 # get the args, duplicate separated with '--next' 266 x2_args = r1.args[1:] 267 x2_args.append('--next') 268 x2_args.extend(proxy_args) 269 r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 270 extra_args=x2_args) 271 r2.check_response(count=2, http_status=200) 272 assert r2.total_connects == 1 273 274 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 275 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 276 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 277 @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported") 278 def test_10_11_noreuse_proxy_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat): 279 # different --proxy-tls13-ciphers, no reuse of connection for https: 280 curl = CurlClient(env=env) 281 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 282 pytest.skip('only supported with nghttp2') 283 url = f'https://localhost:{env.https_port}/data.json' 284 proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel) 285 r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 286 extra_args=proxy_args) 287 r1.check_response(count=1, http_status=200) 288 assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \ 289 if tunnel == 'h2' else 'HTTP/1.1' 290 # get the args, duplicate separated with '--next' 291 x2_args = r1.args[1:] 292 x2_args.append('--next') 293 x2_args.extend(proxy_args) 294 x2_args.extend(['--proxy-tls13-ciphers', 'TLS_AES_256_GCM_SHA384']) 295 r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 296 extra_args=x2_args) 297 r2.check_response(count=2, http_status=200) 298 assert r2.total_connects == 2 299 300 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 301 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 302 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 303 @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported") 304 def test_10_12_noreuse_proxy_http(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat): 305 # different --proxy-tls13-ciphers, no reuse of connection for http: 306 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 307 pytest.skip('only supported with nghttp2') 308 curl = CurlClient(env=env) 309 url = f'http://localhost:{env.http_port}/data.json' 310 proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel) 311 r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 312 extra_args=proxy_args) 313 r1.check_response(count=1, http_status=200) 314 assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \ 315 if tunnel == 'h2' else 'HTTP/1.1' 316 # get the args, duplicate separated with '--next' 317 x2_args = r1.args[1:] 318 x2_args.append('--next') 319 x2_args.extend(proxy_args) 320 x2_args.extend(['--proxy-tls13-ciphers', 'TLS_AES_256_GCM_SHA384']) 321 r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 322 extra_args=x2_args) 323 r2.check_response(count=2, http_status=200) 324 assert r2.total_connects == 2 325 326 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 327 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 328 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 329 @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported") 330 def test_10_13_noreuse_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat): 331 # different --tls13-ciphers on https: same proxy config 332 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 333 pytest.skip('only supported with nghttp2') 334 curl = CurlClient(env=env) 335 url = f'https://localhost:{env.https_port}/data.json' 336 proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel) 337 r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 338 extra_args=proxy_args) 339 r1.check_response(count=1, http_status=200) 340 assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \ 341 if tunnel == 'h2' else 'HTTP/1.1' 342 # get the args, duplicate separated with '--next' 343 x2_args = r1.args[1:] 344 x2_args.append('--next') 345 x2_args.extend(proxy_args) 346 x2_args.extend(['--tls13-ciphers', 'TLS_AES_256_GCM_SHA384']) 347 r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 348 extra_args=x2_args) 349 r2.check_response(count=2, http_status=200) 350 assert r2.total_connects == 2 351 352 # download via https: proxy (no tunnel) using IP address 353 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 354 reason='curl lacks HTTPS-proxy support') 355 @pytest.mark.skipif(condition=Env.curl_uses_lib('bearssl'), reason="ip address cert verification not supported") 356 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 357 def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto, repeat): 358 if proto == 'h2' and not env.curl_uses_lib('nghttp2'): 359 pytest.skip('only supported with nghttp2') 360 curl = CurlClient(env=env) 361 url = f'http://localhost:{env.http_port}/data.json' 362 xargs = curl.get_proxy_args(proto=proto, use_ip=True) 363 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 364 extra_args=xargs) 365 if env.curl_uses_lib('mbedtls') and \ 366 not env.curl_lib_version_at_least('mbedtls', '3.5.0'): 367 r.check_exit_code(60) # CURLE_PEER_FAILED_VERIFICATION 368 else: 369 r.check_response(count=1, http_status=200, 370 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 371
