| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297 |
- #!/usr/bin/env python
- """static - A stupidly simple WSGI way to serve static (or mixed) content.
- (See the docstrings of the various functions and classes.)
- Copyright (C) 2006-2009 Luke Arno - http://lukearno.com/
- This library is free software; you can redistribute it and/or
- modify it under the terms of the GNU Lesser General Public
- License as published by the Free Software Foundation; either
- version 2.1 of the License, or (at your option) any later version.
- This library is distributed in the hope that it will be useful,
- but WITHOUT ANY WARRANTY; without even the implied warranty of
- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- Lesser General Public License for more details.
- You should have received a copy of the GNU Lesser General Public
- License along with this library; if not, write to:
- The Free Software Foundation, Inc.,
- 51 Franklin Street, Fifth Floor,
- Boston, MA 02110-1301, USA.
- Luke Arno can be found at http://lukearno.com/
- """
- import mimetypes
- import time
- import string
- import sys
- from os import path, stat, getcwd
- from email.utils import formatdate, parsedate
- from fnmatch import fnmatch
- from wsgiref import util
- from wsgiref.validate import validator
- from wsgiref.headers import Headers
- from wsgiref.simple_server import make_server
- from optparse import OptionParser
- try: from pkg_resources import resource_filename, Requirement
- except: pass
- try: import kid
- except: pass
- class MagicError(Exception): pass
- class StatusApp:
- """Used by WSGI apps to return some HTTP status."""
- def __init__(self, status, message=None):
- self.status = status
- if message is None:
- self.message = status
- else:
- self.message = message
- def __call__(self, environ, start_response, headers=[]):
- if self.message:
- Headers(headers).add_header('Content-type', 'text/plain')
- start_response(self.status, headers)
- if environ['REQUEST_METHOD'] == 'HEAD':
- return [""]
- else:
- return [self.message]
- class Cling(object):
- """A stupidly simple way to serve static content via WSGI.
- Serve the file of the same path as PATH_INFO in self.datadir.
- Look up the Content-type in self.content_types by extension
- or use 'text/plain' if the extension is not found.
- Serve up the contents of the file or delegate to self.not_found.
- """
- block_size = 16 * 4096
- index_file = 'index.html'
- not_found = StatusApp('404 Not Found')
- not_modified = StatusApp('304 Not Modified', "")
- moved_permanently = StatusApp('301 Moved Permanently')
- method_not_allowed = StatusApp('405 Method Not Allowed')
- success_no_content = StatusApp('204 No Content', "")
- server_error = StatusApp('500 Internal Server Error')
- def __init__(self, root, **kw):
- """Just set the root and any other attribs passes via **kw."""
- self.root = root
- for k, v in kw.iteritems():
- setattr(self, k, v)
- def __call__(self, environ, start_response):
- """Respond to a request when called in the usual WSGI way."""
- path_info = environ.get('PATH_INFO', '')
- full_path = self._full_path(path_info)
- if not self._is_under_root(full_path):
- return self.not_found(environ, start_response)
- if path.isdir(full_path):
- if full_path[-1] <> '/' or full_path == self.root:
- location = util.request_uri(environ, include_query=False) + '/'
- if environ.get('QUERY_STRING'):
- location += '?' + environ.get('QUERY_STRING')
- headers = [('Location', location)]
- return self.moved_permanently(environ, start_response, headers)
- else:
- full_path = self._full_path(path_info + self.index_file)
- try:
- sz = int(environ['CONTENT_LENGTH'])
- except:
- sz = 0
- if environ['REQUEST_METHOD'] == 'PUT' and sz > 0:
- for putglob in self.puttable:
- if fnmatch(path_info, putglob):
- data = environ['wsgi.input'].read(sz)
- try:
- with open(full_path, "wb") as f: f.write(data)
- return self.success_no_content(environ, start_response)
- except:
- print sys.exc_info()[1]
- return self.server_error(environ, start_response)
- if environ['REQUEST_METHOD'] not in ('GET', 'HEAD'):
- headers = [('Allow', 'GET, HEAD')]
- return self.method_not_allowed(environ, start_response, headers)
- content_type = self._guess_type(full_path)
- try:
- etag, last_modified = self._conditions(full_path, environ)
- headers = [('Date', formatdate(time.time())),
- ('Last-Modified', last_modified),
- ('ETag', etag)]
- if_modified = environ.get('HTTP_IF_MODIFIED_SINCE')
- if if_modified and (parsedate(if_modified)
- >= parsedate(last_modified)):
- return self.not_modified(environ, start_response, headers)
- if_none = environ.get('HTTP_IF_NONE_MATCH')
- if if_none and (if_none == '*' or etag in if_none):
- return self.not_modified(environ, start_response, headers)
- file_like = self._file_like(full_path)
- headers.append(('Content-Type', content_type))
- start_response("200 OK", headers)
- if environ['REQUEST_METHOD'] == 'GET':
- return self._body(full_path, environ, file_like)
- else:
- return ['']
- except (IOError, OSError), e:
- print e
- return self.not_found(environ, start_response)
- def _full_path(self, path_info):
- """Return the full path from which to read."""
- return self.root + path_info
- def _is_under_root(self, full_path):
- """Guard against arbitrary file retrieval."""
- if (path.abspath(full_path) + path.sep)\
- .startswith(path.abspath(self.root) + path.sep):
- return True
- else:
- return False
- def _guess_type(self, full_path):
- """Guess the mime type using the mimetypes module."""
- return mimetypes.guess_type(full_path)[0] or 'text/plain'
- def _conditions(self, full_path, environ):
- """Return a tuple of etag, last_modified by mtime from stat."""
- mtime = stat(full_path).st_mtime
- return str(mtime), formatdate(mtime)
- def _file_like(self, full_path):
- """Return the appropriate file object."""
- return open(full_path, 'rb')
- def _body(self, full_path, environ, file_like):
- """Return an iterator over the body of the response."""
- way_to_send = environ.get('wsgi.file_wrapper', iter_and_close)
- return way_to_send(file_like, self.block_size)
- def iter_and_close(file_like, block_size):
- """Yield file contents by block then close the file."""
- while 1:
- try:
- block = file_like.read(block_size)
- if block: yield block
- else: raise StopIteration
- except StopIteration, si:
- file_like.close()
- return
- def cling_wrap(package_name, dir_name, **kw):
- """Return a Cling that serves from the given package and dir_name.
- This uses pkg_resources.resource_filename which is not the
- recommended way, since it extracts the files.
- I think this works fine unless you have some _very_ serious
- requirements for static content, in which case you probably
- shouldn't be serving it through a WSGI app, IMHO. YMMV.
- """
- resource = Requirement.parse(package_name)
- return Cling(resource_filename(resource, dir_name), **kw)
- def command():
- usage = "%prog [--help] [-d DIR] [-l [HOST][:PORT]] [-p GLOB[,GLOB...]]"
- parser = OptionParser(usage=usage, version="static 0.3.6")
- parser.add_option("-d", "--dir", dest="rootdir", default=".",
- help="Root directory to serve. Defaults to '.' .", metavar="DIR")
- parser.add_option("-l", "--listen", dest="listen", default="127.0.0.1:8888",
- help="Listen on this interface (given by its hostname or IP) and port."+
- " HOST defaults to 127.0.0.1. PORT defaults to 8888. "+
- "Leave HOST empty to listen on all interfaces (INSECURE!).",
- metavar="[HOST][:PORT]")
- parser.add_option("-p", "--puttable", dest="puttable", default="",
- help="Comma or space-separated list of request paths for which to"+
- " permit PUT requests. Each path is a glob pattern that may "+
- "contain wildcard characters '*' and/or '?'. "+
- "'*' matches any sequence of characters, including the empty"+
- " string. '?' matches exactly 1 arbitrary character. "+
- "NOTE: Both '*' and '?' match slashes and dots. "+
- "I.e. --puttable=* makes every file under DIR writable!",
- metavar="GLOB[,GLOB...]")
- parser.add_option("--validate", dest="validate", action="store_true",
- default=False,
- help="Enable HTTP validation. You don't need this unless "+
- "you're developing static.py itself.")
- options, args = parser.parse_args()
- if len(args) > 0:
- parser.print_help(sys.stderr)
- sys.exit(1)
- parts = options.listen.split(":")
- if len(parts) == 1:
- try: # if the the listen argument consists only of a port number
- port = int(parts[0])
- host = None
- except: # could not parse as port number => must be a host IP or name
- host = parts[0]
- port = None
- elif len(parts) == 2:
- host, port = parts
- else:
- sys.exit("Invalid host:port specification.")
- if not host:
- host = '0.0.0.0'
- if not port:
- port = 8888
- try:
- port = int(port)
- if port <= 0 or port > 65535: raise ValueError
- except:
- sys.exit("Invalid host:port specification.")
- puttable = set(path.abspath(p) for p in
- options.puttable.replace(","," ").split())
- if puttable and host not in ('127.0.0.1', 'localhost'):
- print("Permitting PUT access for non-localhost connections may be unwise.")
- options.rootdir = path.abspath(options.rootdir)
- for p in puttable:
- if not p.startswith(options.rootdir):
- sys.exit("puttable path '%s' not under root '%s'" % (p, options.rootdir))
- # cut off root prefix from puttable paths
- puttable = set(p[len(options.rootdir):] for p in puttable)
- app = Cling(options.rootdir, puttable=puttable)
- if options.validate:
- app = validator(app)
- try:
- print "Serving %s to http://%s:%d" % (options.rootdir, host, port)
- if puttable:
- print("The following paths (relative to server root) may be "+
- "OVERWRITTEN via HTTP PUT.")
- for p in puttable:
- print p
- make_server(host, port, app).serve_forever()
- except KeyboardInterrupt, ki:
- print "Ciao, baby!"
- except:
- sys.exit("Problem initializing server: %s" % sys.exc_info()[1])
- if __name__ == '__main__':
- command()
|