feat(aap): add support for downstream requests analysis (#14481)

OPEN FOR REVIEW, DO NOT MERGE YET

API 10 support for downstream request analysis.

- update ssrf rasp instrumentation to include api10 for urllib (standard
cpython api)
- update threat tests for request header and body analysis
- update telemetry for new env var to configure the new feature

System tests not done yet.

## 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

@@ -105,6 +105,7 @@ def __init__(self, span: Optional[Span] = None, rc_products: str = ""):
         self.finalized: bool = False
         self.api_security_reported: int = 0
         self.rc_products: str = rc_products
+        self.downstream_requests: int = 0
 
 
 def _get_asm_context() -> Optional[ASM_Environment]:
@@ -140,6 +141,25 @@ def get_entry_span() -> Optional[Span]:
     return env.entry_span
 
 
+KNUTH_FACTOR: int = 11400714819323199488
+UINT64_MAX: int = (1 << 64) - 1
+
+
+class DownstreamRequests:
+    counter: int = 0
+    sampling_rate: int = int(asm_config._dr_sample_rate * UINT64_MAX)
+
+
+def should_analyze_body_response(env) -> bool:
+    """Must be called only after should_analyze_downstream returned True."""
+    DownstreamRequests.counter += 1
+    env.downstream_requests += 1
+    return (
+        env.downstream_requests <= asm_config._dr_body_limit_per_request
+        and (DownstreamRequests.counter * KNUTH_FACTOR) % UINT64_MAX <= DownstreamRequests.sampling_rate
+    )
+
+
 def get_framework() -> str:
     env = _get_asm_context()
     if env is None:

ddtrace/appsec/_common_module_patches.py

@@ -1,11 +1,14 @@
 # This module must not import other modules unconditionally that require iast
 import ctypes
+import io
+import json
 import os
 from typing import Any
 from typing import Callable
 from typing import Dict
 from typing import Iterable
 from typing import List
+from typing import Tuple
 from typing import Union
 
 from wrapt import FunctionWrapper
@@ -49,6 +52,7 @@ def _(module):
 
     try_wrap_function_wrapper("builtins", "open", wrapped_open_CFDDB7ABBA9081B6)
     try_wrap_function_wrapper("urllib.request", "OpenerDirector.open", wrapped_open_ED4CF71136E15EBF)
+    try_wrap_function_wrapper("http.client", "HTTPConnection.request", wrapped_request)
     core.on("asm.block.dbapi.execute", execute_4C9BAC8E228EB347)
     log.debug("Patching common modules: builtins and urllib.request")
     _is_patched = True
@@ -134,18 +138,71 @@ def wrapped_open_CFDDB7ABBA9081B6(original_open_callable, instance, args, kwargs
         )
 
 
+def _build_headers(lst: Iterable[Tuple[str, str]]) -> Dict[str, Union[str, List[str]]]:
+    res: Dict[str, Union[str, List[str]]] = {}
+    for a, b in lst:
+        if a in res:
+            v = res[a]
+            if isinstance(v, str):
+                res[a] = [v, b]
+            else:
+                v.append(b)
+        else:
+            res[a] = b
+    return res
+
+
+def wrapped_request(original_request_callable, instance, args, kwargs):
+    from ddtrace.appsec._asm_request_context import call_waf_callback
+
+    full_url = core.get_item("full_url")
+    if full_url is not None:
+        use_body = core.get_item("use_body", False)
+        method = args[0] if len(args) > 0 else kwargs.get("method", None)
+        body = args[2] if len(args) > 2 else kwargs.get("body", None)
+        headers = args[3] if len(args) > 3 else kwargs.get("headers", {})
+        addresses = {EXPLOIT_PREVENTION.ADDRESS.SSRF: full_url, "DOWN_REQ_METHOD": method, "DOWN_REQ_HEADERS": headers}
+        content_type = headers.get("Content-Type", None) or headers.get("content-type", None)
+        if use_body and content_type == "application/json":
+            try:
+                addresses["DOWN_REQ_BODY"] = json.loads(body)
+            except Exception:
+                pass  # nosec
+        res = call_waf_callback(
+            addresses,
+            crop_trace="wrapped_open_ED4CF71136E15EBF",
+            rule_type=EXPLOIT_PREVENTION.TYPE.SSRF,
+        )
+        if res and _must_block(res.actions):
+            raise BlockingException(get_blocked(), EXPLOIT_PREVENTION.BLOCKING, EXPLOIT_PREVENTION.TYPE.SSRF, full_url)
+    return original_request_callable(*args, **kwargs)
+
+
+def _parse_http_response_body(response):
+    try:
+        if response.length and response.headers.get("content-type", None) == "application/json":
+            length = response.length
+            body = response.read()
+            response.fp = io.BytesIO(body)
+            response.length = length
+            return json.loads(body)
+    except Exception:
+        return None
+    return None
+
+
 def wrapped_open_ED4CF71136E15EBF(original_open_callable, instance, args, kwargs):
     """
     wrapper for open url function
     """
     if asm_config._iast_enabled:
         # TODO: IAST SSRF sink to be added
         pass
-
     if _get_rasp_capability("ssrf"):
         try:
+            from ddtrace.appsec._asm_request_context import _get_asm_context
             from ddtrace.appsec._asm_request_context import call_waf_callback
-            from ddtrace.appsec._asm_request_context import in_asm_context
+            from ddtrace.appsec._asm_request_context import should_analyze_body_response
         except ImportError:
             # open is used during module initialization
             # and shouldn't be changed at that time
@@ -155,19 +212,42 @@ def wrapped_open_ED4CF71136E15EBF(original_open_callable, instance, args, kwargs
         url = args[0] if args else kwargs.get("fullurl", None)
         if url.__class__.__name__ == "Request":
             url = url.get_full_url()
-        if isinstance(url, str) and url:
-            if in_asm_context():
-                res = call_waf_callback(
-                    {EXPLOIT_PREVENTION.ADDRESS.SSRF: url},
-                    crop_trace="wrapped_open_ED4CF71136E15EBF",
-                    rule_type=EXPLOIT_PREVENTION.TYPE.SSRF,
-                )
-                if res and _must_block(res.actions):
-                    raise BlockingException(
-                        get_blocked(), EXPLOIT_PREVENTION.BLOCKING, EXPLOIT_PREVENTION.TYPE.SSRF, url
-                    )
-            else:
-                _report_rasp_skipped(EXPLOIT_PREVENTION.TYPE.SSRF, False)
+        valid_url = isinstance(url, str) and bool(url)
+        if valid_url and url and (ctx := _get_asm_context()):
+            use_body = should_analyze_body_response(ctx)
+            with core.context_with_data("url_open_analysis", full_url=url, use_body=use_body):
+                # API10, doing all request calls in HTTPConnection.request
+                try:
+                    response = original_open_callable(*args, **kwargs)
+                    # api10 response handler for regular reponses
+                    if response.__class__.__name__ == "HTTPResponse":
+                        addresses = {
+                            "DOWN_RES_STATUS": response.status,
+                            "DOWN_RES_HEADERS": _build_headers(response.getheaders()),
+                        }
+                        if use_body:
+                            addresses["DOWN_RES_BODY"] = _parse_http_response_body(response)
+                        call_waf_callback(addresses, rule_type=EXPLOIT_PREVENTION.TYPE.SSRF)
+                    return response
+                except Exception as e:
+                    # api10 response handler for error reponses
+                    if e.__class__.__name__ == "HTTPError":
+                        try:
+                            status_code = e.code
+                        except Exception:
+                            status_code = None
+                        try:
+                            response_headers = _build_headers(e.headers.items())
+                        except Exception:
+                            response_headers = None
+                        if status_code is not None or response_headers is not None:
+                            call_waf_callback(
+                                {"DOWN_RES_STATUS": status_code, "DOWN_RES_HEADERS": response_headers},
+                                rule_type=EXPLOIT_PREVENTION.TYPE.SSRF,
+                            )
+                    raise
+        elif valid_url:
+            _report_rasp_skipped(EXPLOIT_PREVENTION.TYPE.SSRF, False)
     return original_open_callable(*args, **kwargs)
 
 

ddtrace/appsec/_constants.py

@@ -235,6 +235,12 @@ class WAF_DATA_NAMES(metaclass=Constant_Class):
     SQLI_SYSTEM_ADDRESS: Literal["server.db.system"] = "server.db.system"
     LOGIN_FAILURE: Literal["server.business_logic.users.login.failure"] = "server.business_logic.users.login.failure"
     LOGIN_SUCCESS: Literal["server.business_logic.users.login.success"] = "server.business_logic.users.login.success"
+    DOWN_REQ_HEADERS: Literal["server.io.net.request.headers"] = "server.io.net.request.headers"
+    DOWN_REQ_METHOD: Literal["server.io.net.request.method"] = "server.io.net.request.method"
+    DOWN_REQ_BODY: Literal["server.io.net.request.body"] = "server.io.net.request.body"
+    DOWN_RES_STATUS: Literal["server.io.net.response.status"] = "server.io.net.response.status"
+    DOWN_RES_HEADERS: Literal["server.io.net.response.headers"] = "server.io.net.response.headers"
+    DOWN_RES_BODY: Literal["server.io.net.response.body"] = "server.io.net.response.body"
 
 
 class SPAN_DATA_NAMES(metaclass=Constant_Class):

ddtrace/settings/asm.py

@@ -187,6 +187,16 @@ class ASMConfig(DDConfig):
     # Timeout for the request body reading in seconds.
     _fast_api_async_body_timeout = DDConfig.var(float, "DD_FASTAPI_ASYNC_BODY_TIMEOUT_SECONDS", default=0.1)
 
+    # DOWNSTREAM REQUESTS INSTRUMENTATION
+    # sample rate for body analysis
+    _dr_sample_rate: float = DDConfig.var(
+        float, "DD_API_SECURITY_DOWNSTREAM_REQUEST_BODY_ANALYSIS_SAMPLE_RATE", default=0.5
+    )
+    # max number of downstream requests analysis  with bodies per request
+    _dr_body_limit_per_request: int = DDConfig.var(
+        int, "DD_API_SECURITY_MAX_DOWNSTREAM_REQUEST_BODY_ANALYSIS", default=1
+    )
+
     # for tests purposes
     _asm_config_keys = [
         "_asm_enabled",
@@ -217,6 +227,8 @@ class ASMConfig(DDConfig):
         "_api_security_enabled",
         "_api_security_sample_delay",
         "_api_security_parse_response_body",
+        "_dr_sample_rate",
+        "_dr_body_limit_per_request",
         "_waf_timeout",
         "_iast_redaction_enabled",
         "_iast_redaction_name_pattern",

releasenotes/notes/api10-ed802d03b6067b57.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    AAP: This introduces downstream request analysis (API10).

tests/appsec/appsec/rules-rasp.json

@@ -245,6 +245,75 @@
       "on_match": [
         "stack_trace"
       ]
+    },
+    {
+      "id": "apiA-100-001",
+      "name": "API 10 tag rule on headers",
+      "tags": {
+        "type": "api10 request headers",
+        "category": "attack_attempt"
+      },
+      "conditions": [
+        {
+          "parameters": {
+            "inputs": [
+              {
+                "address": "server.io.net.request.headers",
+                "key_path": [
+                  "Host"
+                ]
+              }
+            ],
+            "list": ["www.datadoghq.com"]
+          },
+          "operator": "exact_match"
+        }
+      ],
+      "output": {
+        "event": true,
+        "keep": true,
+        "attributes": {
+          "_dd.appsec.trace.mark": {
+            "value": "TAG_API10_HEADER"
+          }
+        }
+      },
+      "on_match": []
+    },
+    {
+      "id": "apiA-100-002",
+      "name": "API 10 tag rule on body",
+      "tags": {
+        "type": "api10 request body",
+        "category": "attack_attempt"
+      },
+      "conditions": [
+        {
+          "parameters": {
+            "inputs": [
+              {
+                "address": "server.io.net.request.body",
+                "key_path": [
+                  "payload"
+                ]
+              }
+            ],
+            "list": ["qw2jedrkjerbgol23ewpfirj2qw3or"]
+          },
+          "operator": "exact_match"
+        }
+      ],
+      "output": {
+        "event": true,
+        "keep": true,
+        "attributes": {
+          "_dd.appsec.trace.mark": {
+            "value": "TAG_API10_BODY"
+          }
+        }
+      },
+      "on_match": []
     }
+
   ]
 }

