source: mod_gnutls/test/https-test-client.py @ 618ee14

asyncioproxy-ticket
Last change on this file since 618ee14 was 618ee14, checked in by Fiona Klute <fiona.klute@…>, 15 months ago

Experimental Python-based HTTPS test client

The script uses gnutls-cli to provide a TLS transport for the Python
http.client library. Not used in any test yet, but it might replace
the static file based HTTP request/response test approach later.

  • Property mode set to 100755
File size: 4.4 KB
Line 
1#!/usr/bin/python3
2# PYTHON_ARGCOMPLETE_OK
3
4# Copyright 2019 Fiona Klute
5#
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17
18import socket
19import subprocess
20
21from http.client import HTTPConnection
22from time import sleep
23
24class HTTPSubprocessConnection(HTTPConnection):
25    def __init__(self, command, host, port=None,
26                 timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
27                 blocksize=8192):
28        super(HTTPSubprocessConnection, self).__init__(host, port, timeout,
29                                                       source_address=None,
30                                                       blocksize=blocksize)
31        # "command" must be a list containing binary and command line
32        # parameters
33        self.command = command
34        # This will be the subprocess reference when connected
35        self._sproc = None
36        # The subprocess return code is stored here on close()
37        self.returncode = None
38        # The set_tunnel method of the super class is not supported
39        # (see exception doc)
40        self.set_tunnel = None
41
42    def connect(self):
43        s_local, s_remote = socket.socketpair(socket.AF_UNIX,
44                                              socket.SOCK_STREAM)
45        s_local.settimeout(self.timeout)
46
47        # TODO: Maybe capture stderr?
48        self._sproc = subprocess.Popen(self.command, stdout=s_remote,
49                                       stdin=s_remote, close_fds=True)
50        s_remote.close()
51        self.sock = s_local
52
53    def close(self):
54        super().close()
55        # Wait for the process to stop, send SIGTERM/SIGKILL if
56        # necessary
57        self.returncode = self._sproc.wait(self.timeout)
58        if self.returncode == None:
59            self._sproc.terminate()
60            self.returncode = self._sproc.wait(self.timeout)
61            if self.returncode == None:
62                self._sproc.kill()
63                self.returncode = self._sproc.wait(self.timeout)
64
65
66
67def format_response(resp):
68    print('{} {}'.format(resp.status, resp.reason))
69    for name, value in resp.getheaders():
70        print('{}: {}'.format(name, value))
71    print()
72    print(resp.read().decode())
73
74
75
76if __name__ == "__main__":
77    import argparse
78    parser = argparse.ArgumentParser(
79        description='Send HTTP requests through gnutls-cli',
80        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
81    parser.add_argument('host', nargs='?', help='Access the specified host',
82                        default='localhost')
83    parser.add_argument('--insecure', action='store_true',
84                        help='do not validate the server certificate')
85    parser.add_argument('-p', '--port', type=int,
86                        help='Access the specified port', default='8000')
87    parser.add_argument('--x509cafile', type=str,
88                        help='Use the specified CA to validate the '
89                        'server certificate')
90
91    # enable bash completion if argcomplete is available
92    try:
93        import argcomplete
94        argcomplete.autocomplete(parser)
95    except ImportError:
96        pass
97
98    args = parser.parse_args()
99
100    # note: "--logfile" option requires GnuTLS version >= 3.6.7
101    command = ['gnutls-cli', '--logfile=/dev/stderr']
102    if args.insecure:
103        command.append('--insecure')
104    if args.x509cafile:
105        command.append('--x509cafile')
106        command.append(args.x509cafile)
107    command = command + ['-p', str(args.port), args.host]
108
109    conn = HTTPSubprocessConnection(command, args.host, port=args.port,
110                                    timeout=6.0)
111    # Maybe call connect() here to detect handshake errors before
112    # sending the request?
113
114    # Add headers={'Host': 'test.host'} to provoke "421 Misdirected
115    # Request"
116    conn.request('GET', '/')
117    resp = conn.getresponse()
118    format_response(resp)
119
120    # This could be used to test keepalive behavior
121    #sleep(2)
122
123    conn.request('GET', '/test.txt')
124    resp = conn.getresponse()
125    format_response(resp)
126
127    conn.close()
128    exit(conn.returncode)
Note: See TracBrowser for help on using the repository browser.