#!/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 rfc822 import time import string import sys from os import path, stat, getcwd 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', rfc822.formatdate(time.time())), ('Last-Modified', last_modified), ('ETag', etag)] if_modified = environ.get('HTTP_IF_MODIFIED_SINCE') if if_modified and (rfc822.parsedate(if_modified) >= rfc822.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), rfc822.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()