mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
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.
This commit is contained in:
Yucheng He
parent
1009976c49
commit
e47f07e44c
3 changed files with 494 additions and 30 deletions
|
|
@ -11,10 +11,13 @@ import sys
|
|||
sys.path.insert(
|
||||
0, os.path.abspath("../..")
|
||||
) # Adds the parent directory to the system path
|
||||
import functools
|
||||
import configparser
|
||||
import contextlib
|
||||
import re
|
||||
import tempfile
|
||||
from collections.abc import Generator, 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 +436,74 @@ _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"
|
||||
|
||||
# A .py suffix keeps detect_secrets' own config transformers off this file (they only fire on
|
||||
# FileType.OTHER and FileType.YAML) while leaving every plugin's regex set unchanged.
|
||||
_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 _quoted_assignments(text: str) -> tuple[str, ...]:
|
||||
"""Rewrites the bare ``key = value`` assignments in ``text`` as quoted ones.
|
||||
|
||||
detect_secrets does this itself, but only when its first pass over the raw text found
|
||||
nothing, so one vendor-prefixed key in a message hides every unquoted assignment beside
|
||||
it. Interpolation stays off so that no emitted value is one the message never contained.
|
||||
"""
|
||||
parser: Final = configparser.ConfigParser(interpolation=None)
|
||||
# Keys keep their case, exactly as detect_secrets' own parser does.
|
||||
parser.optionxform = str # pyright: ignore[reportAttributeAccessIssue] # configparser types optionxform as a method
|
||||
try:
|
||||
parser.read_string(f"[{_CONFIG_SECTION}]\n{text}")
|
||||
except (configparser.Error, UnicodeDecodeError):
|
||||
return ()
|
||||
|
||||
return tuple(
|
||||
f'{key} = "{value}"'
|
||||
for section in parser
|
||||
for key, values in parser.items(section)
|
||||
for value in values.splitlines()
|
||||
if value and '"' not in value
|
||||
)
|
||||
|
||||
|
||||
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 +514,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 +541,18 @@ 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,
|
||||
# detected_secrets is ordered longest value first, so the alternation redacts a
|
||||
# secret that contains another one as a whole rather than in pieces.
|
||||
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,55 @@
|
|||
"""
|
||||
This plugin searches for credential-shaped values assigned to a credential-named key.
|
||||
"""
|
||||
|
||||
import re
|
||||
from collections.abc import Generator, Mapping
|
||||
from typing import Final
|
||||
|
||||
from detect_secrets.plugins.keyword import KeywordDetector
|
||||
|
||||
_CREDENTIAL_VALUE: Final = re.compile(r"[A-Za-z0-9_.~+/-]+={0,2}")
|
||||
_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]+)+")
|
||||
|
||||
|
||||
class CredentialKeywordDetector(KeywordDetector): # pyright: ignore[reportUntypedBaseClass] # detect_secrets ships no type information
|
||||
"""Yields the values ``KeywordDetector`` matches that are one credential-shaped token of
|
||||
at least ``minimum_length`` characters, dropping the ones that name a credential rather
|
||||
than holding one."""
|
||||
|
||||
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:
|
||||
return (
|
||||
len(value) >= self.minimum_length
|
||||
and _CREDENTIAL_VALUE.fullmatch(value) is not None
|
||||
and _ENVIRONMENT_REFERENCE.fullmatch(value) is None
|
||||
and _ENVIRONMENT_VARIABLE_NAME.fullmatch(value) is None
|
||||
and _LOWERCASE_WORD_SEQUENCE.fullmatch(value) is None
|
||||
)
|
||||
|
||||
def analyze_string(
|
||||
self,
|
||||
string: str,
|
||||
denylist_regex_to_group: Mapping[re.Pattern[str], int] | None = None,
|
||||
) -> Generator[str, None, None]:
|
||||
yield from (
|
||||
value
|
||||
for value in super().analyze_string(string, denylist_regex_to_group)
|
||||
if self._is_credential(value)
|
||||
)
|
||||
|
|
@ -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,323 @@ 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"),
|
||||
],
|
||||
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",
|
||||
],
|
||||
)
|
||||
def test_scan_message_redacts_credentials_assigned_to_credential_keys(content, secret):
|
||||
guardrail = _guardrail()
|
||||
|
||||
assert secret not in guardrail.redact_text(content)
|
||||
|
||||
|
||||
@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",
|
||||
"api_key: hunter2!brahms",
|
||||
"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",
|
||||
"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",
|
||||
],
|
||||
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",
|
||||
"non-credential-charset",
|
||||
"model-name",
|
||||
"namespaced-model-name",
|
||||
"media-type",
|
||||
"hyphenated-english",
|
||||
"hyphenated-phrase",
|
||||
"hyphenated-question",
|
||||
"url-under-credential-key",
|
||||
"path-under-credential-key",
|
||||
"snake-case-argument",
|
||||
"snake-case-assignment",
|
||||
],
|
||||
)
|
||||
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):
|
||||
"""An operator config that names the plugin without sizing it keeps the same floor."""
|
||||
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
|
||||
|
||||
|
||||
@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):
|
||||
"""A bad value in an operator config has to fail while the guardrail is being built;
|
||||
reaching the scan with one turns every single request into a 500."""
|
||||
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,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",
|
||||
),
|
||||
],
|
||||
ids=["flat-assignment", "env-file", "continuation-line"],
|
||||
)
|
||||
def test_scan_message_still_sees_assignments_sharing_a_message_with_a_vendor_key(
|
||||
content, secret
|
||||
):
|
||||
"""detect_secrets stops quoting assignments as soon as its first pass matches."""
|
||||
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():
|
||||
"""The exclusion lives on the plugin, so it cannot suppress a vendor detector's hit."""
|
||||
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):
|
||||
"""The retry pass parses attacker-controlled text, so it must not raise, hang, or
|
||||
leave the prompt behind in a temp file."""
|
||||
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):
|
||||
"""The rewritten copy is parsed with interpolation off, so no reported value can be one
|
||||
the parser assembled rather than read; reporting one would mask unrelated text."""
|
||||
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():
|
||||
"""A value the parser could assemble also appears verbatim in a benign sentence; redacting
|
||||
it would destroy the sentence while leaving the line it came from untouched."""
|
||||
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}"
|
||||
|
|
@ -102,6 +429,34 @@ 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):
|
||||
"""A backtracking blow-up on these runs takes minutes, so the bound is loose enough
|
||||
to stay green on a loaded CI box."""
|
||||
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 +474,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