mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
fix(hide-secrets): restore credential coverage lost to the 4.5 entropy limit (#40190)
* fix(hide-secrets): restore credential coverage lost to the 4.5 entropy limit Shannon entropy is bounded by log2(length), so the 4.5 limit #39879 shipped cannot score any value shorter than 23 characters, and it catches a random 32-character base64 credential only about two thirds of the time. A line like REDIS_PASSWORD=aB3dE6gH9jK2mN5p therefore reaches the provider in the clear. Add a keyword plugin that yields the credential-shaped value assigned to a credential-named key, reusing detect_secrets' own maintained denylist so camelCase, snake_case and SCREAMING_CASE all work with no local word list, and re-run the assignment-quoting transform detect_secrets skips once its first pass has matched. The entropy limits are untouched, so #39879's false-positive fix still holds. * fix(hide-secrets): read the assignments in a prompt that is mostly prose configparser aborts the whole parse on the first line it cannot read, so a message like "Here is my config, can you review it?" followed by REDIS_PASSWORD=... lost every assignment to that one prose line. Hand the parser only the lines it can read, dedent the assignments inside a pasted config, and keep each key distinct by line number so a config naming api_key once per model keeps every value instead of only the last. * fix(hide-secrets): drop the plugin docstrings and pin the block-scalar shapes * fix(hide-secrets): keep a comment or an indented header from closing an open value * fix(hide-secrets): drop the explanatory comments from the new scan helpers * fix(hide-secrets): accept punctuation in a credential value The value filter only allowed the URL-safe Base64 alphabet, so a password such as hunter2!brahms or p@ssw0rd!2026 passed through unredacted while the upstream keyword plugin had already matched it. The filter now rejects only whitespace and brackets, which keeps function calls, subscripts and sentences out while letting symbol-heavy passwords through. * fix(hide-secrets): redact every credential on a line and skip timestamps and plain urls replaces the inherited first-match scan with finditer over every keyword match, drops iso 8601 timestamps and userinfo-free urls from credential values, and threads the parser's open-option state through itertools.accumulate instead of rebinding it Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): scan the first token of an assignment and ignore surrounding punctuation Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * refactor(hide-secrets): drop the unreachable configparser error fallback Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): keep prose after a credential key out of the keyword detector A bare value followed by ordinary words (secret_sauce: Worcestershire sauce) is prose, so the synthetic assignment is only built when the value stands alone or is followed by a shell operator, comment, or another assignment Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): scan the first token of shell-style assignments regardless of what follows Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): keep spaced assignments in scope when shell text follows the value Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(hide-secrets): drop docstrings that restate the test names Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): stop reading a comparison operator as a trailing assignment Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(hide-secrets): keep dashed flags as assignment trailers Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng-berri
GitHub
parent
359b7a8489
commit
e4706fa409
3 changed files with 764 additions and 33 deletions
|
|
@ -11,10 +11,14 @@ import sys
|
|||
sys.path.insert(
|
||||
0, os.path.abspath("../..")
|
||||
) # Adds the parent directory to the system path
|
||||
import functools
|
||||
import configparser
|
||||
import contextlib
|
||||
import itertools
|
||||
import re
|
||||
import tempfile
|
||||
from collections.abc import Generator, Iterator, Sequence
|
||||
from contextvars import ContextVar
|
||||
from typing import TYPE_CHECKING, ClassVar, Literal, Optional
|
||||
from typing import TYPE_CHECKING, ClassVar, Final, Literal, Optional
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.caching.caching import DualCache
|
||||
|
|
@ -433,12 +437,101 @@ _default_detect_secrets_config = {
|
|||
"name": "ZendeskSecretKeyDetector",
|
||||
"path": _custom_plugins_path + "/zendesk_secret_key.py",
|
||||
},
|
||||
{
|
||||
"name": "CredentialKeywordDetector",
|
||||
"path": _custom_plugins_path + "/credential_keyword.py",
|
||||
},
|
||||
{"name": "Base64HighEntropyString", "limit": 4.5},
|
||||
{"name": "HexHighEntropyString", "limit": 3.0},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
_CONFIG_SECTION: Final = "litellm-prompt"
|
||||
|
||||
_ASSIGNMENT_LINE: Final = re.compile(r"[^\s\[#;:=][^:=]*[:=]")
|
||||
|
||||
_SHELL_ASSIGNMENT: Final = re.compile(r"(?P<key>[^\s\[#;:=](?:[^:=]*[^\s:=])?)=(?P<value>\S+)")
|
||||
|
||||
_SHELL_OPERATORS: Final = ";&|"
|
||||
|
||||
_SHELL_TRAILER: Final = re.compile(r"\\|#.*|-*\w[\w.-]*=\S*")
|
||||
|
||||
_SCAN_SUFFIX: Final = ".py"
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _temp_file(text: str) -> Generator[str, None, None]:
|
||||
temp_file: Final = tempfile.NamedTemporaryFile(suffix=_SCAN_SUFFIX, delete=False)
|
||||
try:
|
||||
temp_file.write(text.encode("utf-8"))
|
||||
temp_file.close()
|
||||
yield temp_file.name
|
||||
finally:
|
||||
temp_file.close()
|
||||
os.remove(temp_file.name)
|
||||
|
||||
|
||||
def _scan_lines(lines: Sequence[str]) -> frozenset[tuple[str, str]]:
|
||||
from detect_secrets import SecretsCollection
|
||||
|
||||
secrets: Final = SecretsCollection()
|
||||
with _temp_file("\n".join(lines)) as path:
|
||||
secrets.scan_file(path)
|
||||
|
||||
return frozenset(
|
||||
(found_secret.secret_value, found_secret.type)
|
||||
for file in secrets.files
|
||||
for found_secret in secrets[file]
|
||||
if found_secret.secret_value is not None
|
||||
)
|
||||
|
||||
|
||||
def _classify_line(state: tuple[bool, str | None], numbered: tuple[int, str]) -> tuple[bool, str | None]:
|
||||
open_option: Final = state[0]
|
||||
number, line = numbered
|
||||
stripped: Final = line.strip()
|
||||
if not stripped or stripped[0] in "#;":
|
||||
return open_option, None
|
||||
shell_assignment: Final = _SHELL_ASSIGNMENT.match(stripped)
|
||||
if shell_assignment is not None:
|
||||
return True, f"{shell_assignment['key']}_{number}={shell_assignment['value']}"
|
||||
assignment: Final = _ASSIGNMENT_LINE.match(stripped)
|
||||
if assignment is not None:
|
||||
return True, f"{assignment.group()[:-1].strip()}_{number}{stripped[assignment.end() - 1 :]}"
|
||||
if line[0].isspace() and open_option:
|
||||
return True, line
|
||||
return False, None
|
||||
|
||||
|
||||
def _parseable_lines(text: str) -> Iterator[str]:
|
||||
states: Final = itertools.accumulate(enumerate(text.splitlines()), _classify_line, initial=(False, None))
|
||||
return (line for _, line in states if line is not None)
|
||||
|
||||
|
||||
def _lone_value(line: str) -> str | None:
|
||||
tokens: Final = line.split()
|
||||
if not tokens or '"' in tokens[0]:
|
||||
return None
|
||||
value: Final = tokens[0].rstrip(_SHELL_OPERATORS)
|
||||
if len(tokens) == 1 or value != tokens[0] or _SHELL_TRAILER.fullmatch(tokens[1]) is not None:
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _quoted_assignments(text: str) -> tuple[str, ...]:
|
||||
parser: Final = configparser.ConfigParser(interpolation=None)
|
||||
parser.optionxform = str # pyright: ignore[reportAttributeAccessIssue] # configparser types optionxform as a method
|
||||
parser.read_string(f"[{_CONFIG_SECTION}]\n" + "\n".join(_parseable_lines(text)))
|
||||
return tuple(
|
||||
f'{key} = "{value}"'
|
||||
for section in parser
|
||||
for key, values in parser.items(section)
|
||||
for line in values.splitlines()
|
||||
if (value := _lone_value(line)) is not None
|
||||
)
|
||||
|
||||
|
||||
class _ENTERPRISE_SecretDetection(CustomGuardrail):
|
||||
# Keeps proxied traffic on async_pre_call_hook (the unified apply_guardrail
|
||||
# path skips should_run_check and never sees data["prompt"]).
|
||||
|
|
@ -449,35 +542,21 @@ class _ENTERPRISE_SecretDetection(CustomGuardrail):
|
|||
super().__init__(**kwargs)
|
||||
|
||||
def scan_message_for_secrets(self, message_content: str):
|
||||
from detect_secrets import SecretsCollection
|
||||
from detect_secrets.settings import transient_settings
|
||||
|
||||
temp_file = tempfile.NamedTemporaryFile(delete=False)
|
||||
temp_file.write(message_content.encode("utf-8"))
|
||||
temp_file.close()
|
||||
|
||||
secrets = SecretsCollection()
|
||||
|
||||
detect_secrets_config = (
|
||||
self.user_defined_detect_secrets_config or _default_detect_secrets_config
|
||||
)
|
||||
with transient_settings(detect_secrets_config):
|
||||
secrets.scan_file(temp_file.name)
|
||||
|
||||
os.remove(temp_file.name)
|
||||
found: Final = _scan_lines(
|
||||
(*message_content.splitlines(), *_quoted_assignments(message_content))
|
||||
)
|
||||
|
||||
return [
|
||||
{"type": found_secret.type, "value": found_secret.secret_value}
|
||||
for file in sorted(secrets.files)
|
||||
for found_secret in sorted(
|
||||
secrets[file],
|
||||
key=lambda secret: (
|
||||
-len(secret.secret_value or ""),
|
||||
secret.type,
|
||||
secret.secret_value or "",
|
||||
),
|
||||
{"type": secret_type, "value": value}
|
||||
for value, secret_type in sorted(
|
||||
found, key=lambda pair: (-len(pair[0]), pair[1], pair[0])
|
||||
)
|
||||
if found_secret.secret_value is not None
|
||||
]
|
||||
|
||||
def redact_text(self, text: str, source: str = "message") -> str:
|
||||
|
|
@ -490,15 +569,16 @@ class _ENTERPRISE_SecretDetection(CustomGuardrail):
|
|||
if counts is not None:
|
||||
for secret in detected_secrets:
|
||||
counts[secret["type"]] = counts.get(secret["type"], 0) + 1
|
||||
secret_types = [secret["type"] for secret in detected_secrets]
|
||||
secret_types: Final = sorted(
|
||||
dict.fromkeys(secret["type"] for secret in detected_secrets)
|
||||
)
|
||||
verbose_proxy_logger.warning(
|
||||
f"Detected and redacted secrets in {source}: {secret_types}"
|
||||
"Detected and redacted secrets in %s: %s", source, secret_types
|
||||
)
|
||||
return functools.reduce(
|
||||
lambda redacted, secret: redacted.replace(secret["value"], "[REDACTED]"),
|
||||
detected_secrets,
|
||||
text,
|
||||
pattern: Final = re.compile(
|
||||
"|".join(re.escape(secret["value"]) for secret in detected_secrets)
|
||||
)
|
||||
return pattern.sub("[REDACTED]", text)
|
||||
|
||||
async def should_run_check(self, user_api_key_dict: UserAPIKeyAuth) -> bool:
|
||||
if user_api_key_dict.permissions is not None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,63 @@
|
|||
import re
|
||||
from collections.abc import Generator, Mapping
|
||||
from string import punctuation
|
||||
from typing import Final
|
||||
|
||||
from detect_secrets.plugins.keyword import (
|
||||
QUOTES_REQUIRED_DENYLIST_REGEX_TO_GROUP,
|
||||
KeywordDetector,
|
||||
)
|
||||
|
||||
_CREDENTIAL_VALUE: Final = re.compile(r"[^\s()\[\]]+")
|
||||
_ENVIRONMENT_REFERENCE: Final = re.compile(r"os\.environ/\w+", re.IGNORECASE)
|
||||
_ENVIRONMENT_VARIABLE_NAME: Final = re.compile(r"[A-Z][A-Z0-9]*(?:_[A-Z0-9]+)+")
|
||||
_LOWERCASE_WORD_SEQUENCE: Final = re.compile(r"[a-z]+(?:[-._/][a-z]+)+")
|
||||
_ISO_8601_TIMESTAMP: Final = re.compile(
|
||||
r"\d{4}-\d{2}-\d{2}(?:T\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:?\d{2})?)?"
|
||||
)
|
||||
_URL_WITHOUT_USERINFO_OR_QUERY: Final = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*://[^\s@?]*")
|
||||
_BENIGN_VALUES: Final = (
|
||||
_ENVIRONMENT_REFERENCE,
|
||||
_ENVIRONMENT_VARIABLE_NAME,
|
||||
_LOWERCASE_WORD_SEQUENCE,
|
||||
_ISO_8601_TIMESTAMP,
|
||||
_URL_WITHOUT_USERINFO_OR_QUERY,
|
||||
)
|
||||
|
||||
|
||||
class CredentialKeywordDetector(KeywordDetector): # pyright: ignore[reportUntypedBaseClass] # detect_secrets ships no type information
|
||||
secret_type = "Credential Keyword"
|
||||
|
||||
def __init__(self, minimum_length: int = 12, keyword_exclude: str | None = None) -> None:
|
||||
if (
|
||||
not isinstance(minimum_length, int) # pyright: ignore[reportUnnecessaryIsInstance] # the value comes from an operator's YAML
|
||||
or minimum_length < 1
|
||||
):
|
||||
raise ValueError(f"minimum_length must be a positive integer, got {minimum_length!r}")
|
||||
super().__init__(keyword_exclude=keyword_exclude)
|
||||
self.minimum_length = minimum_length
|
||||
|
||||
def _is_credential(self, value: str) -> bool:
|
||||
core: Final = value.strip(punctuation)
|
||||
return (
|
||||
len(value) >= self.minimum_length
|
||||
and _CREDENTIAL_VALUE.fullmatch(value) is not None
|
||||
and all(benign.fullmatch(core) is None for benign in _BENIGN_VALUES)
|
||||
)
|
||||
|
||||
def analyze_string(
|
||||
self,
|
||||
string: str,
|
||||
denylist_regex_to_group: Mapping[re.Pattern[str], int] | None = None,
|
||||
) -> Generator[str, None, None]:
|
||||
if self.keyword_exclude is not None and self.keyword_exclude.search(string):
|
||||
return
|
||||
regex_to_group: Final = (
|
||||
QUOTES_REQUIRED_DENYLIST_REGEX_TO_GROUP if denylist_regex_to_group is None else denylist_regex_to_group
|
||||
)
|
||||
yield from (
|
||||
match.group(group)
|
||||
for regex, group in regex_to_group.items()
|
||||
for match in regex.finditer(string)
|
||||
if self._is_credential(match.group(group))
|
||||
)
|
||||
|
|
@ -10,12 +10,15 @@ Covers the three defects from the ticket:
|
|||
handling live only on the native path).
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm_enterprise.enterprise_callbacks.secret_detection import (
|
||||
_ENTERPRISE_SecretDetection,
|
||||
_default_detect_secrets_config,
|
||||
_masked_entity_count,
|
||||
)
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
|
|
@ -29,6 +32,13 @@ URL_ENCODED_KEY = "Bearer%20sk-Ab3dEf6Gh7Ij8Kl9Mn0Pq2Rs3Tu4Vw5X"
|
|||
AWS_KEYS = [f"AKIAIOSFODNN7EXAMPL{suffix}" for suffix in "FEDCBA"]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_masked_entity_count():
|
||||
token = _masked_entity_count.set(None)
|
||||
yield
|
||||
_masked_entity_count.reset(token)
|
||||
|
||||
|
||||
def _guardrail() -> _ENTERPRISE_SecretDetection:
|
||||
return _ENTERPRISE_SecretDetection(guardrail_name="hide-secrets", event_hook="pre_call", default_on=True)
|
||||
|
||||
|
|
@ -58,6 +68,561 @@ def test_scan_message_preserves_quoted_benign_identifiers():
|
|||
assert guardrail.redact_text(content) == content
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content,secret",
|
||||
[
|
||||
("REDIS_PASSWORD=aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
("SESSION_SECRET=Kp7Nq2Wz9Bt4Xr6Vm1Ls", "Kp7Nq2Wz9Bt4Xr6Vm1Ls"),
|
||||
('{"db_password": "Tq8Zm2XpLv9KdNbRcYw3"}', "Tq8Zm2XpLv9KdNbRcYw3"),
|
||||
("api_secret: Zx4Kp9Lm2Qr7Ns3Vt", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("password = hunter2brahms9x", "hunter2brahms9x"),
|
||||
("client_secret=Hq7Zm3XkLp9Wd2Nb", "Hq7Zm3XkLp9Wd2Nb"),
|
||||
('apiKey: "aB3dE6gH9jK2mN5p"', "aB3dE6gH9jK2mN5p"),
|
||||
('{"clientSecret": "Kp7Nq2Wz9Bt4Xr6Vm1Ls"}', "Kp7Nq2Wz9Bt4Xr6Vm1Ls"),
|
||||
('dbPassword = "Zx4Kp9Lm2Qr7Ns3Vt"', "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("MY_APP_DB_PASSWORD=Kp7Nq2Wz9Bt4Xr6Vm1Ls", "Kp7Nq2Wz9Bt4Xr6Vm1Ls"),
|
||||
("x_api_key: 8f3Kd9Lm2Qr7Ns3Vt", "8f3Kd9Lm2Qr7Ns3Vt"),
|
||||
("password: Zm9vYmFyYmF6+abc/def123=", "Zm9vYmFyYmF6+abc/def123="),
|
||||
("REDIS_PASSWORD=correcthorsebattery", "correcthorsebattery"),
|
||||
('SECRET_KEY = "django-insecure-9v2xk4qw8z"', "django-insecure-9v2xk4qw8z"),
|
||||
(
|
||||
"aws_secret_access_key = wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
||||
"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
|
||||
),
|
||||
("password=aB3dE6gH9jK2", "aB3dE6gH9jK2"),
|
||||
("api_key: hunter2!brahms", "hunter2!brahms"),
|
||||
('db_password: "p@ssw0rd!2026"', "p@ssw0rd!2026"),
|
||||
(
|
||||
'url: "postgresql://user:s3cr3t@db-host:5432/app"',
|
||||
"postgresql://user:s3cr3t@db-host:5432/app",
|
||||
),
|
||||
(
|
||||
'db_password: "postgresql://user:s3cr3t@db-host:5432/app"',
|
||||
"postgresql://user:s3cr3t@db-host:5432/app",
|
||||
),
|
||||
(
|
||||
'signing_secret_url: "https://example.com/cb?sig=Zx4Kp9Lm2Qr7Ns3Vt"',
|
||||
"https://example.com/cb?sig=Zx4Kp9Lm2Qr7Ns3Vt",
|
||||
),
|
||||
(
|
||||
'redis_secret_url: "redis://:Zx4Kp9Lm2Qr7Ns3Vt@cache-host:6379/0"',
|
||||
"Zx4Kp9Lm2Qr7Ns3Vt",
|
||||
),
|
||||
("password=2026-09-08T17:38:40Zbrahms", "2026-09-08T17:38:40Zbrahms"),
|
||||
(
|
||||
'{"password": "YOUR_API_KEY_HERE", "client_secret": "correcthorsebattery"}',
|
||||
"correcthorsebattery",
|
||||
),
|
||||
("docker run -e REDIS_PASSWORD=aB3dE6gH9jK2mN5p \\\n -e REDIS_PORT=6379 redis", "aB3dE6gH9jK2mN5p"),
|
||||
("DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt && echo done", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("password = Zx4Kp9Lm2Qr7Ns3Vt # rotate me", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("my db password: Zx4Kp9Lm2Qr7Ns3Vt.", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("export DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt DB_HOST=db.internal", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt; systemctl restart app", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt | tee creds.txt", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt > setup.log", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("docker run -e DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt --name app postgres", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("password=correcthorsebattery please", "correcthorsebattery"),
|
||||
("DB_PASSWORD = Zx4Kp9Lm2Qr7Ns3Vt; systemctl restart app", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD = Zx4Kp9Lm2Qr7Ns3Vt \\", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD = Zx4Kp9Lm2Qr7Ns3Vt DB_HOST=db.internal", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD = Zx4Kp9Lm2Qr7Ns3Vt --db-host=db.internal", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("DB_PASSWORD = Zx4Kp9Lm2Qr7Ns3Vt DEBUG=", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
],
|
||||
ids=[
|
||||
"env-password",
|
||||
"env-secret",
|
||||
"json-field",
|
||||
"yaml-field",
|
||||
"bare-assignment",
|
||||
"client-secret",
|
||||
"camel-case-key",
|
||||
"camel-case-secret",
|
||||
"camel-case-password",
|
||||
"namespaced-env",
|
||||
"underscored-header",
|
||||
"base64-padding",
|
||||
"digit-free-value",
|
||||
"django-secret-key",
|
||||
"slashed-aws-secret",
|
||||
"shortest-accepted-value",
|
||||
"punctuation-bearing-password",
|
||||
"symbol-heavy-password",
|
||||
"connection-string-under-a-url-key",
|
||||
"connection-string-under-a-credential-key",
|
||||
"signed-url-under-a-credential-key",
|
||||
"password-only-url-under-a-credential-key",
|
||||
"timestamp-prefixed-password",
|
||||
"credential-after-a-rejected-placeholder",
|
||||
"docker-flag-with-a-line-continuation",
|
||||
"shell-command-after-the-value",
|
||||
"inline-comment-after-the-value",
|
||||
"sentence-ending-in-the-value",
|
||||
"second-assignment-after-the-value",
|
||||
"semicolon-after-the-value",
|
||||
"pipe-after-the-value",
|
||||
"redirect-after-the-value",
|
||||
"docker-flag-after-the-value",
|
||||
"prose-after-a-shell-assignment",
|
||||
"spaced-assignment-then-a-shell-command",
|
||||
"spaced-assignment-then-a-line-continuation",
|
||||
"spaced-assignment-then-a-second-assignment",
|
||||
"spaced-assignment-then-a-dashed-flag",
|
||||
"spaced-assignment-then-an-empty-assignment",
|
||||
],
|
||||
)
|
||||
def test_scan_message_redacts_credentials_assigned_to_credential_keys(content, secret):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert secret not in guardrail.redact_text(content)
|
||||
|
||||
|
||||
def test_scan_message_redacts_only_the_first_token_of_a_shell_assignment():
|
||||
guardrail = _guardrail()
|
||||
content = "docker run -e REDIS_PASSWORD=aB3dE6gH9jK2mN5p \\\n -e REDIS_PORT=6379 redis && echo done"
|
||||
|
||||
assert (
|
||||
guardrail.redact_text(content)
|
||||
== "docker run -e REDIS_PASSWORD=[REDACTED] \\\n -e REDIS_PORT=6379 redis && echo done"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("operator", [";", "&&", "|"])
|
||||
def test_scan_message_keeps_a_shell_operator_glued_to_the_value(operator):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert (
|
||||
guardrail.redact_text(f"DB_PASSWORD=Zx4Kp9Lm2Qr7Ns3Vt{operator} systemctl restart app")
|
||||
== f"DB_PASSWORD=[REDACTED]{operator} systemctl restart app"
|
||||
)
|
||||
|
||||
|
||||
def test_scan_message_closes_a_yaml_block_at_the_next_unindented_line():
|
||||
guardrail = _guardrail()
|
||||
content = "api_key: >\n aB3dE6gH9jK2mN5p\nSteps\n Rotate-Before-Friday please"
|
||||
|
||||
assert guardrail.redact_text(content) == "api_key: >\n [REDACTED]\nSteps\n Rotate-Before-Friday please"
|
||||
|
||||
|
||||
def test_scan_message_redacts_every_credential_on_one_line():
|
||||
guardrail = _guardrail()
|
||||
content = '{"db_password": "Tq8Zm2XpLv9KdNbRcYw3", "client_secret": "correcthorsebattery"}'
|
||||
|
||||
assert guardrail.redact_text(content) == '{"db_password": "[REDACTED]", "client_secret": "[REDACTED]"}'
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
"The user forgot their password and asked for a reset link",
|
||||
"Rotate the client secret every 90 days",
|
||||
"The secret: keep it quiet",
|
||||
"My password: correct horse battery staple",
|
||||
"secretary: Maria Gonzalez",
|
||||
"password_reset_email: Please click the link below to reset",
|
||||
'config = {"api_key": "YOUR_API_KEY_HERE"}',
|
||||
"api_key: <your-key-here>",
|
||||
'{"max_tokens": 4096, "model": "gpt-4o-mini"}',
|
||||
'def get_api_key():\n return os.environ["OPENAI_API_KEY"]',
|
||||
' valid_token = UserAPIKeyAuth(user_id="u1")',
|
||||
'password = get_password(user, "prod")',
|
||||
"monkey=aB3dE6gH9jK2mN5p",
|
||||
"idempotency_key: req_2026090712000000",
|
||||
'cache_key = "u1_user_api_key_user_id"',
|
||||
"the key: 2026-09-07T12:00:00Z",
|
||||
"api_key: os.environ/E2B_API_KEY",
|
||||
"langfuse_secret: os.environ/LANGFUSE_PROJECT1_SECRET",
|
||||
"api_key = OPENAI_API_KEY",
|
||||
"password = pwd12345678",
|
||||
"model_key: gpt-4o-mini-2024-07-18",
|
||||
"openrouter/anthropic/claude-3-5-sonnet-20240620",
|
||||
'{"content-type": "application/json"}',
|
||||
"passwordless_login: enabled-for-all-users",
|
||||
'password: "I forgot mine, can you reset it"',
|
||||
"secret_sauce: tomatoes-basil-garlic-oregano",
|
||||
"user_secret_question: what-was-your-first-pet",
|
||||
"password_reset_url: example.com/reset-password/flow",
|
||||
"private_key_path: keys/prod/server-cert.pem",
|
||||
"litellm.completion(model=model, api_key=openai_api_key)",
|
||||
"params['aws_secret_access_key'] = aws_secret_access_key",
|
||||
'api_key = "OPENAI_API_KEY"',
|
||||
"model_list:\n - litellm_params:\n api_key: 'PERPLEXITY_API_KEY'",
|
||||
'config = build(_provider("ve_missing", api_key_env="VE_MISSING_KEY"))',
|
||||
"api_key = get_api_key_from_env()",
|
||||
"api_key = get_secret_str(MISTRAL_OCR_API_KEY_ENV_VAR)",
|
||||
"secret_manager = MagicMock(spec=BaseSecretManager)",
|
||||
"api_key = self.resolve_server_api_key(",
|
||||
"api_key = sys.argv[1]",
|
||||
"password = credentials[environment]",
|
||||
'api_key_created_at: "2026-09-08T17:38:40Z"',
|
||||
'api_key_expires_at: "2026-09-08T17:38:40.123456+05:30"',
|
||||
'password_reset_url: "https://example.com/reset-password/flow"',
|
||||
'secret_docs_url: "https://example.com/reset-password/flow#step-2"',
|
||||
'{"api_key_created_at": "2026-09-08T17:38:40Z", "password_reset_url": "https://example.com/reset/flow"}',
|
||||
"secret_sauce: tomatoes-basil-garlic-oregano.",
|
||||
"secret_docs_url: https://example.com/docs/keys, then rotate",
|
||||
"api_key_created_at: 2026-09-08T17:38:40Z; api_key_env: OPENAI_API_KEY!",
|
||||
"api_key: $OPENAI_API_KEY",
|
||||
'api_key: "${OPENAI_API_KEY}"',
|
||||
"private_key_path: /keys/prod/server-cert.pem",
|
||||
"password_hint: your usual one followed by Ticket-LIT7049-Suffix",
|
||||
"Translate this recipe note into French:\nsecret_sauce: Worcestershire sauce",
|
||||
"api_key = Massachusetts (the state, not a key)",
|
||||
"secret_sauce:Worcestershire sauce",
|
||||
"password: correctHorseBattery != anotherValue",
|
||||
],
|
||||
ids=[
|
||||
"prose-password",
|
||||
"prose-secret",
|
||||
"colon-prose-secret",
|
||||
"colon-prose-password",
|
||||
"secretary",
|
||||
"sentence-after-keyword",
|
||||
"uppercase-placeholder",
|
||||
"templated-placeholder",
|
||||
"max-tokens",
|
||||
"code-paste",
|
||||
"constructor-call",
|
||||
"indirect-reference",
|
||||
"word-ending-in-key",
|
||||
"idempotency-key",
|
||||
"cache-key",
|
||||
"timestamp-after-key",
|
||||
"env-reference",
|
||||
"env-reference-nested",
|
||||
"env-variable-name",
|
||||
"below-minimum-length",
|
||||
"model-name",
|
||||
"namespaced-model-name",
|
||||
"media-type",
|
||||
"hyphenated-english",
|
||||
"quoted-sentence-under-a-credential-key",
|
||||
"hyphenated-phrase",
|
||||
"hyphenated-question",
|
||||
"url-under-credential-key",
|
||||
"path-under-credential-key",
|
||||
"snake-case-argument",
|
||||
"snake-case-assignment",
|
||||
"quoted-env-variable-name",
|
||||
"quoted-env-name-in-a-config",
|
||||
"quoted-env-name-in-a-code-paste",
|
||||
"bare-call",
|
||||
"call-with-an-argument",
|
||||
"keyword-argument-call",
|
||||
"unclosed-call",
|
||||
"positional-subscript",
|
||||
"keyed-subscript",
|
||||
"timestamp-under-a-credential-key",
|
||||
"offset-timestamp-under-a-credential-key",
|
||||
"url-under-a-credential-key",
|
||||
"fragment-url-under-a-credential-key",
|
||||
"metadata-object-under-credential-keys",
|
||||
"hyphenated-english-ending-a-sentence",
|
||||
"url-followed-by-a-clause",
|
||||
"timestamp-and-env-name-with-trailing-punctuation",
|
||||
"shell-variable-reference",
|
||||
"quoted-braced-shell-variable-reference",
|
||||
"absolute-path-under-a-credential-key",
|
||||
"sentence-holding-a-later-mixed-case-token",
|
||||
"capitalized-word-starting-a-phrase",
|
||||
"capitalized-word-before-a-parenthetical",
|
||||
"yaml-scalar-without-a-space-after-the-colon",
|
||||
"comparison-operator-after-the-value",
|
||||
],
|
||||
)
|
||||
def test_scan_message_keeps_benign_values(content):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert guardrail.scan_message_for_secrets(content) == []
|
||||
assert guardrail.redact_text(content) == content
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,redacted",
|
||||
[("aB3dE6gH9jK2", True), ("aB3dE6gH9jK", False)],
|
||||
ids=["at-minimum-length", "below-minimum-length"],
|
||||
)
|
||||
def test_credential_keyword_detector_honours_its_minimum_length(value, redacted):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert (value not in guardrail.redact_text(f"password={value}")) is redacted
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value,redacted",
|
||||
[("aB3dE6gH9jK2", True), ("aB3dE6gH9jK", False)],
|
||||
ids=["at-default-minimum-length", "below-default-minimum-length"],
|
||||
)
|
||||
def test_credential_keyword_detector_defaults_its_minimum_length(value, redacted):
|
||||
guardrail = _ENTERPRISE_SecretDetection(
|
||||
guardrail_name="hide-secrets",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
detect_secrets_config={
|
||||
"plugins_used": [
|
||||
{key: setting for key, setting in plugin.items() if key != "minimum_length"}
|
||||
for plugin in _default_detect_secrets_config["plugins_used"]
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
assert (value not in guardrail.redact_text(f"password={value}")) is redacted
|
||||
|
||||
|
||||
def test_credential_keyword_detector_honours_keyword_exclude():
|
||||
guardrail = _ENTERPRISE_SecretDetection(
|
||||
guardrail_name="hide-secrets",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
detect_secrets_config={
|
||||
"plugins_used": [
|
||||
{**plugin, "keyword_exclude": "fixture_"} if plugin["name"] == "CredentialKeywordDetector" else plugin
|
||||
for plugin in _default_detect_secrets_config["plugins_used"]
|
||||
]
|
||||
},
|
||||
)
|
||||
content = "fixture_password=aB3dE6gH9jK2mN5p\npassword=Kp7Nq2Wz9Bt4Xr6Vm1Ls"
|
||||
|
||||
assert guardrail.redact_text(content) == "fixture_password=aB3dE6gH9jK2mN5p\npassword=[REDACTED]"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("minimum_length", ["12", 0, -1, 1.5], ids=["string", "zero", "negative", "float"])
|
||||
def test_credential_keyword_detector_rejects_an_unusable_minimum_length(minimum_length):
|
||||
guardrail = _ENTERPRISE_SecretDetection(
|
||||
guardrail_name="hide-secrets",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
detect_secrets_config={
|
||||
"plugins_used": [
|
||||
{**plugin, "minimum_length": minimum_length}
|
||||
if plugin["name"] == "CredentialKeywordDetector"
|
||||
else plugin
|
||||
for plugin in _default_detect_secrets_config["plugins_used"]
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="minimum_length"):
|
||||
guardrail.scan_message_for_secrets("password=aB3dE6gH9jK2mN5p")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
"[db\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"[\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"[note] have a look\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"]\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"[]\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
],
|
||||
ids=["unclosed", "bare-bracket", "bracketed-prose", "stray-close", "empty-header"],
|
||||
)
|
||||
def test_scan_message_reads_a_config_with_a_broken_section_header(content):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert "Zx4Kp9Lm2Qr7Ns3Vt" not in guardrail.redact_text(content)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
"=orphan\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
" indented before any key\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"greeting = %(name)s\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
"token = a\x00b\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n",
|
||||
],
|
||||
ids=["empty-key", "leading-continuation", "interpolation", "nul-byte"],
|
||||
)
|
||||
def test_scan_message_reads_lines_that_a_stock_ini_parser_rejects(content):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert "Zx4Kp9Lm2Qr7Ns3Vt" not in guardrail.redact_text(content)
|
||||
|
||||
|
||||
def test_scan_message_reads_a_config_that_repeats_a_section():
|
||||
guardrail = _guardrail()
|
||||
content = "[db]\nhost = localhost\n[db]\npassword = Zx4Kp9Lm2Qr7Ns3Vt\n"
|
||||
|
||||
assert "Zx4Kp9Lm2Qr7Ns3Vt" not in guardrail.redact_text(content)
|
||||
|
||||
|
||||
def test_scan_message_keeps_every_value_when_a_config_repeats_a_key():
|
||||
guardrail = _guardrail()
|
||||
content = (
|
||||
"model_list:\n"
|
||||
" - model_name: gpt-4o\n litellm_params:\n api_key: aB3dE6gH9jK2mN5p\n"
|
||||
" - model_name: claude\n litellm_params:\n api_key: Kp7Nq2Wz9Bt4Xr6Vm1Ls\n"
|
||||
)
|
||||
|
||||
redacted = guardrail.redact_text(content)
|
||||
|
||||
assert "aB3dE6gH9jK2mN5p" not in redacted
|
||||
assert "Kp7Nq2Wz9Bt4Xr6Vm1Ls" not in redacted
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content,secret",
|
||||
[
|
||||
(
|
||||
f"api_key: {OPENAI_KEY}\nREDIS_PASSWORD=aB3dE6gH9jK2mN5p",
|
||||
"aB3dE6gH9jK2mN5p",
|
||||
),
|
||||
(
|
||||
f"OPENAI_API_KEY={OPENAI_KEY}\nDB_PASSWORD=Kp7Nq2Wz9Bt4Xr6Vm1Ls",
|
||||
"Kp7Nq2Wz9Bt4Xr6Vm1Ls",
|
||||
),
|
||||
(
|
||||
f"api_key: {OPENAI_KEY}\npassword =\n Zx4Kp9Lm2Qr7Ns3Vt",
|
||||
"Zx4Kp9Lm2Qr7Ns3Vt",
|
||||
),
|
||||
(
|
||||
"Here is my config, can you review it?\nREDIS_PASSWORD=aB3dE6gH9jK2mN5p",
|
||||
"aB3dE6gH9jK2mN5p",
|
||||
),
|
||||
(
|
||||
"REDIS_PASSWORD=aB3dE6gH9jK2mN5p\nCan you tell me what is wrong with it?",
|
||||
"aB3dE6gH9jK2mN5p",
|
||||
),
|
||||
(
|
||||
"Hi team\nplease rotate this before Friday\ndb_password=Zx4Kp9Lm2Qr7Ns3Vt\nthanks!",
|
||||
"Zx4Kp9Lm2Qr7Ns3Vt",
|
||||
),
|
||||
(
|
||||
"model_list:\n - model_name: gpt-4o\n litellm_params:\n api_key: aB3dE6gH9jK2mN5p\n",
|
||||
"aB3dE6gH9jK2mN5p",
|
||||
),
|
||||
("api_key: >\n aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
("api_key: |-\n aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
("secret= \\\n aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
("password =\n# rotate me\n Zx4Kp9Lm2Qr7Ns3Vt", "Zx4Kp9Lm2Qr7Ns3Vt"),
|
||||
("api_key =\n; rotate me\n aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
(" # pasted from the vault\napi_key=aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
(" [db]\napi_key=aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
(" pasted with a leading indent\napi_key=aB3dE6gH9jK2mN5p", "aB3dE6gH9jK2mN5p"),
|
||||
],
|
||||
ids=[
|
||||
"flat-assignment",
|
||||
"env-file",
|
||||
"continuation-line",
|
||||
"prose-before",
|
||||
"prose-after",
|
||||
"prose-both-sides",
|
||||
"indented-config",
|
||||
"yaml-folded-block",
|
||||
"yaml-literal-block",
|
||||
"backslash-continuation",
|
||||
"comment-inside-a-value",
|
||||
"semicolon-comment-inside-a-value",
|
||||
"indented-comment-above",
|
||||
"indented-section-header-above",
|
||||
"indented-prose-above",
|
||||
],
|
||||
)
|
||||
def test_scan_message_still_sees_assignments_sharing_a_message_with_a_vendor_key(content, secret):
|
||||
guardrail = _guardrail()
|
||||
|
||||
redacted = guardrail.redact_text(content)
|
||||
assert secret not in redacted
|
||||
assert OPENAI_KEY not in redacted
|
||||
|
||||
|
||||
def test_environment_reference_filter_only_drops_the_whole_value():
|
||||
guardrail = _guardrail()
|
||||
|
||||
for reference in ("os.environ/OPENAI_API_KEY", "os.environ/e2b_api_key"):
|
||||
assert guardrail.redact_text(f"password={reference}") == f"password={reference}"
|
||||
assert guardrail.redact_text("password=notos.environ/OPENAI_API_KEY") == ("password=[REDACTED]")
|
||||
|
||||
|
||||
def test_environment_variable_names_are_dropped_only_for_the_keyword_plugin():
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert guardrail.redact_text("password=REDIS_PASSWORD") == "password=REDIS_PASSWORD"
|
||||
assert guardrail.scan_message_for_secrets('k = "ABCD1234_EFGH5678_IJKLMN"') == [
|
||||
{"type": "Base64 High Entropy String", "value": "ABCD1234_EFGH5678_IJKLMN"}
|
||||
]
|
||||
|
||||
|
||||
def test_masked_entity_count_keeps_the_vendor_type_beside_the_entropy_type():
|
||||
guardrail = _guardrail()
|
||||
_masked_entity_count.set({})
|
||||
|
||||
guardrail.redact_text('k = "ghp_abcdefghijklmnopqrstuvwxyzABCDEF1234"')
|
||||
|
||||
assert _masked_entity_count.get() == {
|
||||
"Base64 High Entropy String": 1,
|
||||
"GitHub Token": 1,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
f"api_key: '{OPENAI_KEY}'\n"
|
||||
+ "a: &a ["
|
||||
+ ", ".join(['"x"'] * 9)
|
||||
+ "]\n"
|
||||
+ "".join(f"{chr(98 + i)}: &{chr(98 + i)} [" + ", ".join([f"*{chr(97 + i)}"] * 9) + "]\n" for i in range(7)),
|
||||
f"api_key: '{OPENAI_KEY}'\ndeep: " + "[" * 400 + "]" * 400,
|
||||
f"api_key: '{OPENAI_KEY}'\nbroken: [unclosed",
|
||||
],
|
||||
ids=["anchor-expansion", "deep-nesting", "unparseable"],
|
||||
)
|
||||
def test_scan_message_contains_hostile_config_text(content, monkeypatch, tmp_path):
|
||||
guardrail = _guardrail()
|
||||
monkeypatch.setenv("TMPDIR", str(tmp_path))
|
||||
monkeypatch.setattr(tempfile, "tempdir", None)
|
||||
|
||||
started = time.perf_counter()
|
||||
found = guardrail.scan_message_for_secrets(content)
|
||||
|
||||
assert time.perf_counter() - started < 10.0
|
||||
assert OPENAI_KEY in [secret["value"] for secret in found]
|
||||
assert list(tmp_path.iterdir()) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
f"api_key = '{OPENAI_KEY}'\nbase = abcdefghijkl\npassword = x\n %(base)sZZZZQQQQ\n",
|
||||
"base = abcdefghijkl\npassword = x\n %(base)sZZZZQQQQ\n",
|
||||
f"api_key = '{OPENAI_KEY}'\nbase = Kp7Nq2Wz9Bt4\npassword = x\n"
|
||||
" %(base)s-primary\nnote = Kp7Nq2Wz9Bt4-primary is the hostname\n",
|
||||
'base = "abcdefghijkl"\npassword = "%(base)sZZZZQQQQ"\n',
|
||||
],
|
||||
ids=[
|
||||
"vendor-key-present",
|
||||
"no-vendor-key",
|
||||
"value-echoed-elsewhere",
|
||||
"quoted-interpolation",
|
||||
],
|
||||
)
|
||||
def test_scan_message_never_reports_a_value_the_message_does_not_hold(content):
|
||||
guardrail = _guardrail()
|
||||
|
||||
for secret in guardrail.scan_message_for_secrets(content):
|
||||
assert secret["value"] in content
|
||||
|
||||
|
||||
def test_scan_message_leaves_unrelated_text_alone_when_a_value_is_echoed():
|
||||
guardrail = _guardrail()
|
||||
content = (
|
||||
f"api_key = '{OPENAI_KEY}'\nbase = Kp7Nq2Wz9Bt4\npassword = x\n"
|
||||
" %(base)s-primary\nnote = Kp7Nq2Wz9Bt4-primary is the hostname\n"
|
||||
)
|
||||
|
||||
assert "note = Kp7Nq2Wz9Bt4-primary is the hostname" in guardrail.redact_text(content)
|
||||
|
||||
|
||||
def test_masked_entity_count_counts_each_secret_once():
|
||||
guardrail = _guardrail()
|
||||
_masked_entity_count.set({})
|
||||
|
||||
guardrail.redact_text(f"first {OPENAI_KEY} second {OPENAI_KEY}")
|
||||
|
||||
assert _masked_entity_count.get() == {"Strict OpenAI API Key": 1}
|
||||
|
||||
|
||||
def test_scan_message_redacts_every_openai_key_occurrence():
|
||||
guardrail = _guardrail()
|
||||
content = f"first {OPENAI_KEY}, second {OPENAI_KEY}"
|
||||
|
|
@ -81,9 +646,7 @@ def test_scan_message_requires_ascii_digits_for_openai_like_values():
|
|||
def test_scan_message_redacts_openai_key_after_separator():
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert guardrail.redact_text(f"openai_{OPENAI_KEY} key-{OPENAI_KEY}") == (
|
||||
"openai_[REDACTED] key-[REDACTED]"
|
||||
)
|
||||
assert guardrail.redact_text(f"openai_{OPENAI_KEY} key-{OPENAI_KEY}") == ("openai_[REDACTED] key-[REDACTED]")
|
||||
assert guardrail.redact_text(URL_ENCODED_KEY) == "Bearer%20[REDACTED]"
|
||||
|
||||
|
||||
|
|
@ -102,6 +665,31 @@ def test_scan_message_stays_linear_on_repeated_sk_separators():
|
|||
assert time.perf_counter() - started < 2.0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content",
|
||||
[
|
||||
f"api_key: '{OPENAI_KEY}'\npassword=" + "a-" * 10_000 + "!",
|
||||
f"api_key: '{OPENAI_KEY}'\npassword:" + '"' * 20_000,
|
||||
f"api_key: '{OPENAI_KEY}'\n" + "api_key:" * 10_000,
|
||||
f"api_key: '{OPENAI_KEY}'\nsecret=" + "aB3dE6gH9jK2mN5p " * 2_000,
|
||||
f"api_key: '{OPENAI_KEY}'\n" + "\n".join(f"password{i}=aB3dE6gH9jK2mN5p{i}" for i in range(3_000)),
|
||||
],
|
||||
ids=[
|
||||
"value-run",
|
||||
"quote-run",
|
||||
"keyword-run",
|
||||
"value-repeat",
|
||||
"assignment-flood",
|
||||
],
|
||||
)
|
||||
def test_scan_message_stays_linear_on_adversarial_credential_lines(content):
|
||||
guardrail = _guardrail()
|
||||
|
||||
started = time.perf_counter()
|
||||
guardrail.redact_text(content)
|
||||
assert time.perf_counter() - started < 10.0
|
||||
|
||||
|
||||
def test_scan_message_redacts_whole_stripe_live_key():
|
||||
guardrail = _guardrail()
|
||||
|
||||
|
|
@ -119,8 +707,8 @@ def test_scan_message_replaces_longest_overlapping_match_first():
|
|||
guardrail = _guardrail()
|
||||
content = f'token = "{OPENAI_KEY}/extra"'
|
||||
|
||||
detected = guardrail.scan_message_for_secrets(content)
|
||||
assert [secret["value"] for secret in detected] == [f"{OPENAI_KEY}/extra", OPENAI_KEY]
|
||||
values = [secret["value"] for secret in guardrail.scan_message_for_secrets(content)]
|
||||
assert values == [f"{OPENAI_KEY}/extra", OPENAI_KEY]
|
||||
assert guardrail.redact_text(content) == 'token = "[REDACTED]"'
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue