validators.py 文件源码

def _get_user_from_sso(jwt_token, token):
        try:
            data = jwt.decode(jwt_token, Config.jwt_secret, algorithms=['HS256'])
            username = data["username"]
            name = data.get("firstName", username)
            surname = data.get("lastName", "")
            if username != token:
                BaseHandler.raise_exc(Forbidden, "FORBIDDEN", "Use the same username from the SSO")
            if Database.get_user(username) is None:
                Database.begin()
                Database.add_user(username, name, surname, sso_user=True, autocommit=False)
                for task in Database.get_tasks():
                    Database.add_user_task(username, task["name"], autocommit=False)
                Database.commit()
                Logger.info("NEW_USER", "User %s created from SSO" % username)
            return Database.get_user(username)
        except jwt.exceptions.DecodeError:
            BaseHandler.raise_exc(Forbidden, "FORBIDDEN", "Please login at %s" % Config.sso_url)