Remove non-default request handlers from testserver.py
We've migrated enough tests to EmbeddedTestServer that nothing seems to
actually use the request handlers anymore. This dramatically reduces the
amount of code that needs to be ported to Python 3, notably the bytes vs
str mess.
Bug: 492672, 1248530
Change-Id: I26f9a6549d7702f8f61199de71a1cbf4999b45bd
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3156547
Reviewed-by: Matt Menke <mmenke@chromium.org>
Commit-Queue: David Benjamin <davidben@chromium.org>
Cr-Commit-Position: refs/heads/main@{#921640}
NOKEYCHECK=True
GitOrigin-RevId: abdf28e532f2284e4137d3dea315a7f2bd7f9e87
diff --git a/testserver.py b/testserver.py
index 55d43c5..4aff5bd 100755
--- a/testserver.py
+++ b/testserver.py
@@ -16,24 +16,14 @@
import base64
import BaseHTTPServer
-import cgi
-import hashlib
import logging
import os
-import json
-import random
-import re
import select
import socket
import SocketServer
import ssl
-import struct
import sys
-import threading
-import time
-import urllib
import urlparse
-import zlib
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.dirname(BASE_DIR)))
@@ -204,1186 +194,17 @@
class TestPageHandler(testserver_base.BasePageHandler):
- # Class variables to allow for persistence state between page handler
- # invocations
- rst_limits = {}
- fail_precondition = {}
-
def __init__(self, request, client_address, socket_server):
- connect_handlers = [
- self.RedirectConnectHandler,
- self.ServerAuthConnectHandler,
- self.DefaultConnectResponseHandler]
- get_handlers = [
- self.NoCacheMaxAgeTimeHandler,
- self.NoCacheTimeHandler,
- self.CacheTimeHandler,
- self.CacheExpiresHandler,
- self.CacheProxyRevalidateHandler,
- self.CachePrivateHandler,
- self.CachePublicHandler,
- self.CacheSMaxAgeHandler,
- self.CacheMustRevalidateHandler,
- self.CacheMustRevalidateMaxAgeHandler,
- self.CacheNoStoreHandler,
- self.CacheNoStoreMaxAgeHandler,
- self.CacheNoTransformHandler,
- self.DownloadHandler,
- self.DownloadFinishHandler,
- self.EchoHeader,
- self.EchoHeaderCache,
- self.EchoAllHandler,
- self.ZipFileHandler,
- self.FileHandler,
- self.SetCookieHandler,
- self.ExpectAndSetCookieHandler,
- self.SetHeaderHandler,
- self.AuthBasicHandler,
- self.AuthDigestHandler,
- self.SlowServerHandler,
- self.NoContentHandler,
- self.ServerRedirectHandler,
- self.CrossSiteRedirectHandler,
- self.ClientRedirectHandler,
- self.SSLManySmallRecords,
- self.GetChannelID,
- self.GetClientCert,
- self.ClientCipherListHandler,
- self.CloseSocketHandler,
- self.DefaultResponseHandler]
- post_handlers = [
- self.EchoTitleHandler,
- self.EchoHandler,
- self.PostOnlyFileHandler,
- self.EchoMultipartPostHandler] + get_handlers
- put_handlers = [
- self.EchoTitleHandler,
- self.EchoHandler] + get_handlers
- head_handlers = [
- self.FileHandler,
- self.DefaultResponseHandler]
-
- self._mime_types = {
- 'crx' : 'application/x-chrome-extension',
- 'exe' : 'application/octet-stream',
- 'gif': 'image/gif',
- 'jpeg' : 'image/jpeg',
- 'jpg' : 'image/jpeg',
- 'js' : 'application/javascript',
- 'json': 'application/json',
- 'pdf' : 'application/pdf',
- 'txt' : 'text/plain',
- 'wav' : 'audio/wav',
- 'xml' : 'text/xml'
- }
- self._default_mime_type = 'text/html'
-
+ connect_handlers = [self.DefaultConnectResponseHandler]
+ get_handlers = [self.DefaultResponseHandler]
+ post_handlers = get_handlers
+ put_handlers = get_handlers
+ head_handlers = [self.DefaultResponseHandler]
testserver_base.BasePageHandler.__init__(self, request, client_address,
socket_server, connect_handlers,
get_handlers, head_handlers,
post_handlers, put_handlers)
- def GetMIMETypeFromName(self, file_name):
- """Returns the mime type for the specified file_name. So far it only looks
- at the file extension."""
-
- (_shortname, extension) = os.path.splitext(file_name.split("?")[0])
- if len(extension) == 0:
- # no extension.
- return self._default_mime_type
-
- # extension starts with a dot, so we need to remove it
- return self._mime_types.get(extension[1:], self._default_mime_type)
-
- def NoCacheMaxAgeTimeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and no caching requested."""
-
- if not self._ShouldHandleRequest("/nocachetime/maxage"):
- return False
-
- self.send_response(200)
- self.send_header('Cache-Control', 'max-age=0')
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def NoCacheTimeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and no caching requested."""
-
- if not self._ShouldHandleRequest("/nocachetime"):
- return False
-
- self.send_response(200)
- self.send_header('Cache-Control', 'no-cache')
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheTimeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and allows caching for one minute."""
-
- if not self._ShouldHandleRequest("/cachetime"):
- return False
-
- self.send_response(200)
- self.send_header('Cache-Control', 'max-age=60')
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheExpiresHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and set the page to expire on 1 Jan 2099."""
-
- if not self._ShouldHandleRequest("/cache/expires"):
- return False
-
- self.send_response(200)
- self.send_header('Expires', 'Thu, 1 Jan 2099 00:00:00 GMT')
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheProxyRevalidateHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and allows caching for 60 seconds"""
-
- if not self._ShouldHandleRequest("/cache/proxy-revalidate"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=60, proxy-revalidate')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CachePrivateHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and allows caching for 3 seconds."""
-
- if not self._ShouldHandleRequest("/cache/private"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=3, private')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CachePublicHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and allows caching for 3 seconds."""
-
- if not self._ShouldHandleRequest("/cache/public"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=3, public')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheSMaxAgeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow for caching."""
-
- if not self._ShouldHandleRequest("/cache/s-maxage"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'public, s-maxage = 60, max-age = 0')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheMustRevalidateHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow caching."""
-
- if not self._ShouldHandleRequest("/cache/must-revalidate"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'must-revalidate')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheMustRevalidateMaxAgeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow caching event though max-age of 60
- seconds is specified."""
-
- if not self._ShouldHandleRequest("/cache/must-revalidate/max-age"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=60, must-revalidate')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheNoStoreHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow the page to be stored."""
-
- if not self._ShouldHandleRequest("/cache/no-store"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'no-store')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def CacheNoStoreMaxAgeHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow the page to be stored even though max-age
- of 60 seconds is specified."""
-
- if not self._ShouldHandleRequest("/cache/no-store/max-age"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=60, no-store')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
-
- def CacheNoTransformHandler(self):
- """This request handler yields a page with the title set to the current
- system time, and does not allow the content to transformed during
- user-agent caching"""
-
- if not self._ShouldHandleRequest("/cache/no-transform"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'no-transform')
- self.end_headers()
-
- self.wfile.write('<html><head><title>%s</title></head></html>' %
- time.time())
-
- return True
-
- def EchoHeader(self):
- """This handler echoes back the value of a specific request header."""
-
- return self.EchoHeaderHelper("/echoheader")
-
- def EchoHeaderCache(self):
- """This function echoes back the value of a specific request header while
- allowing caching for 10 hours."""
-
- return self.EchoHeaderHelper("/echoheadercache")
-
- def EchoHeaderHelper(self, echo_header):
- """This function echoes back the value of the request header passed in."""
-
- if not self._ShouldHandleRequest(echo_header):
- return False
-
- query_char = self.path.find('?')
- if query_char != -1:
- header_name = self.path[query_char+1:]
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- if echo_header == '/echoheadercache':
- self.send_header('Cache-control', 'max-age=60000')
- else:
- self.send_header('Cache-control', 'no-cache')
- # insert a vary header to properly indicate that the cachability of this
- # request is subject to value of the request header being echoed.
- if len(header_name) > 0:
- self.send_header('Vary', header_name)
- self.end_headers()
-
- if len(header_name) > 0:
- self.wfile.write(self.headers.getheader(header_name))
-
- return True
-
- def ReadRequestBody(self):
- """This function reads the body of the current HTTP request, handling
- both plain and chunked transfer encoded requests."""
-
- if self.headers.getheader('transfer-encoding') != 'chunked':
- length = int(self.headers.getheader('content-length'))
- return self.rfile.read(length)
-
- # Read the request body as chunks.
- body = ""
- while True:
- line = self.rfile.readline()
- length = int(line, 16)
- if length == 0:
- self.rfile.readline()
- break
- body += self.rfile.read(length)
- self.rfile.read(2)
- return body
-
- def EchoHandler(self):
- """This handler just echoes back the payload of the request, for testing
- form submission."""
-
- if not self._ShouldHandleRequest("/echo"):
- return False
-
- _, _, _, _, query, _ = urlparse.urlparse(self.path)
- query_params = cgi.parse_qs(query, True)
- if 'status' in query_params:
- self.send_response(int(query_params['status'][0]))
- else:
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write(self.ReadRequestBody())
- return True
-
- def EchoTitleHandler(self):
- """This handler is like Echo, but sets the page title to the request."""
-
- if not self._ShouldHandleRequest("/echotitle"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- request = self.ReadRequestBody()
- self.wfile.write('<html><head><title>')
- self.wfile.write(request)
- self.wfile.write('</title></head></html>')
- return True
-
- def EchoAllHandler(self):
- """This handler yields a (more) human-readable page listing information
- about the request header & contents."""
-
- if not self._ShouldHandleRequest("/echoall"):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head><style>'
- 'pre { border: 1px solid black; margin: 5px; padding: 5px }'
- '</style></head><body>'
- '<div style="float: right">'
- '<a href="/echo">back to referring page</a></div>'
- '<h1>Request Body:</h1><pre>')
-
- if self.command == 'POST' or self.command == 'PUT':
- qs = self.ReadRequestBody()
- params = cgi.parse_qs(qs, keep_blank_values=1)
-
- for param in params:
- self.wfile.write('%s=%s\n' % (param, params[param][0]))
-
- self.wfile.write('</pre>')
-
- self.wfile.write('<h1>Request Headers:</h1><pre>%s</pre>' % self.headers)
-
- self.wfile.write('</body></html>')
- return True
-
- def EchoMultipartPostHandler(self):
- """This handler echoes received multipart post data as json format."""
-
- if not (self._ShouldHandleRequest("/echomultipartpost") or
- self._ShouldHandleRequest("/searchbyimage")):
- return False
-
- content_type, parameters = cgi.parse_header(
- self.headers.getheader('content-type'))
- if content_type == 'multipart/form-data':
- post_multipart = cgi.parse_multipart(self.rfile, parameters)
- elif content_type == 'application/x-www-form-urlencoded':
- raise Exception('POST by application/x-www-form-urlencoded is '
- 'not implemented.')
- else:
- post_multipart = {}
-
- # Since the data can be binary, we encode them by base64.
- post_multipart_base64_encoded = {}
- for field, values in post_multipart.items():
- post_multipart_base64_encoded[field] = [base64.b64encode(value)
- for value in values]
-
- result = {'POST_multipart' : post_multipart_base64_encoded}
-
- self.send_response(200)
- self.send_header("Content-type", "text/plain")
- self.end_headers()
- self.wfile.write(json.dumps(result, indent=2, sort_keys=False))
- return True
-
- def DownloadHandler(self):
- """This handler sends a downloadable file with or without reporting
- the size (6K)."""
-
- if self.path.startswith("/download-unknown-size"):
- send_length = False
- elif self.path.startswith("/download-known-size"):
- send_length = True
- else:
- return False
-
- #
- # The test which uses this functionality is attempting to send
- # small chunks of data to the client. Use a fairly large buffer
- # so that we'll fill chrome's IO buffer enough to force it to
- # actually write the data.
- # See also the comments in the client-side of this test in
- # download_uitest.cc
- #
- size_chunk1 = 35*1024
- size_chunk2 = 10*1024
-
- self.send_response(200)
- self.send_header('Content-Type', 'application/octet-stream')
- self.send_header('Cache-Control', 'max-age=0')
- if send_length:
- self.send_header('Content-Length', size_chunk1 + size_chunk2)
- self.end_headers()
-
- # First chunk of data:
- self.wfile.write("*" * size_chunk1)
- self.wfile.flush()
-
- # handle requests until one of them clears this flag.
- self.server.wait_for_download = True
- while self.server.wait_for_download:
- self.server.handle_request()
-
- # Second chunk of data:
- self.wfile.write("*" * size_chunk2)
- return True
-
- def DownloadFinishHandler(self):
- """This handler just tells the server to finish the current download."""
-
- if not self._ShouldHandleRequest("/download-finish"):
- return False
-
- self.server.wait_for_download = False
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-Control', 'max-age=0')
- self.end_headers()
- return True
-
- def _ReplaceFileData(self, data, query_parameters):
- """Replaces matching substrings in a file.
-
- If the 'replace_text' URL query parameter is present, it is expected to be
- of the form old_text:new_text, which indicates that any old_text strings in
- the file are replaced with new_text. Multiple 'replace_text' parameters may
- be specified.
-
- If the parameters are not present, |data| is returned.
- """
-
- query_dict = cgi.parse_qs(query_parameters)
- replace_text_values = query_dict.get('replace_text', [])
- for replace_text_value in replace_text_values:
- replace_text_args = replace_text_value.split(':')
- if len(replace_text_args) != 2:
- raise ValueError(
- 'replace_text must be of form old_text:new_text. Actual value: %s' %
- replace_text_value)
- old_text_b64, new_text_b64 = replace_text_args
- old_text = base64.urlsafe_b64decode(old_text_b64)
- new_text = base64.urlsafe_b64decode(new_text_b64)
- data = data.replace(old_text, new_text)
- return data
-
- def ZipFileHandler(self):
- """This handler sends the contents of the requested file in compressed form.
- Can pass in a parameter that specifies that the content length be
- C - the compressed size (OK),
- U - the uncompressed size (Non-standard, but handled),
- S - less than compressed (OK because we keep going),
- M - larger than compressed but less than uncompressed (an error),
- L - larger than uncompressed (an error)
- Example: compressedfiles/Picture_1.doc?C
- """
-
- prefix = "/compressedfiles/"
- if not self.path.startswith(prefix):
- return False
-
- # Consume a request body if present.
- if self.command == 'POST' or self.command == 'PUT' :
- self.ReadRequestBody()
-
- _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
-
- if not query in ('C', 'U', 'S', 'M', 'L'):
- return False
-
- sub_path = url_path[len(prefix):]
- entries = sub_path.split('/')
- file_path = os.path.join(self.server.data_dir, *entries)
- if os.path.isdir(file_path):
- file_path = os.path.join(file_path, 'index.html')
-
- if not os.path.isfile(file_path):
- print "File not found " + sub_path + " full path:" + file_path
- self.send_error(404)
- return True
-
- f = open(file_path, "rb")
- data = f.read()
- uncompressed_len = len(data)
- f.close()
-
- # Compress the data.
- data = zlib.compress(data)
- compressed_len = len(data)
-
- content_length = compressed_len
- if query == 'U':
- content_length = uncompressed_len
- elif query == 'S':
- content_length = compressed_len / 2
- elif query == 'M':
- content_length = (compressed_len + uncompressed_len) / 2
- elif query == 'L':
- content_length = compressed_len + uncompressed_len
-
- self.send_response(200)
- self.send_header('Content-Type', 'application/msword')
- self.send_header('Content-encoding', 'deflate')
- self.send_header('Connection', 'close')
- self.send_header('Content-Length', content_length)
- self.send_header('ETag', '\'' + file_path + '\'')
- self.end_headers()
-
- self.wfile.write(data)
-
- return True
-
- def FileHandler(self):
- """This handler sends the contents of the requested file. Wow, it's like
- a real webserver!"""
-
- prefix = self.server.file_root_url
- if not self.path.startswith(prefix):
- return False
- return self._FileHandlerHelper(prefix)
-
- def PostOnlyFileHandler(self):
- """This handler sends the contents of the requested file on a POST."""
-
- prefix = urlparse.urljoin(self.server.file_root_url, 'post/')
- if not self.path.startswith(prefix):
- return False
- return self._FileHandlerHelper(prefix)
-
- def _FileHandlerHelper(self, prefix):
- request_body = ''
- if self.command == 'POST' or self.command == 'PUT':
- # Consume a request body if present.
- request_body = self.ReadRequestBody()
-
- _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
- query_dict = cgi.parse_qs(query)
-
- expected_body = query_dict.get('expected_body', [])
- if expected_body and request_body not in expected_body:
- self.send_response(404)
- self.end_headers()
- self.wfile.write('')
- return True
-
- expected_headers = query_dict.get('expected_headers', [])
- for expected_header in expected_headers:
- header_name, expected_value = expected_header.split(':')
- if self.headers.getheader(header_name) != expected_value:
- self.send_response(404)
- self.end_headers()
- self.wfile.write('')
- return True
-
- sub_path = url_path[len(prefix):]
- entries = sub_path.split('/')
- file_path = os.path.join(self.server.data_dir, *entries)
- if os.path.isdir(file_path):
- file_path = os.path.join(file_path, 'index.html')
-
- if not os.path.isfile(file_path):
- print "File not found " + sub_path + " full path:" + file_path
- self.send_error(404)
- return True
-
- f = open(file_path, "rb")
- data = f.read()
- f.close()
-
- data = self._ReplaceFileData(data, query)
-
- old_protocol_version = self.protocol_version
-
- # If file.mock-http-headers exists, it contains the headers we
- # should send. Read them in and parse them.
- headers_path = file_path + '.mock-http-headers'
- if os.path.isfile(headers_path):
- f = open(headers_path, "r")
-
- # "HTTP/1.1 200 OK"
- response = f.readline()
- http_major, http_minor, status_code = re.findall(
- 'HTTP/(\d+).(\d+) (\d+)', response)[0]
- self.protocol_version = "HTTP/%s.%s" % (http_major, http_minor)
- self.send_response(int(status_code))
-
- for line in f:
- header_values = re.findall('(\S+):\s*(.*)', line)
- if len(header_values) > 0:
- # "name: value"
- name, value = header_values[0]
- self.send_header(name, value)
- f.close()
- else:
- # Could be more generic once we support mime-type sniffing, but for
- # now we need to set it explicitly.
-
- range_header = self.headers.get('Range')
- if range_header and range_header.startswith('bytes='):
- # Note this doesn't handle all valid byte range_header values (i.e.
- # left open ended ones), just enough for what we needed so far.
- range_header = range_header[6:].split('-')
- start = int(range_header[0])
- if range_header[1]:
- end = int(range_header[1])
- else:
- end = len(data) - 1
-
- self.send_response(206)
- content_range = ('bytes ' + str(start) + '-' + str(end) + '/' +
- str(len(data)))
- self.send_header('Content-Range', content_range)
- data = data[start: end + 1]
- else:
- self.send_response(200)
-
- self.send_header('Content-Type', self.GetMIMETypeFromName(file_path))
- self.send_header('Accept-Ranges', 'bytes')
- self.send_header('Content-Length', len(data))
- self.send_header('ETag', '\'' + file_path + '\'')
- self.end_headers()
-
- if (self.command != 'HEAD'):
- self.wfile.write(data)
-
- self.protocol_version = old_protocol_version
- return True
-
- def SetCookieHandler(self):
- """This handler just sets a cookie, for testing cookie handling."""
-
- if not self._ShouldHandleRequest("/set-cookie"):
- return False
-
- query_char = self.path.find('?')
- if query_char != -1:
- cookie_values = self.path[query_char + 1:].split('&')
- else:
- cookie_values = ("",)
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- for cookie_value in cookie_values:
- self.send_header('Set-Cookie', '%s' % cookie_value)
- self.end_headers()
- for cookie_value in cookie_values:
- self.wfile.write('%s' % cookie_value)
- return True
-
- def ExpectAndSetCookieHandler(self):
- """Expects some cookies to be sent, and if they are, sets more cookies.
-
- The expect parameter specifies a required cookie. May be specified multiple
- times.
- The set parameter specifies a cookie to set if all required cookies are
- preset. May be specified multiple times.
- The data parameter specifies the response body data to be returned."""
-
- if not self._ShouldHandleRequest("/expect-and-set-cookie"):
- return False
-
- _, _, _, _, query, _ = urlparse.urlparse(self.path)
- query_dict = cgi.parse_qs(query)
- cookies = set()
- if 'Cookie' in self.headers:
- cookie_header = self.headers.getheader('Cookie')
- cookies.update([s.strip() for s in cookie_header.split(';')])
- got_all_expected_cookies = True
- for expected_cookie in query_dict.get('expect', []):
- if expected_cookie not in cookies:
- got_all_expected_cookies = False
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- if got_all_expected_cookies:
- for cookie_value in query_dict.get('set', []):
- self.send_header('Set-Cookie', '%s' % cookie_value)
- self.end_headers()
- for data_value in query_dict.get('data', []):
- self.wfile.write(data_value)
- return True
-
- def SetHeaderHandler(self):
- """This handler sets a response header. Parameters are in the
- key%3A%20value&key2%3A%20value2 format."""
-
- if not self._ShouldHandleRequest("/set-header"):
- return False
-
- query_char = self.path.find('?')
- if query_char != -1:
- headers_values = self.path[query_char + 1:].split('&')
- else:
- headers_values = ("",)
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- for header_value in headers_values:
- header_value = urllib.unquote(header_value)
- (key, value) = header_value.split(': ', 1)
- self.send_header(key, value)
- self.end_headers()
- for header_value in headers_values:
- self.wfile.write('%s' % header_value)
- return True
-
- def AuthBasicHandler(self):
- """This handler tests 'Basic' authentication. It just sends a page with
- title 'user/pass' if you succeed."""
-
- if not self._ShouldHandleRequest("/auth-basic"):
- return False
-
- username = userpass = password = b64str = ""
- expected_password = 'secret'
- realm = 'testrealm'
- set_cookie_if_challenged = False
-
- _, _, url_path, _, query, _ = urlparse.urlparse(self.path)
- query_params = cgi.parse_qs(query, True)
- if 'set-cookie-if-challenged' in query_params:
- set_cookie_if_challenged = True
- if 'password' in query_params:
- expected_password = query_params['password'][0]
- if 'realm' in query_params:
- realm = query_params['realm'][0]
-
- auth = self.headers.getheader('authorization')
- try:
- if not auth:
- raise Exception('no auth')
- b64str = re.findall(r'Basic (\S+)', auth)[0]
- userpass = base64.b64decode(b64str)
- username, password = re.findall(r'([^:]+):(\S+)', userpass)[0]
- if password != expected_password:
- raise Exception('wrong password')
- except Exception, e:
- # Authentication failed.
- self.send_response(401)
- self.send_header('WWW-Authenticate', 'Basic realm="%s"' % realm)
- self.send_header('Content-Type', 'text/html')
- if set_cookie_if_challenged:
- self.send_header('Set-Cookie', 'got_challenged=true')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('<title>Denied: %s</title>' % e)
- self.wfile.write('</head><body>')
- self.wfile.write('auth=%s<p>' % auth)
- self.wfile.write('b64str=%s<p>' % b64str)
- self.wfile.write('username: %s<p>' % username)
- self.wfile.write('userpass: %s<p>' % userpass)
- self.wfile.write('password: %s<p>' % password)
- self.wfile.write('You sent:<br>%s<p>' % self.headers)
- self.wfile.write('</body></html>')
- return True
-
- # Authentication successful. (Return a cachable response to allow for
- # testing cached pages that require authentication.)
- old_protocol_version = self.protocol_version
- self.protocol_version = "HTTP/1.1"
-
- if_none_match = self.headers.getheader('if-none-match')
- if if_none_match == "abc":
- self.send_response(304)
- self.end_headers()
- elif url_path.endswith(".gif"):
- # Using chrome/test/data/google/logo.gif as the test image
- test_image_path = ['google', 'logo.gif']
- gif_path = os.path.join(self.server.data_dir, *test_image_path)
- if not os.path.isfile(gif_path):
- self.send_error(404)
- self.protocol_version = old_protocol_version
- return True
-
- f = open(gif_path, "rb")
- data = f.read()
- f.close()
-
- self.send_response(200)
- self.send_header('Content-Type', 'image/gif')
- self.send_header('Cache-control', 'max-age=60000')
- self.send_header('Etag', 'abc')
- self.end_headers()
- self.wfile.write(data)
- else:
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.send_header('Cache-control', 'max-age=60000')
- self.send_header('Etag', 'abc')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('<title>%s/%s</title>' % (username, password))
- self.wfile.write('</head><body>')
- self.wfile.write('auth=%s<p>' % auth)
- self.wfile.write('You sent:<br>%s<p>' % self.headers)
- self.wfile.write('</body></html>')
-
- self.protocol_version = old_protocol_version
- return True
-
- def GetNonce(self, force_reset=False):
- """Returns a nonce that's stable per request path for the server's lifetime.
- This is a fake implementation. A real implementation would only use a given
- nonce a single time (hence the name n-once). However, for the purposes of
- unittesting, we don't care about the security of the nonce.
-
- Args:
- force_reset: Iff set, the nonce will be changed. Useful for testing the
- "stale" response.
- """
-
- if force_reset or not self.server.nonce_time:
- self.server.nonce_time = time.time()
- return hashlib.md5('privatekey%s%d' %
- (self.path, self.server.nonce_time)).hexdigest()
-
- def AuthDigestHandler(self):
- """This handler tests 'Digest' authentication.
-
- It just sends a page with title 'user/pass' if you succeed.
-
- A stale response is sent iff "stale" is present in the request path.
- """
-
- if not self._ShouldHandleRequest("/auth-digest"):
- return False
-
- stale = 'stale' in self.path
- nonce = self.GetNonce(force_reset=stale)
- opaque = hashlib.md5('opaque').hexdigest()
- password = 'secret'
- realm = 'testrealm'
-
- auth = self.headers.getheader('authorization')
- pairs = {}
- try:
- if not auth:
- raise Exception('no auth')
- if not auth.startswith('Digest'):
- raise Exception('not digest')
- # Pull out all the name="value" pairs as a dictionary.
- pairs = dict(re.findall(r'(\b[^ ,=]+)="?([^",]+)"?', auth))
-
- # Make sure it's all valid.
- if pairs['nonce'] != nonce:
- raise Exception('wrong nonce')
- if pairs['opaque'] != opaque:
- raise Exception('wrong opaque')
-
- # Check the 'response' value and make sure it matches our magic hash.
- # See http://www.ietf.org/rfc/rfc2617.txt
- hash_a1 = hashlib.md5(
- ':'.join([pairs['username'], realm, password])).hexdigest()
- hash_a2 = hashlib.md5(':'.join([self.command, pairs['uri']])).hexdigest()
- if 'qop' in pairs and 'nc' in pairs and 'cnonce' in pairs:
- response = hashlib.md5(':'.join([hash_a1, nonce, pairs['nc'],
- pairs['cnonce'], pairs['qop'], hash_a2])).hexdigest()
- else:
- response = hashlib.md5(':'.join([hash_a1, nonce, hash_a2])).hexdigest()
-
- if pairs['response'] != response:
- raise Exception('wrong password')
- except Exception, e:
- # Authentication failed.
- self.send_response(401)
- hdr = ('Digest '
- 'realm="%s", '
- 'domain="/", '
- 'qop="auth", '
- 'algorithm=MD5, '
- 'nonce="%s", '
- 'opaque="%s"') % (realm, nonce, opaque)
- if stale:
- hdr += ', stale="TRUE"'
- self.send_header('WWW-Authenticate', hdr)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('<title>Denied: %s</title>' % e)
- self.wfile.write('</head><body>')
- self.wfile.write('auth=%s<p>' % auth)
- self.wfile.write('pairs=%s<p>' % pairs)
- self.wfile.write('You sent:<br>%s<p>' % self.headers)
- self.wfile.write('We are replying:<br>%s<p>' % hdr)
- self.wfile.write('</body></html>')
- return True
-
- # Authentication successful.
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('<title>%s/%s</title>' % (pairs['username'], password))
- self.wfile.write('</head><body>')
- self.wfile.write('auth=%s<p>' % auth)
- self.wfile.write('pairs=%s<p>' % pairs)
- self.wfile.write('</body></html>')
-
- return True
-
- def SlowServerHandler(self):
- """Wait for the user suggested time before responding. The syntax is
- /slow?0.5 to wait for half a second."""
-
- if not self._ShouldHandleRequest("/slow"):
- return False
- query_char = self.path.find('?')
- wait_sec = 1.0
- if query_char >= 0:
- try:
- wait_sec = float(self.path[query_char + 1:])
- except ValueError:
- pass
- time.sleep(wait_sec)
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- self.end_headers()
- self.wfile.write("waited %.1f seconds" % wait_sec)
- return True
-
- def NoContentHandler(self):
- """Returns a 204 No Content response."""
-
- if not self._ShouldHandleRequest("/nocontent"):
- return False
- self.send_response(204)
- self.end_headers()
- return True
-
- def ServerRedirectHandler(self):
- """Sends a server redirect to the given URL. The syntax is
- '/server-redirect?http://foo.bar/asdf' to redirect to
- 'http://foo.bar/asdf'"""
-
- test_name = "/server-redirect"
- if not self._ShouldHandleRequest(test_name):
- return False
-
- query_char = self.path.find('?')
- if query_char < 0 or len(self.path) <= query_char + 1:
- self.sendRedirectHelp(test_name)
- return True
- dest = urllib.unquote(self.path[query_char + 1:])
-
- self.send_response(301) # moved permanently
- self.send_header('Location', dest)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('</head><body>Redirecting to %s</body></html>' % dest)
-
- return True
-
- def CrossSiteRedirectHandler(self):
- """Sends a server redirect to the given site. The syntax is
- '/cross-site/hostname/...' to redirect to //hostname/...
- It is used to navigate between different Sites, causing
- cross-site/cross-process navigations in the browser."""
-
- test_name = "/cross-site"
- if not self._ShouldHandleRequest(test_name):
- return False
-
- params = urllib.unquote(self.path[(len(test_name) + 1):])
- slash = params.find('/')
- if slash < 0:
- self.sendRedirectHelp(test_name)
- return True
-
- host = params[:slash]
- path = params[(slash+1):]
- dest = "//%s:%s/%s" % (host, str(self.server.server_port), path)
-
- self.send_response(301) # moved permanently
- self.send_header('Location', dest)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('</head><body>Redirecting to %s</body></html>' % dest)
-
- return True
-
- def ClientRedirectHandler(self):
- """Sends a client redirect to the given URL. The syntax is
- '/client-redirect?http://foo.bar/asdf' to redirect to
- 'http://foo.bar/asdf'"""
-
- test_name = "/client-redirect"
- if not self._ShouldHandleRequest(test_name):
- return False
-
- query_char = self.path.find('?')
- if query_char < 0 or len(self.path) <= query_char + 1:
- self.sendRedirectHelp(test_name)
- return True
- dest = urllib.unquote(self.path[query_char + 1:])
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><head>')
- self.wfile.write('<meta http-equiv="refresh" content="0;url=%s">' % dest)
- self.wfile.write('</head><body>Redirecting to %s</body></html>' % dest)
-
- return True
-
- def SSLManySmallRecords(self):
- """Sends a reply consisting of a variety of small writes. These will be
- translated into a series of small SSL records when used over an HTTPS
- server."""
-
- if not self._ShouldHandleRequest('/ssl-many-small-records'):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- self.end_headers()
-
- # Write ~26K of data, in 1350 byte chunks
- for i in xrange(20):
- self.wfile.write('*' * 1350)
- self.wfile.flush()
- return True
-
- def GetChannelID(self):
- """Send a reply containing the hashed ChannelID that the client provided."""
-
- if not self._ShouldHandleRequest('/channel-id'):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- self.end_headers()
- channel_id = bytes(self.server.tlsConnection.channel_id)
- self.wfile.write(hashlib.sha256(channel_id).digest().encode('base64'))
- return True
-
- def GetClientCert(self):
- """Send a reply whether a client certificate was provided."""
-
- if not self._ShouldHandleRequest('/client-cert'):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- self.end_headers()
-
- cert_chain = self.server.tlsConnection.session.clientCertChain
- if cert_chain != None:
- self.wfile.write('got client cert with fingerprint: ' +
- cert_chain.getFingerprint())
- else:
- self.wfile.write('got no client cert')
- return True
-
- def ClientCipherListHandler(self):
- """Send a reply containing the cipher suite list that the client
- provided. Each cipher suite value is serialized in decimal, followed by a
- newline."""
-
- if not self._ShouldHandleRequest('/client-cipher-list'):
- return False
-
- self.send_response(200)
- self.send_header('Content-Type', 'text/plain')
- self.end_headers()
-
- cipher_suites = self.server.tlsConnection.clientHello.cipher_suites
- self.wfile.write('\n'.join(str(c) for c in cipher_suites))
- return True
-
- def CloseSocketHandler(self):
- """Closes the socket without sending anything."""
-
- if not self._ShouldHandleRequest('/close-socket'):
- return False
-
- self.wfile.close()
- return True
-
def DefaultResponseHandler(self):
"""This is the catch-all response handler for requests that aren't handled
by one of the special handlers above.
@@ -1399,38 +220,6 @@
self.wfile.write(contents)
return True
- def RedirectConnectHandler(self):
- """Sends a redirect to the CONNECT request for www.redirect.com. This
- response is not specified by the RFC, so the browser should not follow
- the redirect."""
-
- if (self.path.find("www.redirect.com") < 0):
- return False
-
- dest = "http://www.destination.com/foo.js"
-
- self.send_response(302) # moved temporarily
- self.send_header('Location', dest)
- self.send_header('Connection', 'close')
- self.end_headers()
- return True
-
- def ServerAuthConnectHandler(self):
- """Sends a 401 to the CONNECT request for www.server-auth.com. This
- response doesn't make sense because the proxy server cannot request
- server authentication."""
-
- if (self.path.find("www.server-auth.com") < 0):
- return False
-
- challenge = 'Basic realm="WallyWorld"'
-
- self.send_response(401) # unauthorized
- self.send_header('WWW-Authenticate', challenge)
- self.send_header('Connection', 'close')
- self.end_headers()
- return True
-
def DefaultConnectResponseHandler(self):
"""This is the catch-all response handler for CONNECT requests that aren't
handled by one of the special handlers above. Real Web servers respond
@@ -1444,22 +233,6 @@
self.wfile.write(contents)
return True
- # called by the redirect handling function when there is no parameter
- def sendRedirectHelp(self, redirect_name):
- self.send_response(200)
- self.send_header('Content-Type', 'text/html')
- self.end_headers()
- self.wfile.write('<html><body><h1>Error: no redirect destination</h1>')
- self.wfile.write('Use <pre>%s?http://dest...</pre>' % redirect_name)
- self.wfile.write('</body></html>')
-
- # called by chunked handling function
- def sendChunkHelp(self, chunk):
- # Each chunk consists of: chunk size (hex), CRLF, chunk body, CRLF
- self.wfile.write('%X\r\n' % len(chunk))
- self.wfile.write(chunk)
- self.wfile.write('\r\n')
-
class ProxyRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
"""A request handler that behaves as a proxy server. Only CONNECT, GET and
@@ -1762,14 +535,6 @@
'path to the file containing the certificate '
'and private key for the server in PEM '
'format')
- self.option_parser.add_option('--cert-serial', dest='cert_serial',
- default=0, type=int,
- help='If non-zero then the generated '
- 'certificate will have this serial number')
- self.option_parser.add_option('--cert-common-name', dest='cert_common_name',
- default="127.0.0.1",
- help='The generated certificate will have '
- 'this common name')
self.option_parser.add_option('--tls-intolerant', dest='tls_intolerant',
default='0', type='int',
help='If nonzero, certain TLS connections '
