feat(iast): xss vulnerability for django applications (#12116)

XSS vulnerability detection.

System tests (merge after this PR):
DataDog/system-tests#3923


## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

---------

Co-authored-by: Federico Mon <[email protected]>

ddtrace/appsec/_iast/_evidence_redaction/_sensitive_handler.py

@@ -10,6 +10,7 @@
 from ..constants import VULN_HEADER_INJECTION
 from ..constants import VULN_SQL_INJECTION
 from ..constants import VULN_SSRF
+from ..constants import VULN_XSS
 from .command_injection_sensitive_analyzer import command_injection_sensitive_analyzer
 from .default_sensitive_analyzer import default_sensitive_analyzer
 from .header_injection_sensitive_analyzer import header_injection_sensitive_analyzer
@@ -45,6 +46,7 @@ def __init__(self):
             VULN_SQL_INJECTION: sql_sensitive_analyzer,
             VULN_SSRF: url_sensitive_analyzer,
             VULN_HEADER_INJECTION: header_injection_sensitive_analyzer,
+            VULN_XSS: default_sensitive_analyzer,
             VULN_CODE_INJECTION: default_sensitive_analyzer,
         }
 

ddtrace/appsec/_iast/_patch_modules.py

@@ -7,6 +7,7 @@
     "header_injection": True,
     "weak_cipher": True,
     "weak_hash": True,
+    "xss": True,
 }
 
 

ddtrace/appsec/_iast/constants.py

@@ -14,6 +14,7 @@
 VULN_CMDI = "COMMAND_INJECTION"
 VULN_HEADER_INJECTION = "HEADER_INJECTION"
 VULN_CODE_INJECTION = "CODE_INJECTION"
+VULN_XSS = "XSS"
 VULN_SSRF = "SSRF"
 VULN_STACKTRACE_LEAK = "STACKTRACE_LEAK"
 

ddtrace/appsec/_iast/taint_sinks/xss.py

@@ -0,0 +1,78 @@
+from typing import Text
+
+from ddtrace.appsec._common_module_patches import try_unwrap
+from ddtrace.appsec._constants import IAST_SPAN_TAGS
+from ddtrace.appsec._iast import oce
+from ddtrace.appsec._iast._iast_request_context import is_iast_request_enabled
+from ddtrace.appsec._iast._metrics import _set_metric_iast_executed_sink
+from ddtrace.appsec._iast._metrics import _set_metric_iast_instrumented_sink
+from ddtrace.appsec._iast._metrics import increment_iast_span_metric
+from ddtrace.appsec._iast._patch import set_and_check_module_is_patched
+from ddtrace.appsec._iast._patch import set_module_unpatched
+from ddtrace.appsec._iast._patch import try_wrap_function_wrapper
+from ddtrace.appsec._iast._taint_tracking._taint_objects import is_pyobject_tainted
+from ddtrace.appsec._iast.constants import VULN_XSS
+from ddtrace.appsec._iast.taint_sinks._base import VulnerabilityBase
+from ddtrace.internal.logger import get_logger
+from ddtrace.settings.asm import config as asm_config
+
+
+log = get_logger(__name__)
+
+
+@oce.register
+class XSS(VulnerabilityBase):
+    vulnerability_type = VULN_XSS
+
+
+def get_version() -> Text:
+    return ""
+
+
+def patch():
+    if not asm_config._iast_enabled:
+        return
+
+    if not set_and_check_module_is_patched("flask", default_attr="_datadog_xss_patch"):
+        return
+    if not set_and_check_module_is_patched("django", default_attr="_datadog_xss_patch"):
+        return
+    if not set_and_check_module_is_patched("fastapi", default_attr="_datadog_xss_patch"):
+        return
+
+    try_wrap_function_wrapper(
+        "django.utils.safestring",
+        "mark_safe",
+        _iast_django_xss,
+    )
+
+    try_wrap_function_wrapper(
+        "django.template.defaultfilters",
+        "mark_safe",
+        _iast_django_xss,
+    )
+
+    _set_metric_iast_instrumented_sink(VULN_XSS)
+
+
+def unpatch():
+    try_unwrap("django.utils.safestring", "mark_safe")
+    try_unwrap("django.template.defaultfilters", "mark_safe")
+
+    set_module_unpatched("flask", default_attr="_datadog_xss_patch")
+    set_module_unpatched("django", default_attr="_datadog_xss_patch")
+    set_module_unpatched("fastapi", default_attr="_datadog_xss_patch")
+
+
+def _iast_django_xss(wrapped, instance, args, kwargs):
+    if args and len(args) >= 1:
+        _iast_report_xss(args[0])
+    return wrapped(*args, **kwargs)
+
+
+def _iast_report_xss(code_string: Text):
+    increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, XSS.vulnerability_type)
+    _set_metric_iast_executed_sink(XSS.vulnerability_type)
+    if is_iast_request_enabled():
+        if is_pyobject_tainted(code_string):
+            XSS.report(evidence_value=code_string)

