"""Miscellaneous WSGI-related Utilities"""

[0] Fix | Delete

[1] Fix | Delete

import posixpath

[2] Fix | Delete

[3] Fix | Delete

__all__ = [

[4] Fix | Delete

'FileWrapper', 'guess_scheme', 'application_uri', 'request_uri',

[5] Fix | Delete

'shift_path_info', 'setup_testing_defaults',

[6] Fix | Delete

]

[7] Fix | Delete

[8] Fix | Delete

[9] Fix | Delete

class FileWrapper:

[10] Fix | Delete

"""Wrapper to convert file-like objects to iterables"""

[11] Fix | Delete

[12] Fix | Delete

def __init__(self, filelike, blksize=8192):

[13] Fix | Delete

self.filelike = filelike

[14] Fix | Delete

self.blksize = blksize

[15] Fix | Delete

if hasattr(filelike,'close'):

[16] Fix | Delete

self.close = filelike.close

[17] Fix | Delete

[18] Fix | Delete

def __getitem__(self,key):

[19] Fix | Delete

import warnings

[20] Fix | Delete

warnings.warn(

[21] Fix | Delete

"FileWrapper's __getitem__ method ignores 'key' parameter. "

[22] Fix | Delete

"Use iterator protocol instead.",

[23] Fix | Delete

DeprecationWarning,

[24] Fix | Delete

stacklevel=2

[25] Fix | Delete

)

[26] Fix | Delete

data = self.filelike.read(self.blksize)

[27] Fix | Delete

if data:

[28] Fix | Delete

return data

[29] Fix | Delete

raise IndexError

[30] Fix | Delete

[31] Fix | Delete

def __iter__(self):

[32] Fix | Delete

return self

[33] Fix | Delete

[34] Fix | Delete

def __next__(self):

[35] Fix | Delete

data = self.filelike.read(self.blksize)

[36] Fix | Delete

if data:

[37] Fix | Delete

return data

[38] Fix | Delete

raise StopIteration

[39] Fix | Delete

[40] Fix | Delete

def guess_scheme(environ):

[41] Fix | Delete

"""Return a guess for whether 'wsgi.url_scheme' should be 'http' or 'https'

[42] Fix | Delete

"""

[43] Fix | Delete

if environ.get("HTTPS") in ('yes','on','1'):

[44] Fix | Delete

return 'https'

[45] Fix | Delete

else:

[46] Fix | Delete

return 'http'

[47] Fix | Delete

[48] Fix | Delete

def application_uri(environ):

[49] Fix | Delete

"""Return the application's base URI (no PATH_INFO or QUERY_STRING)"""

[50] Fix | Delete

url = environ['wsgi.url_scheme']+'://'

[51] Fix | Delete

from urllib.parse import quote

[52] Fix | Delete

[53] Fix | Delete

if environ.get('HTTP_HOST'):

[54] Fix | Delete

url += environ['HTTP_HOST']

[55] Fix | Delete

else:

[56] Fix | Delete

url += environ['SERVER_NAME']

[57] Fix | Delete

[58] Fix | Delete

if environ['wsgi.url_scheme'] == 'https':

[59] Fix | Delete

if environ['SERVER_PORT'] != '443':

[60] Fix | Delete

url += ':' + environ['SERVER_PORT']

[61] Fix | Delete

else:

[62] Fix | Delete

if environ['SERVER_PORT'] != '80':

[63] Fix | Delete

url += ':' + environ['SERVER_PORT']

[64] Fix | Delete

[65] Fix | Delete

url += quote(environ.get('SCRIPT_NAME') or '/', encoding='latin1')

[66] Fix | Delete

return url

[67] Fix | Delete

[68] Fix | Delete

def request_uri(environ, include_query=True):

[69] Fix | Delete

"""Return the full request URI, optionally including the query string"""

[70] Fix | Delete

url = application_uri(environ)

[71] Fix | Delete

from urllib.parse import quote

[72] Fix | Delete

path_info = quote(environ.get('PATH_INFO',''), safe='/;=,', encoding='latin1')

[73] Fix | Delete

if not environ.get('SCRIPT_NAME'):

[74] Fix | Delete

url += path_info[1:]

[75] Fix | Delete

else:

[76] Fix | Delete

url += path_info

[77] Fix | Delete

if include_query and environ.get('QUERY_STRING'):

[78] Fix | Delete

url += '?' + environ['QUERY_STRING']

[79] Fix | Delete

return url

[80] Fix | Delete

[81] Fix | Delete

def shift_path_info(environ):

[82] Fix | Delete

"""Shift a name from PATH_INFO to SCRIPT_NAME, returning it

[83] Fix | Delete

[84] Fix | Delete

If there are no remaining path segments in PATH_INFO, return None.

[85] Fix | Delete

Note: 'environ' is modified in-place; use a copy if you need to keep

[86] Fix | Delete

the original PATH_INFO or SCRIPT_NAME.

[87] Fix | Delete

[88] Fix | Delete

Note: when PATH_INFO is just a '/', this returns '' and appends a trailing

[89] Fix | Delete

'/' to SCRIPT_NAME, even though empty path segments are normally ignored,

[90] Fix | Delete

and SCRIPT_NAME doesn't normally end in a '/'. This is intentional

[91] Fix | Delete

behavior, to ensure that an application can tell the difference between

[92] Fix | Delete

'/x' and '/x/' when traversing to objects.

[93] Fix | Delete

"""

