def login_required()

in lemur/auth/service.py [0:0]

• 47 lines of code
• 15 McCabe index (conditional complexity)

def login_required(f):
    """
    Validates the JWT and ensures that is has not expired and the user is still active.

    :param f:
    :return:
    """

    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not request.headers.get("Authorization"):
            response = jsonify(message="Missing authorization header")
            response.status_code = 401
            return response

        try:
            token = request.headers.get("Authorization").split()[1]
        except Exception as e:
            return dict(message="Token is invalid"), 403
        token_secrets = current_app.config.get("LEMUR_TOKEN_SECRETS", [current_app.config["LEMUR_TOKEN_SECRET"]])
        try:
            header_data = fetch_token_header(token)
            payload = decode_with_multiple_secrets(token, token_secrets, algorithms=[header_data["alg"]])
        except jwt.DecodeError:
            return dict(message="Token is invalid"), 403
        except jwt.ExpiredSignatureError:
            return dict(message="Token has expired"), 403
        except jwt.InvalidTokenError:
            return dict(message="Token is invalid"), 403
        except Exception:  # noqa
            if current_app.config.get("DEBUG", False):
                raise
            return dict(message="Failed to decode token"), 403

        if "aid" in payload:
            access_key = api_key_service.get(payload["aid"])
            if access_key.revoked:
                return dict(message="Token has been revoked"), 403
            if access_key.ttl != -1:
                current_time = datetime.utcnow()
                # API key uses days
                expired_time = datetime.fromtimestamp(access_key.issued_at) + timedelta(days=access_key.ttl)
                if current_time >= expired_time:
                    return dict(message="Token has expired"), 403
            if access_key.application_name:
                g.caller_application = access_key.application_name

        user = user_service.get(payload["sub"])

        if not user.active:
            return dict(message="User is not currently active"), 403

        g.current_user = user

        if not g.current_user:
            return dict(message="You are not logged in"), 403

        # Tell Flask-Principal the identity changed
        identity_changed.send(
            current_app._get_current_object(), identity=Identity(g.current_user.id)
        )

        return f(*args, **kwargs)

    return decorated_function