fix(csrf): resolve all 124 SonarQube CSRF security hotspots

server/routes/atlassian/atlassian_routes.py

@@ -117,7 +117,7 @@ def _validate_jsm_ops(access_token: str, cloud_id: str) -> Optional[Dict[str, An
 # POST /atlassian/connect
 # ------------------------------------------------------------------
 
-@atlassian_bp.route("/connect", methods=["POST", "OPTIONS"])
+@atlassian_bp.route("/connect", methods=["POST"])
 @require_permission("connectors", "write")
 def connect(user_id):
     """Unified connect for Atlassian products (Confluence/Jira/both)."""
@@ -363,7 +363,7 @@ def status(user_id):
 # POST /atlassian/disconnect
 # ------------------------------------------------------------------
 
-@atlassian_bp.route("/disconnect", methods=["POST", "OPTIONS"])
+@atlassian_bp.route("/disconnect", methods=["POST"])
 @require_permission("connectors", "write")
 def disconnect(user_id):
     """Disconnect one or all Atlassian products."""

server/routes/auth_routes.py

@@ -10,7 +10,6 @@
 from utils.db.db_utils import connect_to_db_as_user
 from utils.db.connection_pool import db_pool
 from utils.auth.rbac_decorators import require_auth_only
-from utils.web.cors_utils import create_cors_response
 import os
 
 auth_bp = Blueprint('auth', __name__, url_prefix='/api/auth')
@@ -39,7 +38,7 @@ def add_cors_headers(response):
     response.headers['Access-Control-Allow-Headers'] = 'Content-Type, X-Provider, X-Requested-With, X-User-ID, Authorization'
     return response
 
-@auth_bp.route('/register', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/register', methods=['POST'])
 def register():
     """Register a new organization with its first admin user.
 
@@ -48,8 +47,6 @@ def register():
     - Users within an existing org are created by an admin via
       /api/admin/users (invite-only).
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
 
     try:
         data = request.get_json()
@@ -162,15 +159,13 @@ def register():
         return jsonify({"error": "Registration failed"}), 500
 
 
-@auth_bp.route('/setup-org', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/setup-org', methods=['POST'])
 @require_auth_only
 def setup_org(user_id):
     """Create an organization for an authenticated user who doesn't have one.
 
     Body: { org_name }
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
     try:
         data = request.get_json()
         if not data:
@@ -273,12 +268,9 @@ def setup_org(user_id):
         return jsonify({"error": "Organization setup failed"}), 500
 
 
-@auth_bp.route('/login', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/login', methods=['POST'])
 def login():
     """Authenticate user with email and password."""
-    if request.method == 'OPTIONS':
-        return create_cors_response()
-
     try:
         data = request.get_json()
         if not data:
@@ -339,12 +331,10 @@ def login():
         return jsonify({"error": "Login failed"}), 500
 
 
-@auth_bp.route('/change-password', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/change-password', methods=['POST'])
 @require_auth_only
 def change_password(user_id):
     """Change user password (requires authentication)."""
-    if request.method == 'OPTIONS':
-        return create_cors_response()
     try:
         data = request.get_json()
         if not data:

server/routes/aws/auth.py

@@ -18,7 +18,7 @@
 
 auth_bp = Blueprint("aws_auth_bp", __name__)
 
-@auth_bp.route('/get-credentials', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/get-credentials', methods=['POST'])
 @require_permission("connectors", "read")
 def aws_get_credentials(user_id):
     """Retrieve AWS credentials stored for the user."""
@@ -77,18 +77,15 @@ def aws_get_credentials(user_id):
         return jsonify({"error": "Failed to retrieve AWS credentials"}), 500
 
 
-@auth_bp.route('/auth', methods=['POST', 'OPTIONS'])
+@auth_bp.route('/auth', methods=['POST'])
 @require_permission("connectors", "write")
 def auth(user_id):
     """
     AWS authentication endpoint using IAM role assumption.
-    
+
     Requires External ID that matches the workspace's External ID for security.
     Legacy flow without External ID is no longer supported.
     """
-    if flask.request.method == 'OPTIONS':
-        return create_cors_response()
-
     logging.info("=== AWS AUTH ENDPOINT STARTED ===")
     try:
         data = flask.request.get_json()

server/routes/aws/onboarding.py

@@ -113,21 +113,18 @@ def get_aws_onboarding_links(user_id, workspace_id):
         return jsonify({"error": "Internal server error"}), 500
 
 
-@onboarding_bp.route('/workspaces/<workspace_id>/aws/role', methods=['POST', 'OPTIONS'])
+@onboarding_bp.route('/workspaces/<workspace_id>/aws/role', methods=['POST'])
 @require_permission("connectors", "write")
 def set_aws_role(user_id, workspace_id):
     """
     Manually set the AWS role ARN for a workspace.
-    
+
     Expected payload:
     {
         "roleArn": "arn:aws:iam::123456789012:role/AuroraRole",
         "readOnlyRoleArn": "arn:aws:iam::123456789012:role/AuroraReadOnly"  // optional
     }
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
-
     try:
         workspace = get_workspace_by_id(workspace_id)
         if not workspace:
@@ -306,16 +303,13 @@ def create_user_workspace(authenticated_user_id, user_id):
         return jsonify({"error": "Internal server error"}), 500
 
 
-
-@onboarding_bp.route('/workspaces/<workspace_id>/aws/cleanup', methods=['POST', 'OPTIONS'])
+@onboarding_bp.route('/workspaces/<workspace_id>/aws/cleanup', methods=['POST'])
 @require_permission("connectors", "write")
 def workspace_cleanup(user_id, workspace_id):
     """Disconnect AWS connection by removing it from user_connections (single source of truth).
-    
+
     Users must manually remove IAM roles and other AWS resources in their AWS console.
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
 
     try:
         workspace = get_workspace_by_id(workspace_id)
@@ -398,7 +392,7 @@ def list_aws_accounts(user_id, workspace_id):
         return jsonify({"error": "Internal server error"}), 500
 
 
-@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/bulk', methods=['POST', 'OPTIONS'])
+@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/bulk', methods=['POST'])
 @require_permission("connectors", "write")
 def bulk_register_aws_accounts(user_id, workspace_id):
     """Register multiple AWS accounts at once.
@@ -416,9 +410,6 @@ def bulk_register_aws_accounts(user_id, workspace_id):
     Returns per-account success/failure so partially-successful bulk imports
     are surfaced clearly to the caller.
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
-
     try:
         workspace = get_workspace_by_id(workspace_id)
         if not workspace or workspace['user_id'] != user_id:
@@ -505,13 +496,10 @@ def bulk_register_aws_accounts(user_id, workspace_id):
         return jsonify({"error": "Internal server error"}), 500
 
 
-@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/<account_id>', methods=['DELETE', 'OPTIONS'])
+@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/<account_id>', methods=['DELETE'])
 @require_permission("connectors", "write")
 def delete_aws_account(user_id, workspace_id, account_id):
     """Disconnect a single AWS account from the workspace."""
-    if request.method == 'OPTIONS':
-        return create_cors_response()
-
     try:
         workspace = get_workspace_by_id(workspace_id)
         if not workspace or workspace['user_id'] != user_id:
@@ -554,17 +542,14 @@ def list_inactive_aws_accounts(user_id, workspace_id):
         return jsonify({"error": "Internal server error"}), 500
 
 
-@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/<account_id>/reconnect', methods=['POST', 'OPTIONS'])
+@onboarding_bp.route('/workspaces/<workspace_id>/aws/accounts/<account_id>/reconnect', methods=['POST'])
 @require_permission("connectors", "write")
 def reconnect_aws_account(user_id, workspace_id, account_id):
     """Reconnect a previously disconnected AWS account.
 
     Validates the role still works via STS AssumeRole, then re-activates
     the connection. No CloudFormation redeployment needed.
     """
-    if request.method == 'OPTIONS':
-        return create_cors_response()
-
     try:
         workspace = get_workspace_by_id(workspace_id)
         if not workspace or workspace['user_id'] != user_id:

server/routes/azure/azure_routes.py

@@ -15,11 +15,9 @@
 azure_bp = Blueprint("azure_bp", __name__)
 
 # ---- Azure Routes ------------------------------------------------------#
-@azure_bp.route("/azure/login", methods=["POST", "GET", "OPTIONS"])
+@azure_bp.route("/azure/login", methods=["POST", "GET"])
 @require_permission("connectors", "write")
 def azure_login_route(user_id):
-    if flask.request.method == 'OPTIONS':
-        return create_cors_response()
     return azure_login()
 
 
@@ -76,7 +74,7 @@ def azure_callback_route():
     return azure_callback()
 
 
-@azure_bp.route("/azure/fetch_data", methods=["GET", "POST", "OPTIONS"])
+@azure_bp.route("/azure/fetch_data", methods=["GET", "OPTIONS"])
 @require_permission("connectors", "read")
 def fetch_data(user_id):
     if flask.request.method == 'OPTIONS':
@@ -147,31 +145,38 @@ def azure_clusters(user_id):
         return jsonify({"error": "Failed to fetch AKS clusters"}), 500
 
 
-@azure_bp.route("/api/azure-subscriptions", methods=["GET", "POST", "OPTIONS"])
+@azure_bp.route("/api/azure-subscriptions", methods=["GET", "OPTIONS"])
 @require_permission("connectors", "read")
-def azure_subscriptions(user_id):
+def azure_subscriptions_get(user_id):
     if request.method == "OPTIONS":
         return create_cors_response()
     try:
-        if request.method == "GET":
-            from utils.auth.stateless_auth import get_org_id_from_request
-            org_id = get_org_id_from_request()
-            token_data = get_token_data(user_id, "azure", org_id=org_id)
-            if not token_data:
-                logging.warning(f"[AZURE API] No Azure token data found for user {user_id}")
-                return jsonify({"error": "No Azure credentials found. Please authenticate with Azure."}), 401
-            subscription_id = token_data.get("subscription_id")
-            subscription_name = token_data.get("subscription_name", "Azure Subscription")
-            if not subscription_id:
-                logging.warning(f"[AZURE API] No Azure subscription found for user {user_id}")
-                return jsonify({"error": "No Azure subscription found. Please configure your Azure subscription."}), 401
-            projects = [{"projectId": subscription_id, "name": subscription_name, "enabled": True}]
-            return jsonify({"projects": projects}), 200
-        else:
-            data = request.get_json()
-            projects = data.get("projects", [])
-            logging.info(f"Azure subscription selection update received: {projects}")
-            return jsonify({"status": "success"})
+        from utils.auth.stateless_auth import get_org_id_from_request
+        org_id = get_org_id_from_request()
+        token_data = get_token_data(user_id, "azure", org_id=org_id)
+        if not token_data:
+            logging.warning(f"[AZURE API] No Azure token data found for user {user_id}")
+            return jsonify({"error": "No Azure credentials found. Please authenticate with Azure."}), 401
+        subscription_id = token_data.get("subscription_id")
+        subscription_name = token_data.get("subscription_name", "Azure Subscription")
+        if not subscription_id:
+            logging.warning(f"[AZURE API] No Azure subscription found for user {user_id}")
+            return jsonify({"error": "No Azure subscription found. Please configure your Azure subscription."}), 401
+        projects = [{"projectId": subscription_id, "name": subscription_name, "enabled": True}]
+        return jsonify({"projects": projects}), 200
+    except Exception as e:
+        logging.error("Error in azure_subscriptions_get", exc_info=e)
+        return jsonify({"error": "Failed to process Azure subscriptions"}), 500
+
+
+@azure_bp.route("/api/azure-subscriptions", methods=["POST"])
+@require_permission("connectors", "write")
+def azure_subscriptions_post(user_id):
+    try:
+        data = request.get_json() or {}
+        projects = data.get("projects", [])
+        logging.info(f"Azure subscription selection update received: {projects}")
+        return jsonify({"status": "success"})
     except Exception as e:
-        logging.error("Error in azure_subscriptions", exc_info=e)
+        logging.error("Error in azure_subscriptions_post", exc_info=e)
         return jsonify({"error": "Failed to process Azure subscriptions"}), 500

server/routes/bigpanda/bigpanda_routes.py

@@ -33,7 +33,7 @@ def _get_stored_credentials(user_id: str) -> dict | None:
         return None
 
 
-@bigpanda_bp.route("/connect", methods=["POST", "OPTIONS"])
+@bigpanda_bp.route("/connect", methods=["POST"])
 @require_permission("connectors", "write")
 def connect(user_id):
     data = request.get_json(force=True, silent=True) or {}
@@ -82,7 +82,7 @@ def status(user_id):
     })
 
 
-@bigpanda_bp.route("/disconnect", methods=["POST", "DELETE", "OPTIONS"])
+@bigpanda_bp.route("/disconnect", methods=["POST", "DELETE"])
 @require_permission("connectors", "write")
 def disconnect(user_id):
     try:
@@ -116,11 +116,8 @@ def _verify_webhook_user(user_id: str) -> bool:
         return False
 
 
-@bigpanda_bp.route("/webhook/<user_id>", methods=["POST", "OPTIONS"])
+@bigpanda_bp.route("/webhook/<user_id>", methods=["POST"])
 def webhook(user_id: str):
-    if request.method == "OPTIONS":
-        return create_cors_response()
-
     if not user_id:
         return jsonify({"error": "user_id is required"}), 400
 

server/routes/bitbucket/bitbucket.py

@@ -19,13 +19,10 @@
 FRONTEND_URL = os.getenv("FRONTEND_URL")
 
 
-@bitbucket_bp.route("/login", methods=["POST", "OPTIONS"])
+@bitbucket_bp.route("/login", methods=["POST"])
 @require_permission("connectors", "write")
 def bitbucket_login(user_id):
     """Handle Bitbucket login - either API token or OAuth initiation."""
-    if request.method == "OPTIONS":
-        return create_cors_response()
-
     try:
         data = request.get_json() or {}
 
@@ -272,13 +269,10 @@ def bitbucket_status(user_id):
         return jsonify({"connected": False, "error": "Failed to check Bitbucket status"}), 500
 
 
-@bitbucket_bp.route("/disconnect", methods=["POST", "OPTIONS"])
+@bitbucket_bp.route("/disconnect", methods=["POST"])
 @require_permission("connectors", "write")
 def bitbucket_disconnect(user_id):
     """Disconnect Bitbucket account for a user."""
-    if request.method == "OPTIONS":
-        return create_cors_response()
-
     try:
         from utils.secrets.secret_ref_utils import delete_user_secret