chore(asm): add SSRF support for webbrowser.open

.github/CODEOWNERS

@@ -63,6 +63,7 @@ ddtrace/appsec/                     @DataDog/asm-python
 ddtrace/settings/asm.py             @DataDog/asm-python
 ddtrace/contrib/subprocess/         @DataDog/asm-python
 ddtrace/contrib/flask_login/        @DataDog/asm-python
+ddtrace/contrib/webbrowser          @DataDog/asm-python
 ddtrace/internal/_exceptions.py     @DataDog/asm-python
 tests/appsec/                       @DataDog/asm-python
 tests/contrib/dbapi/test_dbapi_appsec.py    @DataDog/asm-python

ddtrace/appsec/_iast/taint_sinks/ssrf.py

@@ -18,8 +18,23 @@ class SSRF(VulnerabilityBase):
     vulnerability_type = VULN_SSRF
 
 
+_FUNC_TO_URL_ARGUMENT = {
+    "http.client.request": (1, "url"),
+    "requests.sessions.request": (1, "url"),
+    "urllib3._request_methods.request": (1, "url"),
+    "urllib3.request.request": (1, "url"),
+    "webbrowser.open": (0, "url"),
+}
+
+
 def _iast_report_ssrf(func: Callable, *args, **kwargs):
-    report_ssrf = args[1] if len(args) > 1 else kwargs.get("url", None)
+    func_key = func.__module__ + "." + func.__name__
+    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)
+        return
+
+    report_ssrf = args[arg_pos] if len(args) >= arg_pos else kwargs.get(kwarg_name, None)  # type: ignore[arg-type]
 
     if report_ssrf:
         from .._metrics import _set_metric_iast_executed_sink

ddtrace/contrib/webbrowser/__init__.py

@@ -0,0 +1,12 @@
+"""
+Trace the standard library ``webbrowser`` library to trace
+HTTP requests and detect SSRF vulnerabilities. It is enabled by default
+if ``DD_IAST_ENABLED`` is set to ``True`` (for detecting sink points) and/or
+``DD_ASM_ENABLED`` is set to ``True`` (for exploit prevention).
+"""
+from .patch import get_version
+from .patch import patch
+from .patch import unpatch
+
+
+__all__ = ["patch", "unpatch", "get_version"]

ddtrace/contrib/webbrowser/patch.py

@@ -0,0 +1,35 @@
+import webbrowser
+
+from ddtrace.appsec._common_module_patches import wrapped_request_D8CB81E472AF98A2 as _wrap_open
+from ddtrace.appsec._iast._metrics import _set_metric_iast_instrumented_sink
+from ddtrace.appsec._iast.constants import VULN_SSRF
+from ddtrace.settings.asm import config as asm_config
+from ddtrace.vendor.wrapt import wrap_function_wrapper as _w
+
+from ..trace_utils import unwrap as _u
+
+
+def get_version():
+    # type: () -> str
+    return ""
+
+
+def patch():
+    """patch the built-in webbrowser methods for tracing"""
+    if getattr(webbrowser, "__datadog_patch", False):
+        return
+    webbrowser.__datadog_patch = True
+
+    _w("webbrowser", "open", _wrap_open)
+
+    if asm_config._iast_enabled:
+        _set_metric_iast_instrumented_sink(VULN_SSRF)
+
+
+def unpatch():
+    """unpatch any previously patched modules"""
+    if not getattr(webbrowser, "__datadog_patch", False):
+        return
+    webbrowser.__datadog_patch = False
+
+    _u(webbrowser, "open")

tests/.suitespec.json

@@ -262,6 +262,12 @@
         "urllib3": [
             "ddtrace/contrib/urllib3/*"
         ],
+        "webbrowser": [
+            "ddtrace/contrib/webbrowser/*"
+        ],
+        "urllib": [
+            "ddtrace/contrib/urllib/*"
+        ],
         "rq": [
             "ddtrace/contrib/rq/*"
         ],
@@ -1287,6 +1293,22 @@
             "tests/contrib/urllib3/*",
             "tests/snapshots/tests.contrib.urllib3.*"
         ],
