xref: /aosp_15_r20/external/curl/tests/http/test_10_proxy.py (revision 6236dae45794135f37c4eb022389c904c8b0090d)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#***************************************************************************
#                                  _   _ ____  _
#  Project                     ___| | | |  _ \| |
#                             / __| | | | |_) | |
#                            | (__| |_| |  _ <| |___
#                             \___|\___/|_| \_\_____|
#
# Copyright (C) Daniel Stenberg, <[email protected]>, et al.
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution. The terms
# are also available at https://curl.se/docs/copyright.html.
#
# You may opt to use, copy, modify, merge, publish, distribute and/or sell
# copies of the Software, and permit persons to whom the Software is
# furnished to do so, under the terms of the COPYING file.
#
# This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
# KIND, either express or implied.
#
# SPDX-License-Identifier: curl
#
###########################################################################
#
import filecmp
import logging
import os
import re
import pytest

from testenv import Env, CurlClient, ExecResult


log = logging.getLogger(__name__)


class TestProxy:

    @pytest.fixture(autouse=True, scope='class')
    def _class_scope(self, env, httpd, nghttpx_fwd):
        push_dir = os.path.join(httpd.docs_dir, 'push')
        if not os.path.exists(push_dir):
            os.makedirs(push_dir)
        if env.have_nghttpx():
            nghttpx_fwd.start_if_needed()
        env.make_data_file(indir=env.gen_dir, fname="data-100k", fsize=100*1024)
        env.make_data_file(indir=env.gen_dir, fname="data-10m", fsize=10*1024*1024)
        httpd.clear_extra_configs()
        httpd.reload()

    def get_tunnel_proto_used(self, r: ExecResult):
        for line in r.trace_lines:
            m = re.match(r'.* CONNECT tunnel: (\S+) negotiated$', line)
            if m:
                return m.group(1)
        assert False, f'tunnel protocol not found in:\n{"".join(r.trace_lines)}'
        return None

    # download via http: proxy (no tunnel)
    def test_10_01_proxy_http(self, env: Env, httpd, repeat):
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=curl.get_proxy_args(proxys=False))
        r.check_response(count=1, http_status=200)

    # download via https: proxy (no tunnel)
    @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
                        reason='curl lacks HTTPS-proxy support')
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    def test_10_02_proxys_down(self, env: Env, httpd, proto, repeat):
        if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        xargs = curl.get_proxy_args(proto=proto)
        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=xargs)
        r.check_response(count=1, http_status=200,
                         protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')

    # upload via https: with proto (no tunnel)
    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("fname, fcount", [
        ['data.json', 5],
        ['data-100k', 5],
        ['data-1m', 2]
    ])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(),
                        reason="no nghttpx available")
    def test_10_02_proxys_up(self, env: Env, httpd, nghttpx, proto,
                             fname, fcount, repeat):
        if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        count = fcount
        srcfile = os.path.join(httpd.docs_dir, fname)
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/curltest/echo?id=[0-{count-1}]'
        xargs = curl.get_proxy_args(proto=proto)
        r = curl.http_upload(urls=[url], data=f'@{srcfile}', alpn_proto=proto,
                             extra_args=xargs)
        r.check_response(count=count, http_status=200,
                         protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')
        indata = open(srcfile).readlines()
        for i in range(count):
            respdata = open(curl.response_file(i)).readlines()
            assert respdata == indata

    # download http: via http: proxytunnel
    def test_10_03_proxytunnel_http(self, env: Env, httpd, repeat):
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        xargs = curl.get_proxy_args(proxys=False, tunnel=True)
        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=xargs)
        r.check_response(count=1, http_status=200)

    # download http: via https: proxytunnel
    @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
                        reason='curl lacks HTTPS-proxy support')
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_04_proxy_https(self, env: Env, httpd, nghttpx_fwd, repeat):
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        xargs = curl.get_proxy_args(tunnel=True)
        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=xargs)
        r.check_response(count=1, http_status=200)

    # download https: with proto via http: proxytunnel
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    def test_10_05_proxytunnel_http(self, env: Env, httpd, proto, repeat):
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/data.json'
        xargs = curl.get_proxy_args(proxys=False, tunnel=True)
        r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True,
                               extra_args=xargs)
        r.check_response(count=1, http_status=200,
                         protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')

    # download https: with proto via https: proxytunnel
    @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
                        reason='curl lacks HTTPS-proxy support')
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_06_proxytunnel_https(self, env: Env, httpd, nghttpx_fwd, proto, tunnel, repeat):
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/data.json?[0-0]'
        xargs = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True,
                               extra_args=xargs)
        r.check_response(count=1, http_status=200,
                         protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')
        assert self.get_tunnel_proto_used(r) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        srcfile = os.path.join(httpd.docs_dir, 'data.json')
        dfile = curl.download_file(0)
        assert filecmp.cmp(srcfile, dfile, shallow=False)

    # download many https: with proto via https: proxytunnel
    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("fname, fcount", [
        ['data.json', 100],
        ['data-100k', 20],
        ['data-1m', 5]
    ])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_07_pts_down_small(self, env: Env, httpd, nghttpx_fwd, proto,
                                  tunnel, fname, fcount, repeat):
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        count = fcount
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/{fname}?[0-{count-1}]'
        xargs = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r = curl.http_download(urls=[url], alpn_proto=proto, with_stats=True,
                               extra_args=xargs)
        r.check_response(count=count, http_status=200,
                         protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')
        assert self.get_tunnel_proto_used(r) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        srcfile = os.path.join(httpd.docs_dir, fname)
        for i in range(count):
            dfile = curl.download_file(i)
            assert filecmp.cmp(srcfile, dfile, shallow=False)
        assert r.total_connects == 1, r.dump_logs()

    # upload many https: with proto via https: proxytunnel
    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.parametrize("fname, fcount", [
        ['data.json', 50],
        ['data-100k', 20],
        ['data-1m', 5]
    ])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_08_upload_seq_large(self, env: Env, httpd, nghttpx, proto,
                                    tunnel, fname, fcount, repeat):
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        count = fcount
        srcfile = os.path.join(httpd.docs_dir, fname)
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/curltest/echo?id=[0-{count-1}]'
        xargs = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r = curl.http_upload(urls=[url], data=f'@{srcfile}', alpn_proto=proto,
                             extra_args=xargs)
        assert self.get_tunnel_proto_used(r) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        r.check_response(count=count, http_status=200)
        indata = open(srcfile).readlines()
        for i in range(count):
            respdata = open(curl.response_file(i)).readlines()
            assert respdata == indata, f'resonse {i} differs'
        assert r.total_connects == 1, r.dump_logs()

    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_09_reuse_ser(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url1 = f'https://localhost:{env.https_port}/data.json'
        url2 = f'http://localhost:{env.http_port}/data.json'
        xargs = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r = curl.http_download(urls=[url1, url2], alpn_proto='http/1.1', with_stats=True,
                               extra_args=xargs)
        r.check_response(count=2, http_status=200)
        assert self.get_tunnel_proto_used(r) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        if tunnel == 'h2':
            # TODO: we would like to reuse the first connection for the
            # second URL, but this is currently not possible
            # assert r.total_connects == 1
            assert r.total_connects == 2
        else:
            assert r.total_connects == 2

    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    def test_10_10_reuse_proxy(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
        # url twice via https: proxy separated with '--next', will reuse
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/data.json'
        proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=proxy_args)
        r1.check_response(count=1, http_status=200)
        assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        # get the args, duplicate separated with '--next'
        x2_args = r1.args[1:]
        x2_args.append('--next')
        x2_args.extend(proxy_args)
        r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=x2_args)
        r2.check_response(count=2, http_status=200)
        assert r2.total_connects == 1

    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
    def test_10_11_noreuse_proxy_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
        # different --proxy-tls13-ciphers, no reuse of connection for https:
        curl = CurlClient(env=env)
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        url = f'https://localhost:{env.https_port}/data.json'
        proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=proxy_args)
        r1.check_response(count=1, http_status=200)
        assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        # get the args, duplicate separated with '--next'
        x2_args = r1.args[1:]
        x2_args.append('--next')
        x2_args.extend(proxy_args)
        x2_args.extend(['--proxy-tls13-ciphers', 'TLS_AES_256_GCM_SHA384'])
        r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=x2_args)
        r2.check_response(count=2, http_status=200)
        assert r2.total_connects == 2

    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
    def test_10_12_noreuse_proxy_http(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
        # different --proxy-tls13-ciphers, no reuse of connection for http:
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=proxy_args)
        r1.check_response(count=1, http_status=200)
        assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        # get the args, duplicate separated with '--next'
        x2_args = r1.args[1:]
        x2_args.append('--next')
        x2_args.extend(proxy_args)
        x2_args.extend(['--proxy-tls13-ciphers', 'TLS_AES_256_GCM_SHA384'])
        r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=x2_args)
        r2.check_response(count=2, http_status=200)
        assert r2.total_connects == 2

    @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason="curl without SSL")
    @pytest.mark.parametrize("tunnel", ['http/1.1', 'h2'])
    @pytest.mark.skipif(condition=not Env.have_nghttpx(), reason="no nghttpx available")
    @pytest.mark.skipif(condition=not Env.curl_uses_lib('openssl'), reason="tls13-ciphers not supported")
    def test_10_13_noreuse_https(self, env: Env, httpd, nghttpx_fwd, tunnel, repeat):
        # different --tls13-ciphers on https: same proxy config
        if tunnel == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'https://localhost:{env.https_port}/data.json'
        proxy_args = curl.get_proxy_args(tunnel=True, proto=tunnel)
        r1 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=proxy_args)
        r1.check_response(count=1, http_status=200)
        assert self.get_tunnel_proto_used(r1) == 'HTTP/2' \
            if tunnel == 'h2' else 'HTTP/1.1'
        # get the args, duplicate separated with '--next'
        x2_args = r1.args[1:]
        x2_args.append('--next')
        x2_args.extend(proxy_args)
        x2_args.extend(['--tls13-ciphers', 'TLS_AES_256_GCM_SHA384'])
        r2 = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=x2_args)
        r2.check_response(count=2, http_status=200)
        assert r2.total_connects == 2

    # download via https: proxy (no tunnel) using IP address
    @pytest.mark.skipif(condition=not Env.curl_has_feature('HTTPS-proxy'),
                        reason='curl lacks HTTPS-proxy support')
    @pytest.mark.skipif(condition=Env.curl_uses_lib('bearssl'), reason="ip address cert verification not supported")
    @pytest.mark.parametrize("proto", ['http/1.1', 'h2'])
    def test_10_14_proxys_ip_addr(self, env: Env, httpd, proto, repeat):
        if proto == 'h2' and not env.curl_uses_lib('nghttp2'):
            pytest.skip('only supported with nghttp2')
        curl = CurlClient(env=env)
        url = f'http://localhost:{env.http_port}/data.json'
        xargs = curl.get_proxy_args(proto=proto, use_ip=True)
        r = curl.http_download(urls=[url], alpn_proto='http/1.1', with_stats=True,
                               extra_args=xargs)
        if env.curl_uses_lib('mbedtls') and \
                not env.curl_lib_version_at_least('mbedtls', '3.5.0'):
            r.check_exit_code(60) # CURLE_PEER_FAILED_VERIFICATION
        else:
            r.check_response(count=1, http_status=200,
                             protocol='HTTP/2' if proto == 'h2' else 'HTTP/1.1')