jpayne@68: """Miscellaneous WSGI-related Utilities""" jpayne@68: jpayne@68: import posixpath jpayne@68: jpayne@68: __all__ = [ jpayne@68: 'FileWrapper', 'guess_scheme', 'application_uri', 'request_uri', jpayne@68: 'shift_path_info', 'setup_testing_defaults', jpayne@68: ] jpayne@68: jpayne@68: jpayne@68: class FileWrapper: jpayne@68: """Wrapper to convert file-like objects to iterables""" jpayne@68: jpayne@68: def __init__(self, filelike, blksize=8192): jpayne@68: self.filelike = filelike jpayne@68: self.blksize = blksize jpayne@68: if hasattr(filelike,'close'): jpayne@68: self.close = filelike.close jpayne@68: jpayne@68: def __getitem__(self,key): jpayne@68: import warnings jpayne@68: warnings.warn( jpayne@68: "FileWrapper's __getitem__ method ignores 'key' parameter. " jpayne@68: "Use iterator protocol instead.", jpayne@68: DeprecationWarning, jpayne@68: stacklevel=2 jpayne@68: ) jpayne@68: data = self.filelike.read(self.blksize) jpayne@68: if data: jpayne@68: return data jpayne@68: raise IndexError jpayne@68: jpayne@68: def __iter__(self): jpayne@68: return self jpayne@68: jpayne@68: def __next__(self): jpayne@68: data = self.filelike.read(self.blksize) jpayne@68: if data: jpayne@68: return data jpayne@68: raise StopIteration jpayne@68: jpayne@68: def guess_scheme(environ): jpayne@68: """Return a guess for whether 'wsgi.url_scheme' should be 'http' or 'https' jpayne@68: """ jpayne@68: if environ.get("HTTPS") in ('yes','on','1'): jpayne@68: return 'https' jpayne@68: else: jpayne@68: return 'http' jpayne@68: jpayne@68: def application_uri(environ): jpayne@68: """Return the application's base URI (no PATH_INFO or QUERY_STRING)""" jpayne@68: url = environ['wsgi.url_scheme']+'://' jpayne@68: from urllib.parse import quote jpayne@68: jpayne@68: if environ.get('HTTP_HOST'): jpayne@68: url += environ['HTTP_HOST'] jpayne@68: else: jpayne@68: url += environ['SERVER_NAME'] jpayne@68: jpayne@68: if environ['wsgi.url_scheme'] == 'https': jpayne@68: if environ['SERVER_PORT'] != '443': jpayne@68: url += ':' + environ['SERVER_PORT'] jpayne@68: else: jpayne@68: if environ['SERVER_PORT'] != '80': jpayne@68: url += ':' + environ['SERVER_PORT'] jpayne@68: jpayne@68: url += quote(environ.get('SCRIPT_NAME') or '/', encoding='latin1') jpayne@68: return url jpayne@68: jpayne@68: def request_uri(environ, include_query=True): jpayne@68: """Return the full request URI, optionally including the query string""" jpayne@68: url = application_uri(environ) jpayne@68: from urllib.parse import quote jpayne@68: path_info = quote(environ.get('PATH_INFO',''), safe='/;=,', encoding='latin1') jpayne@68: if not environ.get('SCRIPT_NAME'): jpayne@68: url += path_info[1:] jpayne@68: else: jpayne@68: url += path_info jpayne@68: if include_query and environ.get('QUERY_STRING'): jpayne@68: url += '?' + environ['QUERY_STRING'] jpayne@68: return url jpayne@68: jpayne@68: def shift_path_info(environ): jpayne@68: """Shift a name from PATH_INFO to SCRIPT_NAME, returning it jpayne@68: jpayne@68: If there are no remaining path segments in PATH_INFO, return None. jpayne@68: Note: 'environ' is modified in-place; use a copy if you need to keep jpayne@68: the original PATH_INFO or SCRIPT_NAME. jpayne@68: jpayne@68: Note: when PATH_INFO is just a '/', this returns '' and appends a trailing jpayne@68: '/' to SCRIPT_NAME, even though empty path segments are normally ignored, jpayne@68: and SCRIPT_NAME doesn't normally end in a '/'. This is intentional jpayne@68: behavior, to ensure that an application can tell the difference between jpayne@68: '/x' and '/x/' when traversing to objects. jpayne@68: """ jpayne@68: path_info = environ.get('PATH_INFO','') jpayne@68: if not path_info: jpayne@68: return None jpayne@68: jpayne@68: path_parts = path_info.split('/') jpayne@68: path_parts[1:-1] = [p for p in path_parts[1:-1] if p and p != '.'] jpayne@68: name = path_parts[1] jpayne@68: del path_parts[1] jpayne@68: jpayne@68: script_name = environ.get('SCRIPT_NAME','') jpayne@68: script_name = posixpath.normpath(script_name+'/'+name) jpayne@68: if script_name.endswith('/'): jpayne@68: script_name = script_name[:-1] jpayne@68: if not name and not script_name.endswith('/'): jpayne@68: script_name += '/' jpayne@68: jpayne@68: environ['SCRIPT_NAME'] = script_name jpayne@68: environ['PATH_INFO'] = '/'.join(path_parts) jpayne@68: jpayne@68: # Special case: '/.' on PATH_INFO doesn't get stripped, jpayne@68: # because we don't strip the last element of PATH_INFO jpayne@68: # if there's only one path part left. Instead of fixing this jpayne@68: # above, we fix it here so that PATH_INFO gets normalized to jpayne@68: # an empty string in the environ. jpayne@68: if name=='.': jpayne@68: name = None jpayne@68: return name jpayne@68: jpayne@68: def setup_testing_defaults(environ): jpayne@68: """Update 'environ' with trivial defaults for testing purposes jpayne@68: jpayne@68: This adds various parameters required for WSGI, including HTTP_HOST, jpayne@68: SERVER_NAME, SERVER_PORT, REQUEST_METHOD, SCRIPT_NAME, PATH_INFO, jpayne@68: and all of the wsgi.* variables. It only supplies default values, jpayne@68: and does not replace any existing settings for these variables. jpayne@68: jpayne@68: This routine is intended to make it easier for unit tests of WSGI jpayne@68: servers and applications to set up dummy environments. It should *not* jpayne@68: be used by actual WSGI servers or applications, since the data is fake! jpayne@68: """ jpayne@68: jpayne@68: environ.setdefault('SERVER_NAME','127.0.0.1') jpayne@68: environ.setdefault('SERVER_PROTOCOL','HTTP/1.0') jpayne@68: jpayne@68: environ.setdefault('HTTP_HOST',environ['SERVER_NAME']) jpayne@68: environ.setdefault('REQUEST_METHOD','GET') jpayne@68: jpayne@68: if 'SCRIPT_NAME' not in environ and 'PATH_INFO' not in environ: jpayne@68: environ.setdefault('SCRIPT_NAME','') jpayne@68: environ.setdefault('PATH_INFO','/') jpayne@68: jpayne@68: environ.setdefault('wsgi.version', (1,0)) jpayne@68: environ.setdefault('wsgi.run_once', 0) jpayne@68: environ.setdefault('wsgi.multithread', 0) jpayne@68: environ.setdefault('wsgi.multiprocess', 0) jpayne@68: jpayne@68: from io import StringIO, BytesIO jpayne@68: environ.setdefault('wsgi.input', BytesIO()) jpayne@68: environ.setdefault('wsgi.errors', StringIO()) jpayne@68: environ.setdefault('wsgi.url_scheme',guess_scheme(environ)) jpayne@68: jpayne@68: if environ['wsgi.url_scheme']=='http': jpayne@68: environ.setdefault('SERVER_PORT', '80') jpayne@68: elif environ['wsgi.url_scheme']=='https': jpayne@68: environ.setdefault('SERVER_PORT', '443') jpayne@68: jpayne@68: jpayne@68: jpayne@68: _hoppish = { jpayne@68: 'connection', 'keep-alive', 'proxy-authenticate', jpayne@68: 'proxy-authorization', 'te', 'trailers', 'transfer-encoding', jpayne@68: 'upgrade' jpayne@68: }.__contains__ jpayne@68: jpayne@68: def is_hop_by_hop(header_name): jpayne@68: """Return true if 'header_name' is an HTTP/1.1 "Hop-by-Hop" header""" jpayne@68: return _hoppish(header_name.lower())