+        "webbrowser": [
+            "@bootstrap",
+            "@core",
+            "@contrib",
+            "@tracing",
+            "@webbrowser",
+            "tests/appsec/iast/taint_sinks/test_ssrf.py"
+        ],
+        "urllib": [
+            "@bootstrap",
+            "@core",
+            "@contrib",
+            "@tracing",
+            "@urllib",
+            "tests/appsec/iast/taint_sinks/test_ssrf.py"
+        ],
         "vertica": [
             "@bootstrap",
             "@core",

tests/appsec/iast/taint_sinks/test_ssrf.py

@@ -7,8 +7,13 @@
 from ddtrace.appsec._iast._taint_tracking.aspects import add_aspect
 from ddtrace.appsec._iast.constants import VULN_SSRF
 from ddtrace.contrib.httplib.patch import patch as httplib_patch
+from ddtrace.contrib.httplib.patch import unpatch as httplib_unpatch
 from ddtrace.contrib.requests.patch import patch as requests_patch
+from ddtrace.contrib.requests.patch import unpatch as requests_unpatch
 from ddtrace.contrib.urllib3.patch import patch as urllib3_patch
+from ddtrace.contrib.urllib3.patch import unpatch as urllib3_unpatch
+from ddtrace.contrib.webbrowser.patch import patch as webbrowser_patch
+from ddtrace.contrib.webbrowser.patch import unpatch as webbrowser_unpatch
 from ddtrace.internal import core
 from tests.appsec.iast.iast_utils import get_line_and_hash
 from tests.utils import override_global_config
@@ -57,55 +62,84 @@ def _check_report(span_report, tainted_path, label):
 def test_ssrf_requests(tracer, iast_span_defaults):
     with override_global_config(dict(_iast_enabled=True)):
         requests_patch()
-        import requests
-        from requests.exceptions import ConnectionError
-
-        tainted_url, tainted_path = _get_tainted_url()
         try:
-            # label test_ssrf_requests
-            requests.get(tainted_url)
-        except ConnectionError:
-            pass
+            import requests
+            from requests.exceptions import ConnectionError
 
-        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
-        assert span_report
-        _check_report(span_report, tainted_path, "test_ssrf_requests")
+            tainted_url, tainted_path = _get_tainted_url()
+            try:
+                # label test_ssrf_requests
+                requests.get(tainted_url)
+            except ConnectionError:
+                pass
+
+            span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
+            assert span_report
+            _check_report(span_report, tainted_path, "test_ssrf_requests")
+        finally:
+            requests_unpatch()
 
 
 def test_ssrf_urllib3(tracer, iast_span_defaults):
     with override_global_config(dict(_iast_enabled=True)):
         urllib3_patch()
-        import urllib3
-
-        tainted_url, tainted_path = _get_tainted_url()
         try:
-            # label test_ssrf_urllib3
-            urllib3.request(method="GET", url=tainted_url)
-        except urllib3.exceptions.HTTPError:
-            pass
+            import urllib3
 
-        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
-        assert span_report
-        _check_report(span_report, tainted_path, "test_ssrf_urllib3")
+            tainted_url, tainted_path = _get_tainted_url()
+            try:
+                # label test_ssrf_urllib3
+                urllib3.request(method="GET", url=tainted_url)
+            except urllib3.exceptions.HTTPError:
+                pass
+
+            span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
+            assert span_report
+            _check_report(span_report, tainted_path, "test_ssrf_urllib3")
+        finally:
+            urllib3_unpatch()
 
 
 def test_ssrf_httplib(tracer, iast_span_defaults):
     with override_global_config(dict(_iast_enabled=True)):
         httplib_patch()
-        import http.client
+        try:
+            import http.client
 
-        tainted_url, tainted_path = _get_tainted_url()
+            tainted_url, tainted_path = _get_tainted_url()
+            try:
+                conn = http.client.HTTPConnection("localhost")
+                # label test_ssrf_httplib
+                conn.request("GET", tainted_url)
+                conn.getresponse()
+            except ConnectionError:
+                pass
+
+            span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
+            assert span_report
+            _check_report(span_report, tainted_path, "test_ssrf_httplib")
+        finally:
+            httplib_unpatch()
+
+
+def test_ssrf_webbrowser(tracer, iast_span_defaults):
+    with override_global_config(dict(_iast_enabled=True)):
+        webbrowser_patch()
         try:
-            conn = http.client.HTTPConnection("localhost")
-            # label test_ssrf_httplib
-            conn.request("GET", tainted_url)
-            conn.getresponse()
-        except ConnectionError:
-            pass
-
-        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
-        assert span_report
-        _check_report(span_report, tainted_path, "test_ssrf_httplib")
+            import webbrowser
+
+            tainted_url, tainted_path = _get_tainted_url()
+            try:
+                # label test_ssrf_webbrowser
+                webbrowser.open(tainted_url)
+            except ConnectionError:
+                pass
+
+            span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_defaults)
+            assert span_report
+            _check_report(span_report, tainted_path, "test_ssrf_webbrowser")
+        finally:
+            webbrowser_unpatch()
 
 
 def _check_no_report_if_deduplicated(span_report, num_vuln_expected):
