refactor(appsec): extract rel_path and get_caller_frame_info to _patch_utils

ddtrace/appsec/_iast/taint_sinks/_base.py

@@ -1,5 +1,3 @@
-import os
-import sysconfig
 from typing import Optional
 from typing import Union
 
@@ -8,6 +6,7 @@
 from ddtrace.appsec._iast._taint_tracking import get_ranges
 from ddtrace.appsec._iast.sampling.vulnerability_detection import rollback_quota
 from ddtrace.appsec._iast.sampling.vulnerability_detection import should_process_vulnerability
+from ddtrace.appsec._patch_utils import get_caller_frame_info
 from ddtrace.appsec._trace_utils import _asm_manual_keep
 from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger
@@ -19,7 +18,6 @@
 from .._iast_request_context import get_iast_reporter
 from .._iast_request_context import set_iast_reporter
 from .._span_metrics import increment_iast_span_metric
-from .._stacktrace import get_info_frame
 from ..reporter import Evidence
 from ..reporter import IastSpanReporter
 from ..reporter import Location
@@ -28,13 +26,8 @@
 
 log = get_logger(__name__)
 
-CWD = os.path.abspath(os.getcwd())
-
 TEXT_TYPES = Union[str, bytes, bytearray]
 
-PURELIB_PATH = sysconfig.get_path("purelib")
-STDLIB_PATH = sysconfig.get_path("stdlib")
-
 
 class taint_sink_deduplication(deduplication):
     def _check_deduplication(self):
@@ -111,40 +104,6 @@ def _prepare_report(
 
         return True
 
-    @classmethod
-    def _compute_file_line(cls) -> tuple[Optional[str], Optional[int], Optional[str], Optional[str]]:
-        file_name = line_number = function_name = class_name = None
-        frame_info = get_info_frame()
-        if not frame_info or frame_info[0] in ("", -1):
-            return file_name, line_number, function_name, class_name
-
-        file_name, line_number, function_name, class_name = frame_info
-        if not file_name:
-            return None, None, None, None
-
-        file_name = cls._rel_path(file_name)
-        if not file_name:
-            log.debug("Could not relativize vulnerability location path: %s", frame_info[0])
-            return None, None, None, None
-
-        return file_name, line_number, function_name, class_name
-
-    @staticmethod
-    def _rel_path(file_name: str) -> str:
-        file_name_norm = file_name.replace("\\", "/")
-        if file_name_norm.startswith(PURELIB_PATH):
-            return os.path.relpath(file_name_norm, start=PURELIB_PATH)
-
-        if file_name_norm.startswith(STDLIB_PATH):
-            return os.path.relpath(file_name_norm, start=STDLIB_PATH)
-        if file_name_norm.startswith(CWD):
-            return os.path.relpath(file_name_norm, start=CWD)
-        # If the path contains site-packages anywhere, return 'site-packages/<rest>'
-        # Normalize separators to forward slashes for consistency
-        if (idx := file_name_norm.find("/site-packages/")) != -1:
-            return file_name_norm[idx:]
-        return ""
-
     @classmethod
     def _create_evidence_and_report(
         cls,
@@ -177,7 +136,7 @@ def report(cls, evidence_value: TEXT_TYPES = "", dialect: Optional[str] = None)
             file_name = line_number = function_name = class_name = None
 
             if not getattr(cls, "skip_location", False):
-                file_name, line_number, function_name, class_name = cls._compute_file_line()
+                file_name, line_number, function_name, class_name = get_caller_frame_info()
                 if file_name is None:
                     rollback_quota(cls.vulnerability_type)
                     return result

ddtrace/appsec/_iast/taint_sinks/insecure_cookie.py

@@ -13,6 +13,7 @@
 from ddtrace.appsec._iast.constants import VULN_NO_SAMESITE_COOKIE
 from ddtrace.appsec._iast.sampling.vulnerability_detection import should_process_vulnerability
 from ddtrace.appsec._iast.taint_sinks._base import VulnerabilityBase
+from ddtrace.appsec._patch_utils import get_caller_frame_info
 from ddtrace.internal.settings.asm import config as asm_config
 
 
@@ -38,7 +39,7 @@ def report_cookies(cls, evidence_value, insecure_cookie, no_http_only, no_samesi
         """Build a IastSpanReporter instance to report it in the `AppSecIastSpanProcessor` as a string JSON"""
         if insecure_cookie or no_http_only or no_samesite:
             if should_process_vulnerability(InsecureCookie.vulnerability_type):
-                file_name, line_number, function_name, class_name = cls._compute_file_line()
+                file_name, line_number, function_name, class_name = get_caller_frame_info()
                 if file_name is None:
                     return
                 if insecure_cookie:

ddtrace/appsec/_patch_utils.py

@@ -1,4 +1,6 @@
 import ctypes
+import os
+import sysconfig
 from typing import Any
 from typing import Callable
 from typing import Optional
@@ -12,6 +14,64 @@
 
 
 log = get_logger(__name__)
+
+# Cached paths for relativizing file paths (computed once at import time).
+_CWD = os.path.abspath(os.getcwd())
+_PURELIB_PATH = sysconfig.get_path("purelib") or ""
+_STDLIB_PATH = sysconfig.get_path("stdlib") or ""
+
+
+def rel_path(file_name: str) -> str:
+    """Relativize an absolute file path for vulnerability/reachability reporting.
+
+    Used by both IAST and SCA to produce short, readable paths in telemetry
+    payloads.  Tries purelib first, then stdlib, then CWD-relative, then
+    site-packages.  Returns empty string if the path cannot be relativized.
+    """
+    file_name_norm = file_name.replace("\\", "/")
+    if file_name_norm.startswith(_PURELIB_PATH):
+        return os.path.relpath(file_name_norm, start=_PURELIB_PATH)
+
+    if file_name_norm.startswith(_STDLIB_PATH):
+        return os.path.relpath(file_name_norm, start=_STDLIB_PATH)
+    if file_name_norm.startswith(_CWD):
+        return os.path.relpath(file_name_norm, start=_CWD)
+    # If the path contains site-packages anywhere, return 'site-packages/<rest>'
+    # Normalize separators to forward slashes for consistency
+    if (idx := file_name_norm.find("/site-packages/")) != -1:
+        return file_name_norm[idx:]
+    return ""
+
+
+def get_caller_frame_info() -> tuple:
+    """Walk the stack and return (file_name, line_number, function_name, class_name).
+
+    Uses the native C get_info_frame() to skip ddtrace, stdlib, and special
+    frames, then relativizes the path.  Shared by IAST vulnerability
+    reporting and SCA reachability detection.
+
+    Returns (None, None, None, None) when no relevant frame is found.
+    """
+    try:
+        from ddtrace.appsec._shared._stacktrace import get_info_frame
+    except ImportError:
+        return None, None, None, None
+
+    frame_info = get_info_frame()
+    if not frame_info or frame_info[0] in ("", -1, None):
+        return None, None, None, None
+
+    file_name, line_number, function_name, class_name = frame_info
+    if not file_name:
+        return None, None, None, None
+
+    file_name = rel_path(file_name)
+    if not file_name:
+        return None, None, None, None
+
+    return file_name, line_number, function_name, class_name
+
+
 _DD_ORIGINAL_ATTRIBUTES: dict[Any, Any] = {}
 
 

ddtrace/appsec/_iast/_stacktrace.c → ddtrace/appsec/_shared/_stacktrace.c

@@ -359,7 +359,7 @@ static PyMethodDef StacktraceMethods[] = { { "get_info_frame",
                                            { NULL, NULL, 0, NULL } };
 
 static struct PyModuleDef stacktrace = { PyModuleDef_HEAD_INIT,
-                                         "ddtrace.appsec._iast._stacktrace",
+                                         "ddtrace.appsec._shared._stacktrace",
                                          "stacktrace module",
                                          -1,
                                          StacktraceMethods };

ddtrace/appsec/_shared/_stacktrace.pyi

@@ -0,0 +1,3 @@
+from typing import Optional
+
+def get_info_frame() -> tuple[Optional[str], Optional[int], Optional[str], Optional[str]]: ...

setup.py

@@ -1275,9 +1275,9 @@ def get_exts_for(name):
     if platform.system() not in ("Windows", ""):
         ext_modules.append(
             Extension(
-                "ddtrace.appsec._iast._stacktrace",
+                "ddtrace.appsec._shared._stacktrace",
                 sources=[
-                    "ddtrace/appsec/_iast/_stacktrace.c",
+                    "ddtrace/appsec/_shared/_stacktrace.c",
                 ],
                 extra_compile_args=extra_compile_args + debug_compile_args + fast_build_args,
             )

tests/appsec/architectures/test_appsec_loading_modules.py

@@ -17,7 +17,7 @@
 MODULE_IAST_ONLY = [
     "ddtrace.appsec._iast",
     "ddtrace.appsec._iast._taint_tracking._native",
-    "ddtrace.appsec._iast._stacktrace",
+    "ddtrace.appsec._shared._stacktrace",
 ]
 
 

tests/appsec/iast/taint_sinks/test_weak_hash.py

@@ -1,5 +1,4 @@
 from contextlib import contextmanager
-import os
 import sys
 from unittest import mock
 
@@ -113,10 +112,9 @@ def test_safe_weak_hash_hashlib(iast_context_defaults, hash_func, method, usedfo
     ],
 )
 def test_ensure_line_reported_is_minus_one_for_edge_cases(iast_context_defaults, hash_func, method, fake_line):
-    absolute_path = os.path.abspath(WEAK_ALGOS_FIXTURES_PATH)
     with mock.patch(
-        "ddtrace.appsec._iast.taint_sinks._base.get_info_frame",
-        return_value=(absolute_path, fake_line, "", ""),
+        "ddtrace.appsec._iast.taint_sinks._base.get_caller_frame_info",
+        return_value=(WEAK_ALGOS_FIXTURES_PATH, fake_line, "", ""),
     ):
         parametrized_weak_hash(hash_func, method)
 

tests/appsec/iast/test_stacktrace.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-from ddtrace.appsec._iast._stacktrace import get_info_frame
+from ddtrace.appsec._shared._stacktrace import get_info_frame
 
 
 def test_stacktrace():

tests/appsec/iast_memcheck/fixtures/stacktrace.py

@@ -1,6 +1,6 @@
 import os
 
-from ddtrace.appsec._iast._stacktrace import get_info_frame
+from ddtrace.appsec._shared._stacktrace import get_info_frame
 
 
 CWD = os.path.abspath(os.getcwd())

tests/appsec/iast_memcheck/test_iast_mem_check.py

@@ -8,11 +8,11 @@
 from ddtrace.appsec._iast._iast_request_context_base import _iast_finish_request
 from ddtrace.appsec._iast._iast_request_context_base import _iast_start_request
 from ddtrace.appsec._iast._iast_request_context_base import _num_objects_tainted_in_request
-from ddtrace.appsec._iast._stacktrace import get_info_frame
 from ddtrace.appsec._iast._taint_tracking import OriginType
 from ddtrace.appsec._iast._taint_tracking._context import debug_context_array_size
 from ddtrace.appsec._iast._taint_tracking._taint_objects import taint_pyobject
 from ddtrace.appsec._iast._taint_tracking._taint_objects_base import get_tainted_ranges
+from ddtrace.appsec._shared._stacktrace import get_info_frame
 from tests.appsec.iast.iast_utils import _iast_patched_module
 from tests.appsec.iast_memcheck.fixtures.stacktrace import func_1
 

tests/internal/test_serverless.py

@@ -46,7 +46,7 @@ def test_not_azure_function():
     "ddtrace.appsec._iast._ast.iastpatch",
     "ddtrace.appsec._iast._taint_tracking._native",
     "ddtrace.appsec._iast._taint_tracking._vendor",
-    "ddtrace.appsec._iast._stacktrace",
+    "ddtrace.appsec._shared._stacktrace",
     "ddtrace.internal.datadog.profiling.libdd_wrapper",
     "ddtrace.internal.datadog.profiling.ddup._ddup",
     "ddtrace.internal.datadog.profiling.stack._stack",