chore(iast): fix iast in the ci (#15175)

# Summary

This PR extends **PR #15141**'s fork safety implementation to properly
handle IAST in the CI/pytest environment. It fixes two critical issues:

1. **Threading race conditions** in FastAPI tests (Python 3.12) with
anyio's multi-threaded execution
2. **Fork safety in pytest** by ensuring IAST is always disabled in
child processes during test execution

The solution adds a pytest-specific mode flag that enhances the existing
fork handler without affecting production behavior.

## Context

This PR builds on top of **PR #15141** which implemented fork safety for
IAST by disabling it in multiprocess child processes. That PR
successfully prevented segfaults in production multiprocessing scenarios
but revealed additional issues in the CI/pytest environment.

Author: avara1986

ddtrace/appsec/_iast/__init__.py

@@ -46,6 +46,7 @@ def wrapped_function(wrapped, instance, args, kwargs):
 _IAST_TO_BE_LOADED = True
 _iast_propagation_enabled = False
 _fork_handler_registered = False
+_iast_in_pytest_mode = False
 
 
 def _disable_iast_after_fork():
@@ -87,6 +88,15 @@ def _disable_iast_after_fork():
         from ddtrace.appsec._iast._iast_request_context_base import is_iast_request_enabled
         from ddtrace.appsec._iast._taint_tracking._context import clear_all_request_context_slots
 
+        # In pytest mode, always disable IAST in child processes to avoid segfaults
+        # when tests create multiprocesses (e.g., for testing fork behavior)
+        if _iast_in_pytest_mode:
+            log.debug("IAST fork handler: Pytest mode detected, disabling IAST in child process")
+            clear_all_request_context_slots()
+            IAST_CONTEXT.set(None)
+            asm_config._iast_enabled = False
+            return
+
         if not is_iast_request_enabled():
             # No active context - this is an early fork (web framework worker)
             # IAST can be safely initialized fresh in this child process
@@ -168,6 +178,7 @@ def enable_iast_propagation():
     """Add IAST AST patching in the ModuleWatchdog"""
     # DEV: These imports are here to avoid _ast.ast_patching import in the top level
     # because they are slow and affect serverless startup time
+
     if asm_config._iast_propagation_enabled:
         from ddtrace.appsec._iast._ast.ast_patching import _should_iast_patch
         from ddtrace.appsec._iast._loader import _exec_iast_patched_module
@@ -183,19 +194,30 @@ def enable_iast_propagation():
 
 
 def _iast_pytest_activation():
-    global _iast_propagation_enabled
-    if _iast_propagation_enabled:
+    """Configure IAST settings for pytest execution.
+
+    This function sets up IAST configuration but does NOT create a request context.
+    Request contexts should be created per-test or per-request to avoid threading issues.
+
+    Also sets a global flag to indicate we're in pytest mode, which ensures IAST is
+    disabled in forked child processes to prevent segfaults when tests use multiprocessing.
+    """
+    global _iast_in_pytest_mode
+
+    if not asm_config._iast_enabled:
         return
-    os.environ["DD_IAST_ENABLED"] = os.environ.get("DD_IAST_ENABLED") or "1"
+
+    # Mark that we're running in pytest mode
+    # This flag is checked by the fork handler to disable IAST in child processes
+    _iast_in_pytest_mode = True
+
     os.environ["DD_IAST_REQUEST_SAMPLING"] = os.environ.get("DD_IAST_REQUEST_SAMPLING") or "100.0"
     os.environ["_DD_APPSEC_DEDUPLICATION_ENABLED"] = os.environ.get("_DD_APPSEC_DEDUPLICATION_ENABLED") or "false"
     os.environ["DD_IAST_VULNERABILITIES_PER_REQUEST"] = os.environ.get("DD_IAST_VULNERABILITIES_PER_REQUEST") or "1000"
-    os.environ["DD_IAST_MAX_CONCURRENT_REQUESTS"] = os.environ.get("DD_IAST_MAX_CONCURRENT_REQUESTS") or "1000"
 
     asm_config._iast_request_sampling = 100.0
     asm_config._deduplication_enabled = False
     asm_config._iast_max_vulnerabilities_per_requests = 1000
-    asm_config._iast_max_concurrent_requests = 1000
     oce.reconfigure()
 
 

ddtrace/settings/asm.py

@@ -140,6 +140,8 @@ class ASMConfig(DDConfig):
         + r"ey[I-L][\w=-]+\.ey[I-L][\w=-]+(\.[\w.+\/=-]+)?|[\-]{5}BEGIN[a-z\s]+PRIVATE\sKEY"
         + r"[\-]{5}[^\-]+[\-]{5}END[a-z\s]+PRIVATE\sKEY|ssh-rsa\s*[a-z0-9\/\.+]{100,}",
     )
+    # We never use `asm_config._iast_max_concurrent_requests` directly,
+    # but we define it so it can be reported through telemetry, since it’s used from the C files.
     _iast_max_concurrent_requests = DDConfig.var(
         int,
         IAST.DD_IAST_MAX_CONCURRENT_REQUESTS,

tests/appsec/iast/aspects/conftest.py

@@ -8,13 +8,7 @@
 @pytest.fixture(autouse=True)
 def iast_request():
     with override_global_config(
-        dict(
-            _iast_enabled=True,
-            _iast_is_testing=True,
-            _iast_deduplication_enabled=False,
-            _iast_request_sampling=100.0,
-            _iast_max_concurrent_requests=100,
-        )
+        dict(_iast_enabled=True, _iast_is_testing=True, _iast_deduplication_enabled=False, _iast_request_sampling=100.0)
     ):
         context_id = _start_iast_context_and_oce()
         assert context_id is not None

tests/appsec/iast/taint_tracking/conftest.py

@@ -9,13 +9,7 @@
 @pytest.fixture(autouse=True)
 def iast_request():
     with override_global_config(
-        dict(
-            _iast_enabled=True,
-            _iast_is_testing=True,
-            _iast_deduplication_enabled=False,
-            _iast_request_sampling=100.0,
-            _iast_max_concurrent_requests=100,
-        )
+        dict(_iast_enabled=True, _iast_is_testing=True, _iast_deduplication_enabled=False, _iast_request_sampling=100.0)
     ):
         assert debug_context_array_size() == 2
         _start_iast_context_and_oce()