allow_reuse_address = 1 # Seems to make sense in testing environment

def server_bind(self):

"""Override server_bind to store the server name."""

socketserver.TCPServer.server_bind(self)

host, port = self.server_address[:2]

self.server_name = socket.getfqdn(host)

"""HTTP request handler base class.

# The Python system version, truncated to its first component.

sys_version = "Python/" + sys.version.split()[0]

# The server software version. You may want to override this.

# The format is multiple whitespace-separated strings,

# where each string is of the form name[/version].

server_version = "BaseHTTP/" + __version__

error_message_format = DEFAULT_ERROR_MESSAGE

error_content_type = DEFAULT_ERROR_CONTENT_TYPE

# The default request version. This only affects responses up until

# the point where the request line is parsed, so it mainly decides what

# the client gets back when sending a malformed request line.

# Most web servers default to HTTP 0.9, i.e. don't send a status line.

default_request_version = "HTTP/0.9"

def parse_request(self):

"""Parse a request (internal).

self.command = None # set in case of error on the first line

self.request_version = version = self.default_request_version

self.close_connection = True

requestline = str(self.raw_requestline, 'iso-8859-1')

requestline = requestline.rstrip('\r\n')

self.requestline = requestline

words = requestline.split()

if len(words) == 0:

return False

if len(words) >= 3: # Enough to determine protocol version

version = words[-1]

try:

if not version.startswith('HTTP/'):

raise ValueError

base_version_number = version.split('/', 1)[1]

version_number = base_version_number.split(".")

# RFC 2145 section 3.1 says there can be only one "." and

# - major and minor numbers MUST be treated as

# separate integers;

# - HTTP/2.4 is a lower version than HTTP/2.13, which in

# turn is lower than HTTP/12.3;

# - Leading zeros MUST be ignored by recipients.

if len(version_number) != 2:

raise ValueError

if any(not component.isdigit() for component in version_number):

raise ValueError("non digit in http version")

if any(len(component) > 10 for component in version_number):

raise ValueError("unreasonable length http version")

version_number = int(version_number[0]), int(version_number[1])

except (ValueError, IndexError):

self.send_error(

HTTPStatus.BAD_REQUEST,

"Bad request version (%r)" % version)

return False

if version_number >= (1, 1) and self.protocol_version >= "HTTP/1.1":

self.close_connection = False

if version_number >= (2, 0):

self.send_error(

HTTPStatus.HTTP_VERSION_NOT_SUPPORTED,

"Invalid HTTP version (%s)" % base_version_number)

return False

self.request_version = version

if not 2 <= len(words) <= 3:

self.send_error(

HTTPStatus.BAD_REQUEST,

"Bad request syntax (%r)" % requestline)

return False

command, path = words[:2]

if len(words) == 2:

self.close_connection = True

if command != 'GET':

self.send_error(

HTTPStatus.BAD_REQUEST,

"Bad HTTP/0.9 request type (%r)" % command)

return False

self.command, self.path = command, path

# gh-87389: The purpose of replacing '//' with '/' is to protect

# against open redirect attacks possibly triggered if the path starts

# with '//' because http clients treat //path as an absolute URI

# without scheme (similar to http://path) rather than a path.

if self.path.startswith('//'):

self.path = '/' + self.path.lstrip('/') # Reduce to a single /

# Examine the headers and look for a Connection directive.

try:

self.headers = http.client.parse_headers(self.rfile,

_class=self.MessageClass)

except http.client.LineTooLong as err:

self.send_error(

HTTPStatus.REQUEST_HEADER_FIELDS_TOO_LARGE,

"Line too long",

str(err))

return False

except http.client.HTTPException as err:

self.send_error(

HTTPStatus.REQUEST_HEADER_FIELDS_TOO_LARGE,

"Too many headers",

str(err)

)

return False

conntype = self.headers.get('Connection', "")

if conntype.lower() == 'close':

self.close_connection = True

elif (conntype.lower() == 'keep-alive' and

self.protocol_version >= "HTTP/1.1"):

self.close_connection = False

# Examine the headers and look for an Expect directive

expect = self.headers.get('Expect', "")

if (expect.lower() == "100-continue" and

self.protocol_version >= "HTTP/1.1" and

self.request_version >= "HTTP/1.1"):

if not self.handle_expect_100():

return False

return True

def handle_expect_100(self):

"""Decide what to do with an "Expect: 100-continue" header.

self.send_response_only(HTTPStatus.CONTINUE)

self.end_headers()

return True

def handle_one_request(self):

"""Handle a single HTTP request.

