feat(asm): new user event sdk (#13251)

A new public SDK for ATO/user events for AAP.
Available with 
```python
from ddtrace.appsec import track_user_sdk

# to track successful login attempts
track_user_sdk.track_login_success(login, user_id:optional, metadata:optional)
# to track failed login attempts
track_user_sdk.track_login_failure(login, exists, user_id:optional, metadata:optional)
# to track any custom event
track_user_sdk.track_custom_event(event, metatada)
# to track signup events
track_user_sdk.track_signup(login,  user_id:optional, success:optional, metadata:optional)
# to track authentified user (usually in middleware, using auth token)
track_user_sdk.track_user(login,  user_id, session_id:optional, metadata:optional)

```

Also: 
- minor fixes and improvements in current ATO support.
- threat tests added for sdk (span tags and telemetry unit tests)

This will be validated in system tests with
DataDog/system-tests#4565

The documentation page to be updated is
https://docs.datadoghq.com/security/application_security/threats/add-user-info/?tab=loginsuccess&code-lang=python

APPSEC-56663

## 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/_metrics.py

@@ -221,3 +221,17 @@ def _report_rasp_skipped(rule_type: str, import_error: bool) -> None:
             "more_info": f":waf:rasp_rule_skipped:{rule_type}:{import_error}",
         }
         logger.warning(WARNING_TAGS.TELEMETRY_METRICS, extra=extra, exc_info=True)
+
+
+def _report_ato_sdk_usage(event_type: str, v2: bool = True) -> None:
+    version = "v2" if v2 else "v1"
+    try:
+        tags = (("event_type", event_type), ("sdk_version", version))
+        telemetry.telemetry_writer.add_count_metric(TELEMETRY_NAMESPACE.APPSEC, "sdk.event", 1, tags=tags)
+    except Exception:
+        extra = {
+            "product": "appsec",
+            "exec_limit": 6,
+            "more_info": f":waf:sdk.event:{event_type}:{version}",
+        }
+        logger.warning(WARNING_TAGS.TELEMETRY_METRICS, extra=extra, exc_info=True)

ddtrace/appsec/_trace_utils.py

@@ -1,4 +1,5 @@
 from typing import Any
+from typing import Dict
 from typing import Optional
 
 from ddtrace._trace.span import Span
@@ -36,6 +37,27 @@ def _asm_manual_keep(span: Span) -> None:
     span.context._meta[APPSEC.PROPAGATION_HEADER] = "02"
 
 
