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

#!/usr/bin/python2.4
# Copyright (c) 2006-2008 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.
It defaults to living on localhost:8888.
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.
To shut it down properly, visit localhost:8888/kill.
"""

import base64
import BaseHTTPServer
import cgi
import optparse
import os
import re
import shutil
import SocketServer
import sys
import time
import tlslite
import tlslite.api
import pyftpdlib.ftpserver

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

SERVER_HTTP = 0
SERVER_FTP = 1

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 = 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):
    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.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)
      tlsConnection.ignoreAbruptClose = True
      return True
    except tlslite.api.TLSError, error:
      print "Handshake failure:", str(error)
      return False

class ForkingHTTPServer(SocketServer.ForkingMixIn, StoppableHTTPServer):
  """This is a specialization of of StoppableHTTPServer which serves each
  request in a separate process"""
  pass

class ForkingHTTPSServer(SocketServer.ForkingMixIn, HTTPSServer):
  """This is a specialization of of HTTPSServer which serves each
  request in a separate process"""
  pass

class TestPageHandler(BaseHTTPServer.BaseHTTPRequestHandler):

def __init__(self, request, client_address, socket_server):
    self._connect_handlers = [
      self.RedirectConnectHandler,
      self.ServerAuthConnectHandler,
      self.DefaultConnectResponseHandler]
    self._get_handlers = [
      self.KillHandler,
      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.EchoHeaderOverride,
      self.EchoAllHandler,
      self.FileHandler,
      self.RealFileWithCommonHeaderHandler,
      self.RealBZ2FileWithCommonHeaderHandler,
      self.SetCookieHandler,
      self.AuthBasicHandler,
      self.AuthDigestHandler,
      self.SlowServerHandler,
      self.ContentTypeHandler,
      self.ServerRedirectHandler,
      self.ClientRedirectHandler,
      self.DefaultResponseHandler]
    self._post_handlers = [
      self.WriteFile,
      self.EchoTitleHandler,
      self.EchoAllHandler,
      self.EchoHandler] + self._get_handlers
    self._put_handlers = [
      self.WriteFile,
      self.EchoTitleHandler,
      self.EchoAllHandler,
      self.EchoHandler] + self._get_handlers

self._mime_types = {
      'gif': 'image/gif',
      'jpeg' : 'image/jpeg',
      'jpg' : 'image/jpeg',
      'xml' : 'text/xml'
    }
    self._default_mime_type = 'text/html'

BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, request,
                                                   client_address,
                                                   socket_server)

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 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)
    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 KillHandler(self):
    """This request handler kills the server, for use when we're done"
    with the a particular test."""

if (self.path.find("kill") < 0):
      return False

self.send_response(200)
    self.send_header('Content-type', 'text/html')
    self.send_header('Cache-Control', 'max-age=0')
    self.end_headers()
    if options.never_die:
      self.wfile.write('I cannot die!! BWAHAHA')
    else:
      self.wfile.write('Goodbye cruel world!')
      self.server.stop = True

return True

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