You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Hi.
This Poorya.
I wrote an error handler for my Flask application. This application uses Flask JwT Extended and Flask Socket-io libraries for authentication real-time communication. My error handler doesn't work and gives an error.(@vimalloc, could you please take a look at this issue?)
My code:
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager, jwt_required, create_access_token, get_jwt_identity
from flask import Flask
from flask_socketio import SocketIO, emit, disconnect
from flask_jwt_extended.exceptions import (
NoAuthorizationError,
InvalidHeaderError,
)
# Initialize Flask app
app = Flask(__name__)
port = xxxx
host = "xxx.xxx.xxx.xxxx"
# Configurations
app.config['SECRET_KEY'] = "sdkjfh234"
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'sdkjfh234' # Change this to a random secret
app.config['PROPAGATE_EXCEPTIONS'] = True
# Initialize extensions
db = SQLAlchemy(app)
migrate = Migrate(app, db)
jwt = JWTManager(app)
socketio = SocketIO(app)
# User model
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password = db.Column(db.String(128), nullable=False)
# Route for user registration
@app.route('/signup', methods=['POST'])
def signup():
data = request.get_json()
username = data.get('username')
password = data.get('password')
# Check if username already exists
if User.query.filter_by(username=username).first():
return jsonify({"msg": "Username already exists"}), 400
# Create a new user
new_user = User(username=username, password=password) # Hash the password in a real app
db.session.add(new_user)
db.session.commit()
return jsonify({"msg": "User created successfully"}), 201
# Route for user login
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
username = data.get('username')
password = data.get('password')
user = User.query.filter_by(username=username).first()
# Check if user exists and password is correct
if user and user.password == password: # Hash comparison needed in a real app
access_token = create_access_token(identity={'username': user.username})
return jsonify(access_token=access_token), 200
return jsonify({"msg": "Bad username or password"}), 401
# Protected route
@app.route('/protected', methods=['GET'])
@jwt_required
def protected():
current_user = get_jwt_identity()
return jsonify(logged_in_as=current_user), 200
# @jwt.unauthorized_loader
# def handle_no_authorization_error(error):
# print(error)
# print("***************************************ERROR********************************")
# return jsonify({"error": "Permission Denied"}), 403
# @jwt.invalid_token_loader
# def handle_no_authorization_error(error):
# print(error)
# print("***************************************ERROR********************************")
# return jsonify({"error": "Permission Denied"}), 403
# @app.errorhandler(InvalidHeaderError)
# def handle_no_authorization_error(error):
# print(error)
# print("***************************************ERROR********************************")
# return jsonify({"error": "Permission Denied"}), 403
# Handle missing or invalid JWTs
@jwt.unauthorized_loader
def handle_no_authorization_error(error):
return jsonify({"error": "Permission Denied"}), 403
@jwt.invalid_token_loader
def handle_invalid_token_error(error):
return jsonify({"error": "Invalid Token"}), 403
# SocketIO event for authenticated connections
@socketio.on('connect')
@jwt_required()
def handle_connect():
try:
print("connected!!!")
identity = get_jwt_identity()
print(f"User {identity} connected")
emit('message', {'message': 'User connected'})
except Exception as e:
print(e)
disconnect()
return False
except jwt.invalid_token_loader as e:
print("Error:", e)
disconnect(namespace="message")
return jsonify({"Error": "Invalid Token!!!"}), 403
except jwt.token_verification_failed_loader as e:
print("Error:", e)
disconnect(namespace="message")
return jsonify({"Error": "Token Verification Failed!!!"}), 403
# Websocket on message
@socketio.on("message")
# @jwt_required()
def handle_message(message):
print("Recieved message:", message)
emit('message', message)
# Main entry point
if __name__ == '__main__':
# db.create_all() # Create database tables
# with app.app_context():
# app.run(debug=True, host=host, port=port)
socketio.run(app, host=host, port=port, debug=True)
And the error I got:
message handler error
Traceback (most recent call last):
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/engineio/server.py", line 434, in run_handler
return self.handlers[event](*args)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/socketio/server.py", line 643, in _handle_eio_message
self._handle_connect(eio_sid, pkt.namespace, pkt.data)
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/socketio/server.py", line 537, in _handle_connect
success = self._trigger_event(
^^^^^^^^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/socketio/server.py", line 614, in _trigger_event
return handler(*args)
^^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/flask_socketio/__init__.py", line 282, in _handler
return self._handle_event(handler, message, namespace, sid,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/flask_socketio/__init__.py", line 823, in _handle_event
ret = handler(auth)
^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/flask_jwt_extended/view_decorators.py", line 167, in decorator
verify_jwt_in_request(
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/flask_jwt_extended/view_decorators.py", line 94, in verify_jwt_in_request
jwt_data, jwt_header, jwt_location = _decode_jwt_from_request(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/root/Workspace/truck-sensors-dashboard/.env/lib/python3.12/site-packages/flask_jwt_extended/view_decorators.py", line 358, in _decode_jwt_from_request
raise NoAuthorizationError(errors[0])
flask_jwt_extended.exceptions.NoAuthorizationError: Missing Authorization Header
The text was updated successfully, but these errors were encountered:
Hi.
This Poorya.
I wrote an error handler for my Flask application. This application uses Flask JwT Extended and Flask Socket-io libraries for authentication real-time communication. My error handler doesn't work and gives an error.(@vimalloc, could you please take a look at this issue?)
My code:
And the error I got:
The text was updated successfully, but these errors were encountered: