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 85 86class TestUnix: 87 88 @pytest.fixture(scope="class") 89 def uds_faker(self, env: Env) -> UDSFaker: 90 uds_path = os.path.join(env.gen_dir, 'uds_11.sock') 91 faker = UDSFaker(path=uds_path) 92 faker.start() 93 yield faker 94 faker.stop() 95 96 # download http: via unix socket 97 def test_11_01_unix_connect_http(self, env: Env, httpd, uds_faker, repeat): 98 curl = CurlClient(env=env) 99 url = f'http://{env.domain1}:{env.http_port}/data.json' 100 r = curl.http_download(urls=[url], with_stats=True, 101 extra_args=[ 102 '--unix-socket', uds_faker.path, 103 ]) 104 r.check_response(count=1, http_status=200) 105 106 # download https: via unix socket 107 @pytest.mark.skipif(condition=not Env.have_ssl_curl(), reason=f"curl without SSL") 108 def test_11_02_unix_connect_http(self, env: Env, httpd, uds_faker, repeat): 109 curl = CurlClient(env=env) 110 url = f'https://{env.domain1}:{env.https_port}/data.json' 111 r = curl.http_download(urls=[url], with_stats=True, 112 extra_args=[ 113 '--unix-socket', uds_faker.path, 114 ]) 115 r.check_response(exitcode=35, http_status=None) 116 117 # download HTTP/3 via unix socket 118 @pytest.mark.skipif(condition=not Env.have_h3(), reason='h3 not supported') 119 def test_11_03_unix_connect_quic(self, env: Env, httpd, uds_faker, repeat): 120 curl = CurlClient(env=env) 121 url = f'https://{env.domain1}:{env.https_port}/data.json' 122 r = curl.http_download(urls=[url], with_stats=True, 123 alpn_proto='h3', 124 extra_args=[ 125 '--unix-socket', uds_faker.path, 126 ]) 127 r.check_response(exitcode=96, http_status=None) 128