Use synthetic SCA advisories for regular and Lambda weblogs

manifests/python_lambda.yml

@@ -266,7 +266,7 @@ manifest:
   tests/appsec/test_reports.py::Test_Info: v7.112.0
   tests/appsec/test_reports.py::Test_RequestHeaders: v7.112.0
   tests/appsec/test_reports.py::Test_StatusCode: v7.112.0
-  tests/appsec/test_sca_reachability.py: incomplete_test_app (Lambda SCA dependency fixtures and vulnerable-call endpoints are not implemented)
+  tests/appsec/test_sca_reachability.py: missing_feature (requires Lambda SCA reachability support in dd-trace-py and datadog-lambda-python)
   tests/appsec/test_trace_tagging.py::Test_TraceTaggingRules: v8.113.0
   tests/appsec/test_traces.py::Test_AppSecEventSpanTags: v7.112.0
   tests/appsec/test_traces.py::Test_AppSecEventSpanTags::test_header_collection: bug (APPSEC-58202)

tests/appsec/sca/java/advisories.json

@@ -0,0 +1,18 @@
+{
+  "version": 1,
+  "entries": [
+    {
+      "vuln_id": "system-tests-sca",
+      "artifact": "com.github.junrar:junrar",
+      "version_ranges": [
+        ">=0"
+      ],
+      "symbols": [
+        {
+          "class": "com/github/junrar/LocalFolderExtractor",
+          "method": "createDirectory"
+        }
+      ]
+    }
+  ]
+}

tests/appsec/sca/python/advisories.json

@@ -0,0 +1,20 @@
+{
+  "targets": [
+    {
+      "id": "system-tests-sca",
+      "lang": "python",
+      "dependency_name": "requests",
+      "package_versions": [
+        ">=0"
+      ],
+      "targets": [
+        "requests.sessions:Session.send"
+      ],
+      "vulnerability": {
+        "id": "system-tests-sca",
+        "severity": "LOW",
+        "description": "Synthetic requests target for system-tests SCA reachability coverage."
+      }
+    }
+  ]
+}

tests/appsec/test_sca_reachability.py

@@ -18,23 +18,28 @@
 # Only populate entries once the language's tracer supports SCA reachability
 # and the expected values are confirmed. Missing languages gracefully
 # degrade: structural assertions still run, but value comparisons are skipped.
