diff --git a/enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py b/enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py index bfbfd7bfb15..f0f85178672 100644 --- a/enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py +++ b/enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py @@ -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[^\s\[#;:=](?:[^:=]*[^\s:=])?)=(?P\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: diff --git a/enterprise/litellm_enterprise/enterprise_callbacks/secrets_plugins/credential_keyword.py b/enterprise/litellm_enterprise/enterprise_callbacks/secrets_plugins/credential_keyword.py new file mode 100644 index 00000000000..b69e347ded5 --- /dev/null +++ b/enterprise/litellm_enterprise/enterprise_callbacks/secrets_plugins/credential_keyword.py @@ -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)) + ) diff --git a/tests/test_litellm/enterprise/enterprise_callbacks/test_secret_detection.py b/tests/test_litellm/enterprise/enterprise_callbacks/test_secret_detection.py index c1c3569c0e2..bf9eeeb9968 100644 --- a/tests/test_litellm/enterprise/enterprise_callbacks/test_secret_detection.py +++ b/tests/test_litellm/enterprise/enterprise_callbacks/test_secret_detection.py @@ -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: ", + '{"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]"'