text: #!/usr/bin/env python3
  
  # Copyright 2014 Matthew Wright. All rights reserved.
  # Copyright 2019 Yannick Kirschen. All rights reserved.
  # Use of this source code is governed by the GNU-GPL
  # license that can be found in the LICENSE file.
  
  # Forked from https://github.com/mattupstate/flask-jwt
  
  import logging
  import warnings
  
  from collections import OrderedDict
  from datetime import datetime, timedelta
  from functools import wraps
  
  import jwt
  
  from flask import current_app, request, jsonify, _request_ctx_stack
  from werkzeug.local import LocalProxy
  
  logger = logging.getLogger(__name__)
  
  current_identity = LocalProxy(lambda: getattr(_request_ctx_stack.top, 'current_identity', None))
  
  _jwt = LocalProxy(lambda: current_app.extensions['jwt'])
  
  CONFIG_DEFAULTS = {
      'JWT_DEFAULT_REALM': 'Login Required',
      'JWT_AUTH_URL_RULE': '/auth',
      'JWT_AUTH_ENDPOINT': 'jwt',
      'JWT_AUTH_USERNAME_KEY': 'username',
      'JWT_AUTH_PASSWORD_KEY': 'password',
      'JWT_ALGORITHM': 'HS256',
      'JWT_LEEWAY': timedelta(seconds=10),
      'JWT_AUTH_HEADER_PREFIX': 'JWT',
      'JWT_EXPIRATION_DELTA': timedelta(seconds=300),
      'JWT_NOT_BEFORE_DELTA': timedelta(seconds=0),
      'JWT_VERIFY_CLAIMS': ['signature', 'exp', 'nbf', 'iat'],
      'JWT_REQUIRED_CLAIMS': ['exp', 'iat', 'nbf']
  }
  
  
  def _default_jwt_headers_handler(identity):
      return None
  
  
  def _default_jwt_payload_handler(identity):
      iat = datetime.utcnow()
      exp = iat + current_app.config.get('JWT_EXPIRATION_DELTA')
      nbf = iat + current_app.config.get('JWT_NOT_BEFORE_DELTA')
  
      try:
          identity = getattr(identity, 'id')
      except AttributeError:
          identity = identity['id']
  
      return {'exp': exp, 'iat': iat, 'nbf': nbf, 'identity': identity}
  
  
  def _default_jwt_encode_handler(identity):
      secret = current_app.config['JWT_SECRET_KEY']
      algorithm = current_app.config['JWT_ALGORITHM']
      required_claims = current_app.config['JWT_REQUIRED_CLAIMS']
  
      payload = _jwt.jwt_payload_callback(identity)
      missing_claims = list(set(required_claims) - set(payload.keys()))
  
      if missing_claims:
          raise RuntimeError('Payload is missing required claims: %s' % ', '.join(missing_claims))
  
      headers = _jwt.jwt_headers_callback(identity)
  
      return jwt.encode(payload, secret, algorithm=algorithm, headers=headers)
  
  
  def _default_jwt_decode_handler(token):
      secret = current_app.config['JWT_SECRET_KEY']
      algorithm = current_app.config['JWT_ALGORITHM']
      leeway = current_app.config['JWT_LEEWAY']
  
      verify_claims = current_app.config['JWT_VERIFY_CLAIMS']
      required_claims = current_app.config['JWT_REQUIRED_CLAIMS']
  
      options = {
          'verify_' + claim: True
          for claim in verify_claims
      }
  
      options.update({
          'require_' + claim: True
          for claim in required_claims
      })
  
      return jwt.decode(token, secret, options=options, algorithms=[algorithm], leeway=leeway)
  
  
  def _default_request_handler():
      auth_header_value = request.headers.get('Authorization', None)
      auth_header_prefix = current_app.config['JWT_AUTH_HEADER_PREFIX']
  
      if not auth_header_value:
          return
  
      parts = auth_header_value.split()
  
      if parts[0].lower() != auth_header_prefix.lower():
          raise JWTError('Invalid JWT header', 'Unsupported authorization type')
      elif len(parts) == 1:
          raise JWTError('Invalid JWT header', 'Token missing')
      elif len(parts) > 2:
          raise JWTError('Invalid JWT header', 'Token contains spaces')
  
      return parts[1]
  
  
  def _default_auth_request_handler():
      data = request.get_json()
      username = data.get(current_app.config.get('JWT_AUTH_USERNAME_KEY'), None)
      password = data.get(current_app.config.get('JWT_AUTH_PASSWORD_KEY'), None)
      criterion = [username, password, len(data) == 2]
  
      if not all(criterion):
          raise JWTError('Bad Request', 'Invalid credentials')
  
      identity = _jwt.authentication_callback(username, password)
  
      if identity:
          access_token = _jwt.jwt_encode_callback(identity)
          return _jwt.auth_response_callback(access_token, identity)
      else:
          raise JWTError('Bad Request', 'Invalid credentials')
  
  
  def _default_auth_response_handler(access_token, identity):
      return jsonify({'access_token': access_token.decode('utf-8')})
  
  
  def _default_jwt_error_handler(error):
      logger.error(error)
      return jsonify(OrderedDict([
          ('status_code', error.status_code),
          ('error', error.error),
          ('description', error.description),
      ])), error.status_code, error.headers
  
  
  def _jwt_required(realm):
      """Does the actual work of verifying the JWT data in the current request.
      This is done automatically for you by `jwt_required()` but you could call it manually.
      Doing so would be useful in the context of optional JWT access in your APIs.
  
      :param realm: an optional realm
      """
      token = _jwt.request_callback()
  
      if token is None:
          raise JWTError('Authorization Required', 'Request does not contain an access token',
                         headers={'WWW-Authenticate': 'JWT realm="%s"' % realm})
  
      try:
          payload = _jwt.jwt_decode_callback(token)
      except jwt.InvalidTokenError as e:
          raise JWTError('Invalid token', str(e))
  
      _request_ctx_stack.top.current_identity = identity = _jwt.identity_callback(payload)
  
      if identity is None:
          raise JWTError('Invalid JWT', 'User does not exist')
  
  
  def jwt_required(realm=None):
      """View decorator that requires a valid JWT token to be present in the request
  
      :param realm: an optional realm
      """
      def wrapper(fn):
          @wraps(fn)
          def decorator(*args, **kwargs):
              _jwt_required(realm or current_app.config['JWT_DEFAULT_REALM'])
              return fn(*args, **kwargs)
          return decorator
      return wrapper
  
  
  class JWTError(Exception):
      def __init__(self, error, description, status_code=401, headers=None):
          self.error = error
          self.description = description
          self.status_code = status_code
          self.headers = headers
  
      def __repr__(self):
          return 'JWTError: %s' % self.error
  
      def __str__(self):
          return '%s. %s' % (self.error, self.description)
  
  
  def encode_token():
      return _jwt.encode_callback(_jwt.header_callback(), _jwt.payload_callback())
  
  
  class JWT(object):
      def __init__(self, app=None, authentication_handler=None, identity_handler=None):
          self.authentication_callback = authentication_handler
          self.identity_callback = identity_handler
  
          self.auth_response_callback = _default_auth_response_handler
          self.auth_request_callback = _default_auth_request_handler
          self.jwt_encode_callback = _default_jwt_encode_handler
          self.jwt_decode_callback = _default_jwt_decode_handler
          self.jwt_headers_callback = _default_jwt_headers_handler
          self.jwt_payload_callback = _default_jwt_payload_handler
          self.jwt_error_callback = _default_jwt_error_handler
          self.request_callback = _default_request_handler
  
          if app is not None:
              self.init_app(app)
  
      def init_app(self, app):
          for k, v in CONFIG_DEFAULTS.items():
              app.config.setdefault(k, v)
          app.config.setdefault('JWT_SECRET_KEY', app.config['SECRET_KEY'])
  
          auth_url_rule = app.config.get('JWT_AUTH_URL_RULE', None)
  
          if auth_url_rule:
              if self.auth_request_callback == _default_auth_request_handler:
                  assert self.authentication_callback is not None, (
                      'an authentication_handler function must be defined when using the built in '
                      'authentication resource')
  
              auth_url_options = app.config.get('JWT_AUTH_URL_OPTIONS', {'methods': ['POST']})
              auth_url_options.setdefault('view_func', self.auth_request_callback)
              app.add_url_rule(auth_url_rule, **auth_url_options)
  
          app.errorhandler(JWTError)(self._jwt_error_callback)
  
          if not hasattr(app, 'extensions'):  # pragma: no cover
              app.extensions = {}
  
          app.extensions['jwt'] = self
  
      def _jwt_error_callback(self, error):
          return self.jwt_error_callback(error)
  
      def authentication_handler(self, callback):
          """Specifies the identity handler function. This function receives two positional
          arguments. The first being the username the second being the password. It should return an
          object representing an authenticated identity. Example::
  
              @jwt.authentication_handler
              def authenticate(username, password):
                  user = User.query.filter(User.username == username).scalar()
                  if bcrypt.check_password_hash(user.password, password):
                      return user
  
          :param callback: the identity handler function
          """
          self.authentication_callback = callback
          return callback
  
      def identity_handler(self, callback):
          """Specifies the identity handler function. This function receives one positional argument
          being the JWT payload. For example::
  
              @jwt.identity_handler
              def identify(payload):
                  return User.query.filter(User.id == payload['identity']).scalar()
  
          :param callback: the identity handler function
          """
          self.identity_callback = callback
          return callback
  
      def jwt_error_handler(self, callback):
          """Specifies the error handler function. Example::
  
              @jwt.error_handler
              def error_handler(e):
                  return "Something bad happened", 400
  
          :param callback: the error handler function
          """
          self.jwt_error_callback = callback
          return callback
  
      def auth_response_handler(self, callback):
          """Specifies the authentication response handler function.
  
          :param callable callback: the auth response handler function
          """
          self.auth_response_callback = callback
          return callback
  
      def auth_request_handler(self, callback):
          """Specifies the authentication response handler function.
  
          :param callable callback: the auth request handler function
  
          .. deprecated
          """
          warnings.warn("This handler is deprecated. The recommended approach to have control over "
                        "the authentication resource is to disable the built-in  resource by "
                        "setting JWT_AUTH_URL_RULE=None and registering your own authentication "
                        "resource directly on your application.", DeprecationWarning, stacklevel=2)
          self.auth_request_callback = callback
          return callback
  
      def request_handler(self, callback):
          """Specifieds the request handler function. This function returns a JWT from the current
          request.
  
          :param callable callback: the request handler function
          """
          self.request_callback = callback
          return callback
  
      def jwt_encode_handler(self, callback):
          """Specifies the encoding handler function. This function receives a payload and signs it.
  
          :param callable callback: the encoding handler function
          """
          self.jwt_encode_callback = callback
          return callback
  
      def jwt_decode_handler(self, callback):
          """Specifies the decoding handler function. This function receives a
          signed payload and decodes it.
  
          :param callable callback: the decoding handler function
          """
          self.jwt_decode_callback = callback
          return callback
  
      def jwt_payload_handler(self, callback):
          """Specifies the JWT payload handler function. This function receives the return value from
          the ``identity_handler`` function
  
          Example::
  
              @jwt.payload_handler
              def make_payload(identity):
                  return {'user_id': identity.id}
  
          :param callable callback: the payload handler function
          """
          self.jwt_payload_callback = callback
          return callback
  
      def jwt_headers_handler(self, callback):
          """Specifies the JWT header handler function. This function receives the return value from
          the ``identity_handler`` function.
  
          Example::
  
              @jwt.payload_handler
              def make_payload(identity):
                  return {'user_id': identity.id}
  
          :param callable callback: the payload handler function
          """
          self.jwt_headers_callback = callback
          return callback
  