Cursor demo PR: Add Appsec SSI tests

tests/auto_inject/test_auto_inject_install.py

@@ -83,6 +83,32 @@ def test_profiling(self):
         self._test_install(context.virtual_machine, profile=True)
 
 
+@features.simple_auto_injection_appsec
+@scenarios.simple_auto_injection_appsec
+class TestSimpleAutoInjectManualAppSec(base.AutoInjectBaseTest):
+    def test_appsec(self):
+        logger.info(f"Launching test_install for : [{context.vm_name}]...")
+        self._test_install(context.virtual_machine, appsec=True)
+        logger.info(f"Done test_install for : [{context.vm_name}]")
+
+
+@features.host_auto_installation_script_appsec
+@scenarios.host_auto_injection_install_script_appsec
+class TestHostAutoInjectInstallScriptAppSec(base.AutoInjectBaseTest):
+    @missing_feature(context.vm_os_branch == "windows", reason="Not implemented on Windows")
+    def test_appsec(self):
+        logger.info(f"Launching test_install for : [{context.vm_name}]...")
+        self._test_install(context.virtual_machine, appsec=True)
+        logger.info(f"Done test_install for : [{context.vm_name}]")
+
+
+@features.container_auto_installation_script_appsec
+@scenarios.container_auto_injection_install_script_appsec
+class TestContainerAutoInjectInstallScriptAppSec(base.AutoInjectBaseTest):
+    def test_appsec(self):
+        self._test_install(context.virtual_machine, appsec=True)
+
+
 @features.installer_auto_instrumentation
 @scenarios.installer_auto_injection
 @irrelevant(condition=context.weblog_variant == "test-app-dotnet-iis")

tests/auto_inject/utils.py

@@ -6,22 +6,55 @@
 from threading import Timer
 
 
+def _validate_appsec_trace(trace_id, trace_data):
+    """Validator function for AppSec traces - checks for required AppSec metrics and meta tags"""
+    logger.info(f"Validating AppSec trace {trace_id}")
+
+    try:
+        root_id = trace_data["trace"]["root_id"]
+        root_span = trace_data["trace"]["spans"][root_id]
+
+        meta = root_span.get("meta", {})
+        metrics = root_span.get("metrics", {})
+
+        # Check for _dd.appsec.enabled metric set to 1
+        appsec_enabled_found = "_dd.appsec.enabled" in metrics and metrics["_dd.appsec.enabled"] == 1
+
+        # Check for appsec.event meta tag set to "true"
+        appsec_event_found = "appsec.event" in meta and meta["appsec.event"] == "true"
+
+        if appsec_enabled_found:
+            logger.info("AppSec trace validation successful - found _dd.appsec.enabled=1 in root span")
+            if appsec_event_found:
+                logger.info("AppSec event validation successful - found appsec.event=true in root span")
+            return True
+        else:
+            logger.error("AppSec trace validation failed - _dd.appsec.enabled=1 not found in root span")
+            logger.debug(f"Root span metrics: {metrics}")
+            logger.debug(f"Root span meta: {meta}")
+            return False
+
+    except Exception as e:
+        logger.error(f"Error validating AppSec trace: {e}")
+        return False
+
+
 class AutoInjectBaseTest:
-    def _test_install(self, virtual_machine, *, profile: bool = False):
+    def _test_install(self, virtual_machine, *, profile: bool = False, appsec: bool = False):
         """If there is a multicontainer app, we need to make a request to each app"""
 
         if virtual_machine.get_deployed_weblog().app_type == "multicontainer":
             for app in virtual_machine.get_deployed_weblog().multicontainer_apps:
                 vm_context_url = (
                     f"http://{virtual_machine.get_ip()}:{virtual_machine.deffault_open_port}{app.app_context_url}"
                 )
-                self._check_install(virtual_machine, vm_context_url, profile=profile)
+                self._check_install(virtual_machine, vm_context_url, profile=profile, appsec=appsec)
 
         else:
             vm_context_url = f"http://{virtual_machine.get_ip()}:{virtual_machine.deffault_open_port}{virtual_machine.get_deployed_weblog().app_context_url}"
-            self._check_install(virtual_machine, vm_context_url, profile=profile)
+            self._check_install(virtual_machine, vm_context_url, profile=profile, appsec=appsec)
 
