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

#!/usr/bin/python2.4
# Copyright (c) 2011 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 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 asyncore
import base64
import BaseHTTPServer
import cgi
import errno
import optparse
import os
import re
import select
import simplejson
import SocketServer
import socket
import sys
import struct
import time
import urlparse
import warnings

# Ignore deprecation warnings, they make our output more cluttered.
warnings.filterwarnings("ignore", category=DeprecationWarning)

import pyftpdlib.ftpserver
import tlslite
import tlslite.api

try:
  import hashlib
  _new_md5 = hashlib.md5
except ImportError:
  import md5
  _new_md5 = md5.new

if sys.platform == 'win32':
  import msvcrt

SERVER_HTTP = 0
SERVER_FTP = 1
SERVER_SYNC = 2

# Using debug() seems to cause hangs on XP: see http://crbug.com/64515 .
debug_output = sys.stderr
def debug(str):
  debug_output.write(str + "\n")
  debug_output.flush()

class StoppableHTTPServer(BaseHTTPServer.HTTPServer):
  """This is a specialization of of BaseHTTPServer to allow it
  to be exited cleanly (by setting its "stop" member to True)."""

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

class HTTPSServer(tlslite.api.TLSSocketServerMixIn, StoppableHTTPServer):
  """This is a specialization of StoppableHTTPerver that add https support."""

def __init__(self, server_address, request_hander_class, cert_path,
               ssl_client_auth, ssl_client_cas, ssl_bulk_ciphers):
    s = open(cert_path).read()
    x509 = tlslite.api.X509()
    x509.parse(s)
    self.cert_chain = tlslite.api.X509CertChain([x509])
    s = open(cert_path).read()
    self.private_key = tlslite.api.parsePEMKey(s, private=True)
    self.ssl_client_auth = ssl_client_auth
    self.ssl_client_cas = []
    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

self.session_cache = tlslite.api.SessionCache()
    StoppableHTTPServer.__init__(self, server_address, request_hander_class)

def handshake(self, tlsConnection):
    """Creates the SSL connection."""
    try:
      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)
      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 SyncHTTPServer(StoppableHTTPServer):
  """An HTTP server that handles sync commands."""

def __init__(self, server_address, request_handler_class):
    # We import here to avoid pulling in chromiumsync's dependencies
    # unless strictly necessary.
    import chromiumsync
    import xmppserver
    StoppableHTTPServer.__init__(self, server_address, request_handler_class)
    self._sync_handler = chromiumsync.TestServer()
    self._xmpp_socket_map = {}
    self._xmpp_server = xmppserver.XmppServer(
      self._xmpp_socket_map, ('localhost', 0))
    self.xmpp_port = self._xmpp_server.getsockname()[1]

def HandleCommand(self, query, raw_request):
    return self._sync_handler.HandleCommand(query, raw_request)

def HandleRequestNoBlock(self):
    """Handles a single request.

Copied from SocketServer._handle_request_noblock().
    """
    try:
      request, client_address = self.get_request()
    except socket.error:
      return
    if self.verify_request(request, client_address):
      try:
        self.process_request(request, client_address)
      except:
        self.handle_error(request, client_address)
        self.close_request(request)

def serve_forever(self):
    """This is a merge of asyncore.loop() and SocketServer.serve_forever().
    """

def HandleXmppSocket(fd, socket_map, handler):
      """Runs the handler for the xmpp connection for fd.

Adapted from asyncore.read() et al.
      """
      xmpp_connection = socket_map.get(fd)
      # This could happen if a previous handler call caused fd to get
      # removed from socket_map.
      if xmpp_connection is None:
        return
      try:
        handler(xmpp_connection)
      except (asyncore.ExitNow, KeyboardInterrupt, SystemExit):
        raise
      except:
        xmpp_connection.handle_error()

while True:
      read_fds = [ self.fileno() ]
      write_fds = []
      exceptional_fds = []

for fd, xmpp_connection in self._xmpp_socket_map.items():
        is_r = xmpp_connection.readable()
        is_w = xmpp_connection.writable()
        if is_r:
          read_fds.append(fd)
        if is_w:
          write_fds.append(fd)
        if is_r or is_w:
          exceptional_fds.append(fd)

try:
        read_fds, write_fds, exceptional_fds = (
          select.select(read_fds, write_fds, exceptional_fds))
      except select.error, err:
        if err.args[0] != errno.EINTR:
          raise
        else:
          continue

for fd in read_fds:
        if fd == self.fileno():
          self.HandleRequestNoBlock()
          continue
        HandleXmppSocket(fd, self._xmpp_socket_map,
                         asyncore.dispatcher.handle_read_event)

for fd in write_fds:
        HandleXmppSocket(fd, self._xmpp_socket_map,
                         asyncore.dispatcher.handle_write_event)

for fd in exceptional_fds:
        HandleXmppSocket(fd, self._xmpp_socket_map,
                         asyncore.dispatcher.handle_expt_event)

class BasePageHandler(BaseHTTPServer.BaseHTTPRequestHandler):

def __init__(self, request, client_address, socket_server,
               connect_handlers, get_handlers, post_handlers, put_handlers):
    self._connect_handlers = connect_handlers
    self._get_handlers = get_handlers
    self._post_handlers = post_handlers
    self._put_handlers = put_handlers
    BaseHTTPServer.BaseHTTPRequestHandler.__init__(
      self, request, client_address, socket_server)

def log_request(self, *args, **kwargs):
    # Disable request logging to declutter test log output.
    pass

def _ShouldHandleRequest(self, handler_name):
    """Determines if the path can be handled by the handler.

We consider a handler valid if the path begins with the
    handler name. It can optionally be followed by "?*", "/*".
    """

pattern = re.compile('%s($|\?|/).*' % handler_name)
    return pattern.match(self.path)

def do_CONNECT(self):
    for handler in self._connect_handlers:
      if handler():
        return

def do_GET(self):
    for handler in self._get_handlers:
      if handler():
        return

def do_POST(self):
    for handler in self._post_handlers:
      if handler():
        return

def do_PUT(self):
    for handler in self._put_handlers:
      if handler():
        return

class TestPageHandler(BasePageHandler):

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.FileHandler,
      self.SetCookieHandler,
      self.AuthBasicHandler,
      self.AuthDigestHandler,
      self.SlowServerHandler,
      self.ContentTypeHandler,
      self.NoContentHandler,
      self.ServerRedirectHandler,
      self.ClientRedirectHandler,
      self.MultipartHandler,
      self.DefaultResponseHandler]
    post_handlers = [
      self.EchoTitleHandler,
      self.EchoAllHandler,
      self.EchoHandler,
      self.DeviceManagementHandler] + get_handlers
    put_handlers = [
      self.EchoTitleHandler,
      self.EchoAllHandler,
      self.EchoHandler] + get_handlers

self._mime_types = {
      'crx' : 'application/x-chrome-extension',
      'exe' : 'application/octet-stream',
      'gif': 'image/gif',
      'jpeg' : 'image/jpeg',
      'jpg' : 'image/jpeg',
      'pdf' : 'application/pdf',
      'xml' : 'text/xml'
    }
    self._default_mime_type = 'text/html'

BasePageHandler.__init__(self, request, client_address, socket_server,
                             connect_handlers, get_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()