@@ -120,52 +154,81 @@ def _check_no_report_if_deduplicated(span_report, num_vuln_expected):
 @pytest.mark.parametrize("num_vuln_expected", [1, 0, 0])
 def test_ssrf_requests_deduplication(num_vuln_expected, tracer, iast_span_deduplication_enabled):
     requests_patch()
-    import requests
-    from requests.exceptions import ConnectionError
+    try:
+        import requests
+        from requests.exceptions import ConnectionError
 
-    tainted_url, tainted_path = _get_tainted_url()
-    for _ in range(0, 5):
-        try:
-            # label test_ssrf_requests_deduplication
-            requests.get(tainted_url)
-        except ConnectionError:
-            pass
+        tainted_url, tainted_path = _get_tainted_url()
+        for _ in range(0, 5):
+            try:
+                # label test_ssrf_requests_deduplication
+                requests.get(tainted_url)
+            except ConnectionError:
+                pass
 
-    span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
-    _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
+        _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+    finally:
+        requests_unpatch()
 
 
 @pytest.mark.parametrize("num_vuln_expected", [1, 0, 0])
 def test_ssrf_urllib3_deduplication(num_vuln_expected, tracer, iast_span_deduplication_enabled):
     urllib3_patch()
-    import urllib3
+    try:
+        import urllib3
 
-    tainted_url, tainted_path = _get_tainted_url()
-    for _ in range(0, 5):
-        try:
-            # label test_ssrf_urllib3_deduplication
-            urllib3.request(method="GET", url=tainted_url)
-        except urllib3.exceptions.HTTPError:
-            pass
+        tainted_url, tainted_path = _get_tainted_url()
+        for _ in range(0, 5):
+            try:
+                # label test_ssrf_urllib3_deduplication
+                urllib3.request(method="GET", url=tainted_url)
+            except urllib3.exceptions.HTTPError:
+                pass
 
-    span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
-    _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
+        _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+    finally:
+        requests_unpatch()
 
 
 @pytest.mark.parametrize("num_vuln_expected", [1, 0, 0])
 def test_ssrf_httplib_deduplication(num_vuln_expected, tracer, iast_span_deduplication_enabled):
     httplib_patch()
-    import http.client
+    try:
+        import http.client
 
-    tainted_url, tainted_path = _get_tainted_url()
-    for _ in range(0, 5):
-        try:
-            conn = http.client.HTTPConnection("localhost")
-            # label test_ssrf_httplib_deduplication
-            conn.request("GET", tainted_url)
-            conn.getresponse()
-        except ConnectionError:
-            pass
-
-    span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
-    _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+        tainted_url, tainted_path = _get_tainted_url()
+        for _ in range(0, 5):
+            try:
+                conn = http.client.HTTPConnection("localhost")
+                # label test_ssrf_httplib_deduplication
+                conn.request("GET", tainted_url)
+                conn.getresponse()
+            except ConnectionError:
+                pass
+
+        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
+        _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+    finally:
+        httplib_unpatch()
+
+
+@pytest.mark.parametrize("num_vuln_expected", [1, 0, 0])
+def test_ssrf_webbrowser_deduplication(num_vuln_expected, tracer, iast_span_deduplication_enabled):
+    webbrowser_patch()
+    try:
+        import webbrowser
+
+        tainted_url, tainted_path = _get_tainted_url()
+        for _ in range(0, 5):
+            try:
+                # label test_ssrf_webbrowser_deduplication
+                webbrowser.open(tainted_url)
+            except ConnectionError:
+                pass
+
+        span_report = core.get_item(IAST.CONTEXT_KEY, span=iast_span_deduplication_enabled)
+        _check_no_report_if_deduplicated(span_report, num_vuln_expected)
+    finally:
+        webbrowser_unpatch()