feat(sca): runtime sca detection

Author: avara1986

ddtrace/appsec/sca/_cve_data.json

@@ -2,10 +2,10 @@
   "targets": [
     {
       "id": "vuln-001",
-      "target": "requests.sessions:Session.send",
+      "targets": ["requests.sessions:Session.send"],
       "lang": "python",
       "dependency_name": "requests",
-      "package_versions": "<2.32.0",
+      "package_versions": ["<2.32.0"],
       "vulnerability": {
         "id": "CVE-2024-35195",
         "severity": "MEDIUM",
@@ -14,10 +14,10 @@
     },
     {
       "id": "vuln-002",
-      "target": "urllib3._collections:HTTPHeaderDict.__setitem__",
+      "targets": ["urllib3._collections:HTTPHeaderDict.__setitem__"],
       "lang": "python",
       "dependency_name": "urllib3",
-      "package_versions": "<2.2.2",
+      "package_versions": ["<2.2.2"],
       "vulnerability": {
         "id": "CVE-2024-37891",
         "severity": "MEDIUM",
@@ -26,10 +26,10 @@
     },
     {
       "id": "vuln-003",
-      "target": "jinja2.filters:do_xmlattr",
+      "targets": ["jinja2.filters:do_xmlattr"],
       "lang": "python",
       "dependency_name": "jinja2",
-      "package_versions": "<3.1.3",
+      "package_versions": ["<3.1.3"],
       "vulnerability": {
         "id": "CVE-2024-22195",
         "severity": "MEDIUM",
@@ -38,10 +38,10 @@
     },
     {
       "id": "vuln-004",
-      "target": "jinja2.compiler:generate",
+      "targets": ["jinja2.compiler:generate"],
       "lang": "python",
       "dependency_name": "jinja2",
-      "package_versions": "<3.1.5",
+      "package_versions": ["<3.1.5"],
       "vulnerability": {
         "id": "CVE-2024-56201",
         "severity": "HIGH",
@@ -50,10 +50,10 @@
     },
     {
       "id": "vuln-005",
-      "target": "cryptography.hazmat.primitives.serialization.pkcs12:serialize_key_and_certificates",
+      "targets": ["cryptography.hazmat.primitives.serialization.pkcs12:serialize_key_and_certificates"],
       "lang": "python",
       "dependency_name": "cryptography",
-      "package_versions": "<42.0.4",
+      "package_versions": ["<42.0.4"],
       "vulnerability": {
         "id": "CVE-2024-26130",
         "severity": "HIGH",
@@ -62,10 +62,10 @@
     },
     {
       "id": "vuln-006",
-      "target": "cryptography.x509:load_pem_x509_certificate",
+      "targets": ["cryptography.x509:load_pem_x509_certificate"],
       "lang": "python",
       "dependency_name": "cryptography",
-      "package_versions": "<43.0.1",
+      "package_versions": ["<43.0.1"],
       "vulnerability": {
         "id": "CVE-2024-6119",
         "severity": "HIGH",

ddtrace/appsec/sca/_cve_loader.py

@@ -30,7 +30,7 @@ def _parse_version_constraint(constraint: str) -> Optional[tuple]:
         Tuple of (operator_str, Version) or None if parsing fails.
     """
     constraint = constraint.strip()
-    for op in ("<=", ">=", "==", "!=", "<", ">"):
+    for op in ("<=", ">=", "==", "!=", "<", ">", "="):
         if constraint.startswith(op):
             ver_str = constraint[len(op) :].strip()
             try:
@@ -79,9 +79,33 @@ def _version_matches(installed_version: str, constraint: str) -> bool:
         return installed == target
     elif op == "!=":
         return installed != target
+    elif op == "=":
+        return installed == target
     return False
 
 
+def _compound_constraint_matches(installed_version: str, compound: str) -> bool:
+    """Check if an installed version satisfies a compound constraint.
+
+    A compound constraint is a comma-separated list of sub-constraints
+    that must ALL match (AND logic), e.g. ">=42.2.0, <42.2.28".
+    """
+    parts = [p.strip() for p in compound.split(",") if p.strip()]
+    if not parts:
+        return False
+    return all(_version_matches(installed_version, part) for part in parts)
+
+
+def _any_version_matches(installed_version: str, constraints: list[str]) -> bool:
+    """Check if an installed version satisfies ANY constraint in the list (OR logic).
+
+    Each constraint may itself be a compound constraint (comma-separated AND).
+    """
+    if not constraints:
+        return False
+    return any(_compound_constraint_matches(installed_version, c) for c in constraints)
+
+
 def load_cve_targets(installed_packages: dict[str, str]) -> list[dict[str, Any]]:
     """Load CVE targets from the static JSON, filtering by installed versions.
 
@@ -110,37 +134,41 @@ def load_cve_targets(installed_packages: dict[str, str]) -> list[dict[str, Any]]
             # Package not installed — skip
             continue
 
-        constraint = entry.get("package_versions", "")
-        if not _version_matches(installed_ver, constraint):
+        constraints = entry.get("package_versions", [])
+        if not _any_version_matches(installed_ver, constraints):
             log.debug(
                 "Skipping %s: installed %s does not match %s",
                 entry.get("vulnerability", {}).get("id", "?"),
                 installed_ver,
-                constraint,
+                constraints,
             )
             continue
 
         vuln = entry.get("vulnerability", {})
         cve_id = vuln.get("id", "")
-        target_name = entry.get("target", "")
+        targets = entry.get("targets", [])
 
-        if not target_name or not cve_id:
+        if not targets or not cve_id:
             continue
 
-        applicable_targets.append(
-            {
-                "target": target_name,
-                "dependency_name": dep_name,
-                "cve_id": cve_id,
-            }
-        )
-        log.debug(
-            "CVE %s applies to %s %s (constraint %s)",
-            cve_id,
-            dep_name,
-            installed_ver,
-            constraint,
-        )
+        for target_name in targets:
+            if not target_name:
+                continue
+            applicable_targets.append(
+                {
+                    "target": target_name,
+                    "dependency_name": dep_name,
+                    "cve_id": cve_id,
+                }
+            )
+            log.debug(
+                "CVE %s applies to %s %s (constraint %s, target %s)",
+                cve_id,
+                dep_name,
+                installed_ver,
+                constraints,
+                target_name,
+            )
 
     log.debug("Loaded %d applicable CVE targets out of %d total", len(applicable_targets), len(data.get("targets", [])))
     return applicable_targets

ddtrace/appsec/sca/_instrumenter.py

@@ -52,10 +52,10 @@ def _first_instr_line(code: types.CodeType) -> int:
 def _get_caller_info() -> tuple[str, int, str]:
     """Walk the stack to find the first user-code caller frame.
 
-    Returns (path, line, method) where:
+    Returns (path, line, symbol) where:
     - path: relative file path of the caller
     - line: line number in the caller
-    - method: function (or Class.method) name of the caller
+    - symbol: function (or Class.method) name of the caller
 
     AIDEV-NOTE: Delegates to the shared get_caller_frame_info() which uses
     IAST's native C get_info_frame() + rel_path().  SCA just reshapes the
@@ -66,8 +66,8 @@ def _get_caller_info() -> tuple[str, int, str]:
         if not file_name:
             return "", 0, ""
 
-        method = f"{class_name}.{function_name}" if class_name else (function_name or "")
-        return file_name, line_number or 0, method
+        symbol = f"{class_name}.{function_name}" if class_name else (function_name or "")
+        return file_name, line_number or 0, symbol
     except Exception:
         log.debug("Failed to get caller info via get_caller_frame_info", exc_info=True)
         return "", 0, ""
@@ -153,23 +153,23 @@ def sca_detection_hook(qualified_name: str) -> None:
         writer = _get_telemetry_writer()
         # AIDEV-NOTE: Walk the stack to find the user-code frame that called
         # the vulnerable function, similar to IAST's _compute_file_line.
-        # Reports the caller's path/line/method, not the target function's.
-        caller_path, caller_line, caller_method = _get_caller_info()
+        # Reports the caller's path/line/symbol, not the target function's.
+        caller_path, caller_line, caller_symbol = _get_caller_info()
 
         # AIDEV-NOTE: If the native frame walker can't find user code (e.g.,
         # deep wrapt/gevent stack), fall back to the target's own qualified
         # name so the backend knows the function was reached.  Without this,
         # add_metadata's `if path` guard silently drops the finding and
         # reached stays [].
         if not caller_path:
-            # Use "module.path:Class.method" as path, the method part after ":"
+            # Use "module.path:Class.method" as path, the symbol part after ":"
             parts = qualified_name.split(":", 1)
             caller_path = parts[0] if parts else qualified_name
-            caller_method = parts[1] if len(parts) > 1 else ""
+            caller_symbol = parts[1] if len(parts) > 1 else ""
             caller_line = 0
 
         for cve_id in target_info.cve_ids:
-            writer.attach_dependency_metadata(target_info.package_name, cve_id, caller_path, caller_method, caller_line)
+            writer.attach_dependency_metadata(target_info.package_name, cve_id, caller_path, caller_symbol, caller_line)
 
     except Exception:
         log.debug("SCA detection hook error for %s", qualified_name, exc_info=True)

ddtrace/internal/telemetry/dependency.py

@@ -40,7 +40,7 @@ class ReachabilityMetadata:
     def cve_id(self) -> Optional[str]:
         return self.value.get("id")
 
-    def add_reached_entry(self, path: str, method: str, line: int) -> bool:
+    def add_reached_entry(self, path: str, symbol: str, line: int) -> bool:
         """Add a hit entry to the reached list.
 
         Returns True if the entry was added, False if the list is already
@@ -49,7 +49,7 @@ def add_reached_entry(self, path: str, method: str, line: int) -> bool:
         reached = self.value["reached"]  # always initialized at construction
         if len(reached) >= self._MAX_REACHED_ENTRIES:
             return False
-        reached.append({"path": path, "method": method, "line": line})
+        reached.append({"path": path, "symbol": symbol, "line": line})
         self._sent = False
         return True
 
@@ -134,7 +134,7 @@ def mark_all_metadata_sent(self) -> None:
         for m in self.metadata:
             m._mark_sent()
 
-    def add_metadata(self, cve_id: str, path: str = "", method: str = "", line: int = 0) -> bool:
+    def add_metadata(self, cve_id: str, path: str = "", symbol: str = "", line: int = 0) -> bool:
         """Add or update reachability metadata for a CVE.
 
         AIDEV-NOTE: RFC v3 — one metadata entry per CVE.  If the CVE already
@@ -145,7 +145,7 @@ def add_metadata(self, cve_id: str, path: str = "", method: str = "", line: int
         Args:
             cve_id: CVE identifier (required).
             path: Caller file path (empty for registration-only).
-            method: Caller method name (empty for registration-only).
+            symbol: Caller symbol name (empty for registration-only).
             line: Caller line number (0 for registration-only).
 
         Returns:
@@ -161,15 +161,15 @@ def add_metadata(self, cve_id: str, path: str = "", method: str = "", line: int
         for existing in self.metadata:
             if existing.cve_id == cve_id:
                 # CVE already registered — add hit if call-site info provided
-                if path and existing.add_reached_entry(path, method, line):
+                if path and existing.add_reached_entry(path, symbol, line):
                     return True
                 return False
 
         if len(self.metadata) >= self._MAX_METADATA_ENTRIES:
             return False
 
         # Create new metadata entry for this CVE
-        reached = [{"path": path, "method": method, "line": line}] if path else []
+        reached = [{"path": path, "symbol": symbol, "line": line}] if path else []
         meta = ReachabilityMetadata(
             type="reachability",
             value={"id": cve_id, "reached": reached},
@@ -203,7 +203,7 @@ def attach_reachability_metadata(
     package_name: str,
     cve_id: str,
     path: str,
-    method: str,
+    symbol: str,
     line: int,
 ) -> bool:
     """Attach reachability metadata to an already-tracked dependency.
@@ -220,7 +220,7 @@ def attach_reachability_metadata(
         log.debug("Cannot attach metadata: package %r not yet tracked", package_name)
         return False
 
-    return entry.add_metadata(cve_id, path, method, line)
+    return entry.add_metadata(cve_id, path, symbol, line)
 
 
 def register_cve_metadata(

ddtrace/internal/telemetry/dependency_tracker.py

@@ -125,7 +125,7 @@ def attach_metadata(
         package_name: str,
         cve_id: str,
         path: str,
-        method: str,
+        symbol: str,
         line: int,
     ) -> bool:
         """Attach reachability metadata to an imported dependency.
@@ -141,7 +141,7 @@ def attach_metadata(
         """
         with self._lock:
             self._ensure_entry(package_name)
-            return attach_reachability_metadata(self._imported_dependencies, package_name, cve_id, path, method, line)
+            return attach_reachability_metadata(self._imported_dependencies, package_name, cve_id, path, symbol, line)
 
     def register_cve(self, package_name: str, cve_id: str) -> bool:
         """Register a CVE on a dependency with reached=[].

ddtrace/internal/telemetry/writer.py

@@ -349,14 +349,14 @@ def attach_dependency_metadata(
         package_name: str,
         cve_id: str,
         path: str,
-        method: str,
+        symbol: str,
         line: int,
     ) -> bool:
         """Attach reachability metadata to an imported dependency.
 
         Delegates to DependencyTracker.attach_metadata().
         """
-        return self._dependency_tracker.attach_metadata(package_name, cve_id, path, method, line)
+        return self._dependency_tracker.attach_metadata(package_name, cve_id, path, symbol, line)
 
     def register_cve_metadata(self, package_name: str, cve_id: str) -> bool:
         """Register a CVE on a dependency with reached=[].