fix(iast): weak hash error if vulnerability is outside the context [backport 3.17] (#15038)

dd-octo-sts[bot] · avara1986 · web-flow · commit 1906fcda9040 · 2025-10-27T08:16:34.000-04:00
Backport 8940186 from #15029 to 3.17. ## Description This PR addresses an issue where using weak hashing or cipher algorithms outside of a request context (e.g., during application startup) could raise an unhandled exception. The fix ensures proper error handling when IAST operations are performed without an active request context. ### Root Cause The issue occurred in the [has_quota](cci:1://file:///home/alberto.vara/projects/dd-python/dd-trace-py/ddtrace/appsec/_iast/taint_sinks/_base.py:7:4-12:20) method of the vulnerability base class, which was not properly handling cases where there was no active request context. When IAST operations were performed outside of a request (e.g., during application startup or in unsupported frameworks), the code would attempt to access the vulnerability budget from a non-existent context, leading to an unhandled exception. ### Changes 1. Added null check for IAST context in [has_quota](cci:1://file:///home/alberto.vara/projects/dd-python/dd-trace-py/ddtrace/appsec/_iast/taint_sinks/_base.py:7:4-12:20) method to safely handle cases with no active request 2. Return `False` when no context is available, preventing further processing of vulnerabilities 3. Added test cases to verify the fix works in various scenarios 4. Updated the release notes to document the fix Co-authored-by: Alberto Vara <alberto.vara@datadoghq.com>
diff --git a/ddtrace/appsec/_iast/_span_metrics.py b/ddtrace/appsec/_iast/_span_metrics.py
@@ -29,8 +29,9 @@ def _set_span_tag_iast_executed_sink(span):
 
 
 def get_iast_span_metrics() -> Dict:
-    env = _get_iast_env()
-    return env.iast_span_metrics if env else dict()
+    if env := _get_iast_env():
+        return env.iast_span_metrics
+    return dict()
 
 
 def reset_iast_span_metrics() -> None:
diff --git a/ddtrace/appsec/_iast/taint_sinks/_base.py b/ddtrace/appsec/_iast/taint_sinks/_base.py
@@ -65,8 +65,9 @@ class VulnerabilityBase:
 
     @staticmethod
     def has_quota():
-        context = _get_iast_env()
-        return context.vulnerability_budget < asm_config._iast_max_vulnerabilities_per_requests
+        if context := _get_iast_env():
+            return context.vulnerability_budget < asm_config._iast_max_vulnerabilities_per_requests
+        return False
 
     @classmethod
     @taint_sink_deduplication
diff --git a/ddtrace/appsec/_iast/taint_sinks/ast_taint.py b/ddtrace/appsec/_iast/taint_sinks/ast_taint.py
@@ -3,6 +3,7 @@
 
 from ddtrace.appsec._constants import IAST_SPAN_TAGS
 from ddtrace.appsec._iast._iast_request_context_base import is_iast_request_enabled
+from ddtrace.appsec._iast._logs import iast_error
 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.constants import DEFAULT_COMMAND_INJECTION_FUNCTIONS
@@ -22,54 +23,51 @@ def ast_function(
     *args: Any,
     **kwargs: Any,
 ) -> Any:
-    instance = getattr(func, "__self__", None)
-    func_name = getattr(func, "__name__", None)
-    cls_name = ""
-    if instance is not None and func_name:
-        try:
-            cls_name = instance.__class__.__name__
-        except AttributeError:
-            pass
+    try:
+        instance = getattr(func, "__self__", None)
+        func_name = getattr(func, "__name__", None)
+        cls_name = ""
+        if instance is not None and func_name:
+            try:
+                cls_name = instance.__class__.__name__
+            except AttributeError:
+                pass
 
-    if flag_added_args > 0:
-        args = args[flag_added_args:]
+        if flag_added_args > 0:
+            args = args[flag_added_args:]
 
-    # print(f"func! {func}")
-    # if hasattr(func, "__module__"):
-    #     print(f"func_name: {func_name}, module: {func.__module__}")
-    #     print(DEFAULT_SSRF_FUNCTIONS.get(func.__module__))
-    #     print(func_name in DEFAULT_SSRF_FUNCTIONS.get(func.__module__, ""))
+        if (
+            instance.__class__.__module__ == "random"
+            and cls_name == "Random"
+            and func_name in DEFAULT_WEAK_RANDOMNESS_FUNCTIONS
+        ):
+            if is_iast_request_enabled():
+                if WeakRandomness.has_quota():
+                    WeakRandomness.report(evidence_value=cls_name + "." + func_name)
 
-    if (
-        instance.__class__.__module__ == "random"
-        and cls_name == "Random"
-        and func_name in DEFAULT_WEAK_RANDOMNESS_FUNCTIONS
-    ):
-        if is_iast_request_enabled():
-            if WeakRandomness.has_quota():
-                WeakRandomness.report(evidence_value=cls_name + "." + func_name)
+                # Reports Span Metrics
+                increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakRandomness.vulnerability_type)
+                # Report Telemetry Metrics
+                _set_metric_iast_executed_sink(WeakRandomness.vulnerability_type)
 
