chore(asm/appsec): add rate limiter and waf timeout (backport #3575) (#3634)

This is an automatic backport of pull request #3575 done by [Mergify](https://mergify.com).


---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.com/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.com/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.com
</details>

Author: mergify[bot]

ddtrace/appsec/processor.py

@@ -15,6 +15,7 @@
 from ddtrace.internal import _context
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.processor import SpanProcessor
+from ddtrace.internal.rate_limiter import RateLimiter
 
 
 if TYPE_CHECKING:
@@ -24,6 +25,8 @@
 
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
 DEFAULT_RULES = os.path.join(ROOT_DIR, "rules.json")
+DEFAULT_TRACE_RATE_LIMIT = 100
+DEFAULT_WAF_TIMEOUT = 20  # ms
 
 log = get_logger(__name__)
 
@@ -81,12 +84,24 @@ def _set_headers(span, headers):
             span._set_str_tag(_normalize_tag_name("request", k), headers[k])
 
 
+def _get_rate_limiter():
+    # type: () -> RateLimiter
+    return RateLimiter(int(os.getenv("DD_APPSEC_TRACE_RATE_LIMIT", DEFAULT_TRACE_RATE_LIMIT)))
+
+
+def _get_waf_timeout():
+    # type: () -> int
+    return int(os.getenv("DD_APPSEC_WAF_TIMEOUT", DEFAULT_WAF_TIMEOUT))
+
+
 @attr.s(eq=False)
 class AppSecSpanProcessor(SpanProcessor):
 
     rules = attr.ib(type=str, factory=get_rules)
     _ddwaf = attr.ib(type=DDWaf, default=None)
     _addresses_to_keep = attr.ib(type=Set[str], factory=set)
+    _rate_limiter = attr.ib(type=RateLimiter, factory=_get_rate_limiter)
+    _waf_timeout = attr.ib(type=int, factory=_get_waf_timeout)
 
     @property
     def enabled(self):
@@ -188,8 +203,14 @@ def on_span_finish(self, span):
                 data[_Addresses.SERVER_RESPONSE_HEADERS_NO_COOKIES] = _no_cookies(response_headers)
 
         log.debug("[DDAS-001-00] Executing AppSec In-App WAF with parameters: %s", data)
-        res = self._ddwaf.run(data)  # res is a serialized json
+        res = self._ddwaf.run(data, self._waf_timeout)  # res is a serialized json
         if res is not None:
+            # We run the rate limiter only if there is an attack, its goal is to limit the number of collected asm
+            # events
+            allowed = self._rate_limiter.is_allowed(span.start_ns)
+            if not allowed:
+                # TODO: add metric collection to keep an eye (when it's name is clarified)
+                return
             if _Addresses.SERVER_REQUEST_HEADERS_NO_COOKIES in data:
                 _set_headers(span, data[_Addresses.SERVER_REQUEST_HEADERS_NO_COOKIES])
             # Partial DDAS-011-00

tests/appsec/test_processor.py

@@ -107,3 +107,27 @@ def test_appsec_span_tags_snapshot(tracer):
         set_http_meta(span, {}, raw_uri="http://example.com/.git", status_code="404")
 
     assert "triggers" in json.loads(span.get_tag("_dd.appsec.json"))
+
+
+def test_appsec_span_rate_limit(tracer):
+    with override_env(dict(DD_APPSEC_TRACE_RATE_LIMIT="1")):
+        _enable_appsec(tracer)
+
+        # we have 2 spans going through with a rate limit of 1: this is because the first span will update the rate
+        # limiter last update timestamp. In other words, we need a first call to reset the rate limiter's clock
+        # DEV: aligning rate limiter clock with this span (this
+        #      span will go through as it is linked to the init window)
+        with tracer.trace("test", span_type=SpanTypes.WEB) as span1:
+            set_http_meta(span1, {}, raw_uri="http://example.com/.git", status_code="404")
+
+        with tracer.trace("test", span_type=SpanTypes.WEB) as span2:
+            set_http_meta(span2, {}, raw_uri="http://example.com/.git", status_code="404")
+            span2.start_ns = span1.start_ns + 1
+
+        with tracer.trace("test", span_type=SpanTypes.WEB) as span3:
+            set_http_meta(span3, {}, raw_uri="http://example.com/.git", status_code="404")
+            span2.start_ns = span1.start_ns + 2
+
+        assert span1.get_tag("_dd.appsec.json") is not None
+        assert span2.get_tag("_dd.appsec.json") is not None
+        assert span3.get_tag("_dd.appsec.json") is None