How to disable CSRF in flask?
from ghodawalaaman@programming.dev to programming@programming.dev on 17 Mar 10:32
https://programming.dev/post/47338902
from ghodawalaaman@programming.dev to programming@programming.dev on 17 Mar 10:32
https://programming.dev/post/47338902
Hello,
it seems like an easy question but I tried everything google and AI told me but flask still giving me CSRF token mismatched error. I don’t know how to disable it. I threw everything I found online to disable CSRF but I can’t disable it. it’s so annoying. here is the code:
import mysql.connector
from mysql.connector import Error
from flask import Flask, request, jsonify,redirect, url_for
from authlib.integrations.flask_client import OAuth
import os
from flask_cors import CORS
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
# from flask_wtf.csrf import csrf_exempt
import hashlib
from flask import Flask
from flask_wtf import CSRFProtect
app = Flask(__name__)
app.config['WTF_CSRF_ENABLED'] = False # Disable CSRF globally
csrf = CSRFProtect(app) # This will now be disabled
try:
print("TESTING CONNECTION TO MYSQL DATABASE...")
connection = mysql.connector.connect(
host='localhost',
database='test',
user='root',
password='MySql@123'
)
if connection.is_connected():
print("Connected to MySQL database")
cur = connection.cursor()
cur.execute("SELECT DATABASE();")
record = cur.fetchone()
print("You're connected to database: ", record)
except Error as e:
print("Error while connecting to MySQL", e)
exit(1)
finally:
if connection.is_connected():
cur.close()
connection.close()
print("MySQL connection is closed")
print("TESTING DONE")
app.secret_key = "somethings_secret92387492837492387498"
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
app.config['SESSION_COOKIE_SECURE'] = False
app.config['SESSION_COOKIE_HTTPONLY'] = True
CORS(app)
app.config['JWT_SECRET_KEY'] = "your_jwt_secret_key123487236428374628374628736"
jwt = JWTManager(app)
# OAuth configuration
oauth = OAuth(app)
google = oauth.register(
name='google',
client_id="CLIENT_ID",
client_secret="CLIENT_SECRET",
server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
client_kwargs={
'scope': 'openid email profile'
}
)
@app.errorhandler(Exception)
def handle_exception(e):
return jsonify({"error": str(e)}), 500
@app.route("/",)
@jwt_required()
def hello_world():
return "<p>Hello, World!</p>"
@app.route("/register_by_email", methods=["POST"])
def register():
username = request.form.get("username")
email = request.form.get("email")
password = request.form.get("password")
with mysql.connector.connect(
host='localhost',
database='test',
user='root',
password='MySql@123'
) as connection:
with connection.cursor() as cursor:
cursor.execute("INSERT INTO users (username, email) VALUES (%s, %s)", (username, email))
cursor.execute("SELECT LAST_INSERT_ID()")
user_id = cursor.fetchone()[0]
password_hash = hashlib.sha256(password.encode()).hexdigest()
cursor.execute("INSERT INTO user_passwords (user_id, password_hash) VALUES (%s, %s)", (user_id, password_hash))
connection.commit()
return jsonify({"message": "User registered successfully", "user_id": user_id}), 201
@app.route("/login_by_email", methods=["POST"])
def login():
email = request.form.get("email")
password = request.form.get("password")
with mysql.connector.connect(
host='localhost',
database='test',
user='root',
password='MySql@123'
) as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT id FROM users WHERE email = %s", (email,))
user = cursor.fetchone()
if not user:
return jsonify({"error": "User not found"}), 404
user_id = user[0]
password_hash = hashlib.sha256(password.encode()).hexdigest()
cursor.execute("SELECT * FROM user_passwords WHERE user_id = %s AND password_hash = %s", (user_id, password_hash))
if cursor.fetchone():
return jsonify({"message": "Login successful", "user_id": user_id, "access_token": create_access_token(identity=email)}), 200
else:
return jsonify({"error": "Invalid credentials"}), 401
@app.route("/google_oauth_url",methods = ["GET"])
def login_with_google():
redirect_uri = url_for('callback', _external=True)
return google.create_authorization_url(redirect_uri)
@app.route("/callback",methods = ["GET"])
# @csrf_exempt
def callback():
token = google.authorize_access_token()
user_info = token.get("userinfo")
return jsonify(user_info)
if __name__ == "__main__":
app.run(debug=True)
#programming
threaded - newest