fix: mask secrets in long strings and dict keys in code variables

.sampo/changesets/mask-long-strings-and-dict-keys.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Code variable masking now searches strings of up to 2,048 characters for known credential formats, not only strings of up to 200 characters. A key inside a longer string, such as a SQL query that inlines an access key, is now redacted. Dict keys are now masked too. A key is replaced with a `$$_posthog_redacted_key_<n>_$$` placeholder when it matches a mask pattern but is not a plain field name, when it looks like a secret, when a non-string key holds a part that masking redacts or can't check, when its text can't be read, or when it is too long to scan. URL credentials are removed from string keys, and keys that end up with the same text keep separate entries.

posthog/exception_utils.py

@@ -1144,7 +1144,9 @@ def build(
 # Well-known credential formats, matched regardless of entropy. High-confidence,
 # distinctive-prefix patterns adapted from the gitleaks / detect-secrets rule sets.
 _KNOWN_SECRET_PATTERNS = [
-    # AI / LLM providers
+    # AI / LLM providers. The `sk-` prefix is not anchored to a word boundary, because a
+    # key often follows a letter or digit, e.g. `%3Dsk-...` in a percent-encoded URL or
+    # `\nsk-...` in escaped text. This over-redacts words such as `disk-usage-...`.
     r"sk-ant-[A-Za-z0-9_-]{16,}",  # Anthropic
     r"sk-(?:proj-)?[A-Za-z0-9_-]{20,}",  # OpenAI
     r"hf_[A-Za-z0-9]{34}",  # Hugging Face
@@ -1185,7 +1187,9 @@ def build(
 _KNOWN_SECRET_RE = re.compile("|".join(_KNOWN_SECRET_PATTERNS))
 
 _PEM_PRIVATE_KEY_MARKER = "PRIVATE KEY-----"  # covers RSA/EC/OpenSSH/PKCS8
-_KNOWN_SECRET_MAX_SCAN_LENGTH = 200
+# Same cap as the other pattern checks, so a key embedded in a longer string (a SQL query
+# with credentials, a config dump) is still found.
+_KNOWN_SECRET_MAX_SCAN_LENGTH = _MAX_VALUE_LENGTH_FOR_PATTERN_MATCH
 
 
 def _looks_like_path_or_url(value):
@@ -1370,27 +1374,124 @@ def _masked_type_members(value, config):
     return masked
 
 
+# A mapping can have several redacted keys, so each placeholder carries a number to
+# keep the keys unique.
+_REDACTED_KEY_TEMPLATE = "$$_posthog_redacted_key_{}_$$"
+
+# A string key that matches a mask pattern is kept only when it has this shape. Text with
+# other characters, such as `password=hunter2` or a SQL query, can hold the value itself.
+_FIELD_NAME_RE = re.compile(r"[\w.\-]+")
+
+_CIRCULAR_REF_VALUE = "<circular ref>"
+
+# Markers that show a key probe could not vouch for every part of the key.
+_KEY_PROBE_MARKERS = (
+    CODE_VARIABLES_REDACTED_VALUE,
+    CODE_VARIABLES_TOO_LONG_VALUE,
+    _CIRCULAR_REF_VALUE,
+)
+
+
+def _redacted_key(result):
+    """Return a placeholder key that no key already in ``result`` uses."""
+    n = 0
+    while (candidate := _REDACTED_KEY_TEMPLATE.format(n)) in result:
+        n += 1
+    return candidate
+
+
+def _is_field_name(key, key_is_json_safe):
+    """True when a key that matches a mask pattern names a field, so it is safe to keep.
+    A number or None can't hold a credential, and a string must look like an identifier.
+    Any other key reaches the output as a repr, which can embed field values."""
+    if not key_is_json_safe:
+        return False
+    if isinstance(key, str):
+        return _FIELD_NAME_RE.fullmatch(key) is not None
+    return True
+
+
+class _KeyProbeSeen:
+    """The ``seen`` set for a key probe. The probe stops at every object that the
+    traversal or an earlier probe visited, and its visits count against the same node
+    budget. It records its visits under a separate tag, so an object that a key shares
+    with a value is still masked in full where the value holds it."""
+
+    _TAG = "key_probe"
+
+    def __init__(self, seen):
+        self._seen = seen
+
+    def __contains__(self, obj_id):
+        return obj_id in self._seen or (self._TAG, obj_id) in self._seen
+
+    def add(self, obj_id):
+        self._seen.add((self._TAG, obj_id))
+
+    def __len__(self):
+        return len(self._seen)
+
+
+def _key_parts_fail_masking(key, config, seen, depth):
+    """True when masking ``key`` as a value redacts any part of it, reaches an object
+    that the traversal already visited, or raises. The quotes and brackets of a repr turn
+    off the entropy check, so the parts are checked one by one. A part that was already
+    visited is not checked again, so its text in the key's repr can't be vouched for."""
+    probe_seen = seen if isinstance(seen, _KeyProbeSeen) else _KeyProbeSeen(seen)
+    try:
+        rendered = str(_mask_value(key, config, probe_seen, depth + 1))
+    except Exception:
+        return True
+    return any(marker in rendered for marker in _KEY_PROBE_MARKERS)
+
+
 def _mask_mapping(items, config, seen, depth):
     """Mask a sequence of ``(key, value)`` pairs into a dict. A key matching the mask
     redacts its value; surviving values recurse through ``_mask_value``. Keys are kept
-    JSON-serializable."""
-    result = {}
+    JSON-serializable, and a key whose own text can hold a secret is replaced by a
+    placeholder, because the key text reaches the output as-is."""
+    result: Dict[Any, Any] = {}
     for key, value in items:
         if type(key) is str:
             out_key = key_str = key
+            key_is_json_safe = True
         else:
-            key_str = key if isinstance(key, str) else str(key)
+            try:
+                key_str = key if isinstance(key, str) else str(key)
+            except Exception:
+                # Without the key text, nothing shows whether the key names a secret, so
+                # the value is redacted too.
+                result[_redacted_key(result)] = CODE_VARIABLES_REDACTED_VALUE
+                continue
             # json.dumps only accepts str/int/float/bool/None keys; coerce anything else to
-            # its string form so one exotic key can't make json.dumps fail.
+            # its string form so one exotic key can't make json.dumps fail. That string form
+            # is a repr, which can embed field values, not only a name.
             key_is_json_safe = (
                 key is None
                 or isinstance(key, (str, int))  # bool is an int subclass
                 or (isinstance(key, float) and math.isfinite(key))
             )
             out_key = key if key_is_json_safe else key_str
         if len(key_str) > _MAX_VALUE_LENGTH_FOR_PATTERN_MATCH:
-            result[out_key] = CODE_VARIABLES_TOO_LONG_VALUE
-        elif _matcher_matches(key_str, config.mask):
+            # Too long to scan, so the key text can't be vouched for either.
+            result[_redacted_key(result)] = CODE_VARIABLES_TOO_LONG_VALUE
+            continue
+        key_matches_mask = _matcher_matches(key_str, config.mask)
+        if key_matches_mask and not _is_field_name(key, key_is_json_safe):
+            # A name such as `password` is safe to keep, but other text that matches can
+            # hold the value itself, e.g. `BasicAuth(login='u', password='...')`.
+            out_key = _redacted_key(result)
+        elif config.detect_secrets and _looks_like_secret(key_str):
+            out_key = _redacted_key(result)
+        elif not key_is_json_safe and _key_parts_fail_masking(key, config, seen, depth):
+            out_key = _redacted_key(result)
+        elif config.mask_url_credentials and isinstance(out_key, str):
+            out_key = _redact_url_credentials(out_key)
+        if out_key in result:
+            # Two keys can end up with the same text, for example URLs that differ only in
+            # their credentials. A placeholder keeps the later entry from overwriting.
+            out_key = _redacted_key(result)
+        if key_matches_mask:
             result[out_key] = CODE_VARIABLES_REDACTED_VALUE
         else:
             result[out_key] = _mask_value(value, config, seen, depth + 1)
@@ -1439,7 +1540,7 @@ def _mask_value(value, config, seen=None, depth=0):
         seen = set()
     obj_id = id(value)
     if obj_id in seen:
-        return "<circular ref>"
+        return _CIRCULAR_REF_VALUE
     seen.add(obj_id)
 
     if len(seen) > _MAX_TOTAL_NODES_TO_MASK: