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