testserver.py - Android社区 - https://www.androidos.net.cn/

#!/usr/bin/env python
# Copyright 2013 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""This is a simple HTTP/FTP/TCP/UDP/BASIC_AUTH_PROXY/WEBSOCKET server used for
testing Chrome.

It supports several test URLs, as specified by the handlers in TestPageHandler.
By default, it listens on an ephemeral port and sends the port number back to
the originating process over a pipe. The originating process can specify an
explicit port if necessary.
It can use https if you specify the flag --https=CERT where CERT is the path
to a pem file containing the certificate and private key that should be used.
"""

import base64
import BaseHTTPServer
import cgi
import hashlib
import logging
import minica
import os
import json
import random
import re
import select
import socket
import SocketServer
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)))

import echo_message
import testserver_base

# Append at the end of sys.path, it's fine to use the system library.
sys.path.append(os.path.join(ROOT_DIR, 'third_party', 'pyftpdlib', 'src'))
sys.path.append(os.path.join(ROOT_DIR, 'third_party', 'tlslite'))
import pyftpdlib.ftpserver
import tlslite
import tlslite.api

# Insert at the beginning of the path, we want this to be used
# unconditionally.
sys.path.insert(0, os.path.join(ROOT_DIR, 'third_party', 'pywebsocket', 'src'))
from mod_pywebsocket.standalone import WebSocketServer

SERVER_HTTP = 0
SERVER_FTP = 1
SERVER_TCP_ECHO = 2
SERVER_UDP_ECHO = 3
SERVER_BASIC_AUTH_PROXY = 4
SERVER_WEBSOCKET = 5

# Default request queue size for WebSocketServer.
_DEFAULT_REQUEST_QUEUE_SIZE = 128

class WebSocketOptions:
  """Holds options for WebSocketServer."""

def __init__(self, host, port, data_dir):
    self.request_queue_size = _DEFAULT_REQUEST_QUEUE_SIZE
    self.server_host = host
    self.port = port
    self.websock_handlers = data_dir
    self.scan_dir = None
    self.allow_handlers_outside_root_dir = False
    self.websock_handlers_map_file = None
    self.cgi_directories = []
    self.is_executable_method = None
    self.allow_draft75 = False
    self.strict = True

self.use_tls = False
    self.private_key = None
    self.certificate = None
    self.tls_client_auth = False
    self.tls_client_ca = None
    self.use_basic_auth = False

class RecordingSSLSessionCache(object):
  """RecordingSSLSessionCache acts as a TLS session cache and maintains a log of
  lookups and inserts in order to test session cache behaviours."""

def __init__(self):
    self.log = []

def __getitem__(self, sessionID):
    self.log.append(('lookup', sessionID))
    raise KeyError()

def __setitem__(self, sessionID, session):
    self.log.append(('insert', sessionID))

class HTTPServer(testserver_base.ClientRestrictingServerMixIn,
                 testserver_base.BrokenPipeHandlerMixIn,
                 testserver_base.StoppableHTTPServer):
  """This is a specialization of StoppableHTTPServer that adds client
  verification."""

pass

class OCSPServer(testserver_base.ClientRestrictingServerMixIn,
                 testserver_base.BrokenPipeHandlerMixIn,
                 BaseHTTPServer.HTTPServer):
  """This is a specialization of HTTPServer that serves an
  OCSP response"""

def serve_forever_on_thread(self):
    self.thread = threading.Thread(target = self.serve_forever,
                                   name = "OCSPServerThread")
    self.thread.start()

def stop_serving(self):
    self.shutdown()
    self.thread.join()

class HTTPSServer(tlslite.api.TLSSocketServerMixIn,
                  testserver_base.ClientRestrictingServerMixIn,
                  testserver_base.BrokenPipeHandlerMixIn,
                  testserver_base.StoppableHTTPServer):
  """This is a specialization of StoppableHTTPServer that add https support and
  client verification."""

def __init__(self, server_address, request_hander_class, pem_cert_and_key,
               ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers,
               record_resume_info, tls_intolerant, signed_cert_timestamps,
               fallback_scsv_enabled, ocsp_response):
    self.cert_chain = tlslite.api.X509CertChain().parseChain(pem_cert_and_key)
    # Force using only python implementation - otherwise behavior is different
    # depending on whether m2crypto Python module is present (error is thrown
    # when it is). m2crypto uses a C (based on OpenSSL) implementation under
    # the hood.
    self.private_key = tlslite.api.parsePEMKey(pem_cert_and_key,
                                               private=True,
                                               implementations=['python'])
    self.ssl_client_auth = ssl_client_auth
    self.ssl_client_cas = []
    self.tls_intolerant = tls_intolerant
    self.signed_cert_timestamps = signed_cert_timestamps
    self.fallback_scsv_enabled = fallback_scsv_enabled
    self.ocsp_response = ocsp_response

for ca_file in ssl_client_cas:
      s = open(ca_file).read()
      x509 = tlslite.api.X509()
      x509.parse(s)
      self.ssl_client_cas.append(x509.subject)
    self.ssl_handshake_settings = tlslite.api.HandshakeSettings()
    if ssl_bulk_ciphers is not None:
      self.ssl_handshake_settings.cipherNames = ssl_bulk_ciphers

if record_resume_info:
      # If record_resume_info is true then we'll replace the session cache with
      # an object that records the lookups and inserts that it sees.
      self.session_cache = RecordingSSLSessionCache()
    else:
      self.session_cache = tlslite.api.SessionCache()
    testserver_base.StoppableHTTPServer.__init__(self,
                                                 server_address,
                                                 request_hander_class)

def handshake(self, tlsConnection):
    """Creates the SSL connection."""

try:
      self.tlsConnection = tlsConnection
      tlsConnection.handshakeServer(certChain=self.cert_chain,
                                    privateKey=self.private_key,
                                    sessionCache=self.session_cache,
                                    reqCert=self.ssl_client_auth,
                                    settings=self.ssl_handshake_settings,
                                    reqCAs=self.ssl_client_cas,
                                    tlsIntolerant=self.tls_intolerant,
                                    signedCertTimestamps=
                                    self.signed_cert_timestamps,
                                    fallbackSCSV=self.fallback_scsv_enabled,
                                    ocspResponse = self.ocsp_response)
      tlsConnection.ignoreAbruptClose = True
      return True
    except tlslite.api.TLSAbruptCloseError:
      # Ignore abrupt close.
      return True
    except tlslite.api.TLSError, error:
      print "Handshake failure:", str(error)
      return False

class FTPServer(testserver_base.ClientRestrictingServerMixIn,
                pyftpdlib.ftpserver.FTPServer):
  """This is a specialization of FTPServer that adds client verification."""

pass

class TCPEchoServer(testserver_base.ClientRestrictingServerMixIn,
                    SocketServer.TCPServer):
  """A TCP echo server that echoes back what it has received."""

def server_bind(self):
    """Override server_bind to store the server name."""

SocketServer.TCPServer.server_bind(self)
    host, port = self.socket.getsockname()[:2]
    self.server_name = socket.getfqdn(host)
    self.server_port = port

def serve_forever(self):
    self.stop = False
    self.nonce_time = None
    while not self.stop:
      self.handle_request()
    self.socket.close()

class UDPEchoServer(testserver_base.ClientRestrictingServerMixIn,
                    SocketServer.UDPServer):
  """A UDP echo server that echoes back what it has received."""

def server_bind(self):
    """Override server_bind to store the server name."""

SocketServer.UDPServer.server_bind(self)
    host, port = self.socket.getsockname()[:2]
    self.server_name = socket.getfqdn(host)
    self.server_port = port

def serve_forever(self):
    self.stop = False
    self.nonce_time = None
    while not self.stop:
      self.handle_request()
    self.socket.close()

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.SetManyCookiesHandler,
      self.ExpectAndSetCookieHandler,
      self.SetHeaderHandler,
      self.AuthBasicHandler,
      self.AuthDigestHandler,
      self.SlowServerHandler,
      self.ChunkedServerHandler,
      self.ContentTypeHandler,
      self.NoContentHandler,
      self.ServerRedirectHandler,
      self.ClientRedirectHandler,
      self.MultipartHandler,
      self.GetSSLSessionCacheHandler,
      self.SSLManySmallRecords,
      self.GetChannelID,
      self.CloseSocketHandler,
      self.RangeResetHandler,
      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',
      'json': 'application/json',
      'pdf' : 'application/pdf',
      'txt' : 'text/plain',
      'wav' : 'audio/wav',
      'xml' : 'text/xml'
    }
    self._default_mime_type = 'text/html'

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()