summaryrefslogtreecommitdiffstats
path: root/testing/tools/proxyserver/proxyserver.py
blob: d7b2ef16109096cb207dd724cf4d8e8ab509d72f (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""
Caching HTTP Proxy for use with the Talos pageload tests
Author: Rob Arnold

This file implements a multithreaded caching http 1.1 proxy. HEAD and GET
methods are supported; POST is not yet.
   
Each incoming request is put onto a new thread; python does not have a thread
pool library, so a new thread is spawned for each request. I have tried to use
the python 2.4 standard library wherever possible.

Caching:
The cache is implemented in the Cache class. Items can only be added to the
cache. The only way to remove items from the cache is to blow it all away,
either by deleting the file (default: proxy_cache.db) or passing the -c or
--clear-cache flags on the command line. It is technically possible to remove
items individually from the cache, but there has been no need to do so so far.

The cache is implemented with the shelve module. The key is the combination of
host, port and request (path + params + fragment) and the values stored are the
http status code, headers and content that were received from the remote server.

Access to the cache is guarded by a semaphore which allows concurrent read
access. The semaphore is guarded by a simple mutex which prevents a deadlock
from occuring when two threads try to add an item to the cache at the same time.

Memory usage is kept to a minimum by the shelve module; only items in the cache
that are currently being served stay in memory.

Proxy:
The BaseHTTPServer.BaseHTTPRequestHandler takes care of parsing incoming
requests and managing the socket connection. See the documentation of the
BaseHTTPServer module for more information. When do_HEAD or do_GET is called,
the url that we are supposed to fetch is in self.path.

TODO:
* Implement POST requests. This requires implementing the do_POST method and
  passing the post data along.
* Implement different cache policies
* Added an interface to allow administrators to probe the cache and remove
  items from the database and such.
"""

__version__ = "0.1"

import os
import sys
import time
import threading
import shelve
from optparse import OptionParser, OptionValueError

import SocketServer
import BaseHTTPServer
import socket
import httplib
from urlparse import urlsplit, urlunsplit

class HTTPRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
  server_version = "TalosProxy/" + __version__
  protocol_version = "HTTP/1.1"

  def do_GET(self):
    content = self.send_head()
    if content:
      try:
        self.wfile.write(content)
      except socket.error, e:
        if options.verbose:
          print "Got socket error %s" % e
    #self.close_connection = 1
  def do_HEAD(self):
    self.send_head()

  def getHeaders(self):
    h = {}
    for name in self.headers.keys():
      h[name] = self.headers[name]

    return h

  def send_head(self, method="GET"): 
    o = urlsplit(self.path)

    #sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    headers = self.getHeaders()
    for k in "Proxy-Connection", "Connection":
      if k in headers:
        headers[k] = "Close"
    if "Keep-Alive" in headers:
      del headers["Keep-Alive"]

    reqstring = urlunsplit(('','',o.path, o.query, o.fragment))

    if options.no_cache:
      cache_result = None
    else:
      cache_result = cache.get(o.hostname, o.port, reqstring)

    if not cache_result:
      if options.localonly:
        self.send_error(404, "Object not in cache")
        return None
      else:
        if options.verbose:
          print "Object %s was not in the cache" % self.path
        conn = httplib.HTTPConnection(o.netloc)
        conn.request("GET", reqstring, headers=headers)
        res = conn.getresponse()

        content = res.read()
        conn.close()

        status, headers = res.status, res.getheaders()

        if not options.no_cache:
          cache.add(o.hostname, o.port, reqstring, status, headers, content)
    else:
      status, headers, content = cache_result

    try:
      self.send_response(status)
      for name, value in headers:
        # kill the transfer-encoding header because we don't support it when
        # we send data to the client
        if name not in ('transfer-encoding',):
          self.send_header(name, value)
      if "Content-Length" not in headers:
        self.send_header("Content-Length", str(len(content)))
      self.end_headers()
    except socket.error, e:
      if options.verbose:
        print "Got socket error %s" % e
      return None
    return content
  def log_message(self, format, *args):
    if options.verbose:
      BaseHTTPServer.BaseHTTPRequestHandler.log_message(self, format, *args)

class HTTPServer(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer):
  def __init__(self, address, handler):
    BaseHTTPServer.HTTPServer.__init__(self, address, handler)

class Cache(object):
  """Multithreaded cache uses the shelve module to store pages"""
  # 20 concurrent threads ought to be enough for one browser
  max_concurrency = 20
  def __init__(self, name='', max_concurrency=20):
    name = name or options.cache or "proxy_cache.db"
    self.name = name
    self.max_concurrency = max_concurrency
    self.entries = {}
    self.sem = threading.Semaphore(self.max_concurrency)
    self.semlock = threading.Lock()
    if options.clear_cache:
      flag = 'n'
    else:
      flag = 'c'
    self.db = shelve.DbfilenameShelf(name, flag)

  def __del__(self):
    if hasattr(self, 'db'):
      self.db.close()

  def get_key(self, host, port, resource):
    return '%s:%s/%s' % (host, port, resource)

  def get(self, host, port, resource):
    key = self.get_key(host, port, resource)
    self.semlock.acquire()
    self.sem.acquire()
    self.semlock.release()
    try:
      if not self.db.has_key(key):
        return None
      # returns status, headers, content
      return self.db[key]
    finally:
      self.sem.release()
  def add(self, host, port, resource, status, headers, content):
    key = self.get_key(host, port, resource)
    self.semlock.acquire()
    for i in range(self.max_concurrency):
      self.sem.acquire()
    self.semlock.release()
    try:
      self.db[key] = (status, headers, content)
      self.db.sync()
    finally:
      for i in range(self.max_concurrency):
        self.sem.release()

class Options(object):
  port = 8000
  localonly = False
  clear_cache = False
  no_cache = False
  cache = 'proxy_cache.db'
  verbose = False

def _parseOptions():
  def port_callback(option, opt, value, parser):
    if value > 0 and value < (2 ** 16 - 1):
      setattr(parser.values, option.dest, value)
    else:
      raise OptionValueError("Port number is out of range")

  global options
  parser = OptionParser(version="Talos Proxy " + __version__)
  parser.add_option("-p", "--port", dest="port",
    help="The port to run the proxy server on", metavar="PORT", type="int",
    action="callback", callback=port_callback)
  parser.add_option("-v", "--verbose", action="store_true", dest="verbose",
    help="Include additional debugging information")
  parser.add_option("-l", "--localonly", action="store_true", dest="localonly",
    help="Only serve pages from the local database")
  parser.add_option("-c", "--clear", action="store_true", dest="clear_cache",
    help="Clear the cache on startup")
  parser.add_option("-n", "--no-cache", action="store_true", dest="no_cache",
    help="Do not use a cache")
  parser.add_option("-u", "--use-cache", dest="cache",
    help="The filename of the cache to use", metavar="NAME.db")
  parser.set_defaults(verbose=Options.verbose,
                      port=Options.port,
                      localonly=Options.localonly,
                      clear_cache=Options.clear_cache,
                      no_cache=Options.no_cache,
                      cache=Options.cache)
  options, args = parser.parse_args()

"""Configures the proxy server. This should be called before run_proxy. It can be
called afterwards, but note that it is not threadsafe and some options (namely
port) will not take effect"""
def configure_proxy(**kwargs):
  global options
  options = Options()
  for key in kwargs:
    setattr(options, key, kwargs[key])

def _run():
  global cache
  cache = Cache()
  server_address = ('', options.port)
  httpd = HTTPServer(server_address, HTTPRequestHandler)
  httpd.serve_forever()

"""Starts the proxy; it runs on a separate daemon thread"""
def run_proxy():
  thr = threading.Thread(target=_run)
  # now when we die, the daemon thread will die too
  thr.setDaemon(1)
  thr.start()

if __name__ == '__main__':
  _parseOptions()
  try:
    run_proxy()
    # thr.join() doesn't terminate on keyboard interrupt
    while 1: time.sleep(1)
  except KeyboardInterrupt:
    if options.verbose:
      print "Quittin' time..."

__all__ = ['run_proxy', 'configure_proxy']