try:

self.raw_requestline = self.rfile.readline(65537)

if len(self.raw_requestline) > 65536:

self.requestline = ''

self.request_version = ''

self.command = ''

self.send_error(HTTPStatus.REQUEST_URI_TOO_LONG)

return

if not self.raw_requestline:

self.close_connection = True

return

if not self.parse_request():

# An error code has been sent, just exit

return

mname = 'do_' + self.command

if not hasattr(self, mname):

self.send_error(

HTTPStatus.NOT_IMPLEMENTED,

"Unsupported method (%r)" % self.command)

return

method = getattr(self, mname)

method()

self.wfile.flush() #actually send the response if not already done.

except TimeoutError as e:

#a read or a write timed out. Discard this connection

self.log_error("Request timed out: %r", e)

self.close_connection = True

return

def handle(self):

"""Handle multiple requests if necessary."""

self.close_connection = True

self.handle_one_request()

while not self.close_connection:

self.handle_one_request()

def send_error(self, code, message=None, explain=None):

"""Send and log an error reply.

try:

shortmsg, longmsg = self.responses[code]

except KeyError:

shortmsg, longmsg = '???', '???'

if message is None:

message = shortmsg

if explain is None:

explain = longmsg

self.log_error("code %d, message %s", code, message)

self.send_response(code, message)

self.send_header('Connection', 'close')

# Message body is omitted for cases described in:

# - RFC7230: 3.3. 1xx, 204(No Content), 304(Not Modified)

# - RFC7231: 6.3.6. 205(Reset Content)

body = None

if (code >= 200 and

code not in (HTTPStatus.NO_CONTENT,

HTTPStatus.RESET_CONTENT,

HTTPStatus.NOT_MODIFIED)):

# HTML encode to prevent Cross Site Scripting attacks

# (see bug #1100201)

content = (self.error_message_format % {

'code': code,

'message': html.escape(message, quote=False),

'explain': html.escape(explain, quote=False)

})

body = content.encode('UTF-8', 'replace')

self.send_header("Content-Type", self.error_content_type)

self.send_header('Content-Length', str(len(body)))

self.end_headers()

if self.command != 'HEAD' and body:

self.wfile.write(body)

def send_response(self, code, message=None):

"""Add the response header to the headers buffer and log the

self.log_request(code)

self.send_response_only(code, message)

self.send_header('Server', self.version_string())

self.send_header('Date', self.date_time_string())

def send_response_only(self, code, message=None):

"""Send the response header only."""

if self.request_version != 'HTTP/0.9':

if message is None:

if code in self.responses:

message = self.responses[code][0]

else:

message = ''

if not hasattr(self, '_headers_buffer'):

self._headers_buffer = []

self._headers_buffer.append(("%s %d %s\r\n" %

(self.protocol_version, code, message)).encode(

'latin-1', 'strict'))

def send_header(self, keyword, value):

"""Send a MIME header to the headers buffer."""

if self.request_version != 'HTTP/0.9':

if not hasattr(self, '_headers_buffer'):

self._headers_buffer = []

self._headers_buffer.append(

("%s: %s\r\n" % (keyword, value)).encode('latin-1', 'strict'))

if keyword.lower() == 'connection':

if value.lower() == 'close':

self.close_connection = True

elif value.lower() == 'keep-alive':

self.close_connection = False

def end_headers(self):

"""Send the blank line ending the MIME headers."""

if self.request_version != 'HTTP/0.9':

self._headers_buffer.append(b"\r\n")

self.flush_headers()

def flush_headers(self):

if hasattr(self, '_headers_buffer'):

self.wfile.write(b"".join(self._headers_buffer))

self._headers_buffer = []

def log_request(self, code='-', size='-'):

"""Log an accepted request.

if isinstance(code, HTTPStatus):

code = code.value

self.log_message('"%s" %s %s',

self.requestline, str(code), str(size))

def log_error(self, format, *args):

"""Log an error.

self.log_message(format, *args)

# https://en.wikipedia.org/wiki/List_of_Unicode_characters#Control_codes

_control_char_table = str.maketrans(

{c: fr'\x{c:02x}' for c in itertools.chain(range(0x20), range(0x7f,0xa0))})

_control_char_table[ord('\\')] = r'\\'

def log_message(self, format, *args):

