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