docker-compose.yml

@@ -198,7 +198,6 @@ services:
         ports:
             - "127.0.0.1:8321:8321"
         environment:
-            - DD_APPSEC_ENABLED=true
             - DD_IAST_ENABLED=true
             - DD_IAST_REQUEST_SAMPLING=100
             - DD_IAST_VULNERABILITIES_PER_REQUEST=100

releasenotes/notes/iast-feat-xss-django-6781a8b9a4092832.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Code Security (IAST): XSS detection for Django applications, 
+    which will be displayed on your DataDog Vulnerability Explorer dashboard.
+    See the `Application Vulnerability Management <https://docs.datadoghq.com/security/application_security/vulnerability_management/>`_ documentation for more information about this feature.

tests/appsec/iast/taint_sinks/_taint_sinks_utils.py

@@ -22,6 +22,7 @@ def get_parametrize(vuln_type, ignore_list=None):
             "$1 - Tainted range based redaction - multiple ranges",
             "Redacted source that needs to be truncated",
             "Query with single quoted string literal and null source",
+            "No redacted that needs to be truncated - whole text",
         ):
             continue
 

tests/appsec/iast/taint_sinks/test_xss_redacted.py

@@ -0,0 +1,48 @@
+import os
+
+import pytest
+
+from ddtrace.appsec._iast._taint_tracking import origin_to_str
+from ddtrace.appsec._iast._taint_tracking import str_to_origin
+from ddtrace.appsec._iast.constants import VULN_XSS
+from ddtrace.appsec._iast.taint_sinks.xss import XSS
+from tests.appsec.iast.taint_sinks._taint_sinks_utils import _taint_pyobject_multiranges
+from tests.appsec.iast.taint_sinks._taint_sinks_utils import get_parametrize
+from tests.appsec.iast.taint_sinks.conftest import _get_iast_data
+
+
+ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+@pytest.mark.parametrize(
+    "evidence_input, sources_expected, vulnerabilities_expected,element", list(get_parametrize(VULN_XSS))
+)
+def test_xss_redaction_suite(
+    evidence_input, sources_expected, vulnerabilities_expected, iast_context_defaults, element
+):
+    tainted_object = evidence_input_value = evidence_input.get("value", "")
+    if evidence_input_value:
+        tainted_object = _taint_pyobject_multiranges(
+            evidence_input_value,
+            [
+                (
+                    input_ranges["iinfo"]["parameterName"],
+                    input_ranges["iinfo"]["parameterValue"],
+                    str_to_origin(input_ranges["iinfo"]["type"]),
+                    input_ranges["start"],
+                    input_ranges["end"] - input_ranges["start"],
+                )
+                for input_ranges in evidence_input.get("ranges", {})
+            ],
+        )
+
+    XSS.report(tainted_object)
+
+    data = _get_iast_data()
+    vulnerability = list(data["vulnerabilities"])[0]
+    source = list(data["sources"])[0]
+    source["origin"] = origin_to_str(source["origin"])
+
+    assert vulnerability["type"] == VULN_XSS
+    assert vulnerability["evidence"] == vulnerabilities_expected["evidence"]
+    assert source == sources_expected

tests/appsec/integrations/django_tests/conftest.py

