lint.py (9464B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\werkzeug\contrib\lint.py 2 from urlparse import urlparse 3 from warnings import warn 4 from werkzeug import Headers, FileWrapper, is_entity_header 5 6 class WSGIWarning(Warning): 7 pass 8 9 10 class HTTPWarning(Warning): 11 pass 12 13 14 def check_string(context, obj, stacklevel = 3): 15 if type(obj) is not str: 16 warn(WSGIWarning('%s requires bytestrings, got %s' % (context, obj.__class__.__name__))) 17 18 19 class InputStream(object): 20 21 def __init__(self, stream): 22 self._stream = stream 23 24 def read(self, *args): 25 if len(args) == 0: 26 warn(WSGIWarning('wsgi does not guarantee an EOF marker on the input stream, thus making calls to wsgi.input.read() unsafe. Conforming servers may never return from this call.'), stacklevel=2) 27 elif len(args) != 1: 28 warn(WSGIWarning('too many parameters passed to wsgi.input.read()'), stacklevel=2) 29 return self._stream.read(*args) 30 31 def readline(self, *args): 32 if len(args) == 0: 33 warn(WSGIWarning('Calls to wsgi.input.readline() without arguments are unsafe. Use wsgi.input.read() instead.'), stacklevel=2) 34 elif len(args) == 1: 35 warn(WSGIWarning("wsgi.input.readline() was called with a size hint. WSGI does not support this, although it's available on all major servers."), stacklevel=2) 36 else: 37 raise TypeError('too many arguments passed to wsgi.input.readline()') 38 return self._stream.readline(*args) 39 40 def __iter__(self): 41 try: 42 return iter(self._stream) 43 except TypeError: 44 warn(WSGIWarning('wsgi.input is not iterable.'), stacklevel=2) 45 return iter(()) 46 47 def close(self): 48 warn(WSGIWarning('application closed the input stream!'), stacklevel=2) 49 self._stream.close() 50 51 52 class ErrorStream(object): 53 54 def __init__(self, stream): 55 self._stream = stream 56 57 def write(self, s): 58 check_string('wsgi.error.write()', s) 59 self._stream.write(s) 60 61 def flush(self): 62 self._stream.flush() 63 64 def writelines(self, seq): 65 for line in seq: 66 self.write(seq) 67 68 def close(self): 69 warn(WSGIWarning('application closed the error stream!'), stacklevel=2) 70 self._stream.close() 71 72 73 class GuardedWrite(object): 74 75 def __init__(self, write, chunks): 76 self._write = write 77 self._chunks = chunks 78 79 def __call__(self, s): 80 check_string('write()', s) 81 self._write.write(s) 82 self._chunks.append(len(s)) 83 84 85 class GuardedIterator(object): 86 87 def __init__(self, iterator, headers_set, chunks): 88 self._iterator = iterator 89 self._next = iter(iterator).next 90 self.closed = False 91 self.headers_set = headers_set 92 self.chunks = chunks 93 94 def __iter__(self): 95 return self 96 97 def next(self): 98 if self.closed: 99 warn(WSGIWarning('iterated over closed app_iter'), stacklevel=2) 100 rv = self._next() 101 if not self.headers_set: 102 warn(WSGIWarning('Application returned before it started the response'), stacklevel=2) 103 check_string('application iterator items', rv) 104 self.chunks.append(len(rv)) 105 return rv 106 107 def close(self): 108 self.closed = True 109 if hasattr(self._iterator, 'close'): 110 self._iterator.close() 111 if self.headers_set: 112 status_code, headers = self.headers_set 113 bytes_sent = sum(self.chunks) 114 content_length = headers.get('content-length', type=int) 115 if status_code == 304: 116 for key, value in headers: 117 key = key.lower() 118 if key not in ('expires', 'content-location') and is_entity_header(key): 119 warn(HTTPWarning('entity header %r found in 304 response' % key)) 120 121 if bytes_sent: 122 warn(HTTPWarning('304 responses must not have a body')) 123 elif 100 <= status_code < 200 or status_code == 204: 124 if content_length != 0: 125 warn(HTTPWarning('%r responses must have an empty content length') % status_code) 126 if bytes_sent: 127 warn(HTTPWarning('%r responses must not have a body' % status_code)) 128 elif content_length is not None and content_length != bytes_sent: 129 warn(WSGIWarning('Content-Length and the number of bytes sent to the client do not match.')) 130 131 def __del__(self): 132 if not self.closed: 133 try: 134 warn(WSGIWarning('Iterator was garbage collected before it was closed.')) 135 except: 136 pass 137 138 139 class LintMiddleware(object): 140 141 def __init__(self, app): 142 self.app = app 143 144 def check_environ(self, environ): 145 if type(environ) is not dict: 146 warn(WSGIWarning('WSGI environment is not a standard python dict.'), stacklevel=4) 147 for key in ('REQUEST_METHOD', 'SERVER_NAME', 'SERVER_PORT', 'wsgi.version', 'wsgi.input', 'wsgi.errors', 'wsgi.multithread', 'wsgi.multiprocess', 'wsgi.run_once'): 148 if key not in environ: 149 warn(WSGIWarning('required environment key %r not found' % key), stacklevel=3) 150 151 if environ['wsgi.version'] != (1, 0): 152 warn(WSGIWarning('environ is not a WSGI 1.0 environ'), stacklevel=3) 153 script_name = environ.get('SCRIPT_NAME', '') 154 if script_name and script_name[:1] != '/': 155 warn(WSGIWarning('SCRIPT_NAME does not start with a slash: %r' % script_name), stacklevel=3) 156 path_info = environ.get('PATH_INFO', '') 157 if path_info[:1] != '/': 158 warn(WSGIWarning('PATH_INFO does not start with a slash: %r' % path_info), stacklevel=3) 159 160 def check_start_response(self, status, headers, exc_info): 161 check_string('status', status) 162 status_code = status.split(None, 1)[0] 163 if len(status_code) != 3 or not status_code.isdigit(): 164 warn(WSGIWarning('Status code must be three digits'), stacklevel=3) 165 if len(status) < 4 or status[3] != ' ': 166 warn(WSGIWarning('Invalid value for status %r. Valid status strings are three digits, a space and a status explanation'), stacklevel=3) 167 status_code = int(status_code) 168 if status_code < 100: 169 warn(WSGIWarning('status code < 100 detected'), stacklevel=3) 170 if type(headers) is not list: 171 warn(WSGIWarning('header list is not a list'), stacklevel=3) 172 for item in headers: 173 if type(item) is not tuple or len(item) != 2: 174 warn(WSGIWarning('Headers must tuple 2-item tuples'), stacklevel=3) 175 name, value = item 176 if type(name) is not str or type(value) is not str: 177 warn(WSGIWarning('header items must be strings'), stacklevel=3) 178 if name.lower() == 'status': 179 warn(WSGIWarning('The status header is not supported due to conflicts with the CGI spec.'), stacklevel=3) 180 181 if exc_info is not None and not isinstance(exc_info, tuple): 182 warn(WSGIWarning('invalid value for exc_info'), stacklevel=3) 183 headers = Headers(headers) 184 self.check_headers(headers) 185 return (status_code, headers) 186 187 def check_headers(self, headers): 188 etag = headers.get('etag') 189 if etag is not None: 190 if etag.startswith('w/'): 191 etag = etag[2:] 192 if not etag[:1] == etag[-1:] == '"': 193 warn(HTTPWarning('unquoted etag emitted.'), stacklevel=4) 194 location = headers.get('location') 195 if location is not None: 196 if not urlparse(location).netloc: 197 warn(HTTPWarning('absolute URLs required for location header'), stacklevel=4) 198 199 def check_iterator(self, app_iter): 200 if isinstance(app_iter, basestring): 201 warn(WSGIWarning('application returned string. Response will send character for character to the client which will kill the performance. Return a list or iterable instead.'), stacklevel=3) 202 203 def __call__(self, *args, **kwargs): 204 if len(args) != 2: 205 warn(WSGIWarning('Two arguments to WSGI app required'), stacklevel=2) 206 if kwargs: 207 warn(WSGIWarning('No keyword arguments to WSGI app allowed'), stacklevel=2) 208 environ, start_response = args 209 self.check_environ(environ) 210 environ['wsgi.input'] = InputStream(environ['wsgi.input']) 211 environ['wsgi.errors'] = ErrorStream(environ['wsgi.errors']) 212 environ['wsgi.file_wrapper'] = FileWrapper 213 headers_set = [] 214 chunks = [] 215 216 def checking_start_response(*args, **kwargs): 217 if len(args) not in (2, 3): 218 warn(WSGIWarning('Invalid number of arguments: %s, expected 2 or 3' % len(args), stacklevel=2)) 219 if kwargs: 220 warn(WSGIWarning('no keyword arguments allowed.')) 221 status, headers = args[:2] 222 if len(args) == 3: 223 exc_info = args[2] 224 else: 225 exc_info = None 226 headers_set[:] = self.check_start_response(status, headers, exc_info) 227 return GuardedWrite(start_response(status, headers, exc_info), chunks) 228 229 app_iter = self.app(environ, checking_start_response) 230 self.check_iterator(app_iter) 231 return GuardedIterator(app_iter, headers_set, chunks)