chore(aap): add visibility for rc products (#13800)

- add a new private root span tag `_dd.appsec.rc_products` to track the
number of paths currently active for each appsec remote config product
- update unit tests to check for that tag
- make sure appsec.enabled is on root span

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: christophe-papazian

ddtrace/appsec/_asm_request_context.py

@@ -81,7 +81,7 @@ class ASM_Environment:
     It is contained into a ContextVar.
     """
 
-    def __init__(self, span: Optional[Span] = None):
+    def __init__(self, span: Optional[Span] = None, rc_products: str = ""):
         self.root = not in_asm_context()
         if self.root:
             core.add_suppress_exception(BlockingException)
@@ -105,6 +105,7 @@ def __init__(self, span: Optional[Span] = None):
         self.blocked: Optional[Dict[str, Any]] = None
         self.finalized: bool = False
         self.api_security_reported: int = 0
+        self.rc_products: str = rc_products
 
 
 def _get_asm_context() -> Optional[ASM_Environment]:
@@ -260,6 +261,8 @@ def finalize_asm_env(env: ASM_Environment) -> None:
         res_headers = waf_adresses.get(SPAN_DATA_NAMES.RESPONSE_HEADERS_NO_COOKIES, {})
         if res_headers:
             _set_headers(root_span, res_headers, kind="response")
+        if env.rc_products:
+            root_span.set_tag_str(APPSEC.RC_PRODUCTS, env.rc_products)
 
     core.discard_local_item(_ASM_CONTEXT)
 
@@ -513,10 +516,10 @@ def store_waf_results_data(data) -> None:
     env.waf_triggers.extend(data)
 
 
-def start_context(span: Span):
+def start_context(span: Span, rc_products: str):
     if asm_config._asm_enabled:
         # it should only be called at start of a core context, when ASM_Env is not set yet
-        core.set_item(_ASM_CONTEXT, ASM_Environment(span=span))
+        core.set_item(_ASM_CONTEXT, ASM_Environment(span=span, rc_products=rc_products))
         asm_request_context_set(
             core.get_local_item("remote_addr"),
             core.get_local_item("headers"),

ddtrace/appsec/_constants.py

@@ -66,6 +66,7 @@ class APPSEC(metaclass=Constant_Class):
     RASP_DURATION_EXT: Literal["_dd.appsec.rasp.duration_ext"] = "_dd.appsec.rasp.duration_ext"
     RASP_RULE_EVAL: Literal["_dd.appsec.rasp.rule.eval"] = "_dd.appsec.rasp.rule.eval"
     RASP_TIMEOUTS: Literal["_dd.appsec.rasp.timeout"] = "_dd.appsec.rasp.timeout"
+    RC_PRODUCTS: Literal["_dd.appsec.rc_products"] = "_dd.appsec.rc_products"
     TRUNCATION_STRING_LENGTH: Literal["_dd.appsec.truncated.string_length"] = "_dd.appsec.truncated.string_length"
     TRUNCATION_CONTAINER_SIZE: Literal["_dd.appsec.truncated.container_size"] = "_dd.appsec.truncated.container_size"
     TRUNCATION_CONTAINER_DEPTH: Literal["_dd.appsec.truncated.container_depth"] = "_dd.appsec.truncated.container_depth"

ddtrace/appsec/_ddwaf/waf.py

@@ -5,6 +5,7 @@
 from typing import List
 from typing import Optional
 from typing import Sequence
+from typing import Set
 from typing import Tuple
 
 from ddtrace.appsec._constants import DEFAULT
@@ -77,6 +78,8 @@ def __init__(
             )
         self._default_ruleset = ruleset_map_object
         metrics.ddwaf_version = version()
+        self._rc_products: Dict[str, Set[str]] = {}
+        self._rc_products_str: str = ""
 
     @property
     def required_data(self) -> List[str]:
@@ -115,9 +118,13 @@ def update_rules(
         ok = True
         for product, path in removals:
             ok &= py_remove_config(self._builder, path)
+            self._rc_products.get(product, set()).discard(path)
             if product == "ASM_DD":
                 self._asm_dd_cache.discard(path)
         for product, path, rules in updates:
+            if product not in self._rc_products:
+                self._rc_products[product] = set()
+            self._rc_products[product].add(path)
             if product == "ASM_DD":
                 if ASM_DD_DEFAULT in self._asm_dd_cache:
                     # we need to remove the default ruleset before adding the new one
@@ -138,6 +145,7 @@ def update_rules(
             ddwaf_object_free(diagnostics)
             self._asm_dd_cache.add(ASM_DD_DEFAULT)
         new_handle = py_ddwaf_builder_build_instance(self._builder)
+        self._rc_products_str = ",".join(f"{p}:{len(v)}" for p, v in self._rc_products.items() if v)
         if new_handle:
             self._handle = new_handle
         return ok
@@ -146,6 +154,7 @@ def _at_request_start(self) -> Optional[ddwaf_context_capsule]:
         ctx = None
         if self._handle:
             ctx = py_ddwaf_context_init(self._handle)
+            ctx.rc_products = self._rc_products_str
         if not ctx:
             LOGGER.debug("DDWaf._at_request_start: failure to create the context.")
         return ctx

ddtrace/appsec/_ddwaf/waf_stubs.py

@@ -49,6 +49,7 @@ class ddwaf_context_capsule(Generic[T]):
     def __init__(self, ctx: Type[T], free_fn: Callable[[Type[T]], None]) -> None:
         self.ctx = ctx
         self.free_fn = free_fn
+        self.rc_products: str = ""
 
     def __del__(self):
         if self.ctx:

ddtrace/appsec/_processor.py

@@ -181,18 +181,18 @@ def on_span_start(self, span: Span) -> None:
         if span.span_type not in asm_config._asm_processed_span_types:
             return
 
-        _asm_request_context.start_context(span)
-
         ctx = self._ddwaf._at_request_start()
+        _asm_request_context.start_context(span, ctx.rc_products if ctx is not None else "")
         peer_ip = _asm_request_context.get_ip()
         headers = _asm_request_context.get_headers()
         headers_case_sensitive = _asm_request_context.get_headers_case_sensitive()
+        root_span = span._local_root or span
 
-        span.set_metric(APPSEC.ENABLED, 1.0)
-        span.set_tag_str(_RUNTIME_FAMILY, "python")
+        root_span.set_metric(APPSEC.ENABLED, 1.0)
+        root_span.set_tag_str(_RUNTIME_FAMILY, "python")
 
         def waf_callable(custom_data=None, **kwargs):
-            return self._waf_action(span._local_root or span, ctx, custom_data, **kwargs)
+            return self._waf_action(root_span, ctx, custom_data, **kwargs)
 
         _asm_request_context.set_waf_callback(waf_callable)
         _asm_request_context.add_context_callback(self.metrics._set_waf_request_metrics)

tests/appsec/appsec/test_processor.py

@@ -258,6 +258,7 @@ def test_ip_update_rules_and_block(tracer):
 
     assert get_waf_addresses("http.request.remote_ip") == rules._IP.BLOCKED
     assert is_blocked(span1)
+    assert (span._local_root or span).get_tag(APPSEC.RC_PRODUCTS) == "ASM:1"
 
 
 def test_ip_update_rules_expired_no_block(tracer):
@@ -290,6 +291,7 @@ def test_ip_update_rules_expired_no_block(tracer):
 
     assert get_waf_addresses("http.request.remote_ip") == rules._IP.BLOCKED
     assert is_blocked(span) is False
+    assert (span._local_root or span).get_tag(APPSEC.RC_PRODUCTS) == "ASM:1"
 
 
 @snapshot(
@@ -658,7 +660,7 @@ def test_asm_context_registration(tracer):
 CUSTOM_RULE_METHOD = [
     (
         "ASM",
-        "Datadog/1/ASM/data",
+        "Datadog/3421/ASM/data",
         {
             "custom_rules": [
                 {
@@ -743,11 +745,11 @@ def test_ephemeral_addresses(mock_run, persistent, ephemeral):
     from ddtrace.appsec._utils import _observator
     from ddtrace.trace import tracer
 
-    processor = AppSecSpanProcessor()
-    processor._update_rules([], CUSTOM_RULE_METHOD)
     mock_run.return_value = DDWaf_result(0, [], {}, 0.0, 0.0, False, _observator(), {})
 
-    with asm_context(tracer=tracer, config=config_asm) as span:
+    with asm_context(tracer=tracer, config=config_asm, rc_payload=CUSTOM_RULE_METHOD) as span:
+        processor = tracer._appsec_processor
+        assert processor
         # first call must send all data to the waf
         processor._waf_action(span, None, {persistent: {"key_1": "value_1"}, ephemeral: {"key_2": "value_2"}})
         assert mock_run.call_args[0][1] == {WAF_DATA_NAMES[persistent]: {"key_1": "value_1"}}
@@ -758,3 +760,4 @@ def test_ephemeral_addresses(mock_run, persistent, ephemeral):
         assert mock_run.call_args[1]["ephemeral_data"] == {
             WAF_DATA_NAMES[ephemeral]: {"key_2": "value_3"},
         }
+    assert (span._local_root or span).get_tag(APPSEC.RC_PRODUCTS) == "ASM:1"

tests/appsec/utils.py

@@ -59,6 +59,7 @@ def asm_context(
     block_request_callable: typing.Optional[typing.Callable[[], bool]] = None,
     service: typing.Optional[str] = None,
     config=None,
+    rc_payload=None,
 ) -> typing.Iterator[Span]:
     with override_global_config(config) if config else contextlib.nullcontext():
         if tracer is None:
@@ -68,6 +69,10 @@ def asm_context(
             tracer._span_aggregator.writer._api_version = "v0.4"
             tracer._recreate()
         patch_for_waf_addresses()
+        if rc_payload:
+            processor = tracer._appsec_processor
+            if processor:
+                processor._update_rules([], rc_payload)
         with core.context_with_data(
             "test.asm",
             remote_addr=ip_addr,