/usr/lib/python3/dist-packages/paste/errordocument.py is in python3-paste 2.0.3+dfsg-4ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 | # (c) 2005-2006 James Gardner <james@pythonweb.org>
# This module is part of the Python Paste Project and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
"""
Middleware to display error documents for certain status codes
The middleware in this module can be used to intercept responses with
specified status codes and internally forward the request to an appropriate
URL where the content can be displayed to the user as an error document.
"""
import warnings
import sys
from six.moves.urllib import parse as urlparse
from paste.recursive import ForwardRequestException, RecursiveMiddleware, RecursionLoop
from paste.util import converters
from paste.response import replace_header
import six
def forward(app, codes):
"""
Intercepts a response with a particular status code and returns the
content from a specified URL instead.
The arguments are:
``app``
The WSGI application or middleware chain.
``codes``
A dictionary of integer status codes and the URL to be displayed
if the response uses that code.
For example, you might want to create a static file to display a
"File Not Found" message at the URL ``/error404.html`` and then use
``forward`` middleware to catch all 404 status codes and display the page
you created. In this example ``app`` is your exisiting WSGI
applicaiton::
from paste.errordocument import forward
app = forward(app, codes={404:'/error404.html'})
"""
for code in codes:
if not isinstance(code, int):
raise TypeError('All status codes should be type int. '
'%s is not valid'%repr(code))
def error_codes_mapper(code, message, environ, global_conf, codes):
if code in codes:
return codes[code]
else:
return None
#return _StatusBasedRedirect(app, error_codes_mapper, codes=codes)
return RecursiveMiddleware(
StatusBasedForward(
app,
error_codes_mapper,
codes=codes,
)
)
class StatusKeeper(object):
def __init__(self, app, status, url, headers):
self.app = app
self.status = status
self.url = url
self.headers = headers
def __call__(self, environ, start_response):
def keep_status_start_response(status, headers, exc_info=None):
for header, value in headers:
if header.lower() == 'set-cookie':
self.headers.append((header, value))
else:
replace_header(self.headers, header, value)
return start_response(self.status, self.headers, exc_info)
parts = self.url.split('?')
environ['PATH_INFO'] = parts[0]
if len(parts) > 1:
environ['QUERY_STRING'] = parts[1]
else:
environ['QUERY_STRING'] = ''
#raise Exception(self.url, self.status)
try:
return self.app(environ, keep_status_start_response)
except RecursionLoop as e:
line = 'Recursion error getting error page: %s\n' % e
if six.PY3:
line = line.encode('utf8')
environ['wsgi.errors'].write(line)
keep_status_start_response('500 Server Error', [('Content-type', 'text/plain')], sys.exc_info())
body = ('Error: %s. (Error page could not be fetched)'
% self.status)
if six.PY3:
body = body.encode('utf8')
return [body]
class StatusBasedForward(object):
"""
Middleware that lets you test a response against a custom mapper object to
programatically determine whether to internally forward to another URL and
if so, which URL to forward to.
If you don't need the full power of this middleware you might choose to use
the simpler ``forward`` middleware instead.
The arguments are:
``app``
The WSGI application or middleware chain.
``mapper``
A callable that takes a status code as the
first parameter, a message as the second, and accepts optional environ,
global_conf and named argments afterwards. It should return a
URL to forward to or ``None`` if the code is not to be intercepted.
``global_conf``
Optional default configuration from your config file. If ``debug`` is
set to ``true`` a message will be written to ``wsgi.errors`` on each
internal forward stating the URL forwarded to.
``**params``
Optional, any other configuration and extra arguments you wish to
pass which will in turn be passed back to the custom mapper object.
Here is an example where a ``404 File Not Found`` status response would be
redirected to the URL ``/error?code=404&message=File%20Not%20Found``. This
could be useful for passing the status code and message into another
application to display an error document:
.. code-block:: python
from paste.errordocument import StatusBasedForward
from paste.recursive import RecursiveMiddleware
from urllib import urlencode
def error_mapper(code, message, environ, global_conf, kw)
if code in [404, 500]:
params = urlencode({'message':message, 'code':code})
url = '/error?'%(params)
return url
else:
return None
app = RecursiveMiddleware(
StatusBasedForward(app, mapper=error_mapper),
)
"""
def __init__(self, app, mapper, global_conf=None, **params):
if global_conf is None:
global_conf = {}
# @@: global_conf shouldn't really come in here, only in a
# separate make_status_based_forward function
if global_conf:
self.debug = converters.asbool(global_conf.get('debug', False))
else:
self.debug = False
self.application = app
self.mapper = mapper
self.global_conf = global_conf
self.params = params
def __call__(self, environ, start_response):
url = []
def change_response(status, headers, exc_info=None):
status_code = status.split(' ')
try:
code = int(status_code[0])
except (ValueError, TypeError):
raise Exception(
'StatusBasedForward middleware '
'received an invalid status code %s'%repr(status_code[0])
)
message = ' '.join(status_code[1:])
new_url = self.mapper(
code,
message,
environ,
self.global_conf,
**self.params
)
if not (new_url == None or isinstance(new_url, str)):
raise TypeError(
'Expected the url to internally '
'redirect to in the StatusBasedForward mapper'
'to be a string or None, not %r' % new_url)
if new_url:
url.append([new_url, status, headers])
# We have to allow the app to write stuff, even though
# we'll ignore it:
return [].append
else:
return start_response(status, headers, exc_info)
app_iter = self.application(environ, change_response)
if url:
if hasattr(app_iter, 'close'):
app_iter.close()
def factory(app):
return StatusKeeper(app, status=url[0][1], url=url[0][0],
headers=url[0][2])
raise ForwardRequestException(factory=factory)
else:
return app_iter
def make_errordocument(app, global_conf, **kw):
"""
Paste Deploy entry point to create a error document wrapper.
Use like::
[filter-app:main]
use = egg:Paste#errordocument
next = real-app
500 = /lib/msg/500.html
404 = /lib/msg/404.html
"""
map = {}
for status, redir_loc in kw.items():
try:
status = int(status)
except ValueError:
raise ValueError('Bad status code: %r' % status)
map[status] = redir_loc
forwarder = forward(app, map)
return forwarder
__pudge_all__ = [
'forward',
'make_errordocument',
'empty_error',
'make_empty_error',
'StatusBasedForward',
]
###############################################################################
## Deprecated
###############################################################################
def custom_forward(app, mapper, global_conf=None, **kw):
"""
Deprectated; use StatusBasedForward instead.
"""
warnings.warn(
"errordocuments.custom_forward has been deprecated; please "
"use errordocuments.StatusBasedForward",
DeprecationWarning, 2)
if global_conf is None:
global_conf = {}
return _StatusBasedRedirect(app, mapper, global_conf, **kw)
class _StatusBasedRedirect(object):
"""
Deprectated; use StatusBasedForward instead.
"""
def __init__(self, app, mapper, global_conf=None, **kw):
warnings.warn(
"errordocuments._StatusBasedRedirect has been deprecated; please "
"use errordocuments.StatusBasedForward",
DeprecationWarning, 2)
if global_conf is None:
global_conf = {}
self.application = app
self.mapper = mapper
self.global_conf = global_conf
self.kw = kw
self.fallback_template = """
<html>
<head>
<title>Error %(code)s</title>
</html>
<body>
<h1>Error %(code)s</h1>
<p>%(message)s</p>
<hr>
<p>
Additionally an error occurred trying to produce an
error document. A description of the error was logged
to <tt>wsgi.errors</tt>.
</p>
</body>
</html>
"""
def __call__(self, environ, start_response):
url = []
code_message = []
try:
def change_response(status, headers, exc_info=None):
new_url = None
parts = status.split(' ')
try:
code = int(parts[0])
except (ValueError, TypeError):
raise Exception(
'_StatusBasedRedirect middleware '
'received an invalid status code %s'%repr(parts[0])
)
message = ' '.join(parts[1:])
new_url = self.mapper(
code,
message,
environ,
self.global_conf,
self.kw
)
if not (new_url == None or isinstance(new_url, str)):
raise TypeError(
'Expected the url to internally '
'redirect to in the _StatusBasedRedirect error_mapper'
'to be a string or None, not %s'%repr(new_url)
)
if new_url:
url.append(new_url)
code_message.append([code, message])
return start_response(status, headers, exc_info)
app_iter = self.application(environ, change_response)
except:
try:
import sys
error = str(sys.exc_info()[1])
except:
error = ''
try:
code, message = code_message[0]
except:
code, message = ['', '']
environ['wsgi.errors'].write(
'Error occurred in _StatusBasedRedirect '
'intercepting the response: '+str(error)
)
return [self.fallback_template
% {'message': message, 'code': code}]
else:
if url:
url_ = url[0]
new_environ = {}
for k, v in environ.items():
if k != 'QUERY_STRING':
new_environ['QUERY_STRING'] = urlparse.urlparse(url_)[4]
else:
new_environ[k] = v
class InvalidForward(Exception):
pass
def eat_start_response(status, headers, exc_info=None):
"""
We don't want start_response to do anything since it
has already been called
"""
if status[:3] != '200':
raise InvalidForward(
"The URL %s to internally forward "
"to in order to create an error document did not "
"return a '200' status code." % url_
)
forward = environ['paste.recursive.forward']
old_start_response = forward.start_response
forward.start_response = eat_start_response
try:
app_iter = forward(url_, new_environ)
except InvalidForward:
code, message = code_message[0]
environ['wsgi.errors'].write(
'Error occurred in '
'_StatusBasedRedirect redirecting '
'to new URL: '+str(url[0])
)
return [
self.fallback_template%{
'message':message,
'code':code,
}
]
else:
forward.start_response = old_start_response
return app_iter
else:
return app_iter
|