feat(asm): standalone code security

.github/workflows/system-tests.yml

@@ -133,6 +133,11 @@ jobs:
           docker load < images_artifacts/${{ matrix.weblog-variant}}_weblog_${{ github.sha }}.tar.gz
           docker load < images_artifacts/agent_${{ github.sha }}.tar.gz
 
+      # TODO: Enable once https://github.com/DataDog/system-tests/pull/3506 is merged
+      # - name: Run IAST_STANDALONE
+      #   if: always() && steps.docker_load.outcome == 'success' && matrix.scenario == 'appsec-1'
+      #   run: ./run.sh IAST_STANDALONE
+
       - name: Run DEFAULT
         if: always() && steps.docker_load.outcome == 'success' && matrix.scenario == 'other'
         run: ./run.sh DEFAULT

ddtrace/_trace/tracer.py

@@ -233,9 +233,10 @@ def __init__(
         # _user_sampler is the backup in case we need to revert from remote config to local
         self._user_sampler: Optional[BaseSampler] = DatadogSampler()
         self._asm_enabled = asm_config._asm_enabled
+        self._iast_enabled = asm_config._iast_enabled
         self._appsec_standalone_enabled = asm_config._appsec_standalone_enabled
         self._dogstatsd_url = agent.get_stats_url() if dogstatsd_url is None else dogstatsd_url
-        self._apm_opt_out = self._asm_enabled and self._appsec_standalone_enabled
+        self._apm_opt_out = (self._asm_enabled or self._iast_enabled) and self._appsec_standalone_enabled
         if self._apm_opt_out:
             self.enabled = False
             # Disable compute stats (neither agent or tracer should compute them)
@@ -267,7 +268,6 @@ def __init__(
         self._partial_flush_min_spans = config._partial_flush_min_spans
         # Direct link to the appsec processor
         self._appsec_processor = None
-        self._iast_enabled = asm_config._iast_enabled
         self._endpoint_call_counter_span_processor = EndpointCallCounterProcessor()
         self._span_processors, self._appsec_processor, self._deferred_processors = _default_span_processors_factory(
             self._filters,
@@ -498,7 +498,7 @@ def configure(
         if appsec_standalone_enabled is not None:
             self._appsec_standalone_enabled = asm_config._appsec_standalone_enabled = appsec_standalone_enabled
 
-        if self._appsec_standalone_enabled and self._asm_enabled:
+        if self._appsec_standalone_enabled and (self._asm_enabled or self._iast_enabled):
             self._apm_opt_out = True
             self.enabled = False
             # Disable compute stats (neither agent or tracer should compute them)

ddtrace/appsec/_iast/_iast_request_context.py

@@ -18,7 +18,6 @@
 from ddtrace.appsec._iast._metrics import _set_span_tag_iast_executed_sink
 from ddtrace.appsec._iast._metrics import _set_span_tag_iast_request_tainted
 from ddtrace.appsec._iast.reporter import IastSpanReporter
-from ddtrace.appsec._trace_utils import _asm_manual_keep
 from ddtrace.constants import ORIGIN_KEY
 from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger
@@ -132,7 +131,6 @@ def _iast_end_request(ctx=None, span=None, *args, **kwargs):
                 if report_data:
                     report_data.build_and_scrub_value_parts()
                     req_span.set_tag_str(IAST.JSON, report_data._to_str())
-                    _asm_manual_keep(req_span)
                 _set_metric_iast_request_tainted()
                 _set_span_tag_iast_request_tainted(req_span)
                 _set_span_tag_iast_executed_sink(req_span)

ddtrace/appsec/_iast/taint_sinks/_base.py

@@ -5,6 +5,7 @@
 from typing import Text
 
 from ddtrace import tracer
+from ddtrace.appsec._trace_utils import _asm_manual_keep
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.utils.cache import LFUCache
 
@@ -90,6 +91,11 @@ def _prepare_report(cls, vulnerability_type, evidence, file_name, line_number):
             span = tracer.current_root_span()
             if span:
                 span_id = span.span_id
+                # Mark the span as kept to avoid being dropped by the agent.
+                #
+                # It is important to do it as soon as the vulnerability is reported
+                # to ensure that any downstream propagation performed has the new priority.
+                _asm_manual_keep(span)
 
         vulnerability = Vulnerability(
             type=vulnerability_type,

releasenotes/notes/feat-code-security-standalone-0fc5993ded38e83e.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Code Security: This introduces "Standalone Code Security", a feature that disables APM in the tracer but keeps Code Security (IAST) enabled. In order to enable it, set the environment variables ``DD_IAST_ENABLED=1`` and ``DD_EXPERIMENTAL_APPSEC_STANDALONE_ENABLED=1``.

tests/appsec/appsec/test_asm_standalone.py

@@ -7,27 +7,35 @@
 
 @pytest.fixture(
     params=[
-        {"appsec_enabled": True, "appsec_standalone_enabled": True},
-        {"appsec_enabled": True, "appsec_standalone_enabled": False},
-        {"appsec_enabled": False, "appsec_standalone_enabled": False},
-        {"appsec_enabled": False, "appsec_standalone_enabled": True},
+        {"iast_enabled": True, "appsec_enabled": True, "appsec_standalone_enabled": True},
+        {"iast_enabled": True, "appsec_enabled": True, "appsec_standalone_enabled": False},
+        {"iast_enabled": True, "appsec_enabled": False, "appsec_standalone_enabled": False},
+        {"iast_enabled": True, "appsec_enabled": False, "appsec_standalone_enabled": True},
+        {"iast_enabled": False, "appsec_enabled": True, "appsec_standalone_enabled": True},
+        {"iast_enabled": False, "appsec_enabled": True, "appsec_standalone_enabled": False},
+        {"iast_enabled": False, "appsec_enabled": False, "appsec_standalone_enabled": False},
+        {"iast_enabled": False, "appsec_enabled": False, "appsec_standalone_enabled": True},
         {"appsec_enabled": True},
         {"appsec_enabled": False},
+        {"iast_enabled": True},
+        {"iast_enabled": False},
     ]
 )
 def tracer_appsec_standalone(request, tracer):
     tracer.configure(api_version="v0.4", **request.param)
     yield tracer, request.param
     # Reset tracer configuration
-    tracer.configure(api_version="v0.4", appsec_enabled=False, appsec_standalone_enabled=False)
+    tracer.configure(api_version="v0.4", appsec_enabled=False, appsec_standalone_enabled=False, iast_enabled=False)
 
 
 def test_appsec_standalone_apm_enabled_metric(tracer_appsec_standalone):
     tracer, args = tracer_appsec_standalone
     with tracer.trace("test", span_type=SpanTypes.WEB) as span:
         set_http_meta(span, {}, raw_uri="http://example.com/.git", status_code="404")
 
-    if args == {"appsec_enabled": True, "appsec_standalone_enabled": True}:
+    if args.get("appsec_standalone_enabled", None) and (
+        args.get("appsec_enabled", None) or args.get("iast_enabled", None)
+    ):
         assert span.get_metric("_dd.apm.enabled") == 0.0
     else:
         assert span.get_metric("_dd.apm.enabled") is None

tests/tracer/test_propagation.py

@@ -317,8 +317,15 @@ def test_extract(tracer):  # noqa: F811
         assert len(context.get_all_baggage_items()) == 3
 
 
-def test_asm_standalone_minimum_trace_per_minute_has_no_downstream_propagation(tracer):  # noqa: F811
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
+def test_asm_standalone_minimum_trace_per_minute_has_no_downstream_propagation(
+    tracer, appsec_enabled, iast_enabled  # noqa: F811
+):
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
+    tracer.configure(appsec_enabled=appsec_enabled, appsec_standalone_enabled=True, iast_enabled=iast_enabled)
     try:
         headers = {
             "x-datadog-trace-id": "1234",
@@ -362,8 +369,15 @@ def test_asm_standalone_minimum_trace_per_minute_has_no_downstream_propagation(t
         tracer.configure(appsec_enabled=False, appsec_standalone_enabled=False)
 
 
-def test_asm_standalone_missing_propagation_tags_no_appsec_event_trace_dropped(tracer):  # noqa: F811
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
+def test_asm_standalone_missing_propagation_tags_no_appsec_event_trace_dropped(
+    tracer, appsec_enabled, iast_enabled  # noqa: F811
+):
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
+    tracer.configure(appsec_enabled=appsec_enabled, appsec_standalone_enabled=True, iast_enabled=iast_enabled)
     try:
         with tracer.trace("local_root_span0"):
             # First span should be kept, as we keep 1 per min
@@ -428,10 +442,15 @@ def test_asm_standalone_missing_propagation_tags_appsec_event_present_trace_kept
         tracer.configure(appsec_enabled=False, appsec_standalone_enabled=False)
 
 
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
 def test_asm_standalone_missing_appsec_tag_no_appsec_event_propagation_resets(
-    tracer,  # noqa: F811
+    tracer, appsec_enabled, iast_enabled  # noqa: F811
 ):
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
+    tracer.configure(appsec_enabled=appsec_enabled, appsec_standalone_enabled=True, iast_enabled=iast_enabled)
     try:
         with tracer.trace("local_root_span0"):
             # First span should be kept, as we keep 1 per min
@@ -526,10 +545,15 @@ def test_asm_standalone_missing_appsec_tag_appsec_event_present_trace_kept(
 
 
 @pytest.mark.parametrize("upstream_priority", ["1", "2"])
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
 def test_asm_standalone_present_appsec_tag_no_appsec_event_propagation_set_to_user_keep(
-    tracer, upstream_priority  # noqa: F811
+    tracer, upstream_priority, appsec_enabled, iast_enabled  # noqa: F811
 ):
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
+    tracer.configure(appsec_enabled=appsec_enabled, appsec_standalone_enabled=True, iast_enabled=iast_enabled)
     try:
         with tracer.trace("local_root_span0"):
             # First span should be kept, as we keep 1 per min
@@ -585,10 +609,15 @@ def test_asm_standalone_present_appsec_tag_no_appsec_event_propagation_set_to_us
 
 
 @pytest.mark.parametrize("upstream_priority", ["1", "2"])
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
 def test_asm_standalone_present_appsec_tag_appsec_event_present_propagation_force_keep(
-    tracer, upstream_priority  # noqa: F811
+    tracer, upstream_priority, appsec_enabled, iast_enabled  # noqa: F811
 ):
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
+    tracer.configure(appsec_enabled=appsec_enabled, appsec_standalone_enabled=True, iast_enabled=iast_enabled)
     try:
         with tracer.trace("local_root_span0"):
             # First span should be kept, as we keep 1 per min

tests/tracer/test_tracer.py

@@ -2043,10 +2043,19 @@ def test_import_ddtrace_tracer_not_module():
     assert isinstance(tracer, Tracer)
 
 
-def test_asm_standalone_configuration():
+@pytest.mark.parametrize("appsec_enabled", [True, False])
+@pytest.mark.parametrize("iast_enabled", [True, False])
+def test_asm_standalone_configuration(appsec_enabled, iast_enabled):
+    if not appsec_enabled and not iast_enabled:
+        pytest.skip("AppSec or IAST must be enabled")
+
     tracer = ddtrace.Tracer()
-    tracer.configure(appsec_enabled=True, appsec_standalone_enabled=True)
-    assert tracer._asm_enabled is True
+    tracer.configure(appsec_enabled=appsec_enabled, iast_enabled=iast_enabled, appsec_standalone_enabled=True)
+    if appsec_enabled:
+        assert tracer._asm_enabled is True
+    if iast_enabled:
+        assert tracer._iast_enabled is True
+
     assert tracer._appsec_standalone_enabled is True
     assert tracer._apm_opt_out is True
     assert tracer.enabled is False
@@ -2057,7 +2066,7 @@ def test_asm_standalone_configuration():
 
     assert tracer._compute_stats is False
     # reset tracer values
-    tracer.configure(appsec_enabled=False, appsec_standalone_enabled=False)
+    tracer.configure(appsec_enabled=False, iast_enabled=False, appsec_standalone_enabled=False)
 
 
 def test_gc_not_used_on_root_spans():