-    def _check_install(self, virtual_machine, vm_context_url, *, profile: bool = False):
+    def _check_install(self, virtual_machine, vm_context_url, *, profile: bool = False, appsec: bool = False):
         """We can easily install agent and lib injection software from agent installation script. Given a  sample application we can enable tracing using local environment variables.
         After starting application we can see application HTTP requests traces in the backend.
         Using the agent installation script we can install different versions of the software (release or beta) in different OS.
@@ -47,7 +80,12 @@ def _check_install(self, virtual_machine, vm_context_url, *, profile: bool = Fal
             logger.info(f"Http request done with uuid: [{request_uuid}] for ip [{vm_ip}]")
 
         try:
-            wait_backend_trace_id(request_uuid, profile=profile)
+            # Use AppSec validator if appsec=True, otherwise use existing behavior
+            if appsec:
+                logger.info("Using AppSec validator for trace validation")
+                wait_backend_trace_id(request_uuid, validator=_validate_appsec_trace)
+            else:
+                wait_backend_trace_id(request_uuid, profile=profile)
         except (TimeoutError, AssertionError) as e:
             self._log_trace_debug_message(e, request_uuid)
             raise

utils/_context/_scenarios/__init__.py

@@ -829,6 +829,36 @@ class _Scenarios:
         github_workflow="aws_ssi",
     )
 
+    simple_auto_injection_appsec = InstallerAutoInjectionScenario(
+        "SIMPLE_AUTO_INJECTION_APPSEC",
+        "Onboarding Single Step Instrumentation scenario with AppSec activated by the app env var",
+        app_env={
+            "DD_APPSEC_ENABLED": "true",
+        },
+        scenario_groups=[scenario_groups.all, scenario_groups.simple_onboarding_appsec],
+        github_workflow="aws_ssi",
+    )
+    host_auto_injection_install_script_appsec = InstallerAutoInjectionScenario(
+        "HOST_AUTO_INJECTION_INSTALL_SCRIPT_APPSEC",
+        doc=(
+            "Onboarding Host Single Step Instrumentation scenario using agent "
+            "auto install script with AppSec activating by the installation process"
+        ),
+        vm_provision="host-auto-inject-install-script",
+        agent_env={"DD_APPSEC_ENABLED": "true"},
+        scenario_groups=[scenario_groups.all],
+        github_workflow="aws_ssi",
+    )
+
+    container_auto_injection_install_script_appsec = InstallerAutoInjectionScenario(
+        "CONTAINER_AUTO_INJECTION_INSTALL_SCRIPT_APPSEC",
+        "Onboarding Container Single Step Instrumentation AppSec scenario using agent auto install script",
+        vm_provision="container-auto-inject-install-script",
+        agent_env={"DD_APPSEC_ENABLED": "true"},
+        scenario_groups=[scenario_groups.all],
+        github_workflow="aws_ssi",
+    )
+
     demo_aws = InstallerAutoInjectionScenario(
         "DEMO_AWS",
         "Demo aws scenario",

utils/_context/_scenarios/core.py

@@ -41,6 +41,7 @@ class _ScenarioGroups:
     onboarding = ScenarioGroup()
     simple_onboarding = ScenarioGroup()
     simple_onboarding_profiling = ScenarioGroup()
+    simple_onboarding_appsec = ScenarioGroup()
     docker_ssi = ScenarioGroup()
     essentials = ScenarioGroup()
     external_processing = ScenarioGroup()

utils/_features.py

@@ -2122,6 +2122,33 @@ def container_auto_instrumentation_profiling(test_object):
         pytest.mark.features(feature_id=310)(test_object)
         return test_object
 
+    @staticmethod
+    def simple_auto_injection_appsec(test_object):
+        """AppSec works when enabled through environment variables in SSI environments
+
+        https://feature-parity.us1.prod.dog/#/?feature=478
+        """
+        pytest.mark.features(feature_id=478)(test_object)
+        return test_object
+
+    @staticmethod
+    def host_auto_installation_script_appsec(test_object):
+        """AppSec works when enabled through the agent installer script in Host environments
+
+        https://feature-parity.us1.prod.dog/#/?feature=479
+        """
+        pytest.mark.features(feature_id=479)(test_object)
+        return test_object
+
+    @staticmethod
+    def container_auto_installation_script_appsec(test_object):
+        """AppSec works when enabled through the agent installer script in Container environments
+
+        https://feature-parity.us1.prod.dog/#/?feature=480
+        """
+        pytest.mark.features(feature_id=480)(test_object)
+        return test_object
+
     @staticmethod
     def host_guardrail(test_object):
         """When in SSI, bail out if our version of language is incompatible.

utils/scripts/ci_orchestrators/aws_ssi.json

@@ -215,6 +215,89 @@
                 }
             ]
         },
+        {
+            "scenarios": [
+                "SIMPLE_AUTO_INJECTION_APPSEC"
+            ],
+            "weblogs": [
+                {
+                    "nodejs": [
+                        "test-app-nodejs",
+                        "test-app-nodejs-container"
+                    ],
+                    "java": [
+                        "test-app-java",
+                        "test-app-java-container",
+                        "test-app-java-alpine"
+                    ],
+                    "python": [
+                        "test-app-python",
+                        "test-app-python-container",
+                        "test-app-python-alpine"
+                    ],
+                    "dotnet": [
+                        "test-app-dotnet",
+                        "test-app-dotnet-container"
+                    ],
+                    "php": [
+                        "test-app-php",
+                        "test-app-php-container-83",
+                        "test-app-php-alpine"
+                    ]
+                }
+            ]
+        },
+        {
+            "scenarios": [
+                "HOST_AUTO_INJECTION_INSTALL_SCRIPT_APPSEC"
+            ],
+            "weblogs": [
+                {
+                    "nodejs": [
+                        "test-app-nodejs"
+                    ],
+                    "java": [
+                        "test-app-java"
+                    ],
+                    "python": [
+                        "test-app-python"
+                    ],
+                    "dotnet": [
+                        "test-app-dotnet"
+                    ],
+                    "php": [
+                        "test-app-php"
+                    ]
+                }
+            ]
+        },
+        {
+            "scenarios": [
+                "CONTAINER_AUTO_INJECTION_INSTALL_SCRIPT_APPSEC"
+            ],
+            "weblogs": [
+                {
+                    "nodejs": [
+                        "test-app-nodejs-multicontainer"
+                    ],
+                    "java": [
+                        "test-app-java-multicontainer",
+                        "test-app-java-multialpine"
+                    ],
+                    "python": [
+                        "test-app-python-container",
+                        "test-app-python-alpine"
+                    ],
+                    "dotnet": [
+                        "test-app-dotnet-container"
+                    ],
+                    "php": [
+                        "test-app-php-container-83",
+                        "test-app-php-alpine"
+                    ]
+                }
+            ]
+        },
         {
             "scenarios": [
                 "SIMPLE_INSTALLER_AUTO_INJECTION"