-            # Reports Span Metrics
-            increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakRandomness.vulnerability_type)
-            # Report Telemetry Metrics
-            _set_metric_iast_executed_sink(WeakRandomness.vulnerability_type)
-
-    elif (
-        hasattr(func, "__module__")
-        and DEFAULT_PATH_TRAVERSAL_FUNCTIONS.get(func.__module__)
-        and func_name in DEFAULT_PATH_TRAVERSAL_FUNCTIONS[func.__module__]
-    ):
-        check_and_report_path_traversal(*args, **kwargs)
-    elif (
-        hasattr(func, "__module__")
-        and DEFAULT_COMMAND_INJECTION_FUNCTIONS.get(func.__module__)
-        and func_name in DEFAULT_COMMAND_INJECTION_FUNCTIONS[func.__module__]
-    ):
-        _iast_report_cmdi(func_name, *args, **kwargs)
-    elif (
-        hasattr(func, "__module__")
-        and DEFAULT_SSRF_FUNCTIONS.get(func.__module__)
-        and func_name in DEFAULT_SSRF_FUNCTIONS[func.__module__]
-    ):
-        _iast_report_ssrf(func_name, func.__module__, *args, **kwargs)
+        elif (
+            hasattr(func, "__module__")
+            and DEFAULT_PATH_TRAVERSAL_FUNCTIONS.get(func.__module__)
+            and func_name in DEFAULT_PATH_TRAVERSAL_FUNCTIONS[func.__module__]
+        ):
+            check_and_report_path_traversal(*args, **kwargs)
+        elif (
+            hasattr(func, "__module__")
+            and DEFAULT_COMMAND_INJECTION_FUNCTIONS.get(func.__module__)
+            and func_name in DEFAULT_COMMAND_INJECTION_FUNCTIONS[func.__module__]
+        ):
+            _iast_report_cmdi(func_name, *args, **kwargs)
+        elif (
+            hasattr(func, "__module__")
+            and DEFAULT_SSRF_FUNCTIONS.get(func.__module__)
+            and func_name in DEFAULT_SSRF_FUNCTIONS[func.__module__]
+        ):
+            _iast_report_ssrf(func_name, func.__module__, *args, **kwargs)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in ast_function", e)
     return func(*args, **kwargs)
diff --git a/ddtrace/appsec/_iast/taint_sinks/weak_cipher.py b/ddtrace/appsec/_iast/taint_sinks/weak_cipher.py
@@ -15,6 +15,7 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.settings.asm import config as asm_config
 
+from .._logs import iast_error
 from .._metrics import _set_metric_iast_executed_sink
 from .._metrics import _set_metric_iast_instrumented_sink
 from .._patch_modules import WrapFunctonsForIAST
@@ -124,52 +125,58 @@ def wrapped_aux_blowfish_function(wrapped, instance, args, kwargs):
 
 
 def wrapped_rc4_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        if WeakCipher.has_quota():
-            WeakCipher.report(
-                evidence_value="RC4",
-            )
-        # Reports Span Metrics
-        increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakCipher.vulnerability_type)
-        # Report Telemetry Metrics
-        _set_metric_iast_executed_sink(WeakCipher.vulnerability_type)
-
-    if hasattr(wrapped, "__func__"):
-        return wrapped.__func__(instance, *args, **kwargs)
-    return wrapped(*args, **kwargs)
-
-
-def wrapped_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        if hasattr(instance, "_dd_weakcipher_algorithm"):
+    try:
+        if is_iast_request_enabled():
             if WeakCipher.has_quota():
-                evidence = instance._dd_weakcipher_algorithm + "_" + str(instance.__class__.__name__)
-                WeakCipher.report(evidence_value=evidence)
-
+                WeakCipher.report(
+                    evidence_value="RC4",
+                )
             # Reports Span Metrics
             increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakCipher.vulnerability_type)
             # Report Telemetry Metrics
             _set_metric_iast_executed_sink(WeakCipher.vulnerability_type)
-
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_cipher.wrapped_rc4_function", e)
     if hasattr(wrapped, "__func__"):
         return wrapped.__func__(instance, *args, **kwargs)
     return wrapped(*args, **kwargs)
 
 
