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