-# `cve_id` is a list to support retrocompatibility across tracer versions that
-# may emit different identifiers (e.g. CVE id vs GHSA id) for the same advisory.
+# Regular and Lambda scenarios use the same synthetic advisory for each language.
 _LANG_CONFIG: dict[str, dict[str, Any]] = {
     "python": {
-        "cve_id": ["CVE-2024-35195", "GHSA-652x-xj99-gmcc"],
+        "cve_id": ["system-tests-sca"],
         "vulnerable_dep": "requests",
         "path": "app.py",
         "symbol": "sca_requests_vulnerable_call",
     },
     "java": {
-        "cve_id": ["GHSA-hf5p-q87m-crj7"],
+        "cve_id": ["system-tests-sca"],
         "vulnerable_dep": "junrar",
         "path": "com.datadoghq.system_tests.springboot.ScaReachability",
         "symbol": "junrarHit",
     },
+    "python_lambda": {
+        "cve_id": ["system-tests-sca"],
+        "vulnerable_dep": "requests",
+        "path": "handler.py",
+        "symbol": "sca_requests_vulnerable_call",
+    },
     "java_lambda": {
-        "cve_id": ["GHSA-hf5p-q87m-crj7"],
+        "cve_id": ["system-tests-sca"],
         "vulnerable_dep": "junrar",
         "path": "com.datadoghq.lambda.ScaReachability",
         "symbol": "junrarHit",

utils/_context/_scenarios/_core/sca.py

@@ -0,0 +1,27 @@
+from utils._context.containers import TestedContainer
+
+
+def configure_sca_advisories(container: TestedContainer) -> None:
+    library = container.image.labels["system-tests-library"]
+    match library:
+        case "python_lambda":
+            catalog_path = "/opt/python/lib/python3.13/site-packages/ddtrace/appsec/sca/_cve_data.json"
+        case "python" if any(volume["bind"] == "/volumes/dd-trace-py" for volume in container.volumes.values()):
+            catalog_path = "/volumes/dd-trace-py/ddtrace/appsec/sca/_cve_data.json"
+        case "python":
+            assert container.image.env is not None
+            python_version = ".".join(container.image.env["PYTHON_VERSION"].split(".")[:2])
+            catalog_path = f"/usr/local/lib/python{python_version}/site-packages/ddtrace/appsec/sca/_cve_data.json"
+        case "java" | "java_lambda":
+            catalog_path = "/sca/sca_cves.json"
+            # The agent checks bootstrap resources before its bundled classpath resources.
+            java_options = (
+                container.environment.get("JAVA_TOOL_OPTIONS", (container.image.env or {}).get("JAVA_TOOL_OPTIONS", ""))
+                or ""
+            )
+            container.environment["JAVA_TOOL_OPTIONS"] = f"{java_options} -Xbootclasspath/a:/sca".strip()
+        case _:
+            return
+
+    language = library.removesuffix("_lambda")
+    container.volumes[f"./tests/appsec/sca/{language}/advisories.json"] = {"bind": catalog_path, "mode": "ro"}

utils/_context/_scenarios/appsec.py

@@ -1,6 +1,7 @@
 import pytest
 
 from ._core import DdTraceEndToEndScenario, EndToEndScenario, scenario_groups
+from ._core.sca import configure_sca_advisories
 from utils._context.constants import WeblogCategory
 from utils._context.containers import PostgresContainer
 
@@ -152,6 +153,10 @@ def __init__(self) -> None:
             scenario_groups=[scenario_groups.appsec],
         )
 
+    def configure(self, config: pytest.Config) -> None:
+        super().configure(config)
+        configure_sca_advisories(self.weblog_infra.library_container)
+
 
 class IastDeduplication(DdTraceEndToEndScenario):
     def __init__(self) -> None:

utils/_context/_scenarios/aws_lambda.py

@@ -2,6 +2,7 @@
 from utils import interfaces
 from utils.interfaces._core import ProxyBasedInterfaceValidator
 from ._core import DockerScenario, scenario_groups, scenario_groups as all_scenario_groups, ScenarioGroup
+from ._core.sca import configure_sca_advisories
 from utils._context.constants import WeblogCategory
 from utils._context.containers import LambdaProxyContainer, LambdaWeblogContainer, InternalServerContainer
 from utils._logger import logger
@@ -215,11 +216,18 @@ class AppsecLambdaSca(_LambdaScenario):
     def __init__(self) -> None:
         super().__init__(
             "APPSEC_LAMBDA_SCA",
-            weblog_env={"DD_APPSEC_ENABLED": "false", "DD_APPSEC_SCA_ENABLED": "true"},
+            weblog_env={
+                "DD_APPSEC_ENABLED": "false",
+                "DD_APPSEC_SCA_ENABLED": "true",
+            },
             doc="SCA dependency metadata and runtime reachability in Lambda with AppSec disabled",
             scenario_groups=[scenario_groups.appsec, scenario_groups.appsec_lambda],
         )
 
+    def configure(self, config: pytest.Config) -> None:
+        super().configure(config)
+        configure_sca_advisories(self.lambda_weblog)
+
 
 class AppsecLambdaRasp(_LambdaScenario):
     def __init__(self) -> None:

utils/build/docker/java/spring-boot/src/main/java/com/datadoghq/system_tests/springboot/ScaReachability.java

@@ -15,7 +15,7 @@ public class ScaReachability {
   @PostConstruct
   private void preloadVulnerableClass() {
     // Force LocalFolderExtractor class load during Spring startup so the SCA transformer can
-    // register GHSA-hf5p-q87m-crj7 with reached=[] before any HTTP request triggers junrarHit().
+    // register system-tests-sca with reached=[] before any HTTP request triggers junrarHit().
     // Without this, the class loads lazily on the first request, so registerCve() and recordHit()
     // fire in the same request with no heartbeat window between them.
     try {

utils/build/docker/java_lambda/src/main/java/com/datadoghq/lambda/ScaReachability.java

@@ -9,7 +9,7 @@ public class ScaReachability {
 
     public static void preloadVulnerableClass() {
         // Force LocalFolderExtractor class load during Lambda initialization so the SCA transformer can
-        // register GHSA-hf5p-q87m-crj7 with reached=[] before any HTTP request triggers junrarHit().
+        // register system-tests-sca with reached=[] before any HTTP request triggers junrarHit().
         // Without this, the class loads lazily on the first request, so registerCve() and recordHit()
         // fire in the same request with no heartbeat window between them.
         try {

utils/build/docker/python/flask/app.py

@@ -2386,7 +2386,7 @@ async def llm():
 
 @app.route("/sca/vulnerable-call")
 def sca_requests_vulnerable_call():
-    """Trigger a call to requests.sessions.Session.send (CVE-2024-35195 target)."""
+    """Trigger the synthetic advisory's requests.sessions.Session.send target."""
     session = requests.Session()
     try:
         session.get("http://localhost:1")
@@ -2397,7 +2397,7 @@ def sca_requests_vulnerable_call():
 
 @app.route("/sca/vulnerable-call-alt")
 def sca_requests_vulnerable_call_alt():
-    """Alternate call site for same CVE-2024-35195 target (first-hit-wins test)."""
+    """Alternate call site for the same synthetic advisory (first-hit-wins test)."""
     session = requests.Session()
     try:
         session.post("http://localhost:1", data="x")

utils/build/docker/python/flask/app.sh

@@ -1,10 +1,5 @@
 #!/bin/bash
 
-if [[ ${DD_APPSEC_SCA_ENABLED:-} = "1" || ${DD_APPSEC_SCA_ENABLED:-} = "true" ]]; then
-    # We need a requirement with a CVE to test SCA features
-    python -m pip install requests==2.31.0
-fi
-
 echo "--- PIP FREEZE ---"
 python -m pip freeze
 echo "------------------"

utils/build/docker/python/flask/uwsgi_app.sh

@@ -1,10 +1,5 @@
 #!/bin/bash
 
-if [[ ${DD_APPSEC_SCA_ENABLED:-} = "1" || ${DD_APPSEC_SCA_ENABLED:-} = "true" ]]; then
-    # We need a requirement with a CVE to test SCA features
-    python -m pip install requests==2.31.0
-fi
-
 echo "--- PIP FREEZE ---"
 python -m pip freeze
 echo "------------------"

utils/build/docker/python_lambda/function/handler.py

@@ -47,6 +47,28 @@
 _TRACK_CUSTOM_APPSEC_EVENT_NAME = "system_tests_appsec_event"
 
 
+@app.get("/sca/vulnerable-call")
+def sca_requests_vulnerable_call() -> str:
+    """Exercise the synthetic advisory's Session.send target."""
+    with requests.Session() as session:
+        try:
+            session.send(requests.Request("GET", "http://127.0.0.1:1").prepare(), timeout=1)
+        except requests.exceptions.ConnectionError:
+            pass
+    return "OK"
+
+
+@app.get("/sca/vulnerable-call-alt")
+def sca_requests_vulnerable_call_alt() -> str:
+    """Use a distinct caller for the same target to check first-hit deduplication."""
+    with requests.Session() as session:
+        try:
+            session.send(requests.Request("GET", "http://127.0.0.1:1").prepare(), timeout=1)
+        except requests.exceptions.ConnectionError:
+            pass
+    return "OK"
+
+
 def version_info():
     return {
         "status": "ok",

utils/scripts/libraries_and_scenarios_rules.yml

@@ -228,6 +228,10 @@ patterns:
         scenario_groups: docker_fixtures
     utils/_context/_scenarios/_core/endtoend.py:
         scenario_groups: end_to_end
+    utils/_context/_scenarios/_core/sca.py:
+        scenario_groups: null
+        scenarios: [APPSEC_LAMBDA_SCA, RUNTIME_SCA_REACHABILITY]
+        libraries: [java, java_lambda, python, python_lambda]
     utils/_context/_scenarios/_core/*:  # base classes shared by many scenarios: runs everything
     utils/_context/_scenarios/*/*:  # any other subfolder: not a definition file, runs everything
     utils/_context/_scenarios/open_telemetry.py:  # its scenarios only run with OTel libraries
@@ -589,7 +593,17 @@ patterns:
 
     tests/appsec/test_sca_reachability.py:
         scenario_groups: null
-        libraries: [java, java_lambda, python]
+        libraries: [java, java_lambda, python, python_lambda]
+
+    tests/appsec/sca/java/advisories.json:
+        scenario_groups: null
+        scenarios: [APPSEC_LAMBDA_SCA, RUNTIME_SCA_REACHABILITY]
+        libraries: [java, java_lambda]
+
+    tests/appsec/sca/python/advisories.json:
+        scenario_groups: null
+        scenarios: [APPSEC_LAMBDA_SCA, RUNTIME_SCA_REACHABILITY]
+        libraries: [python, python_lambda]
 
 ################################# Default rules ################################
 # The rules in this section are only applied if no other rule is applicable.