"""Log an arbitrary message.

message = format % args

sys.stderr.write("%s - - [%s] %s\n" %

(self.address_string(),

self.log_date_time_string(),

message.translate(self._control_char_table)))

def version_string(self):

"""Return the server software version string."""

return self.server_version + ' ' + self.sys_version

def date_time_string(self, timestamp=None):

"""Return the current date and time formatted for a message header."""

if timestamp is None:

timestamp = time.time()

return email.utils.formatdate(timestamp, usegmt=True)

def log_date_time_string(self):

"""Return the current time formatted for logging."""

now = time.time()

year, month, day, hh, mm, ss, x, y, z = time.localtime(now)

s = "%02d/%3s/%04d %02d:%02d:%02d" % (

day, self.monthname[month], year, hh, mm, ss)

return s

weekdayname = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']

monthname = [None,

'Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun',

'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']

def address_string(self):

"""Return the client address."""

return self.client_address[0]

# Essentially static class variables

# The version of the HTTP protocol we support.

# Set this to HTTP/1.1 to enable automatic keepalive

protocol_version = "HTTP/1.0"

# MessageClass used to parse headers

MessageClass = http.client.HTTPMessage

# hack to maintain backwards compatibility

responses = {

v: (v.phrase, v.description)

for v in HTTPStatus.__members__.values()

"""Simple HTTP request handler with GET and HEAD commands.

server_version = "SimpleHTTP/" + __version__

index_pages = ("index.html", "index.htm")

extensions_map = _encodings_map_default = {

'.gz': 'application/gzip',

'.Z': 'application/octet-stream',

'.bz2': 'application/x-bzip2',

'.xz': 'application/x-xz',

}

def __init__(self, *args, directory=None, **kwargs):

if directory is None:

directory = os.getcwd()

self.directory = os.fspath(directory)

super().__init__(*args, **kwargs)

def do_GET(self):

"""Serve a GET request."""

f = self.send_head()

if f:

try:

self.copyfile(f, self.wfile)

finally:

f.close()

def do_HEAD(self):

"""Serve a HEAD request."""

f = self.send_head()

if f:

f.close()

def send_head(self):

"""Common code for GET and HEAD commands.

path = self.translate_path(self.path)

f = None

if os.path.isdir(path):

parts = urllib.parse.urlsplit(self.path)

if not parts.path.endswith('/'):

# redirect browser - doing basically what apache does

self.send_response(HTTPStatus.MOVED_PERMANENTLY)

new_parts = (parts[0], parts[1], parts[2] + '/',

parts[3], parts[4])

new_url = urllib.parse.urlunsplit(new_parts)

self.send_header("Location", new_url)

self.send_header("Content-Length", "0")

self.end_headers()

return None

for index in self.index_pages:

index = os.path.join(path, index)

if os.path.isfile(index):

path = index

break

else:

return self.list_directory(path)

ctype = self.guess_type(path)

# check for trailing "/" which should return 404. See Issue17324

# The test for this was added in test_httpserver.py

# However, some OS platforms accept a trailingSlash as a filename

# See discussion on python-dev and Issue34711 regarding

# parsing and rejection of filenames with a trailing slash

if path.endswith("/"):

self.send_error(HTTPStatus.NOT_FOUND, "File not found")

return None

try:

f = open(path, 'rb')

except OSError:

self.send_error(HTTPStatus.NOT_FOUND, "File not found")

return None

try:

fs = os.fstat(f.fileno())

# Use browser cache if possible

if ("If-Modified-Since" in self.headers

and "If-None-Match" not in self.headers):

# compare If-Modified-Since and time of last file modification

try:

ims = email.utils.parsedate_to_datetime(

self.headers["If-Modified-Since"])

except (TypeError, IndexError, OverflowError, ValueError):

# ignore ill-formed values

pass

else:

if ims.tzinfo is None:

# obsolete format with no timezone, cf.

# https://tools.ietf.org/html/rfc7231#section-7.1.1.1

ims = ims.replace(tzinfo=datetime.timezone.utc)

if ims.tzinfo is datetime.timezone.utc:

# compare to UTC datetime of last modification

last_modif = datetime.datetime.fromtimestamp(

fs.st_mtime, datetime.timezone.utc)

# remove microseconds, like in If-Modified-Since

last_modif = last_modif.replace(microsecond=0)

if last_modif <= ims:

self.send_response(HTTPStatus.NOT_MODIFIED)

self.end_headers()

f.close()

return None

self.send_response(HTTPStatus.OK)

self.send_header("Content-type", ctype)

self.send_header("Content-Length", str(fs[6]))

self.send_header("Last-Modified",

self.date_time_string(fs.st_mtime))

self.end_headers()

return f

except:

f.close()

raise

def list_directory(self, path):

"""Helper to produce a directory listing (absent index.html).