-def wrapped_cryptography_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        algorithm_name = instance.algorithm.name.lower()
-        if algorithm_name in get_weak_cipher_algorithms():
-            if WeakCipher.has_quota():
-                WeakCipher.report(
-                    evidence_value=algorithm_name,
-                )
+def wrapped_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
+    try:
+        if is_iast_request_enabled():
+            if hasattr(instance, "_dd_weakcipher_algorithm"):
+                if WeakCipher.has_quota():
+                    evidence = instance._dd_weakcipher_algorithm + "_" + str(instance.__class__.__name__)
+                    WeakCipher.report(evidence_value=evidence)
+
+                # Reports Span Metrics
+                increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakCipher.vulnerability_type)
+                # Report Telemetry Metrics
+                _set_metric_iast_executed_sink(WeakCipher.vulnerability_type)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_cipher.wrapped_function", e)
+    if hasattr(wrapped, "__func__"):
+        return wrapped.__func__(instance, *args, **kwargs)
+    return wrapped(*args, **kwargs)
 
-            # Reports Span Metrics
-            increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakCipher.vulnerability_type)
-            # Report Telemetry Metrics
-            _set_metric_iast_executed_sink(WeakCipher.vulnerability_type)
 
+def wrapped_cryptography_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
+    try:
+        if is_iast_request_enabled():
+            algorithm_name = instance.algorithm.name.lower()
+            if algorithm_name in get_weak_cipher_algorithms():
+                if WeakCipher.has_quota():
+                    WeakCipher.report(
+                        evidence_value=algorithm_name,
+                    )
+
+                # Reports Span Metrics
+                increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakCipher.vulnerability_type)
+                # Report Telemetry Metrics
+                _set_metric_iast_executed_sink(WeakCipher.vulnerability_type)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_cipher.wrapped_cryptography_function", e)
     if hasattr(wrapped, "__func__"):
         return wrapped.__func__(instance, *args, **kwargs)
     return wrapped(*args, **kwargs)
diff --git a/ddtrace/appsec/_iast/taint_sinks/weak_hash.py b/ddtrace/appsec/_iast/taint_sinks/weak_hash.py
@@ -11,6 +11,7 @@
 from .._iast_request_context_base import get_hash_object_tracking
 from .._iast_request_context_base import is_iast_request_enabled
 from .._iast_request_context_base import set_hash_object_tracking
+from .._logs import iast_error
 from .._metrics import _set_metric_iast_executed_sink
 from .._metrics import _set_metric_iast_instrumented_sink
 from .._patch_modules import WrapFunctonsForIAST
