o i*@sdZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl mZdd lmZeeZd ad ad ad ad d ZddZddZddZddZddZgdZddZdS)aIAST (Interactive Application Security Testing) analyzes code for security vulnerabilities. To add new vulnerabilities analyzers (Taint sink) we should update `IAST_PATCH` in `ddtrace/appsec/iast/_patch_modules.py` Create new file with the same name: `ddtrace/appsec/iast/taint_sinks/[my_new_vulnerability].py` Then, implement the `patch()` function and its wrappers. In order to have the better performance, the Overhead control engine (OCE) helps us to control the overhead of our wrapped functions. We should create a class that inherit from `ddtrace.appsec._iast.taint_sinks._base.VulnerabilityBase` and register with `ddtrace.appsec._iast.oce`. @oce.register class MyVulnerability(VulnerabilityBase): vulnerability_type = "MyVulnerability" evidence_type = "kind_of_Vulnerability" Before that, we should decorate our wrappers with `wrap` method and report the vulnerabilities with `report` method. OCE will manage the number of requests, number of vulnerabilities to reduce the overhead. @WeakHash.wrap def wrapped_function(wrapped, instance, args, kwargs): WeakHash.report( evidence_value=evidence, ) return wrapped(*args, **kwargs) N)forksafe) get_logger)ModuleWatchdog)config) iast_listen)oceTFc CstjsdSz>ddlm}ddlm}tr$td|ddt_WdS|s4td|dWdStd|ddt_WdSt y^}ztjd |d d WYd}~dSd}~ww) u Handle IAST state after fork to prevent segmentation faults. This fork handler works in conjunction with the C++ pthread_atfork handler (registered in native.cpp) to provide complete fork-safety: **C++ pthread_atfork handler (automatic, runs first):** - Clears all taint maps (removes stale PyObject pointers from parent) - Resets taint_engine_context (recreates context array with fresh state) - Resets initializer (recreates memory pools) - Happens automatically for ALL forks before this Python handler runs **Python fork handler (this function, runs second):** - Detects fork type (early vs late) - Manages Python-level IAST_CONTEXT - Conditionally disables IAST for late forks (multiprocessing) Fork types: 1. **Early forks (web framework workers)**: Gunicorn, uvicorn, etc. fork BEFORE IAST has active request contexts. Native state was reset by pthread_atfork. IAST remains enabled and works correctly in workers. 2. **Late forks (multiprocessing)**: fork AFTER IAST has active contexts. Native state was reset by pthread_atfork, but we disable IAST in these processes for multiprocessing compatibility and to avoid confusion. Detection logic: - If is_iast_request_enabled() → Late fork → Disable IAST - If not is_iast_request_enabled() → Early fork → Keep IAST enabled This prevents segmentation faults in ALL fork scenarios while maintaining IAST coverage in web framework workers. Nr) IAST_CONTEXT)is_iast_request_enabledzHIAST fork handler: Pytest mode detected, disabling IAST in child processFz~IAST fork handler: No active context (early fork/web worker). Native state auto-reset by pthread_atfork. IAST remains enabled.zIAST fork handler: Active context (late fork/multiprocessing). Native state auto-reset by pthread_atfork. Disabling IAST in child.zError in IAST fork handler: %sTexc_info) asm_config _iast_enabled/ddtrace.appsec._iast._iast_request_context_baser r _iast_in_pytest_modelogdebugset Exception)r r erQ/home/ubuntu/.local/lib/python3.10/site-packages/ddtrace/appsec/_iast/__init__.py_disable_iast_after_fork4s2"       rcCs.tstjrttdatddSdSdS)z Register the fork handler if IAST is enabled and it hasn't been registered yet. This is called during IAST initialization to ensure the fork handler is only registered once and only when IAST is actually being used. Tz#IAST fork safety handler registeredN)_fork_handler_registeredr rrregisterrrrrrrr_register_fork_handlers rcCstjsdSddl}ddlm}|jjd}tj |}z||\}}Wnt y5t j dddYdSw|s?t d dSt ||d }t|}|j|j|jt||jdS) ap Patch the code inside the Flask main app source code file (typically "app.py") so Runtime Code Analysis (IAST) works also for the functions and methods defined inside it. This must be called on the top level or inside the `if __name__ == "__main__"` and must be before the `app.run()` call. It also requires `DD_IAST_ENABLED` to be activated. Nrr)astpatch_module__name__z'Unexpected exception while AST patchingTr z9Main flask module not patched, probably it was not neededexec)r rinspect_ast.ast_patchingr currentframef_back f_globalssysmodulesrrrcompiletypes ModuleType__dict__clearupdater)rr module_namemodule module_path patched_ast compiled_code new_modulerrrddtrace_iast_flask_patchs(       r2cCs`tjr.ddlm}ddlm}ddlm}trdSt dt ||da|t dSdS)z+Add IAST AST patching in the ModuleWatchdogr_should_iast_patch_exec_iast_patched_module)initialize_native_stateNz$iast::instrumentation::starting IASTT) r _iast_propagation_enabled&ddtrace.appsec._iast._ast.ast_patchingr4ddtrace.appsec._iast._loaderr6$ddtrace.appsec._iast._taint_trackingr7rrrregister_pre_exec_module_hookr)r4r6r7rrrenable_iast_propagations      r=cCsntjsdSdatjdpdtjd<tjdpdtjd<tjdp$dtjd<d t_d t_d t_t dS) aConfigure 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. NTDD_IAST_REQUEST_SAMPLINGz100.0 _DD_APPSEC_DEDUPLICATION_ENABLEDfalse#DD_IAST_VULNERABILITIES_PER_REQUEST1000gY@Fi) r rrosenvironget_iast_request_sampling_deduplication_enabled&_iast_max_vulnerabilities_per_requestsr reconfigurerrrr_iast_pytest_activations  rJcCsZddlm}ddlm}tsdSz t||WdadSty,t dYdadSw)zKRemove IAST AST patching from the ModuleWatchdog. Only for testing proposesrr3r5Nz;IAST is already disabled and it's not in the ModuleWatchdogF) r9r4r:r6r8rremove_pre_exec_module_hookKeyErrorrwarning)r4r6rrrdisable_iast_propagations   rN)r2r=rNcCstr ttdadSdS)z&Lazily load the iast module listeners.FN)_IAST_TO_BE_LOADEDrrrrrr load_iast s rP)__doc__rCr$r'ddtrace.internalrddtrace.internal.loggerrddtrace.internal.modulerddtrace.internal.settings.asmrr _listenerr_overhead_control_enginerrrrOr8rrrrr2r=rJrN__all__rPrrrrs.      V(