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