@@ -120,26 +121,32 @@ def wrapped_init_function(wrapped: Callable, instance: Any, args: Any, kwargs: A
         res = wrapped.__func__(instance, *args, **kwargs)
     else:
         res = wrapped(*args, **kwargs)
-    if is_iast_request_enabled():
-        set_hash_object_tracking(res, kwargs.get("usedforsecurity", None) is False)
+    try:
+        if is_iast_request_enabled():
+            set_hash_object_tracking(res, kwargs.get("usedforsecurity", None) is False)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_hash.wrapped_init_function", e)
     return res
 
 
 def wrapped_digest_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        if (
-            WeakHash.has_quota()
-            and instance.name.lower() in get_weak_hash_algorithms()
-            and get_hash_object_tracking(instance) is False
-        ):
-            WeakHash.report(
-                evidence_value=instance.name,
-            )
-
-        # Reports Span Metrics
-        increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
-        # Report Telemetry Metrics
-        _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    try:
+        if is_iast_request_enabled():
+            if (
+                WeakHash.has_quota()
+                and instance.name.lower() in get_weak_hash_algorithms()
+                and get_hash_object_tracking(instance) is False
+            ):
+                WeakHash.report(
+                    evidence_value=instance.name,
+                )
+
+            # Reports Span Metrics
+            increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
+            # Report Telemetry Metrics
+            _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_hash.wrapped_digest_function", e)
 
     if hasattr(wrapped, "__func__"):
         return wrapped.__func__(instance, *args, **kwargs)
@@ -155,31 +162,37 @@ def wrapped_sha1_function(wrapped: Callable, instance: Any, args: Any, kwargs: A
 
 
 def wrapped_new_function(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        if WeakHash.has_quota() and args[0].lower() in get_weak_hash_algorithms():
-            WeakHash.report(
-                evidence_value=args[0].lower(),
-            )
-        # Reports Span Metrics
-        increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
-        # Report Telemetry Metrics
-        _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    try:
+        if is_iast_request_enabled():
+            if WeakHash.has_quota() and args[0].lower() in get_weak_hash_algorithms():
+                WeakHash.report(
+                    evidence_value=args[0].lower(),
+                )
+            # Reports Span Metrics
+            increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
+            # Report Telemetry Metrics
+            _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_hash.wrapped_new_function", e)
 
     if hasattr(wrapped, "__func__"):
         return wrapped.__func__(instance, *args, **kwargs)
     return wrapped(*args, **kwargs)
 
 
 def wrapped_function(wrapped: Callable, evidence: str, instance: Any, args: Any, kwargs: Any) -> Any:
-    if is_iast_request_enabled():
-        if WeakHash.has_quota():
-            WeakHash.report(
-                evidence_value=evidence,
-            )
-        # Reports Span Metrics
-        increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
-        # Report Telemetry Metrics
-        _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    try:
+        if is_iast_request_enabled():
+            if WeakHash.has_quota():
+                WeakHash.report(
+                    evidence_value=evidence,
+                )
+            # Reports Span Metrics
+            increment_iast_span_metric(IAST_SPAN_TAGS.TELEMETRY_EXECUTED_SINK, WeakHash.vulnerability_type)
+            # Report Telemetry Metrics
+            _set_metric_iast_executed_sink(WeakHash.vulnerability_type)
+    except Exception as e:
+        iast_error("propagation::sink_point::Error in weak_hash.wrapped_function", e)
 
     if hasattr(wrapped, "__func__"):
         return wrapped.__func__(instance, *args, **kwargs)
diff --git a/releasenotes/notes/iast-fix-error-with-no-context-4b805f224001601f.yaml b/releasenotes/notes/iast-fix-error-with-no-context-4b805f224001601f.yaml
@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    IAST: Fixed an issue where using weak hashing or cipher algorithms outside of a request context 
+    (e.g., during application startup) could raise an unhandled exception. The fix ensures proper error 
+    handling when IAST operations are performed without an active request context.
diff --git a/tests/appsec/iast/taint_sinks/test_vulnerability_detection.py b/tests/appsec/iast/taint_sinks/test_vulnerability_detection.py
@@ -14,6 +14,7 @@
 from ddtrace.appsec._iast.sampling.vulnerability_detection import reset_request_vulnerabilities
 from ddtrace.appsec._iast.sampling.vulnerability_detection import should_process_vulnerability
 from ddtrace.appsec._iast.sampling.vulnerability_detection import update_global_vulnerability_limit
+from ddtrace.appsec._iast.taint_sinks._base import VulnerabilityBase
 from tests.appsec.iast.iast_utils import _end_iast_context_and_oce
 from tests.appsec.iast.iast_utils import _start_iast_context_and_oce
 from tests.utils import override_global_config
@@ -292,3 +293,10 @@ def test_with_modified_max_vulnerabilities_config():
         # Global map should be updated with all processed vulnerabilities
         update_global_vulnerability_limit()
         assert len(_get_global_limit()["GET:/config_test"]) == 3
+
+
+def test_quota_out_of_context():
+    _end_iast_context_and_oce()
+    env = _get_iast_env()
+    assert env is None
+    assert VulnerabilityBase.has_quota() is False
diff --git a/tests/appsec/iast/taint_sinks/test_weak_cipher.py b/tests/appsec/iast/taint_sinks/test_weak_cipher.py
@@ -1,3 +1,5 @@
+from unittest import mock
+
 import pytest
 
 from ddtrace.appsec._iast._iast_request_context import get_iast_reporter
@@ -215,3 +217,18 @@ def test_weak_cipher_secure_multiple_calls_error(iast_context_defaults):
     span_report = get_iast_reporter()
 
     assert span_report is None
+
+
+@mock.patch("ddtrace.appsec._iast.taint_sinks.weak_hash.is_iast_request_enabled")
+@mock.patch("ddtrace.appsec._iast.taint_sinks.weak_hash.increment_iast_span_metric")
+def test_weak_cipher_out_of_context(
+    mock_is_iast_request_enabled, mock_increment_iast_span_metric, iast_context_defaults
+):
+    mock_is_iast_request_enabled.return_value = True
+    mock_increment_iast_span_metric.side_effect = Exception(
+        "increment_iast_span_metric should not be called in this test"
+    )
+    try:
+        cryptography_algorithm("Blowfish")
+    except Exception as e:
+        pytest.fail(f"parametrized_weak_hash raised an exception: {e}")
diff --git a/tests/appsec/iast/taint_sinks/test_weak_hash.py b/tests/appsec/iast/taint_sinks/test_weak_hash.py
diff --git a/tests/contrib/flask/test_flask_pytest_iast.py b/tests/contrib/flask/test_flask_pytest_iast.py
