/usr/lib/python2.7/dist-packages/flower/views/__init__.py is in python-flower 0.8.3+dfsg-3.
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 | from __future__ import absolute_import
import re
import inspect
import traceback
from distutils.util import strtobool
from base64 import b64decode
import tornado
from ..utils import template, bugreport
class BaseHandler(tornado.web.RequestHandler):
def render(self, *args, **kwargs):
functions = self._get_template_functions()
assert not set(map(lambda x: x[0], functions)) & set(kwargs.keys())
kwargs.update(functions)
super(BaseHandler, self).render(*args, **kwargs)
def write_error(self, status_code, **kwargs):
if status_code in (404, 403):
message = None
if 'exc_info' in kwargs and\
kwargs['exc_info'][0] == tornado.web.HTTPError:
message = kwargs['exc_info'][1].log_message
self.render('404.html', message=message)
elif status_code == 500:
error_trace = ""
for line in traceback.format_exception(*kwargs['exc_info']):
error_trace += line
self.render('error.html',
status_code=status_code,
error_trace=error_trace,
bugreport=bugreport())
elif status_code == 401:
self.set_status(status_code)
self.set_header('WWW-Authenticate', 'Basic realm="flower"')
self.finish('Access denied')
else:
message = None
if 'exc_info' in kwargs and\
kwargs['exc_info'][0] == tornado.web.HTTPError:
message = kwargs['exc_info'][1].log_message
self.set_header('Content-Type', 'text/plain')
self.write(message)
self.set_status(status_code)
def get_current_user(self):
# Basic Auth
basic_auth = self.application.options.basic_auth
if basic_auth:
auth_header = self.request.headers.get("Authorization", "")
try:
basic, credentials = auth_header.split()
credentials = b64decode(credentials.encode()).decode()
if basic != 'Basic' or credentials not in basic_auth:
raise tornado.web.HTTPError(401)
except ValueError:
raise tornado.web.HTTPError(401)
# Google OpenID
if not self.application.options.auth:
return True
user = self.get_secure_cookie('user')
if user:
if not isinstance(user, str):
user = user.decode()
if re.search(self.application.options.auth, user):
return user
return None
def get_argument(self, name, default=[], strip=True, type=None):
arg = super(BaseHandler, self).get_argument(name, default, strip)
if type is not None:
try:
if type is bool:
arg = strtobool(str(arg))
else:
arg = type(arg)
except (ValueError, TypeError):
if arg is None and default is None:
return arg
raise tornado.web.HTTPError(
400,
"Invalid argument '%s' of type '%s'" % (
arg, type.__name__))
return arg
@property
def capp(self):
"return Celery application object"
return self.application.capp
@staticmethod
def _get_template_functions():
return inspect.getmembers(template, inspect.isfunction)
|