tests/appsec/contrib_appsec/db.sqlite3

@@ -170,6 +170,32 @@ def rasp(request, endpoint: str):
     return HttpResponse(f"Unknown endpoint: {endpoint}")
 
 
+@csrf_exempt
+def redirect(request, url: str):
+    import urllib.request
+
+    url = "http://" + url
+    body_str = request.body.decode()
+    if body_str:
+        body = json.loads(body_str)
+    else:
+        body = None
+    try:
+        if body:
+            request_urllib = urllib.request.Request(
+                url, method="POST", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}
+            )
+        else:
+            request_urllib = urllib.request.Request(url, method="GET")
+        with urllib.request.urlopen(request_urllib, timeout=0.5) as f:
+            payload = {"payload": f.read().decode(errors="ignore")}
+    except Exception as e:
+        import traceback
+
+        payload = {"error": repr(e), "trace": traceback.format_exc()}
+    return JsonResponse(payload)
+
+
 @csrf_exempt
 def login_user(request):
     from django.contrib.auth import authenticate
@@ -287,6 +313,7 @@ def shutdown(request):
         path("new_service/<str:service_name>", new_service, name="new_service"),
         path("rasp/<str:endpoint>/", rasp, name="rasp"),
         path("rasp/<str:endpoint>", rasp, name="rasp"),
+        path("redirect/<str:url>/", redirect, name="redirect"),
         path(route="login/", view=login_user, name="login"),
         path("login", login_user, name="login"),
         path("login_sdk/", login_user_sdk, name="login_sdk"),
@@ -300,6 +327,7 @@ def shutdown(request):
         path(r"new_service/(?P<service_name>\w+)$", new_service, name="new_service"),
         path(r"rasp/(?P<endpoint>\w+)/$", new_service, name="rasp"),
         path(r"rasp/(?P<endpoint>\w+)$", new_service, name="rasp"),
+        path(r"redirect/(?P<url>\w+)$", redirect, name="redirect"),
         path("login/", login_user, name="login"),
         path("login", login_user, name="login"),
     ]