Prioritize JSON parsing for body

aikido_zen/sources/django/run_init_stage.py

@@ -10,6 +10,13 @@ def run_init_stage(request):
     """Parse request and body, run "init" stage with request_handler"""
     body = None
     try:
+        # Check for JSON
+        if body is None and request.content_type == "application/json":
+            try:
+                body = json.loads(request.body)
+            except Exception:
+                pass
+
         # try-catch loading of form parameters, this is to fix issue with DATA_UPLOAD_MAX_NUMBER_FIELDS :
         try:
             body = request.POST.dict()
@@ -18,12 +25,6 @@ def run_init_stage(request):
         except Exception:
             pass
 
-        # Check for JSON or XML :
-        if body is None and request.content_type == "application/json":
-            try:
-                body = json.loads(request.body)
-            except Exception:
-                pass
         if body is None or len(body) == 0:
             # E.g. XML Data
             body = request.body

aikido_zen/sources/django/run_init_stage_test.py

@@ -120,6 +120,36 @@ def test_run_init_stage_with_empty_body_string(mock_request):
     assert context.body is None
 
 
+def test_run_init_stage_with_json_wrong_content_type(mock_request):
+    """Test run_init_stage with an XML request."""
+    mock_request.content_type = "application/x-www-form-urlencoded"
+    mock_request.body = '{"key": "value"}'  # Example XML body
+    run_init_stage(mock_request)
+    # Assertions
+    context: Context = get_current_context()
+    assert context.body == {"key": "value"}
+
+
+def test_run_init_stage_with_xml_wrong_content_type(mock_request):
+    """Test run_init_stage with an XML request."""
+    mock_request.content_type = "application/json"
+    mock_request.body = "<root><key>value</key></root>"  # Example XML body
+    run_init_stage(mock_request)
+    # Assertions
+    context: Context = get_current_context()
+    assert context.body == "<root><key>value</key></root>"
+
+
+def test_run_init_stage_with_xml_wrong_content_type_form_urlencoded(mock_request):
+    """Test run_init_stage with an XML request."""
+    mock_request.content_type = "application/x-www-form-urlencoded"
+    mock_request.body = "<root><key>value=2</key></root>"  # Example XML body
+    run_init_stage(mock_request)
+    # Assertions
+    context: Context = get_current_context()
+    assert context.body == "<root><key>value=2</key></root>"
+
+
 def test_run_init_stage_with_xml(mock_request):
     """Test run_init_stage with an XML request."""
     mock_request.content_type = "application/xml"

end2end/django_mysql_test.py

@@ -1,6 +1,7 @@
 import time
 import pytest
 import requests
+import json
 from .server.check_events_from_mock import fetch_events_from_mock, validate_started_event, filter_on_event_type, validate_heartbeat
 
 # e2e tests for django_mysql sample app
@@ -45,6 +46,32 @@ def test_dangerous_response_with_firewall():
         'user': None
     }
 
+def test_dangerous_response_with_form_header_but_json_body():
+    headers = {"Content-Type": "application/x-www-form-urlencoded"}
+    json_body = json.dumps({"dog_name": 'Dangerous bobby", 1); -- '})
+
+    res = requests.post(base_url_fw + "/create", headers=headers, data=json_body)
+    assert res.status_code == 500
+    time.sleep(5)
+
+    events = fetch_events_from_mock("http://localhost:5000")
+    attacks = filter_on_event_type(events, "detected_attack")
+
+    assert len(attacks) == 1
+    del attacks[0]["attack"]["stack"]
+    assert attacks[0]["attack"] == {
+        "blocked": True,
+        "kind": "sql_injection",
+        "metadata": {
+            "sql": 'INSERT INTO sample_app_dogs (dog_name, dog_boss) VALUES ("Dangerous bobby", 1); -- ", "N/A")'
+        },
+        "operation": "MySQLdb.Cursor.execute",
+        "pathToPayload": ".dog_name",
+        "payload": '"Dangerous bobby\\", 1); -- "',
+        "source": "body",
+        "user": None,
+    }
+
 def test_dangerous_response_with_firewall_shell():
     dog_name = 'Dangerous bobby", 1); -- '
     res = requests.get(base_url_fw + "/shell/ls -la")

sample-apps/django-mysql/sample_app/urls.py

@@ -5,6 +5,7 @@
 urlpatterns = [
     path("", views.index, name="index"),
     path("dogpage/<int:dog_id>", views.dog_page, name="dog_page"),
+    path("json/create", views.json_create_dog, name="json_create"),
     path("shell/<str:user_command>", views.shell_url, name="shell"),
     path("create", views.create_dogpage, name="create")
 ]

sample-apps/django-mysql/sample_app/views.py

@@ -1,11 +1,12 @@
 from django.shortcuts import render, get_object_or_404
-from django.http import HttpResponse
+from django.http import HttpResponse, JsonResponse
 from django.template import loader
 from .models import Dogs
 from django.db import connection
 from django.views.decorators.csrf import csrf_exempt
 # Create your views here.
 import subprocess
+import json
 
 def index(request):
     dogs = Dogs.objects.all()
@@ -37,3 +38,15 @@
             print("QUERY : ", query)
             cursor.execute(query)
         return HttpResponse("Dog page created")
+
+@csrf_exempt
+def json_create_dog(request):
+    if request.method == 'POST':
+        body = request.body.decode('utf-8')
+        body_json = json.loads(body)
+        dog_name = body_json.get('dog_name')
+
+        with connection.cursor() as cursor:
+            query = 'INSERT INTO sample_app_dogs (dog_name, dog_boss) VALUES ("%s", "N/A")' % dog_name
+            cursor.execute(query)
+        return JsonResponse({"status": "Dog page created"})