chore(iast): reduce ssrf false positives (#13530)

This PR improves the accuracy of SSRF (Server-Side Request Forgery)
vulnerability detection in the IAST module by reducing false positives,
particularly when dealing with URL fragments.

| URL Part | Tainted? | Safe Builder? | Encoded? | Report SSRF? |

|---------------|---------------|------------------------|----------|--------------|
| Protocol/Host/Port | Yes | N/A | N/A | Yes (HIGH) |
| Path | Yes | Yes | Yes | No |
| Path | Yes | No | No | Yes (MEDIUM) |
| Query | Yes | Yes | N/A | No |
| Query | Yes | No | N/A | Yes (MEDIUM) |
| Fragment | Yes | N/A | N/A | No |

### Key Changes
- Enhanced SSRF detection logic to properly handle URL fragments (#)
- Added validation to ignore cases where tainted data only appears in
URL fragments
- Added test cases to verify fragment handling behavior

### Motivation
URL fragments (parts after #) are client-side only and not sent to the
server, making them irrelevant for SSRF vulnerabilities. The current
implementation was generating false positives when tainted data appeared
only in URL fragments, leading to unnecessary alerts.

### Testing Strategy
- Added new test cases in `tests/appsec/iast/taint_sinks/test_ssrf.py`
- Verified behavior with Django integration tests
- Tested various URL fragment scenarios:
  - Tainted data in fragments only
  - Tainted data in both query params and fragments
  - Multiple fragments with tainted data

## 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)

Author: avara1986

ddtrace/appsec/_iast/_patch_modules.py

@@ -6,6 +6,7 @@
 from ddtrace.appsec._iast.secure_marks.sanitizers import path_traversal_sanitizer
 from ddtrace.appsec._iast.secure_marks.sanitizers import sqli_sanitizer
 from ddtrace.appsec._iast.secure_marks.sanitizers import xss_sanitizer
+from ddtrace.appsec._iast.secure_marks.validators import ssrf_validator
 from ddtrace.appsec._iast.secure_marks.validators import unvalidated_redirect_validator
 
 
@@ -32,6 +33,7 @@ def patch_iast(patch_modules=IAST_PATCH):
     for module in (m for m, e in patch_modules.items() if e):
         when_imported("hashlib")(_on_import_factory(module, "ddtrace.appsec._iast.taint_sinks.%s", raise_errors=False))
 
+    # CMDI sanitizers
     when_imported("shlex")(
         lambda _: try_wrap_function_wrapper(
             "shlex",
@@ -40,6 +42,11 @@ def patch_iast(patch_modules=IAST_PATCH):
         )
     )
 
+    # SSRF
+    when_imported("django.utils.http")(
+        lambda _: try_wrap_function_wrapper("django.utils.http", "url_has_allowed_host_and_scheme", ssrf_validator)
+    )
+
     # SQL sanitizers
     when_imported("mysql.connector.conversion")(
         lambda _: try_wrap_function_wrapper(

ddtrace/appsec/_iast/secure_marks/validators.py

@@ -93,3 +93,18 @@ def unvalidated_redirect_validator(wrapped: Callable, instance: Any, args: Seque
         True if validation passed, False otherwise
     """
     return create_validator(VulnerabilityType.UNVALIDATED_REDIRECT, wrapped, instance, args, kwargs)
+
+
+def ssrf_validator(wrapped: Callable, instance: Any, args: Sequence, kwargs: dict) -> bool:
+    """Validator for ssrf functions.
+
+    Args:
+        wrapped: The original validator function
+        instance: The instance the function is bound to (if any)
+        args: Positional arguments
+        kwargs: Keyword arguments
+
+    Returns:
+        True if validation passed, False otherwise
+    """
+    return create_validator(VulnerabilityType.SSRF, wrapped, instance, args, kwargs)

ddtrace/appsec/_iast/taint_sinks/ssrf.py

@@ -2,21 +2,19 @@
 
 from ddtrace.appsec._constants import IAST_SPAN_TAGS
 from ddtrace.appsec._iast._logs import iast_error
+from ddtrace.appsec._iast._logs import iast_propagation_sink_point_debug_log
 from ddtrace.appsec._iast._metrics import _set_metric_iast_executed_sink
 from ddtrace.appsec._iast._span_metrics import increment_iast_span_metric
 from ddtrace.appsec._iast._taint_tracking import VulnerabilityType
+from ddtrace.appsec._iast._taint_tracking import get_ranges
 from ddtrace.appsec._iast.constants import VULN_SSRF
 from ddtrace.appsec._iast.taint_sinks._base import VulnerabilityBase
-from ddtrace.internal.logger import get_logger
 from ddtrace.internal.utils import ArgumentError
 from ddtrace.internal.utils import get_argument_value
 from ddtrace.internal.utils.importlib import func_name
 from ddtrace.settings.asm import config as asm_config
 
 
-log = get_logger(__name__)
-
-
 class SSRF(VulnerabilityBase):
     vulnerability_type = VULN_SSRF
     secure_mark = VulnerabilityType.SSRF
@@ -33,24 +31,41 @@ class SSRF(VulnerabilityBase):
 
 
 def _iast_report_ssrf(func: Callable, *args, **kwargs):
+    """
+    Check and report potential SSRF (Server-Side Request Forgery) vulnerabilities in function calls.
+
+    This function analyzes calls to URL-handling functions to detect potential SSRF vulnerabilities.
+    It checks if the URL argument is tainted (user-controlled) and reports it if conditions are met.
+    URL fragments (parts after #) are handled specially - if all tainted parts are in the fragment,
+    no vulnerability is reported.
+    """
     func_key = func_name(func)
     arg_pos, kwarg_name = _FUNC_TO_URL_ARGUMENT.get(func_key, (None, None))
     if arg_pos is None:
-        log.debug("%s not found in list of functions supported for SSRF", func_key)
+        iast_propagation_sink_point_debug_log("%s not found in list of functions supported for SSRF", func_key)
         return
 
     try:
         kw = kwarg_name if kwarg_name else ""
         report_ssrf = get_argument_value(list(args), kwargs, arg_pos, kw)
     except ArgumentError:
-        log.debug("Failed to get URL argument from _FUNC_TO_URL_ARGUMENT dict for function %s", func_key)
+        iast_propagation_sink_point_debug_log(
+            "Failed to get URL argument from _FUNC_TO_URL_ARGUMENT dict for function %s", func_key
+        )
         return
-
     if report_ssrf:
         if asm_config.is_iast_request_enabled:
             try:
                 if SSRF.has_quota() and SSRF.is_tainted_pyobject(report_ssrf):
-                    SSRF.report(evidence_value=report_ssrf)
+                    valid_to_report = True
+                    fragment_start = report_ssrf.find("#")
+                    taint_ranges = get_ranges(report_ssrf)
+                    if fragment_start != -1:
+                        # If all taint ranges are in the fragment, do not report
+                        if all(r.start >= fragment_start for r in taint_ranges):
+                            valid_to_report = False
+                    if valid_to_report:
+                        SSRF.report(evidence_value=report_ssrf)
 
                 # Reports Span Metrics
                 _set_metric_iast_executed_sink(SSRF.vulnerability_type)

tests/appsec/iast/taint_sinks/test_ssrf.py

@@ -164,7 +164,6 @@ def _check_no_report_if_deduplicated(num_vuln_expected):
 
 def test_ssrf_requests_deduplication(iast_context_deduplication_enabled):
     requests_patch()
-    _end_iast_context_and_oce()
     try:
         import requests
         from requests.exceptions import ConnectionError  # noqa: A004
@@ -187,7 +186,6 @@ def test_ssrf_requests_deduplication(iast_context_deduplication_enabled):
 
 def test_ssrf_urllib3_deduplication(iast_context_deduplication_enabled):
     urllib3_patch()
-    _end_iast_context_and_oce()
     try:
         for num_vuln_expected in [1, 0, 0]:
             _start_iast_context_and_oce()
@@ -209,7 +207,6 @@ def test_ssrf_urllib3_deduplication(iast_context_deduplication_enabled):
 
 def test_ssrf_httplib_deduplication(iast_context_deduplication_enabled):
     httplib_patch()
-    _end_iast_context_and_oce()
     try:
         import http.client
 
@@ -233,7 +230,6 @@ def test_ssrf_httplib_deduplication(iast_context_deduplication_enabled):
 
 def test_ssrf_webbrowser_deduplication(iast_context_deduplication_enabled):
     webbrowser_patch()
-    _end_iast_context_and_oce()
     try:
         import webbrowser
 
@@ -255,7 +251,6 @@ def test_ssrf_webbrowser_deduplication(iast_context_deduplication_enabled):
 
 def test_ssrf_urllib_deduplication(iast_context_deduplication_enabled):
     urllib_patch()
-    _end_iast_context_and_oce()
     try:
         import urllib.request
 

tests/appsec/integrations/django_tests/conftest.py

@@ -7,6 +7,7 @@
 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 as django_patch
+from ddtrace.contrib.internal.requests.patch import patch as requests_patch
 from ddtrace.internal import core
 from ddtrace.trace import Pin
 from tests.appsec.iast.conftest import _end_iast_context_and_oce
@@ -31,6 +32,7 @@ def pytest_configure():
     ), override_env(dict(_DD_IAST_PATCH_MODULES="tests.appsec.integrations")):
         settings.DEBUG = False
         patch_iast()
+        requests_patch()
         django_patch()
         enable_iast_propagation()
         django.setup()

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

@@ -69,6 +69,7 @@ def shutdown(request):
         views.unvalidated_redirect_path_multiple_sources,
         name="unvalidated_redirect_path_multiple_sources",
     ),
+    handler("appsec/ssrf_requests/$", views.ssrf_requests, name="ssrf_requests"),
     handler("appsec/taint-checking-enabled/$", views.taint_checking_enabled_view, name="taint_checking_enabled_view"),
     handler(
         "appsec/taint-checking-disabled/$", views.taint_checking_disabled_view, name="taint_checking_disabled_view"

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

@@ -4,12 +4,15 @@
 
 import hashlib
 from html import escape
+import json
 import os
 from pathlib import Path
 from pathlib import PosixPath
 import shlex
 import subprocess
 from typing import Any
+import urllib
+from urllib.parse import quote
 
 from django.db import connection
 from django.http import HttpResponse
@@ -18,6 +21,8 @@
 from django.shortcuts import render
 from django.utils.http import url_has_allowed_host_and_scheme
 from django.utils.safestring import mark_safe
+import requests
+from requests.exceptions import ConnectionError  # noqa: A004
 
 from ddtrace.appsec import _asm_request_context
 from ddtrace.appsec._iast._taint_tracking import OriginType
@@ -472,3 +477,65 @@ def signup(request):
         User.objects.create_user(username=login, password=passwd)
         return HttpResponse("OK", status=200)
     return HttpResponse("Error", status=400)
+
+
+def ssrf_requests(request):
+    value = request.GET.get("url")
+    option = request.GET.get("option")
+    try:
+        if option == "path":
+            # label ssrf_requests_path
+            _ = requests.get(f"http://localhost:8080/{value}", timeout=1)
+        elif option == "protocol":
+            # label ssrf_requests_protocol
+            _ = requests.get(f"{value}://localhost:8080/", timeout=1)
+        elif option == "host":
+            # label ssrf_requests_host
+            _ = requests.get(f"http://{value}:8080/", timeout=1)
+        elif option == "query":
+            # label ssrf_requests_query
+            _ = requests.get(f"http://localhost:8080/?{value}", timeout=1)
+        elif option == "query_with_fragment":
+            # label ssrf_requests_query_with_fragment
+            _ = requests.get(f"http://localhost:8080/?{value}", timeout=1)
+        elif option == "port":
+            # label ssrf_requests_port
+            _ = requests.get(f"http://localhost:{value}/", timeout=1)
+        elif option == "fragment1":
+            _ = requests.get(f"http://localhost:8080/#section1={value}", timeout=1)
+        elif option == "fragment2":
+            _ = requests.get(f"http://localhost:8080/?param1=value1&param2=value2#section2={value}", timeout=1)
+        elif option == "fragment3":
+            _ = requests.get(
+                "http://localhost:8080/path-to-something/object_identifier?"
+                f"param1=value1&param2=value2#section3={value}",
+                timeout=1,
+            )
+        elif option == "query_param":
+            _ = requests.get("http://localhost:8080/", params={"param1": value}, timeout=1)
+        elif option == "urlencode_single":
+            params = urllib.parse.urlencode({"key1": value})
+            _ = requests.get(f"http://localhost:8080/?{params}", timeout=1)
+        elif option == "urlencode_multiple":
+            params = urllib.parse.urlencode({"key1": value, "key2": "static_value", "key3": "another_value"})
+            _ = requests.get(f"http://localhost:8080/?{params}", timeout=1)
+        elif option == "urlencode_nested":
+            nested_data = {"user": value, "filters": {"type": "report", "format": "json"}}
+            params = urllib.parse.urlencode({"data": json.dumps(nested_data)})
+            _ = requests.get(f"http://localhost:8080/?{params}", timeout=1)
+        elif option == "urlencode_with_fragment":
+            params = urllib.parse.urlencode({"search": value})
+            _ = requests.get(f"http://localhost:8080/?{params}#results", timeout=1)
+        elif option == "urlencode_doseq":
+            params = urllib.parse.urlencode({"ids": [value, "id2", "id3"]}, doseq=True)
+            _ = requests.get(f"http://localhost:8080/?{params}", timeout=1)
+        elif option == "safe_host":
+            if url_has_allowed_host_and_scheme(value, allowed_hosts={request.get_host()}):
+                _ = requests.get(f"http://{value}:8080/", timeout=1)
+            _ = requests.get(f"http://{value}:8080/", timeout=1)
+        elif option == "safe_path":
+            safe_path = quote(value)
+            _ = requests.get(f"http://localhost:8080/{safe_path}", timeout=1)
+    except ConnectionError:
+        pass
+    return HttpResponse("OK", status=200)

tests/appsec/integrations/django_tests/test_iast_django.py

@@ -10,6 +10,7 @@
 from ddtrace.appsec._iast.constants import VULN_HEADER_INJECTION
 from ddtrace.appsec._iast.constants import VULN_INSECURE_COOKIE
 from ddtrace.appsec._iast.constants import VULN_SQL_INJECTION
+from ddtrace.appsec._iast.constants import VULN_SSRF
 from ddtrace.appsec._iast.constants import VULN_STACKTRACE_LEAK
 from ddtrace.appsec._iast.constants import VULN_UNVALIDATED_REDIRECT
 from ddtrace.settings.asm import config as asm_config
@@ -1467,3 +1468,74 @@ def test_django_iast_sampling_by_route_method(client, test_spans_2_vuln_per_requ
     assert (
         len(list_vulnerabilities) == 16
     ), f"Num vulnerabilities: ({len(list_vulnerabilities)}): {list_vulnerabilities}"
+
+
+@pytest.mark.skipif(not asm_config._iast_supported, reason="Python version not supported by IAST")
+def test_django_ssrf_safe_path(client, iast_span, tracer):
+    tainted_value = "path_param"
+    root_span, _ = _aux_appsec_get_root_span(
+        client, iast_span, tracer, url=f"/appsec/ssrf_requests/?url={tainted_value}"
+    )
+    loaded = root_span.get_tag(IAST.JSON)
+    assert loaded is None
+
+
+@pytest.mark.parametrize(
+    ("option", "url", "value_parts"),
+    [
+        ("path", "url-path/", [{"value": "http://localhost:8080/"}, {"value": "url-path/", "source": 0}]),
+        ("safe_path", "url-path/", None),
+        ("protocol", "http", [{"value": "http", "source": 0}, {"value": "://localhost:8080/"}]),
+        ("host", "localhost", [{"value": "http://"}, {"value": "localhost", "source": 0}, {"value": ":8080/"}]),
+        ("urlencode_single", "value1", None),
+        ("urlencode_multiple", "value1", None),
+        ("urlencode_nested", "value1", None),
+        ("urlencode_with_fragment", "value1", None),
+        ("urlencode_doseq", "value1", None),
+        ("safe_host", "localhost", None),
+        ("port", "8080", [{"value": "http://localhost:"}, {"value": "8080", "source": 0}, {"value": "/"}]),
+        (
+            "query",
+            "param1=value1&param2=value2",
+            [
+                {"value": "http://localhost:8080/?"},
+                {"source": 0, "value": "param1="},
+                {"redacted": True, "source": 0, "pattern": "hijklm"},
+            ],
+        ),
+        (
+            "query_with_fragment",
+            "param1=value_with_%23hash%23&param2=value2",
+            [
+                {"value": "http://localhost:8080/?"},
+                {"source": 0, "value": "param1="},
+                {"redacted": True, "source": 0, "pattern": "hijklmnopqr"},
+                {"source": 0, "value": "#hash#"},
+            ],
+        ),
+        ("fragment1", "fragment_value1", None),
+        ("fragment2", "fragment_value1", None),
+        ("fragment3", "fragment_value1", None),
+        ("query_param", "param1=value1&param2=value2", None),
+    ],
+)
+def test_django_ssrf_url(client, iast_span, tracer, option, url, value_parts):
+    root_span, response = _aux_appsec_get_root_span(
+        client, iast_span, tracer, url=f"/appsec/ssrf_requests/?option={option}&url={url}"
+    )
+
+    assert response.status_code == 200
+    assert response.content == b"OK"
+
+    if value_parts is None:
+        assert root_span.get_tag(IAST.JSON) is None
+    else:
+        loaded = json.loads(root_span.get_tag(IAST.JSON))
+
+        line, hash_value = get_line_and_hash(f"ssrf_requests_{option}", VULN_SSRF, filename=TEST_FILE)
+
+        assert loaded["vulnerabilities"][0]["type"] == VULN_SSRF
+        assert loaded["vulnerabilities"][0]["evidence"] == {"valueParts": value_parts}
+        assert loaded["vulnerabilities"][0]["location"]["path"] == TEST_FILE
+        assert loaded["vulnerabilities"][0]["location"]["line"] == line
+        assert loaded["vulnerabilities"][0]["hash"] == hash_value