Я использую Flask-JWT-Extended для защиты своего Flask API.После входа в систему по защищенному маршруту (add-user) я вызываю get_jwt_identity (), но он возвращает None, поэтому я не могу получить удостоверение.
@flask_app.route('/<api_version>/login', methods=['POST'])
def login(api_version):
print(f'login', request.form)
response = None
try:
username = request.form['username']
password = request.form['password']
if not username:
return jsonify({"msg": "Missing username parameter"}), 400
if not password:
return jsonify({"msg": "Missing password parameter"}), 400
user = User.get_with_password(username, password)
if (not user):
e1 = Unauthorized(
'Invalid username or password. Please try again.')
e1.status = 401
raise e1
""" flask_login.login_user(user, remember=True) """
access_token = create_access_token(identity=username)
response = json.dumps({"token": access_token}, cls=CustomJSONEncoder)
except Exception as e:
errMsg = f'Error Logging in user {username if username else ""}: {e}'
status = e.status if hasattr(e, 'status') else 500
print(f'{errMsg}')
traceback.print_exc()
return Response(
json.dumps({"message": errMsg, "status": status, "stack": traceback.format_exc() }), status=status, mimetype='application/json')
resp = Response(response, status=200, mimetype='application/json')
return resp
@flask_app.route('/<api_version>/add-user', methods=['POST'])
@jwt_required
def add_user(api_version):
print(f'add-user', request)
response = None
username = None
password = None
allow_admin = None
try:
data = request.get_json()
print(f'add-user data', data)
if 'username' in data:
username = data['username']
else:
return jsonify({"msg": "Missing username parameter"}), 400
if 'password' in data:
password = data['password']
else:
return jsonify({"msg": "Missing password parameter"}), 400
if 'allow_admin' in data:
allow_admin = data['allow_admin']
""" user = User.get_with_password(username, password)"""
user = get_jwt_identity()
print('user',user)
if (not user):
e1 = Unauthorized(
'Invalid username or password. Please try again.')
e1.status = 401
raise e1
response = json.dumps({"user": user}, cls=CustomJSONEncoder)
except Exception as e:
errMsg = f'Error Adding User {username}: {e}'
status = e.status if hasattr(e, 'status') else 500
print(f'{errMsg}')
traceback.print_exc()
return Response(
json.dumps({"message": errMsg, "status": status, "stack": traceback.format_exc() }), status=status, mimetype='application/json')
resp = Response(response, status=200, mimetype='application/json')
return resp
User.py
class User():
@classmethod
def get_with_password(cls, username, password):
print(f'User get_with_password {username} with password')
user_db = account.get_account(username)
print(f'User returned from DB: {user_db}')
user = User()
if not user_db or not len(user_db) or (not 'password' in user_db):
return None
user.username = username
user.id = username
if bcrypt.check_password_hash(user_db['password'], password):
user.role = user_db['role']
#user.is_authenticated = True
print(
f'loginUser returning {vars(user)} ')
return user
return None