fix(redaction): cover strict URL reference forms

This commit is contained in:
Jan-Stefan Janetzky 2026-07-14 12:28:24 +00:00 committed by Teknium
parent a48315e322
commit 62a00a7391
3 changed files with 122 additions and 4 deletions

View file

@ -11,6 +11,7 @@ import logging
import os
import re
import shlex
from urllib.parse import unquote_plus
logger = logging.getLogger(__name__)
@ -285,6 +286,22 @@ _URL_USERINFO_RE = re.compile(
r"(https?|wss?|ftp)://([^/\s:@]+):([^/\s@]+)@",
)
# Strict provider-egress URL redaction accepts more URL-reference forms than
# the display/log helpers above. Parameter delimiters stay in capture groups so
# redaction preserves the original query/fragment layout byte-for-byte, while
# the key is decoded separately for classification. Values stop at query or
# fragment pair separators; both ``&`` and ``;`` are valid in deployed URLs.
_STRICT_URL_PARAM_RE = re.compile(
r"([?#&;])([A-Za-z0-9_.~+%\-]+)=([^#&;\s\"'<>]*)"
)
# Match userinfo in both absolute (``scheme://user:pass@host``) and
# network-path (``//user:pass@host``) references. The authority boundary stops
# at path/query/fragment delimiters so an ``@`` elsewhere in a URL is ignored.
_STRICT_URL_USERINFO_RE = re.compile(
r"((?:[A-Za-z][A-Za-z0-9+.-]*:)?//)([^/\s?#@]+)@"
)
# HTTP access logs often use a relative request target rather than a full URL:
# `"POST /webhook?password=... HTTP/1.1"`. The full-URL redactor above only
# sees strings containing `://`, so handle request-target query strings too.
@ -411,6 +428,41 @@ def _redact_url_userinfo(text: str) -> str:
)
def _canonical_url_param_name(name: str) -> str:
"""Decode a URL parameter name for bounded, case-insensitive matching."""
decoded = name
for _ in range(3):
next_value = unquote_plus(decoded)
if next_value == decoded:
break
decoded = next_value
return decoded.casefold()
def _redact_strict_url_credentials(text: str) -> str:
"""Redact credentials from absolute, relative, and network URL references.
This is intentionally stricter than display/log redaction and is used only
at explicit secret-egress boundaries. It preserves original keys,
separators, public parameters, hosts, and paths while masking sensitive
values and URL userinfo.
"""
def _redact_param(match: re.Match) -> str:
if _canonical_url_param_name(match.group(2)) not in _SENSITIVE_QUERY_PARAMS:
return match.group(0)
return f"{match.group(1)}{match.group(2)}=***"
def _redact_userinfo(match: re.Match) -> str:
userinfo = match.group(2)
if ":" in userinfo:
username, _, _password = userinfo.partition(":")
return f"{match.group(1)}{username}:***@"
return f"{match.group(1)}***@"
text = _STRICT_URL_PARAM_RE.sub(_redact_param, text)
return _STRICT_URL_USERINFO_RE.sub(_redact_userinfo, text)
def redact_cdp_url(value: object) -> str:
"""Mask secrets in a CDP/browser endpoint URL before it is logged.
@ -672,9 +724,8 @@ def redact_sensitive_text(
# string), so masking it can't break a skill. The ``user:pass@`` form is
# left to pass through per #34029.
if redact_url_credentials and "://" in text:
text = _redact_url_query_params(text)
text = _redact_url_userinfo(text)
if redact_url_credentials:
text = _redact_strict_url_credentials(text)
# Form-urlencoded bodies (only triggers on clean k=v&k=v inputs).
if "&" in text and "=" in text:

View file

@ -588,6 +588,57 @@ class TestWebUrlsNotRedacted:
assert "dbpass" not in result
class TestStrictUrlCredentialRedaction:
@pytest.mark.parametrize(
("text", "secret", "expected"),
[
(
"https://x.test/#access_token=FRAG_SECRET&view=public",
"FRAG_SECRET",
"https://x.test/#access_token=***&view=public",
),
(
"/resume?token=REL_SECRET&view=public",
"REL_SECRET",
"/resume?token=***&view=public",
),
(
"https://x.test/cb?client%5Fsecret=ENC_SECRET&view=public",
"ENC_SECRET",
"https://x.test/cb?client%5Fsecret=***&view=public",
),
(
"https://x.test/cb?client%255Fsecret=DOUBLE_SECRET&view=public",
"DOUBLE_SECRET",
"https://x.test/cb?client%255Fsecret=***&view=public",
),
(
"/resume?token=SEMICOLON_SECRET;view=public",
"SEMICOLON_SECRET",
"/resume?token=***;view=public",
),
(
"//user:NET_SECRET@x.test/path",
"NET_SECRET",
"//user:***@x.test/path",
),
],
)
def test_masks_all_url_reference_forms_only_when_opted_in(
self, text, secret, expected
):
assert redact_sensitive_text(text) == text
result = redact_sensitive_text(text, redact_url_credentials=True)
assert secret not in result
assert result == expected
def test_similarly_named_public_params_remain_unchanged(self):
text = "/metrics?token_count=17&session_id=public"
assert redact_sensitive_text(text, redact_url_credentials=True) == text
class TestBareTokenUserinfoRedaction:
"""Regression tests for #6396 — a bare credential in URL userinfo
(``scheme://TOKEN@host``, no ``user:pass`` colon) is redacted. This is the

View file

@ -116,11 +116,19 @@ def test_provider_context_is_strictly_sanitized_before_plugin_engine(monkeypatch
prefix_secret = "sk-" + "a" * 30
query_secret = "opaque-query-secret"
userinfo_value = "opaque-userinfo-value"
fragment_secret = "FRAG_SECRET"
relative_secret = "REL_SECRET"
encoded_key_secret = "ENC_SECRET"
network_userinfo_secret = "NET_SECRET"
manager = MagicMock()
manager.on_pre_compress.return_value = (
f"api key: {prefix_secret}\n"
f"callback: https://example.test/cb?access_token={query_secret}&state=ok\n"
f"endpoint: https://user:{userinfo_value}@example.test/private"
f"endpoint: https://user:{userinfo_value}@example.test/private\n"
f"fragment: https://x.test/#access_token={fragment_secret}&view=public\n"
f"relative: /resume?token={relative_secret}&view=public\n"
f"encoded: https://x.test/cb?client%5Fsecret={encoded_key_secret}&view=public\n"
f"network: //user:{network_userinfo_secret}@x.test/path"
)
received = []
compressor = MagicMock()
@ -143,8 +151,16 @@ def test_provider_context_is_strictly_sanitized_before_plugin_engine(monkeypatch
assert prefix_secret not in context
assert query_secret not in context
assert userinfo_value not in context
assert fragment_secret not in context
assert relative_secret not in context
assert encoded_key_secret not in context
assert network_userinfo_secret not in context
assert "access_token=***" in context
assert "https://user:***@example.test/private" in context
assert "https://x.test/#access_token=***&view=public" in context
assert "/resume?token=***&view=public" in context
assert "client%5Fsecret=***&view=public" in context
assert "//user:***@x.test/path" in context
def test_provider_context_is_bounded_before_plugin_engine():