+def _handle_metadata(root_span: Span, prefix: str, metadata: dict) -> None:
+    MAX_DEPTH = 6
+    if metadata is None:
+        return
+    stack = [(prefix, metadata, 1)]
+    while stack:
+        prefix, data, level = stack.pop()
+        if isinstance(data, list):
+            if level < MAX_DEPTH:
+                for i, v in enumerate(data):
+                    stack.append((f"{prefix}.{i}", v, level + 1))
+        elif isinstance(data, dict):
+            if level < MAX_DEPTH:
+                for k, v in data.items():
+                    stack.append((f"{prefix}.{k}", v, level + 1))
+        else:
+            if isinstance(data, bool):
+                data = "true" if data else "false"
+            root_span.set_tag_str(f"{prefix}", str(data))
+
+
 def _track_user_login_common(
     tracer: Any,
     success: bool,
@@ -69,8 +91,7 @@ def _track_user_login_common(
 
         tag_metadata_prefix = "%s.%s" % (APPSEC.USER_LOGIN_EVENT_PREFIX_PUBLIC, success_str)
         if metadata is not None:
-            for k, v in metadata.items():
-                span.set_tag_str("%s.%s" % (tag_metadata_prefix, k), str(v))
+            _handle_metadata(span, tag_metadata_prefix, metadata)
 
         if login:
             span.set_tag_str(f"{APPSEC.USER_LOGIN_EVENT_PREFIX_PUBLIC}.{success_str}.usr.login", login)
@@ -97,7 +118,7 @@ def _track_user_login_common(
 
 def track_user_login_success_event(
     tracer: Any,
-    user_id: str,
+    user_id: Optional[str],
     metadata: Optional[dict] = None,
     login: Optional[str] = None,
     name: Optional[str] = None,
@@ -134,9 +155,12 @@ def track_user_login_success_event(
     if real_mode == LOGIN_EVENTS_MODE.ANON and isinstance(user_id, str):
         user_id = _hash_user_id(user_id)
     span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, real_mode)
-    if login_events_mode != LOGIN_EVENTS_MODE.SDK:
-        span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
-    set_user(None, user_id, name, email, scope, role, session_id, propagate, span, may_block=False)
+    if user_id:
+        if login_events_mode != LOGIN_EVENTS_MODE.SDK:
+            span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
+        else:
+            span.set_tag_str(f"{APPSEC.USER_LOGIN_EVENT_PREFIX_PUBLIC}.success.usr.id", str(user_id))
+    set_user(None, user_id or "", name, email, scope, role, session_id, propagate, span, may_block=False)
     if in_asm_context():
         custom_data = {
             "REQUEST_USER_ID": str(initial_user_id) if initial_user_id else None,
@@ -188,7 +212,7 @@ def track_user_login_failure_event(
         if login_events_mode != LOGIN_EVENTS_MODE.SDK:
             span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
         span.set_tag_str("%s.failure.%s" % (APPSEC.USER_LOGIN_EVENT_PREFIX_PUBLIC, user.ID), str(user_id))
-        span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, real_mode)
+    span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, real_mode)
     # if called from the SDK, set the login, email and name
     if login_events_mode in (LOGIN_EVENTS_MODE.SDK, LOGIN_EVENTS_MODE.AUTO):
         if login:
@@ -198,24 +222,43 @@ def track_user_login_failure_event(
         if name:
             span.set_tag_str("%s.failure.username" % APPSEC.USER_LOGIN_EVENT_PREFIX_PUBLIC, name)
     if in_asm_context():
-        call_waf_callback(custom_data={"LOGIN_FAILURE": None})
+        custom_data: dict[str, Any] = {"LOGIN_FAILURE": None}
+        if login:
+            custom_data["REQUEST_USERNAME"] = login
+        res = call_waf_callback(custom_data=custom_data)
+        if res and any(action in [WAF_ACTIONS.BLOCK_ACTION, WAF_ACTIONS.REDIRECT_ACTION] for action in res.actions):
+            raise BlockingException(get_blocked())
 
 
 def track_user_signup_event(
-    tracer: Any, user_id: str, success: bool, login_events_mode: str = LOGIN_EVENTS_MODE.SDK
+    tracer: Any,
+    user_id: Optional[str],
+    success: bool,
+    login: Optional[str] = None,
+    login_events_mode: str = LOGIN_EVENTS_MODE.SDK,
 ) -> None:
     span = core.get_root_span()
     if span:
         success_str = "true" if success else "false"
         span.set_tag_str(APPSEC.USER_SIGNUP_EVENT, success_str)
-        span.set_tag_str(user.ID, str(user_id))
+        if user_id:
+            if login_events_mode == LOGIN_EVENTS_MODE.ANON and isinstance(user_id, str):
+                user_id = _hash_user_id(user_id)
+            span.set_tag_str(user.ID, str(user_id))
+            span.set_tag_str(APPSEC.USER_SIGNUP_EVENT_USERID, str(user_id))
+            span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
+        if login:
+            if login_events_mode == LOGIN_EVENTS_MODE.ANON and isinstance(login, str):
+                login = _hash_user_id(login)
+            span.set_tag_str(APPSEC.USER_SIGNUP_EVENT_USERNAME, str(login))
+            span.set_tag_str(APPSEC.USER_LOGIN_USERNAME, str(login))
         _asm_manual_keep(span)
 
         # This is used to mark if the call was done from the SDK of the automatic login events
         if login_events_mode == LOGIN_EVENTS_MODE.SDK:
             span.set_tag_str("%s.sdk" % APPSEC.USER_SIGNUP_EVENT, "true")
         else:
-            span.set_tag_str("%s.auto.mode" % APPSEC.USER_SIGNUP_EVENT, str(login_events_mode))
+            span.set_tag_str("%s.auto.mode" % APPSEC.USER_SIGNUP_EVENT_MODE, str(login_events_mode))
 
         return
     else:
@@ -226,7 +269,7 @@ def track_user_signup_event(
         )
 
 
-def track_custom_event(tracer: Any, event_name: str, metadata: dict) -> None:
+def track_custom_event(tracer: Any, event_name: str, metadata: Dict[str, Any]) -> None:
     """
     Add a new custom tracking event.
 
@@ -254,14 +297,9 @@ def track_custom_event(tracer: Any, event_name: str, metadata: dict) -> None:
         return
 
     span.set_tag_str("%s.%s.track" % (APPSEC.CUSTOM_EVENT_PREFIX, event_name), "true")
-
-    for k, v in metadata.items():
-        if isinstance(v, bool):
-            str_v = "true" if v else "false"
-        else:
-            str_v = str(v)
-        span.set_tag_str("%s.%s.%s" % (APPSEC.CUSTOM_EVENT_PREFIX, event_name, k), str_v)
-        _asm_manual_keep(span)
+    if metadata:
+        _handle_metadata(span, f"{APPSEC.CUSTOM_EVENT_PREFIX}.{event_name}", metadata)
+    _asm_manual_keep(span)
 
 
 def should_block_user(tracer: Any, userid: str) -> bool:
@@ -312,24 +350,21 @@ def block_request_if_user_blocked(tracer: Any, userid: str, mode: str = "sdk") -
     :param userid: the ID of the user as registered by `set_user`
     :param mode: the mode of the login event ("sdk" by default, "auto" to simulate auto instrumentation)
     """
-    if not asm_config._asm_enabled:
+    if not asm_config._asm_enabled or mode == LOGIN_EVENTS_MODE.DISABLED:
         log.warning("should_block_user call requires ASM to be enabled")
         return
-    span = core.get_root_span()
-    if span:
-        root_span = span._local_root or span
-        if mode == LOGIN_EVENTS_MODE.SDK:
-            root_span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, LOGIN_EVENTS_MODE.SDK)
-        else:
-            if mode == LOGIN_EVENTS_MODE.AUTO:
-                mode = asm_config._user_event_mode
-            if mode == LOGIN_EVENTS_MODE.DISABLED:
-                return
+    if mode == LOGIN_EVENTS_MODE.AUTO:
+        mode = asm_config._user_event_mode
+    root_span = core.get_root_span()
+    if root_span:
+        root_span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, mode)
+        if userid:
             if mode == LOGIN_EVENTS_MODE.ANON:
                 userid = _hash_user_id(str(userid))
             root_span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, mode)
-            root_span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(userid))
-        root_span.set_tag_str(user.ID, str(userid))
+            if mode != LOGIN_EVENTS_MODE.SDK:
+                root_span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(userid))
+            root_span.set_tag_str(user.ID, str(userid))
     if should_block_user(None, userid):
         _asm_request_context.block_request()
 
