| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639 |
- # (c) 2005 Ian Bicking and contributors; written for Paste (http://pythonpaste.org)
- # Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php
- """
- WSGI applications that parse the URL and dispatch to on-disk resources
- """
- import os
- import six
- import sys
- import imp
- import mimetypes
- try:
- import pkg_resources
- except ImportError:
- pkg_resources = None
- from paste import request
- from paste import fileapp
- from paste.util import import_string
- from paste import httpexceptions
- from .httpheaders import ETAG
- from paste.util import converters
- class NoDefault(object):
- pass
- __all__ = ['URLParser', 'StaticURLParser', 'PkgResourcesParser']
- class URLParser(object):
- """
- WSGI middleware
- Application dispatching, based on URL. An instance of `URLParser` is
- an application that loads and delegates to other applications. It
- looks for files in its directory that match the first part of
- PATH_INFO; these may have an extension, but are not required to have
- one, in which case the available files are searched to find the
- appropriate file. If it is ambiguous, a 404 is returned and an error
- logged.
- By default there is a constructor for .py files that loads the module,
- and looks for an attribute ``application``, which is a ready
- application object, or an attribute that matches the module name,
- which is a factory for building applications, and is called with no
- arguments.
- URLParser will also look in __init__.py for special overrides.
- These overrides are:
- ``urlparser_hook(environ)``
- This can modify the environment. Its return value is ignored,
- and it cannot be used to change the response in any way. You
- *can* use this, for example, to manipulate SCRIPT_NAME/PATH_INFO
- (try to keep them consistent with the original URL -- but
- consuming PATH_INFO and moving that to SCRIPT_NAME is ok).
- ``urlparser_wrap(environ, start_response, app)``:
- After URLParser finds the application, it calls this function
- (if present). If this function doesn't call
- ``app(environ, start_response)`` then the application won't be
- called at all! This can be used to allocate resources (with
- ``try:finally:``) or otherwise filter the output of the
- application.
- ``not_found_hook(environ, start_response)``:
- If no file can be found (*in this directory*) to match the
- request, then this WSGI application will be called. You can
- use this to change the URL and pass the request back to
- URLParser again, or on to some other application. This
- doesn't catch all ``404 Not Found`` responses, just missing
- files.
- ``application(environ, start_response)``:
- This basically overrides URLParser completely, and the given
- application is used for all requests. ``urlparser_wrap`` and
- ``urlparser_hook`` are still called, but the filesystem isn't
- searched in any way.
- """
- parsers_by_directory = {}
- # This is lazily initialized
- init_module = NoDefault
- global_constructors = {}
- def __init__(self, global_conf,
- directory, base_python_name,
- index_names=NoDefault,
- hide_extensions=NoDefault,
- ignore_extensions=NoDefault,
- constructors=None,
- **constructor_conf):
- """
- Create a URLParser object that looks at `directory`.
- `base_python_name` is the package that this directory
- represents, thus any Python modules in this directory will
- be given names under this package.
- """
- if global_conf:
- import warnings
- warnings.warn(
- 'The global_conf argument to URLParser is deprecated; '
- 'either pass in None or {}, or use make_url_parser',
- DeprecationWarning)
- else:
- global_conf = {}
- if os.path.sep != '/':
- directory = directory.replace(os.path.sep, '/')
- self.directory = directory
- self.base_python_name = base_python_name
- # This logic here should be deprecated since it is in
- # make_url_parser
- if index_names is NoDefault:
- index_names = global_conf.get(
- 'index_names', ('index', 'Index', 'main', 'Main'))
- self.index_names = converters.aslist(index_names)
- if hide_extensions is NoDefault:
- hide_extensions = global_conf.get(
- 'hide_extensions', ('.pyc', '.bak', '.py~', '.pyo'))
- self.hide_extensions = converters.aslist(hide_extensions)
- if ignore_extensions is NoDefault:
- ignore_extensions = global_conf.get(
- 'ignore_extensions', ())
- self.ignore_extensions = converters.aslist(ignore_extensions)
- self.constructors = self.global_constructors.copy()
- if constructors:
- self.constructors.update(constructors)
- # @@: Should we also check the global options for constructors?
- for name, value in constructor_conf.items():
- if not name.startswith('constructor '):
- raise ValueError(
- "Only extra configuration keys allowed are "
- "'constructor .ext = import_expr'; you gave %r "
- "(=%r)" % (name, value))
- ext = name[len('constructor '):].strip()
- if isinstance(value, (str, unicode)):
- value = import_string.eval_import(value)
- self.constructors[ext] = value
- def __call__(self, environ, start_response):
- environ['paste.urlparser.base_python_name'] = self.base_python_name
- if self.init_module is NoDefault:
- self.init_module = self.find_init_module(environ)
- path_info = environ.get('PATH_INFO', '')
- if not path_info:
- return self.add_slash(environ, start_response)
- if (self.init_module
- and getattr(self.init_module, 'urlparser_hook', None)):
- self.init_module.urlparser_hook(environ)
- orig_path_info = environ['PATH_INFO']
- orig_script_name = environ['SCRIPT_NAME']
- application, filename = self.find_application(environ)
- if not application:
- if (self.init_module
- and getattr(self.init_module, 'not_found_hook', None)
- and environ.get('paste.urlparser.not_found_parser') is not self):
- not_found_hook = self.init_module.not_found_hook
- environ['paste.urlparser.not_found_parser'] = self
- environ['PATH_INFO'] = orig_path_info
- environ['SCRIPT_NAME'] = orig_script_name
- return not_found_hook(environ, start_response)
- if filename is None:
- name, rest_of_path = request.path_info_split(environ['PATH_INFO'])
- if not name:
- name = 'one of %s' % ', '.join(
- self.index_names or
- ['(no index_names defined)'])
- return self.not_found(
- environ, start_response,
- 'Tried to load %s from directory %s'
- % (name, self.directory))
- else:
- environ['wsgi.errors'].write(
- 'Found resource %s, but could not construct application\n'
- % filename)
- return self.not_found(
- environ, start_response,
- 'Tried to load %s from directory %s'
- % (filename, self.directory))
- if (self.init_module
- and getattr(self.init_module, 'urlparser_wrap', None)):
- return self.init_module.urlparser_wrap(
- environ, start_response, application)
- else:
- return application(environ, start_response)
- def find_application(self, environ):
- if (self.init_module
- and getattr(self.init_module, 'application', None)
- and not environ.get('paste.urlparser.init_application') == environ['SCRIPT_NAME']):
- environ['paste.urlparser.init_application'] = environ['SCRIPT_NAME']
- return self.init_module.application, None
- name, rest_of_path = request.path_info_split(environ['PATH_INFO'])
- environ['PATH_INFO'] = rest_of_path
- if name is not None:
- environ['SCRIPT_NAME'] = environ.get('SCRIPT_NAME', '') + '/' + name
- if not name:
- names = self.index_names
- for index_name in names:
- filename = self.find_file(environ, index_name)
- if filename:
- break
- else:
- # None of the index files found
- filename = None
- else:
- filename = self.find_file(environ, name)
- if filename is None:
- return None, filename
- else:
- return self.get_application(environ, filename), filename
- def not_found(self, environ, start_response, debug_message=None):
- exc = httpexceptions.HTTPNotFound(
- 'The resource at %s could not be found'
- % request.construct_url(environ),
- comment=debug_message)
- return exc.wsgi_application(environ, start_response)
- def add_slash(self, environ, start_response):
- """
- This happens when you try to get to a directory
- without a trailing /
- """
- url = request.construct_url(environ, with_query_string=False)
- url += '/'
- if environ.get('QUERY_STRING'):
- url += '?' + environ['QUERY_STRING']
- exc = httpexceptions.HTTPMovedPermanently(
- 'The resource has moved to %s - you should be redirected '
- 'automatically.' % url,
- headers=[('location', url)])
- return exc.wsgi_application(environ, start_response)
- def find_file(self, environ, base_filename):
- possible = []
- """Cache a few values to reduce function call overhead"""
- for filename in os.listdir(self.directory):
- base, ext = os.path.splitext(filename)
- full_filename = os.path.join(self.directory, filename)
- if (ext in self.hide_extensions
- or not base):
- continue
- if filename == base_filename:
- possible.append(full_filename)
- continue
- if ext in self.ignore_extensions:
- continue
- if base == base_filename:
- possible.append(full_filename)
- if not possible:
- #environ['wsgi.errors'].write(
- # 'No file found matching %r in %s\n'
- # % (base_filename, self.directory))
- return None
- if len(possible) > 1:
- # If there is an exact match, this isn't 'ambiguous'
- # per se; it might mean foo.gif and foo.gif.back for
- # instance
- if full_filename in possible:
- return full_filename
- else:
- environ['wsgi.errors'].write(
- 'Ambiguous URL: %s; matches files %s\n'
- % (request.construct_url(environ),
- ', '.join(possible)))
- return None
- return possible[0]
- def get_application(self, environ, filename):
- if os.path.isdir(filename):
- t = 'dir'
- else:
- t = os.path.splitext(filename)[1]
- constructor = self.constructors.get(t, self.constructors.get('*'))
- if constructor is None:
- #environ['wsgi.errors'].write(
- # 'No constructor found for %s\n' % t)
- return constructor
- app = constructor(self, environ, filename)
- if app is None:
- #environ['wsgi.errors'].write(
- # 'Constructor %s return None for %s\n' %
- # (constructor, filename))
- pass
- return app
- def register_constructor(cls, extension, constructor):
- """
- Register a function as a constructor. Registered constructors
- apply to all instances of `URLParser`.
- The extension should have a leading ``.``, or the special
- extensions ``dir`` (for directories) and ``*`` (a catch-all).
- `constructor` must be a callable that takes two arguments:
- ``environ`` and ``filename``, and returns a WSGI application.
- """
- d = cls.global_constructors
- assert extension not in d, (
- "A constructor already exists for the extension %r (%r) "
- "when attemption to register constructor %r"
- % (extension, d[extension], constructor))
- d[extension] = constructor
- register_constructor = classmethod(register_constructor)
- def get_parser(self, directory, base_python_name):
- """
- Get a parser for the given directory, or create one if
- necessary. This way parsers can be cached and reused.
- # @@: settings are inherited from the first caller
- """
- try:
- return self.parsers_by_directory[(directory, base_python_name)]
- except KeyError:
- parser = self.__class__(
- {},
- directory, base_python_name,
- index_names=self.index_names,
- hide_extensions=self.hide_extensions,
- ignore_extensions=self.ignore_extensions,
- constructors=self.constructors)
- self.parsers_by_directory[(directory, base_python_name)] = parser
- return parser
- def find_init_module(self, environ):
- filename = os.path.join(self.directory, '__init__.py')
- if not os.path.exists(filename):
- return None
- return load_module(environ, filename)
- def __repr__(self):
- return '<%s directory=%r; module=%s at %s>' % (
- self.__class__.__name__,
- self.directory,
- self.base_python_name,
- hex(abs(id(self))))
- def make_directory(parser, environ, filename):
- base_python_name = environ['paste.urlparser.base_python_name']
- if base_python_name:
- base_python_name += "." + os.path.basename(filename)
- else:
- base_python_name = os.path.basename(filename)
- return parser.get_parser(filename, base_python_name)
- URLParser.register_constructor('dir', make_directory)
- def make_unknown(parser, environ, filename):
- return fileapp.FileApp(filename)
- URLParser.register_constructor('*', make_unknown)
- def load_module(environ, filename):
- base_python_name = environ['paste.urlparser.base_python_name']
- module_name = os.path.splitext(os.path.basename(filename))[0]
- if base_python_name:
- module_name = base_python_name + '.' + module_name
- return load_module_from_name(environ, filename, module_name,
- environ['wsgi.errors'])
- def load_module_from_name(environ, filename, module_name, errors):
- if module_name in sys.modules:
- return sys.modules[module_name]
- init_filename = os.path.join(os.path.dirname(filename), '__init__.py')
- if not os.path.exists(init_filename):
- try:
- f = open(init_filename, 'w')
- except (OSError, IOError) as e:
- errors.write(
- 'Cannot write __init__.py file into directory %s (%s)\n'
- % (os.path.dirname(filename), e))
- return None
- f.write('#\n')
- f.close()
- fp = None
- if module_name in sys.modules:
- return sys.modules[module_name]
- if '.' in module_name:
- parent_name = '.'.join(module_name.split('.')[:-1])
- base_name = module_name.split('.')[-1]
- parent = load_module_from_name(environ, os.path.dirname(filename),
- parent_name, errors)
- else:
- base_name = module_name
- fp = None
- try:
- fp, pathname, stuff = imp.find_module(
- base_name, [os.path.dirname(filename)])
- module = imp.load_module(module_name, fp, pathname, stuff)
- finally:
- if fp is not None:
- fp.close()
- return module
- def make_py(parser, environ, filename):
- module = load_module(environ, filename)
- if not module:
- return None
- if hasattr(module, 'application') and module.application:
- return getattr(module.application, 'wsgi_application', module.application)
- base_name = module.__name__.split('.')[-1]
- if hasattr(module, base_name):
- obj = getattr(module, base_name)
- if hasattr(obj, 'wsgi_application'):
- return obj.wsgi_application
- else:
- # @@: Old behavior; should probably be deprecated eventually:
- return getattr(module, base_name)()
- environ['wsgi.errors'].write(
- "Cound not find application or %s in %s\n"
- % (base_name, module))
- return None
- URLParser.register_constructor('.py', make_py)
- class StaticURLParser(object):
- """
- Like ``URLParser`` but only serves static files.
- ``cache_max_age``:
- integer specifies Cache-Control max_age in seconds
- """
- # @@: Should URLParser subclass from this?
- def __init__(self, directory, root_directory=None,
- cache_max_age=None):
- self.directory = self.normpath(directory)
- self.root_directory = self.normpath(root_directory or directory)
- self.cache_max_age = cache_max_age
- def normpath(path):
- return os.path.normcase(os.path.abspath(path))
- normpath = staticmethod(normpath)
- def __call__(self, environ, start_response):
- path_info = environ.get('PATH_INFO', '')
- if not path_info:
- return self.add_slash(environ, start_response)
- if path_info == '/':
- # @@: This should obviously be configurable
- filename = 'index.html'
- else:
- filename = request.path_info_pop(environ)
- full = self.normpath(os.path.join(self.directory, filename))
- if not full.startswith(self.root_directory):
- # Out of bounds
- return self.not_found(environ, start_response)
- if not os.path.exists(full):
- return self.not_found(environ, start_response)
- if os.path.isdir(full):
- # @@: Cache?
- return self.__class__(full, root_directory=self.root_directory,
- cache_max_age=self.cache_max_age)(environ,
- start_response)
- if environ.get('PATH_INFO') and environ.get('PATH_INFO') != '/':
- return self.error_extra_path(environ, start_response)
- if_none_match = environ.get('HTTP_IF_NONE_MATCH')
- if if_none_match:
- mytime = os.stat(full).st_mtime
- if str(mytime) == if_none_match:
- headers = []
- ## FIXME: probably should be
- ## ETAG.update(headers, '"%s"' % mytime)
- ETAG.update(headers, mytime)
- start_response('304 Not Modified', headers)
- return [''] # empty body
- fa = self.make_app(full)
- if self.cache_max_age:
- fa.cache_control(max_age=self.cache_max_age)
- return fa(environ, start_response)
- def make_app(self, filename):
- return fileapp.FileApp(filename)
- def add_slash(self, environ, start_response):
- """
- This happens when you try to get to a directory
- without a trailing /
- """
- url = request.construct_url(environ, with_query_string=False)
- url += '/'
- if environ.get('QUERY_STRING'):
- url += '?' + environ['QUERY_STRING']
- exc = httpexceptions.HTTPMovedPermanently(
- 'The resource has moved to %s - you should be redirected '
- 'automatically.' % url,
- headers=[('location', url)])
- return exc.wsgi_application(environ, start_response)
- def not_found(self, environ, start_response, debug_message=None):
- exc = httpexceptions.HTTPNotFound(
- 'The resource at %s could not be found'
- % request.construct_url(environ),
- comment='SCRIPT_NAME=%r; PATH_INFO=%r; looking in %r; debug: %s'
- % (environ.get('SCRIPT_NAME'), environ.get('PATH_INFO'),
- self.directory, debug_message or '(none)'))
- return exc.wsgi_application(environ, start_response)
- def error_extra_path(self, environ, start_response):
- exc = httpexceptions.HTTPNotFound(
- 'The trailing path %r is not allowed' % environ['PATH_INFO'])
- return exc.wsgi_application(environ, start_response)
- def __repr__(self):
- return '<%s %r>' % (self.__class__.__name__, self.directory)
- def make_static(global_conf, document_root, cache_max_age=None):
- """
- Return a WSGI application that serves a directory (configured
- with document_root)
- cache_max_age - integer specifies CACHE_CONTROL max_age in seconds
- """
- if cache_max_age is not None:
- cache_max_age = int(cache_max_age)
- return StaticURLParser(
- document_root, cache_max_age=cache_max_age)
- class PkgResourcesParser(StaticURLParser):
- def __init__(self, egg_or_spec, resource_name, manager=None, root_resource=None):
- if pkg_resources is None:
- raise NotImplementedError("This class requires pkg_resources.")
- if isinstance(egg_or_spec, (six.binary_type, six.text_type)):
- self.egg = pkg_resources.get_distribution(egg_or_spec)
- else:
- self.egg = egg_or_spec
- self.resource_name = resource_name
- if manager is None:
- manager = pkg_resources.ResourceManager()
- self.manager = manager
- if root_resource is None:
- root_resource = resource_name
- self.root_resource = os.path.normpath(root_resource)
- def __repr__(self):
- return '<%s for %s:%r>' % (
- self.__class__.__name__,
- self.egg.project_name,
- self.resource_name)
- def __call__(self, environ, start_response):
- path_info = environ.get('PATH_INFO', '')
- if not path_info:
- return self.add_slash(environ, start_response)
- if path_info == '/':
- # @@: This should obviously be configurable
- filename = 'index.html'
- else:
- filename = request.path_info_pop(environ)
- resource = os.path.normcase(os.path.normpath(
- self.resource_name + '/' + filename))
- if self.root_resource is not None and not resource.startswith(self.root_resource):
- # Out of bounds
- return self.not_found(environ, start_response)
- if not self.egg.has_resource(resource):
- return self.not_found(environ, start_response)
- if self.egg.resource_isdir(resource):
- # @@: Cache?
- child_root = self.root_resource is not None and self.root_resource or \
- self.resource_name
- return self.__class__(self.egg, resource, self.manager,
- root_resource=child_root)(environ, start_response)
- if environ.get('PATH_INFO') and environ.get('PATH_INFO') != '/':
- return self.error_extra_path(environ, start_response)
- type, encoding = mimetypes.guess_type(resource)
- if not type:
- type = 'application/octet-stream'
- # @@: I don't know what to do with the encoding.
- try:
- file = self.egg.get_resource_stream(self.manager, resource)
- except (IOError, OSError) as e:
- exc = httpexceptions.HTTPForbidden(
- 'You are not permitted to view this file (%s)' % e)
- return exc.wsgi_application(environ, start_response)
- start_response('200 OK',
- [('content-type', type)])
- return fileapp._FileIter(file)
- def not_found(self, environ, start_response, debug_message=None):
- exc = httpexceptions.HTTPNotFound(
- 'The resource at %s could not be found'
- % request.construct_url(environ),
- comment='SCRIPT_NAME=%r; PATH_INFO=%r; looking in egg:%s#%r; debug: %s'
- % (environ.get('SCRIPT_NAME'), environ.get('PATH_INFO'),
- self.egg, self.resource_name, debug_message or '(none)'))
- return exc.wsgi_application(environ, start_response)
- def make_pkg_resources(global_conf, egg, resource_name=''):
- """
- A static file parser that loads data from an egg using
- ``pkg_resources``. Takes a configuration value ``egg``, which is
- an egg spec, and a base ``resource_name`` (default empty string)
- which is the path in the egg that this starts at.
- """
- if pkg_resources is None:
- raise NotImplementedError("This function requires pkg_resources.")
- return PkgResourcesParser(egg, resource_name)
- def make_url_parser(global_conf, directory, base_python_name,
- index_names=None, hide_extensions=None,
- ignore_extensions=None,
- **constructor_conf):
- """
- Create a URLParser application that looks in ``directory``, which
- should be the directory for the Python package named in
- ``base_python_name``. ``index_names`` are used when viewing the
- directory (like ``'index'`` for ``'index.html'``).
- ``hide_extensions`` are extensions that are not viewable (like
- ``'.pyc'``) and ``ignore_extensions`` are viewable but only if an
- explicit extension is given.
- """
- if index_names is None:
- index_names = global_conf.get(
- 'index_names', ('index', 'Index', 'main', 'Main'))
- index_names = converters.aslist(index_names)
- if hide_extensions is None:
- hide_extensions = global_conf.get(
- 'hide_extensions', ('.pyc', 'bak', 'py~'))
- hide_extensions = converters.aslist(hide_extensions)
- if ignore_extensions is None:
- ignore_extensions = global_conf.get(
- 'ignore_extensions', ())
- ignore_extensions = converters.aslist(ignore_extensions)
- # There's no real way to set constructors currently...
- return URLParser({}, directory, base_python_name,
- index_names=index_names,
- hide_extensions=hide_extensions,
- ignore_extensions=ignore_extensions,
- **constructor_conf)
|