def check_jwt_authorization():
current_identity = getattr(_request_ctx_stack.top,
'current_identity', None)
if current_identity:
return current_identity
skip_check = False
if current_app.config.get("disable_jwt", False):
skip_check = True
if request.endpoint in current_app.view_functions:
fn = current_app.view_functions[request.endpoint]
# Check Flask-RESTful endpoints for openness
if hasattr(fn, "view_class"):
exempt = getattr(fn.view_class, "no_jwt_check", [])
if request.method in exempt:
skip_check = True
elif fn in _open_endpoints:
skip_check = True
# the static folder is open to all without authentication
if request.endpoint == "static" or request.url.endswith("favicon.ico"):
skip_check = True
# In case the endpoint requires no authorization, and the request does not
# carry any authorization info as well, we will not try to verify any JWT's
if skip_check and 'Authorization' not in request.headers:
return
token, auth_type = get_auth_token_and_type()
current_identity = verify_token(token, auth_type)
if auth_type == "JWT":
# Cache this token
cache_token(current_identity)
# Authorization token has now been converted to a verified payload
_request_ctx_stack.top.current_identity = current_identity
return current_identity
python类top()的实例源码
def use_forwarded_port(graph):
"""
Inject the `X-Forwarded-Port` (if any) into the current URL adapter.
The URL adapter is used by `url_for` to build a URLs.
"""
# There must be a better way!
context = _request_ctx_stack.top
if _request_ctx_stack is None:
return None
# determine the configured overrides
forwarded_host = graph.config.port_forwarding.get("host")
forwarded_port = request.headers.get("X-Forwarded-Port")
if not forwarded_port and not forwarded_host:
return None
# determine the current server name
if ":" in context.url_adapter.server_name:
server_host, server_port = context.url_adapter.server_name.split(":", 1)
else:
server_host = context.url_adapter.server_name
server_port = 443 if context.url_adapter.url_scheme == "https" else 80
# choose a new server name
if forwarded_host:
server_name = forwarded_host
elif server_port:
server_name = "{}:{}".format(server_host, forwarded_port)
else:
server_name = "{}:{}".format(server_host, server_port)
context.url_adapter.server_name = server_name
return server_name
def teardown_request(self, exception):
ctx = _ctx_stack.top
if hasattr(ctx, "mysql_db"):
ctx.mysql_db.close()
def get_db(self):
ctx = _ctx_stack.top
if ctx is not None:
if not hasattr(ctx, "mysql_db"):
ctx.mysql_db = self.connect()
return ctx.mysql_db
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def _generate_request_arguments(url, spec, endpoint, headers, args, kwargs):
# Prepare (g)requests arguments
data = None
params = None
custom_url = url
if hasattr(stack.top, 'call_id'):
headers['KlueCallID'] = stack.top.call_id
if hasattr(stack.top, 'call_path'):
headers['KlueCallPath'] = stack.top.call_path
if endpoint.param_in_path:
# Fill url with values from kwargs, and remove those params from kwargs
custom_url = _format_flask_url(url, kwargs)
if endpoint.param_in_query:
# The query parameters are contained in **kwargs
params = kwargs
# TODO: validate params? or let the server do that...
elif endpoint.param_in_body:
# The body parameter is the first elem in *args
if len(args) != 1:
raise ValidationError("%s expects exactly 1 parameter" % endpoint.handler_client)
data = json.dumps(spec.model_to_json(args[0]))
# Prune undefined parameters that would otherwise be turned into '=None'
# query params
if params:
for k in list(params.keys()):
if params[k] is None:
del params[k]
return custom_url, params, data, headers
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def append_header(self, response):
'''
Adds WWW-Authenticate header with SPNEGO challenge or Kerberos token
'''
token = getattr(stack.top, 'kerberos_token', None)
if response.status_code == 401:
# Negotiate is an additional authenticate method.
response.headers.add('WWW-Authenticate', 'Negotiate')
elif token:
response.headers['WWW-Authenticate'] = 'Negotiate {}'.format(token)
return response
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def __exit__(self, exc_type, exc_value, tb):
self.preserve_context = False
# on exit we want to clean up earlier. Normally the request context
# stays preserved until the next request in the same thread comes
# in. See RequestGlobals.push() for the general behavior.
top = _request_ctx_stack.top
if top is not None and top.preserved:
top.pop()
def before_request(self):
ctx = _request_ctx_stack.top
ctx.mysql_db = self.connect()
def teardown_request(self, exception):
ctx = _request_ctx_stack.top
if hasattr(ctx, "mysql_db"):
ctx.mysql_db.close()
def get_db(self):
ctx = _request_ctx_stack.top
if ctx is not None:
return ctx.mysql_db