@@ -416,16 +451,25 @@ def _on_django_process(result_user, session_key, mode, kwargs, pin, info_retriev
     if (not asm_config._asm_enabled) or mode == LOGIN_EVENTS_MODE.DISABLED:
         return
     user_id, user_extra = get_user_info(info_retriever, django_config, kwargs)
+    user_login = user_extra.get("login")
     res = None
     if result_user and result_user.is_authenticated:
         span = core.get_root_span()
-        if mode == LOGIN_EVENTS_MODE.ANON and isinstance(user_id, str):
-            hash_id = _hash_user_id(user_id)
-            span.set_tag_str(APPSEC.USER_LOGIN_USERID, hash_id)
+        if mode == LOGIN_EVENTS_MODE.ANON:
+            hash_id = ""
+            if isinstance(user_id, str):
+                hash_id = _hash_user_id(user_id)
+                span.set_tag_str(APPSEC.USER_LOGIN_USERID, hash_id)
+            if isinstance(user_login, str):
+                hash_login = _hash_user_id(user_login)
+                span.set_tag_str(APPSEC.USER_LOGIN_USERNAME, hash_login)
             span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, mode)
             set_user(None, hash_id, propagate=True, may_block=False, span=span)
         elif mode == LOGIN_EVENTS_MODE.IDENT:
-            span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
+            if user_id:
+                span.set_tag_str(APPSEC.USER_LOGIN_USERID, str(user_id))
+            if user_login:
+                span.set_tag_str(APPSEC.USER_LOGIN_USERNAME, str(user_login))
             span.set_tag_str(APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, mode)
             set_user(
                 None,
@@ -440,7 +484,7 @@ def _on_django_process(result_user, session_key, mode, kwargs, pin, info_retriev
             real_mode = mode if mode != LOGIN_EVENTS_MODE.AUTO else asm_config._user_event_mode
             custom_data = {
                 "REQUEST_USER_ID": str(user_id) if user_id else None,
-                "REQUEST_USERNAME": user_extra.get("login"),
+                "REQUEST_USERNAME": user_login,
                 "LOGIN_SUCCESS": real_mode,
             }
             if session_key:

ddtrace/appsec/track_user_sdk.py

@@ -0,0 +1,102 @@
+"""
+Public User Tracking SDK Version 2
+
+This module provides a public interface for tracking user events.
+This replaces the previous version of the SDK available in ddtrace.appsec.trace_utils
+Implementation can change in the future, but the interface will remain compatible.
+"""
+
+import typing as t
+
+from ddtrace.appsec import _asm_request_context
+from ddtrace.appsec import _constants
+from ddtrace.appsec import _metrics
+from ddtrace.appsec import _trace_utils
+from ddtrace.appsec._asm_request_context import get_blocked as _get_blocked
+from ddtrace.appsec._constants import WAF_ACTIONS as _WAF_ACTIONS
+import ddtrace.appsec.trace_utils  # noqa: F401
+from ddtrace.internal import core as _core
+from ddtrace.internal._exceptions import BlockingException
+
+
+def track_login_success(login: str, user_id: t.Any = None, metadata: t.Optional[t.Dict[str, t.Any]] = None) -> None:
+    """
+    Track a successful user login event.
+
+    This function should be called when a user successfully logs in to the application.
+    It will create an event that can be used for monitoring and analysis.
+    """
+    _metrics._report_ato_sdk_usage("login_success")
+    _trace_utils.track_user_login_success_event(None, user_id, login=login, metadata=metadata)
+
+
+def track_login_failure(
+    login: str, exists: bool, user_id: t.Any = None, metadata: t.Optional[t.Dict[str, t.Any]] = None
+):
+    """
+    Track a failed user login event.
+
+    This function should be called when a user fails to log in to the application.
+    It will create an event that can be used for monitoring and analysis.
+    """
+    _metrics._report_ato_sdk_usage("login_failure")
+    _trace_utils.track_user_login_failure_event(None, user_id, exists=exists, login=login, metadata=metadata)
+
+
+def track_signup(
+    login: str, user_id: t.Any = None, success: bool = True, metadata: t.Optional[t.Dict[str, t.Any]] = None
+):
+    """
+    Track a user signup event.
+
+    This function should be called when a user successfully signs up for the application.
+    It will create an event that can be used for monitoring and analysis.
+    """
+    _metrics._report_ato_sdk_usage("signup")
+    _trace_utils.track_user_signup_event(None, user_id, success, login=login)
+    if metadata:
+        _trace_utils.track_custom_event(None, "signup_sdk", metadata=metadata)
+
+
+def track_user(
+    login: str, user_id: t.Any = None, session_id=t.Optional[str], metadata: t.Optional[t.Dict[str, t.Any]] = None
+):
+    """
+    Track an authenticated user.
+
+    This function should be called when a user is authenticated in the application."
+    """
+    span = _core.get_root_span()
+    if span is None:
+        return
+    if user_id:
+        span.set_tag_str(_constants.APPSEC.USER_LOGIN_USERID, str(user_id))
+    if login:
+        span.set_tag_str(_constants.APPSEC.USER_LOGIN_USERNAME, str(login))
+
+    _trace_utils.set_user(None, user_id, session_id=session_id, may_block=False)
+    if metadata:
+        _trace_utils.track_custom_event(None, "auth_sdk", metadata=metadata)
+    span.set_tag_str(_constants.APPSEC.AUTO_LOGIN_EVENTS_COLLECTION_MODE, _constants.LOGIN_EVENTS_MODE.SDK)
+    if _asm_request_context.in_asm_context():
+        custom_data = {
+            "REQUEST_USER_ID": str(user_id) if user_id else None,
+            "REQUEST_USERNAME": login,
+            "LOGIN_SUCCESS": "sdk",
+        }
+        if session_id:
+            custom_data["REQUEST_SESSION_ID"] = session_id
+        res = _asm_request_context.call_waf_callback(custom_data=custom_data, force_sent=True)
+        if res and any(action in [_WAF_ACTIONS.BLOCK_ACTION, _WAF_ACTIONS.REDIRECT_ACTION] for action in res.actions):
+            raise BlockingException(_get_blocked())
+
+
+def track_custom_event(event_name: str, metadata: t.Dict[str, t.Any]):
+    """
+    Track a custom user event.
+
+    This function should be called when a custom user event occurs in the application.
+    It will create an event that can be used for monitoring and analysis.
+    """
+    _metrics._report_ato_sdk_usage("custom")
+    _trace_utils.track_custom_event(None, event_name, metadata=metadata)

releasenotes/notes/ato_sdk_v2-35c17dc258c5b690.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    AAP: This introduces a new user event sdk available through ddtrace.appsec.track_user_sdk for manual instrumentation.
+    More information on our documentation page.