@@ -5,6 +5,7 @@
 import pytest
 
 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.trace import Pin
 from tests.appsec.iast.conftest import _end_iast_context_and_oce
@@ -27,11 +28,22 @@ def pytest_configure():
         )
     ):
         settings.DEBUG = False
-        enable_iast_propagation()
+        patch_iast()
         patch()
+        enable_iast_propagation()
         django.setup()
 
 
+@pytest.fixture
+def debug_mode():
+    from django.conf import settings
+
+    original_debug = settings.DEBUG
+    settings.DEBUG = True
+    yield
+    settings.DEBUG = original_debug
+
+
 @pytest.fixture
 def tracer():
     tracer = DummyTracer()

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

@@ -40,7 +40,7 @@
     {
         "BACKEND": "django.template.backends.django.DjangoTemplates",
         "DIRS": [
-            os.path.join(BASE_DIR, "templates"),
+            os.path.join(BASE_DIR, "django_app", "templates"),
         ],
         "APP_DIRS": True,
         "OPTIONS": {

tests/appsec/integrations/django_tests/django_app/templates/index.html

@@ -0,0 +1,5 @@
+<html>
+<body>
+<p>Input: {{ user_input }}</p>
+</body>
+</html>

tests/appsec/integrations/django_tests/django_app/templates/index_autoescape.html

@@ -0,0 +1,7 @@
+<html>
+<body>
+<p>{% autoescape on %}
+    {{ user_input }}
+{% endautoescape %}</p>
+</body>
+</html>

tests/appsec/integrations/django_tests/django_app/templates/index_safe.html

@@ -0,0 +1,5 @@
+<html>
+<body>
+<p>Input: {{ user_input|safe }}</p>
+</body>
+</html>

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

@@ -73,6 +73,10 @@ def shutdown(request):
     handler("appsec/insecure-cookie/test_insecure/$", views.view_insecure_cookies_insecure),
     handler("appsec/insecure-cookie/test_secure/$", views.view_insecure_cookies_secure),
     handler("appsec/insecure-cookie/test_empty_cookie/$", views.view_insecure_cookies_empty),
+    handler("appsec/xss/$", views.xss_http_request_parameter_mark_safe),
+    handler("appsec/xss/secure/$", views.xss_secure),
+    handler("appsec/xss/safe/$", views.xss_http_request_parameter_template_safe),
+    handler("appsec/xss/autoscape/$", views.xss_http_request_parameter_autoscape),
     path(
         "appsec/sqli_http_path_parameter/<str:q_http_path_parameter>/",
         views.sqli_http_path_parameter,

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

@@ -8,6 +8,8 @@
 from django.db import connection
 from django.http import HttpResponse
 from django.http import JsonResponse
+from django.shortcuts import render
+from django.utils.safestring import mark_safe
 
 from ddtrace.appsec import _asm_request_context
 from ddtrace.appsec._iast._taint_tracking import OriginType
@@ -68,6 +70,34 @@ def checkuser_view(request, user_id):
     return HttpResponse(status=200)
 
 
+def xss_http_request_parameter_mark_safe(request):
+    user_input = request.GET.get("input", "")
+
+    # label xss_http_request_parameter_mark_safe
+    return render(request, "index.html", {"user_input": mark_safe(user_input)})
+
+
+def xss_secure(request):
+    user_input = request.GET.get("input", "")
+
+    # label xss_http_request_parameter_mark_safe
+    return render(request, "index.html", {"user_input": user_input})
+
+
+def xss_http_request_parameter_template_safe(request):
+    user_input = request.GET.get("input", "")
+
+    # label xss_http_request_parameter_template_safe
+    return render(request, "index_safe.html", {"user_input": user_input})
+
+
+def xss_http_request_parameter_autoscape(request):
+    user_input = request.GET.get("input", "")
+
+    # label xss_http_request_parameter_autoscape
+    return render(request, "index_autoescape.html", {"user_input": user_input})
+
+
 def sqli_http_request_parameter(request):
     import bcrypt
     from django.contrib.auth.hashers import BCryptSHA256PasswordHasher