try:

list = os.listdir(path)

except OSError:

self.send_error(

HTTPStatus.NOT_FOUND,

"No permission to list directory")

return None

list.sort(key=lambda a: a.lower())

r = []

try:

displaypath = urllib.parse.unquote(self.path,

errors='surrogatepass')

except UnicodeDecodeError:

displaypath = urllib.parse.unquote(self.path)

displaypath = html.escape(displaypath, quote=False)

enc = sys.getfilesystemencoding()

title = f'Directory listing for {displaypath}'

r.append('<!DOCTYPE HTML>')

r.append('<html lang="en">')

r.append('<head>')

r.append(f'<meta charset="{enc}">')

r.append(f'<title>{title}</title>\n</head>')

r.append(f'<body>\n<h1>{title}</h1>')

r.append('<hr>\n<ul>')

for name in list:

fullname = os.path.join(path, name)

displayname = linkname = name

# Append / for directories or @ for symbolic links

if os.path.isdir(fullname):

displayname = name + "/"

linkname = name + "/"

if os.path.islink(fullname):

displayname = name + "@"

# Note: a link to a directory displays with @ and links with /

r.append('<li><a href="%s">%s</a></li>'

% (urllib.parse.quote(linkname,

errors='surrogatepass'),

html.escape(displayname, quote=False)))

r.append('</ul>\n<hr>\n</body>\n</html>\n')

encoded = '\n'.join(r).encode(enc, 'surrogateescape')

f = io.BytesIO()

f.write(encoded)

f.seek(0)

self.send_response(HTTPStatus.OK)

self.send_header("Content-type", "text/html; charset=%s" % enc)

self.send_header("Content-Length", str(len(encoded)))

self.end_headers()

return f

def translate_path(self, path):

"""Translate a /-separated PATH to the local filename syntax.

# abandon query parameters

path = path.split('?',1)[0]

path = path.split('#',1)[0]

# Don't forget explicit trailing slash when normalizing. Issue17324

trailing_slash = path.rstrip().endswith('/')

try:

path = urllib.parse.unquote(path, errors='surrogatepass')

except UnicodeDecodeError:

path = urllib.parse.unquote(path)

path = posixpath.normpath(path)

words = path.split('/')

words = filter(None, words)

path = self.directory

for word in words:

if os.path.dirname(word) or word in (os.curdir, os.pardir):

# Ignore components that are not a simple file/directory name

continue

path = os.path.join(path, word)

if trailing_slash:

path += '/'

return path

def copyfile(self, source, outputfile):

"""Copy all data between two file objects.

shutil.copyfileobj(source, outputfile)

def guess_type(self, path):

"""Guess the type of a file.

base, ext = posixpath.splitext(path)

if ext in self.extensions_map:

return self.extensions_map[ext]

ext = ext.lower()

if ext in self.extensions_map:

return self.extensions_map[ext]

guess, _ = mimetypes.guess_type(path)

if guess:

return guess

"""

# Query component should not be involved.

path, _, query = path.partition('?')

path = urllib.parse.unquote(path)

# Similar to os.path.split(os.path.normpath(path)) but specific to URL

# path semantics rather than local operating system semantics.

path_parts = path.split('/')

head_parts = []

for part in path_parts[:-1]:

if part == '..':

head_parts.pop() # IndexError if more '..' than prior parts

elif part and part != '.':

head_parts.append( part )

if path_parts:

tail_part = path_parts.pop()

if tail_part:

if tail_part == '..':

head_parts.pop()

tail_part = ''

elif tail_part == '.':

tail_part = ''

else:

tail_part = ''

if query:

tail_part = '?'.join((tail_part, query))

splitpath = ('/' + '/'.join(head_parts), tail_part)

collapsed_path = "/".join(splitpath)

"""Internal routine to get nobody's uid"""

global nobody

if nobody:

return nobody

try:

import pwd

except ImportError:

return -1

try:

nobody = pwd.getpwnam('nobody')[2]

except KeyError:

nobody = 1 + max(x[2] for x in pwd.getpwall())

"""Complete HTTP server with GET, HEAD and POST commands.

# Determine platform specifics

have_fork = hasattr(os, 'fork')

# Make rfile unbuffered -- we need to read one line and then pass

# the rest to a subprocess, so we can't use buffered input.

rbufsize = 0

def do_POST(self):