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 os 29import socket 30from threading import Thread 31import pytest 32 33from testenv import Env, CurlClient 34 35 36log = logging.getLogger(__name__) 37 38class UDSFaker: 39 40 def __init__(self, path): 41 self._uds_path = path 42 self._done = False 43 44 @property 45 def path(self): 46 return self._uds_path 47 48 def start(self): 49 def process(self): 50 self._socket.listen(1) 51 self._process() 52 53 try: 54 os.unlink(self._uds_path) 55 except OSError: 56 if os.path.exists(self._uds_path): 57 raise 58 self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 59 self._socket.bind(self._uds_path) 60 self._thread = Thread(target=process, daemon=True, args=[self]) 61 self._thread.start() 62 63 def stop(self): 64 self._done = True 65 self._socket.close() 66 67 def _process(self): 68 while self._done is False: 69 try: 70 c, client_address = self._socket.accept() 71 try: 72 data = c.recv(16) 73 c.sendall("""HTTP/1.1 200 Ok 74Server: UdsFaker 75Content-Type: application/json 76Content-Length: 19 77 78{ "host": "faked" }""".encode()) 79 finally: 80 c.close() 81 82 except ConnectionAbortedError: 83 self._done = True 84 except OSError: 85 self._done = True 86 87 88class TestUnix: 89 90 @pytest.fixture(scope="class") 91 def uds_faker(self, env: Env) -> UDSFaker: 92 uds_path = os.path.join(env.gen_dir, 'uds_11.sock') 93 faker = UDSFaker(path=uds_path) 94 faker.start() 95 yield faker 96 faker.stop() 97 98 # download http: via unix socket 99 def test_11_01_unix_connect_http(self, env: Env, httpd, uds_faker, repeat): 100 curl = CurlClient(env=env) 101 url = f'http://{env.domain1}:{env.http_port}/data.json' 102 r = curl.http_download(urls=[url], with_stats=True, 103 extra_args=[ 104 '--unix-socket', uds_faker.path, 105 ]) 106 r.check_response(count=1, http_status=200) 107 108 # download https: via unix socket 109 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 110 def test_11_02_unix_connect_http(self, env: Env, httpd, uds_faker, repeat): 111 curl = CurlClient(env=env) 112 url = f'https://{env.domain1}:{env.https_port}/data.json' 113 r = curl.http_download(urls=[url], with_stats=True, 114 extra_args=[ 115 '--unix-socket', uds_faker.path, 116 ]) 117 r.check_response(exitcode=35, http_status=None) 118 119 # download HTTP/3 via unix socket 120 @pytest.mark.skipif(condition=not Env.have_h3(), reason='h3 not supported') 121 def test_11_03_unix_connect_quic(self, env: Env, httpd, uds_faker, repeat): 122 curl = CurlClient(env=env) 123 url = f'https://{env.domain1}:{env.https_port}/data.json' 124 r = curl.http_download(urls=[url], with_stats=True, 125 alpn_proto='h3', 126 extra_args=[ 127 '--unix-socket', uds_faker.path, 128 ]) 129 r.check_response(exitcode=96, http_status=None) 130