chore(iast): more sqli redaction tests

Author: avara1986

tests/appsec/integrations/django_tests/conftest.py

@@ -6,12 +6,13 @@
 
 from ddtrace.appsec._iast import enable_iast_propagation
 from ddtrace.appsec._iast._patch_modules import patch_iast
-from ddtrace.contrib.internal.django.patch import patch
+from ddtrace.contrib.internal.django.patch import patch as django_patch
 from ddtrace.trace import Pin
 from tests.appsec.iast.conftest import _end_iast_context_and_oce
 from tests.appsec.iast.conftest import _start_iast_context_and_oce
 from tests.utils import DummyTracer
 from tests.utils import TracerSpanContainer
+from tests.utils import override_env
 from tests.utils import override_global_config
 
 
@@ -26,10 +27,10 @@ def pytest_configure():
             _iast_deduplication_enabled=False,
             _iast_request_sampling=100.0,
         )
-    ):
+    ), override_env(dict(_DD_IAST_PATCH_MODULES="tests.appsec.integrations")):
         settings.DEBUG = False
         patch_iast()
-        patch()
+        django_patch()
         enable_iast_propagation()
         django.setup()
 

tests/appsec/integrations/django_tests/django_app/urls.py

@@ -46,6 +46,11 @@ def shutdown(request):
         views.sqli_http_request_parameter_name_post,
         name="sqli_http_request_parameter_name_post",
     ),
+    handler(
+        "appsec/sqli_query_no_redacted/$",
+        views.sqli_query_no_redacted,
+        name="sqli_query_no_redacted",
+    ),
     handler(
         "appsec/sqli_http_request_header_name/$",
         views.sqli_http_request_header_name,

tests/appsec/integrations/django_tests/django_app/views.py

@@ -129,6 +129,14 @@ def sqli_http_request_parameter_name_post(request):
     return HttpResponse(request.META["HTTP_USER_AGENT"], status=200)
 
 
+def sqli_query_no_redacted(request):
+    obj = request.GET["q"]
+    with connection.cursor() as cursor:
+        # label sqli_query_no_redacted
+        cursor.execute(f"SELECT * FROM {obj} ORDER BY name")
+    return HttpResponse("OK", status=200)
+
+
 def sqli_http_request_header_name(request):
     key = [x for x in request.META.keys() if x == "master"][0]
 

tests/appsec/integrations/django_tests/test_django_appsec_iast.py

@@ -168,7 +168,7 @@ def test_django_tainted_user_agent_iast_disabled(client, test_spans, tracer):
 
 @pytest.mark.django_db()
 @pytest.mark.skipif(not asm_config._iast_supported, reason="Python version not supported by IAST")
-def test_django_tainted_user_agent_iast_enabled_sqli_http_request_parameter(client, test_spans, tracer):
+def test_django_sqli_http_request_parameter(client, test_spans, tracer):
     root_span, response = _aux_appsec_get_root_span(
         client,
         test_spans,
@@ -309,6 +309,46 @@ def test_django_sqli_http_request_parameter_name_post(client, test_spans, tracer
     assert loaded["vulnerabilities"][0]["hash"] == hash_value
 
 
+@pytest.mark.django_db()
+@pytest.mark.skipif(not asm_config._iast_supported, reason="Python version not supported by IAST")
+def test_django_sqli_query_no_redacted(client, test_spans, tracer):
+    root_span, response = _aux_appsec_get_root_span(
+        client,
+        test_spans,
+        tracer,
+        url="/appsec/sqli_query_no_redacted/?q=sqlite_master",
+    )
+
+    vuln_type = "SQL_INJECTION"
+
+    assert response.status_code == 200
+    assert response.content == b"OK"
+
+    loaded = json.loads(root_span.get_tag(IAST.JSON))
+
+    line, hash_value = get_line_and_hash("sqli_query_no_redacted", vuln_type, filename=TEST_FILE)
+
+    assert loaded["sources"] == [
+        {
+            "name": "q",
+            "origin": "http.request.parameter",
+            "value": "sqlite_master",
+        }
+    ]
+
+    assert loaded["vulnerabilities"][0]["type"] == vuln_type
+    assert loaded["vulnerabilities"][0]["evidence"] == {
+        "valueParts": [
+            {"value": "SELECT * FROM "},
+            {"source": 0, "value": "sqlite_master"},
+            {"value": " ORDER BY name"},
+        ]
+    }
+    assert loaded["vulnerabilities"][0]["location"]["path"] == TEST_FILE
+    assert loaded["vulnerabilities"][0]["location"]["line"] == line
+    assert loaded["vulnerabilities"][0]["hash"] == hash_value
+
+
 @pytest.mark.django_db()
 @pytest.mark.skipif(not asm_config._iast_supported, reason="Python version not supported by IAST")
 def test_django_sqli_http_request_header_value(client, test_spans, tracer):