fix(iast): weak hash error if vulnerability is outside the context (#15029)

avara1986 · avara1986 · commit 11b9fbdc311b · 2025-10-27T10:16:26.000+01:00
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
