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 logging 28import re 29import pytest 30 31from testenv import Env, CurlClient, ExecResult 32 33 34log = logging.getLogger(__name__) 35 36 37@pytest.mark.skipif(condition=Env.setup_incomplete(), 38 reason=f"missing: {Env.incomplete_reason()}") 39class TestProxyAuth: 40 41 @pytest.fixture(autouse=True, scope='class') 42 def _class_scope(self, env, httpd, nghttpx_fwd): 43 if env.have_nghttpx(): 44 nghttpx_fwd.start_if_needed() 45 httpd.clear_extra_configs() 46 httpd.set_proxy_auth(True) 47 httpd.reload() 48 yield 49 httpd.set_proxy_auth(False) 50 httpd.reload() 51 52 def get_tunnel_proto_used(self, r: ExecResult): 53 for line in r.trace_lines: 54 m = re.match(r'.* CONNECT tunnel: (\S+) negotiated$', line) 55 if m: 56 return m.group(1) 57 assert False, f'tunnel protocol not found in:\n{"".join(r.trace_lines)}' 58 return None 59 60 # download via http: proxy (no tunnel), no auth 61 def test_13_01_proxy_no_auth(self, env: Env, httpd, repeat): 62 curl = CurlClient(env=env) 63 url = f'http://localhost:{env.http_port}/data.json' 64 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 65 extra_args=curl.get_proxy_args(proxys=False)) 66 r.check_response(count=1, http_status=407) 67 68 # download via http: proxy (no tunnel), auth 69 def test_13_02_proxy_auth(self, env: Env, httpd, repeat): 70 curl = CurlClient(env=env) 71 url = f'http://localhost:{env.http_port}/data.json' 72 xargs = curl.get_proxy_args(proxys=False) 73 xargs.extend(['--proxy-user', 'proxy:proxy']) 74 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 75 extra_args=xargs) 76 r.check_response(count=1, http_status=200) 77 78 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 79 reason='curl lacks HTTPS-proxy support') 80 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 81 def test_13_03_proxys_no_auth(self, env: Env, httpd, nghttpx_fwd, repeat): 82 curl = CurlClient(env=env) 83 url = f'http://localhost:{env.http_port}/data.json' 84 xargs = curl.get_proxy_args(proxys=True) 85 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 86 extra_args=xargs) 87 r.check_response(count=1, http_status=407) 88 89 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 90 reason='curl lacks HTTPS-proxy support') 91 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 92 def test_13_04_proxys_auth(self, env: Env, httpd, nghttpx_fwd, repeat): 93 curl = CurlClient(env=env) 94 url = f'http://localhost:{env.http_port}/data.json' 95 xargs = curl.get_proxy_args(proxys=True) 96 xargs.extend(['--proxy-user', 'proxy:proxy']) 97 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 98 extra_args=xargs) 99 r.check_response(count=1, http_status=200) 100 101 def test_13_05_tunnel_http_no_auth(self, env: Env, httpd, repeat): 102 curl = CurlClient(env=env) 103 url = f'http://localhost:{env.http_port}/data.json' 104 xargs = curl.get_proxy_args(proxys=False, tunnel=True) 105 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 106 extra_args=xargs) 107 # expect "COULD_NOT_CONNECT" 108 r.check_response(exitcode=56, http_status=None) 109 110 def test_13_06_tunnel_http_auth(self, env: Env, httpd, repeat): 111 curl = CurlClient(env=env) 112 url = f'http://localhost:{env.http_port}/data.json' 113 xargs = curl.get_proxy_args(proxys=False, tunnel=True) 114 xargs.extend(['--proxy-user', 'proxy:proxy']) 115 r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True, 116 extra_args=xargs) 117 r.check_response(count=1, http_status=200) 118 119 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 120 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 121 reason='curl lacks HTTPS-proxy support') 122 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 123 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 124 def test_13_07_tunnels_no_auth(self, env: Env, httpd, proto, tunnel, repeat): 125 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 126 pytest.skip('only supported with nghttp2') 127 curl = CurlClient(env=env) 128 url = f'https://localhost:{env.https_port}/data.json' 129 xargs = curl.get_proxy_args(proxys=True, tunnel=True, proto=tunnel) 130 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, 131 extra_args=xargs) 132 # expect "COULD_NOT_CONNECT" 133 r.check_response(exitcode=56, http_status=None) 134 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 135 if tunnel == 'h2' else 'HTTP/1.1' 136 137 @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available") 138 @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'), 139 reason='curl lacks HTTPS-proxy support') 140 @pytest.mark.parametrize("proto", ['http/1.1', 'h2']) 141 @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2']) 142 def test_13_08_tunnels_auth(self, env: Env, httpd, proto, tunnel, repeat): 143 if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'): 144 pytest.skip('only supported with nghttp2') 145 curl = CurlClient(env=env) 146 url = f'https://localhost:{env.https_port}/data.json' 147 xargs = curl.get_proxy_args(proxys=True, tunnel=True, proto=tunnel) 148 xargs.extend(['--proxy-user', 'proxy:proxy']) 149 r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True, 150 extra_args=xargs) 151 r.check_response(count=1, http_status=200, 152 protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1') 153 assert self.get_tunnel_proto_used(r) == 'HTTP/2' \ 154 if tunnel == 'h2' else 'HTTP/1.1' 155