[94] Fix | Delete

path_info = environ.get('PATH_INFO','')

[95] Fix | Delete

if not path_info:

[96] Fix | Delete

return None

[97] Fix | Delete

[98] Fix | Delete

path_parts = path_info.split('/')

[99] Fix | Delete

path_parts[1:-1] = [p for p in path_parts[1:-1] if p and p != '.']

[100] Fix | Delete

name = path_parts[1]

[101] Fix | Delete

del path_parts[1]

[102] Fix | Delete

[103] Fix | Delete

script_name = environ.get('SCRIPT_NAME','')

[104] Fix | Delete

script_name = posixpath.normpath(script_name+'/'+name)

[105] Fix | Delete

if script_name.endswith('/'):

[106] Fix | Delete

script_name = script_name[:-1]

[107] Fix | Delete

if not name and not script_name.endswith('/'):

[108] Fix | Delete

script_name += '/'

[109] Fix | Delete

[110] Fix | Delete

environ['SCRIPT_NAME'] = script_name

[111] Fix | Delete

environ['PATH_INFO'] = '/'.join(path_parts)

[112] Fix | Delete

[113] Fix | Delete

# Special case: '/.' on PATH_INFO doesn't get stripped,

[114] Fix | Delete

# because we don't strip the last element of PATH_INFO

[115] Fix | Delete

# if there's only one path part left. Instead of fixing this

[116] Fix | Delete

# above, we fix it here so that PATH_INFO gets normalized to

[117] Fix | Delete

# an empty string in the environ.

[118] Fix | Delete

if name=='.':

[119] Fix | Delete

name = None

[120] Fix | Delete

return name

[121] Fix | Delete

[122] Fix | Delete

def setup_testing_defaults(environ):

[123] Fix | Delete

"""Update 'environ' with trivial defaults for testing purposes

[124] Fix | Delete

[125] Fix | Delete

This adds various parameters required for WSGI, including HTTP_HOST,

[126] Fix | Delete

SERVER_NAME, SERVER_PORT, REQUEST_METHOD, SCRIPT_NAME, PATH_INFO,

[127] Fix | Delete

and all of the wsgi.* variables. It only supplies default values,

[128] Fix | Delete

and does not replace any existing settings for these variables.

[129] Fix | Delete

[130] Fix | Delete

This routine is intended to make it easier for unit tests of WSGI

[131] Fix | Delete

servers and applications to set up dummy environments. It should *not*

[132] Fix | Delete

be used by actual WSGI servers or applications, since the data is fake!

[133] Fix | Delete

"""

[134] Fix | Delete

[135] Fix | Delete

environ.setdefault('SERVER_NAME','127.0.0.1')

[136] Fix | Delete

environ.setdefault('SERVER_PROTOCOL','HTTP/1.0')

[137] Fix | Delete

[138] Fix | Delete

environ.setdefault('HTTP_HOST',environ['SERVER_NAME'])

[139] Fix | Delete

environ.setdefault('REQUEST_METHOD','GET')

[140] Fix | Delete

[141] Fix | Delete

if 'SCRIPT_NAME' not in environ and 'PATH_INFO' not in environ:

[142] Fix | Delete

environ.setdefault('SCRIPT_NAME','')

[143] Fix | Delete

environ.setdefault('PATH_INFO','/')

[144] Fix | Delete

[145] Fix | Delete

environ.setdefault('wsgi.version', (1,0))

[146] Fix | Delete

environ.setdefault('wsgi.run_once', 0)

[147] Fix | Delete

environ.setdefault('wsgi.multithread', 0)

[148] Fix | Delete

environ.setdefault('wsgi.multiprocess', 0)

[149] Fix | Delete

[150] Fix | Delete

from io import StringIO, BytesIO

[151] Fix | Delete

environ.setdefault('wsgi.input', BytesIO())

[152] Fix | Delete

environ.setdefault('wsgi.errors', StringIO())

[153] Fix | Delete

environ.setdefault('wsgi.url_scheme',guess_scheme(environ))

[154] Fix | Delete

[155] Fix | Delete

if environ['wsgi.url_scheme']=='http':

[156] Fix | Delete

environ.setdefault('SERVER_PORT', '80')

[157] Fix | Delete

elif environ['wsgi.url_scheme']=='https':

[158] Fix | Delete

environ.setdefault('SERVER_PORT', '443')

[159] Fix | Delete

[160] Fix | Delete

[161] Fix | Delete

[162] Fix | Delete

_hoppish = {

[163] Fix | Delete

'connection', 'keep-alive', 'proxy-authenticate',

[164] Fix | Delete

'proxy-authorization', 'te', 'trailers', 'transfer-encoding',

[165] Fix | Delete

'upgrade'

[166] Fix | Delete

}.__contains__

[167] Fix | Delete

[168] Fix | Delete

def is_hop_by_hop(header_name):

[169] Fix | Delete

"""Return true if 'header_name' is an HTTP/1.1 "Hop-by-Hop" header"""

[170] Fix | Delete

return _hoppish(header_name.lower())

[171] Fix | Delete

[172] Fix | Delete