fix(security): reject path traversal and control chars in key_alias-derived secret names (LIT-4201)

key_alias becomes the secret name written to HashiCorp Vault and CyberArk
Conjur when store_virtual_keys is enabled. Vault's get_url concatenated
secret_name directly into the request URL and Conjur's
_ensure_variable_exists interpolated it unescaped into a YAML policy body,
so a malicious key_alias could path-traverse the Vault write or inject
extra Conjur policy statements. The only existing guard was opt-in
(enable_key_alias_format_validation, default off) and its charset still
permitted ".." even when enabled.

Add raise_if_unsafe_secret_name, an unconditional check rejecting ".."
sequences and control characters (including the Unicode line breaks YAML
treats the same as "\n": NEL, LINE SEPARATOR, PARAGRAPH SEPARATOR), applied
at both vulnerable sinks and at the API boundary in
_validate_key_alias_format, independent of the opt-in flag.

Also close two adjacent gaps found in internal review: Vault's get_url now
percent-encodes secret_name (preserving "/" and "@") so "#"/"?" can't turn
into a URL fragment/query string instead of a literal path segment, and
Conjur's policy YAML is now built with a real YAML serializer
(yaml.safe_dump forced to double-quoted style) instead of raw string
interpolation, so metacharacters that aren't on the traversal/control-char
denylist (a bare ":" or "#") can no longer change the parsed policy
structure.
This commit is contained in:
yucheng-berriai 2026-07-03 17:56:45 -07:00
parent 1e7dbe52c1
commit 7d9ac8d043
8 changed files with 291 additions and 9 deletions

View file

@ -111,6 +111,7 @@ from litellm.repositories.verification_token_repository import (
VerificationTokenRepository,
)
from litellm.router import Router
from litellm.secret_managers.base_secret_manager import raise_if_unsafe_secret_name
from litellm.secret_managers.main import get_secret
from litellm.types.proxy.management_endpoints.key_management_endpoints import (
BulkUpdateKeyRequest,
@ -6267,8 +6268,14 @@ def _validate_key_alias_format(key_alias: Optional[str]) -> None:
"""
Validate the format of the key_alias.
Gated behind ``litellm.enable_key_alias_format_validation`` (default **False**).
When disabled, no validation is performed so existing workflows are not broken.
key_alias can become the secret name written to an external secret manager
(HashiCorp Vault, CyberArk Conjur) when store_virtual_keys is enabled, so the
path-traversal / control-character check below always runs, regardless of
``litellm.enable_key_alias_format_validation``.
The remaining charset/length rules are gated behind
``litellm.enable_key_alias_format_validation`` (default **False**). When disabled,
only the security check above is performed, so existing workflows are not broken.
Rules (when enabled):
- None is OK (no alias).
@ -6276,10 +6283,20 @@ def _validate_key_alias_format(key_alias: Optional[str]) -> None:
- start/end with alphanumeric
- only allow a-zA-Z0-9_-/.@
"""
if not litellm.enable_key_alias_format_validation:
if key_alias is None:
return
if key_alias is None:
try:
raise_if_unsafe_secret_name(key_alias)
except ValueError as e:
raise ProxyException(
message=f"Invalid key_alias: {e}",
type=ProxyErrorTypes.bad_request_error,
param="key_alias",
code=400,
)
if not litellm.enable_key_alias_format_validation:
return
if not _KEY_ALIAS_PATTERN.match(key_alias):

View file

@ -1,3 +1,4 @@
import re
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, Union
@ -5,6 +6,26 @@ import httpx
from litellm import verbose_logger
# U+0085 (NEL), U+2028 (LINE SEPARATOR), and U+2029 (PARAGRAPH SEPARATOR) are line
# breaks under the YAML spec, same as backslash-n or backslash-r, so a denylist
# restricted to ASCII control characters is bypassable for YAML-based sinks.
_UNSAFE_SECRET_NAME_PATTERN = re.compile(r"\.\.|[\x00-\x1f\x7f-\x9f…]")
def raise_if_unsafe_secret_name(secret_name: str) -> None:
"""
Reject secret names that could path-traverse a secret manager's API (e.g. HashiCorp
Vault, which builds its request URL by string concatenation) or inject control
characters into a policy document (e.g. CyberArk Conjur, which embeds the secret
name directly into a YAML policy body).
Rejects any ".." substring rather than only a ".." path segment, since that is
the only traversal-safe answer for Vault's raw string-concatenated URL; a
version-like name such as "release-1.0..2" is rejected too.
"""
if _UNSAFE_SECRET_NAME_PATTERN.search(secret_name):
raise ValueError(f"Unsafe secret_name {secret_name!r}: must not contain '..' or control characters")
class BaseSecretManager(ABC):
"""

View file

@ -4,6 +4,7 @@ from typing import Any, Dict, Optional, Union
from urllib.parse import quote
import httpx
import yaml
import litellm
from litellm._logging import verbose_logger
@ -15,7 +16,7 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.proxy._types import KeyManagementSystem
from .base_secret_manager import BaseSecretManager
from .base_secret_manager import BaseSecretManager, raise_if_unsafe_secret_name
from .main import str_to_bool
@ -125,8 +126,13 @@ class CyberArkSecretManager(BaseSecretManager):
"""
# In production, we'd check if the variable exists first
# For now, we'll attempt to create it and ignore if it already exists
raise_if_unsafe_secret_name(secret_name)
policy_url = f"{self.conjur_addr}/policies/{self.conjur_account}/policy/root"
policy_yaml = f"- !variable {secret_name}\n"
# Force double-quoted scalar style so any YAML metacharacter in secret_name
# (newlines, other Unicode line breaks, colons, '#' comments) is escaped
# instead of being interpreted as new policy syntax.
quoted_name = yaml.safe_dump(secret_name, default_style='"').strip()
policy_yaml = f"- !variable {quoted_name}\n"
try:
client = _get_httpx_client(params={"ssl_verify": self.ssl_verify})

View file

@ -1,5 +1,6 @@
import os
from typing import Any, Dict, Optional, Union
from urllib.parse import quote
import httpx
@ -14,7 +15,7 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.proxy._types import KeyManagementSystem
from .base_secret_manager import BaseSecretManager
from .base_secret_manager import BaseSecretManager, raise_if_unsafe_secret_name
class HashicorpSecretManager(BaseSecretManager):
@ -220,6 +221,7 @@ class HashicorpSecretManager(BaseSecretManager):
- With custom mount: http://127.0.0.1:8200/v1/kv/data/mykey
- With path prefix: http://127.0.0.1:8200/v1/secret/data/myapp/mykey
"""
raise_if_unsafe_secret_name(secret_name)
resolved_namespace = self._sanitize_path_component(namespace if namespace is not None else self.vault_namespace)
resolved_mount = self._sanitize_path_component(mount_name if mount_name is not None else self.vault_mount_name)
if resolved_mount is None:
@ -234,7 +236,10 @@ class HashicorpSecretManager(BaseSecretManager):
_url += f"{resolved_mount}/data/"
if resolved_path_prefix:
_url += f"{resolved_path_prefix}/"
_url += secret_name
# Preserve "/" (hierarchical secret paths) and "@" (e.g. emails in aliases)
# unencoded; percent-encode everything else so a secret_name containing "#"
# or "?" can't turn into a URL fragment/query string instead of a path segment.
_url += quote(secret_name, safe="/@")
return _url
def _sanitize_plain_value(self, value: Optional[Union[str, int]]) -> Optional[str]:

View file

@ -5,6 +5,7 @@ Integration test for CyberArk Conjur Secret Manager.
import os
import sys
import pytest
import yaml
from dotenv import load_dotenv
load_dotenv()
@ -42,6 +43,90 @@ def create_mock_response(status_code: int, text: str = ""):
return mock_response
@pytest.mark.asyncio
async def test_cyberark_write_secret_rejects_yaml_injection():
"""
Regression test: secret_name (derived from user-controlled key_alias) used to be
interpolated unescaped into the Conjur policy YAML body
(`f"- !variable {secret_name}\n"`), so a key_alias containing a newline plus
another policy directive would be sent as additional Conjur policy statements.
async_write_secret must reject it before any policy/value HTTP call is made.
"""
with patch("litellm.proxy.proxy_server.premium_user", True):
malicious_secret_name = "foo\n- !grant\n role: !!admin\n member: attacker"
mock_sync_client = MagicMock()
mock_async_client = AsyncMock()
with (
patch(
"litellm.secret_managers.cyberark_secret_manager._get_httpx_client",
return_value=mock_sync_client,
),
patch(
"litellm.secret_managers.cyberark_secret_manager.get_async_httpx_client",
return_value=mock_async_client,
),
):
cyberark_manager = CyberArkSecretManager()
response = await cyberark_manager.async_write_secret(
secret_name=malicious_secret_name,
secret_value="sk-1234",
)
assert response["status"] == "error"
assert "Unsafe secret_name" in response["message"]
# The malicious policy YAML must never reach the wire.
mock_sync_client.client.post.assert_not_called()
mock_async_client.post.assert_not_called()
@pytest.mark.parametrize(
"secret_name",
[
"foo: bar", # colon: would otherwise parse as a mapping, not a scalar
"foo # bar", # unquoted '#' starts a YAML comment mid-scalar
"plain-alias",
"team/user@example.com",
],
)
def test_cyberark_ensure_variable_exists_escapes_yaml_metacharacters(secret_name):
"""
Regression test: secret_name is embedded in a hand-built single-line YAML
entry (`f"- !variable {secret_name}\\n"`). Characters that don't match
raise_if_unsafe_secret_name's traversal/control-character denylist (a bare
':' or '#', no newline needed) can still change the parsed YAML structure
-- ':' turns the scalar into a mapping, '#' truncates it at a comment.
_ensure_variable_exists must escape the scalar (not just denylist-check it)
so the policy body always parses back to exactly one '!variable' scalar
node holding the untouched secret_name.
"""
with patch("litellm.proxy.proxy_server.premium_user", True):
captured = {}
def _capture_post(url, headers=None, content=None):
captured["content"] = content
return create_mock_response(status_code=201, text="")
mock_sync_client = MagicMock()
mock_sync_client.client.post.side_effect = _capture_post
with patch(
"litellm.secret_managers.cyberark_secret_manager._get_httpx_client",
return_value=mock_sync_client,
):
cyberark_manager = CyberArkSecretManager()
cyberark_manager._ensure_variable_exists(secret_name)
policy_yaml = captured["content"]
parsed = yaml.compose(policy_yaml)
assert len(parsed.value) == 1
node = parsed.value[0]
assert node.tag == "!variable"
assert node.value == secret_name
@pytest.mark.asyncio
async def test_cyberark_write_and_read_secret():
"""

View file

@ -409,6 +409,61 @@ def test_hashicorp_custom_mount_and_prefix(hashicorp_secret_manager):
hashicorp_secret_manager.vault_namespace = original_namespace
@pytest.mark.parametrize(
"malicious_secret_name",
[
"../../../other-app/creds",
"litellm/../../secret",
"foo\nbar",
"foobar", # YAML LINE SEPARATOR: same class of break as "\n"
"foobar", # YAML PARAGRAPH SEPARATOR
"foo\x85bar", # YAML NEL
],
)
def test_hashicorp_get_url_rejects_path_traversal(monkeypatch, malicious_secret_name):
"""
Regression test: secret_name (derived from user-controlled key_alias) is
concatenated directly into the Vault request URL with no sanitization. get_url
must reject '..' sequences and control characters (including the Unicode line
breaks YAML treats the same as "\\n") instead of building a URL that could
escape the configured mount/path_prefix.
Uses monkeypatch + a directly-constructed manager (not the shared
hashicorp_secret_manager fixture) so this runs in CI without real Vault
credentials configured; get_url performs no I/O.
"""
monkeypatch.setenv("HCP_VAULT_TOKEN", "test-token-for-get-url-only")
manager = HashicorpSecretManager()
with pytest.raises(ValueError):
manager.get_url(malicious_secret_name)
def test_hashicorp_get_url_encodes_reserved_url_characters(monkeypatch):
"""
Regression test: secret_name used to be concatenated raw into the URL, so a
'#' or '?' would be interpreted as a URL fragment/query separator by the HTTP
client instead of a literal character in the Vault KV path. get_url must
percent-encode them while still preserving '/' (hierarchical paths) and '@'
(e.g. emails in aliases) unencoded, matching existing usage.
"""
monkeypatch.setenv("HCP_VAULT_TOKEN", "test-token-for-get-url-only")
manager = HashicorpSecretManager()
manager.vault_namespace = None
manager.vault_path_prefix = None
url = manager.get_url("foo#bar")
assert "#" not in url
assert url.endswith("foo%23bar")
url = manager.get_url("foo?evil=1")
assert "?" not in url.split("/data/", 1)[1]
assert url.endswith("foo%3Fevil%3D1")
url = manager.get_url("team/user@example.com")
assert url.endswith("team/user@example.com")
mock_old_vault_response = {
"request_id": "80fafb6a-e96a-4c5b-29fa-ff505ac72201",
"lease_id": "",

View file

@ -8969,7 +8969,7 @@ class TestValidateKeyAliasFormat:
litellm.enable_key_alias_format_validation = False
def test_validation_skipped_when_flag_disabled(self):
"""When enable_key_alias_format_validation is False (default), no validation occurs."""
"""When enable_key_alias_format_validation is False (default), no charset/length validation occurs."""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_validate_key_alias_format,
)
@ -8980,6 +8980,48 @@ class TestValidateKeyAliasFormat:
_validate_key_alias_format("!invalid!")
_validate_key_alias_format("a" * 256)
@pytest.mark.parametrize(
"unsafe_alias",
[
"../../../other-app/creds",
"litellm/../../secret",
"foo\n- !grant\n role: !!admin\n member: attacker",
"foo\rbar",
"foo\x00bar",
],
)
def test_validate_key_alias_format_rejects_traversal_and_control_chars_even_when_flag_disabled(
self, unsafe_alias
):
"""
Regression test: key_alias becomes the secret name written to HashiCorp Vault
(path-concatenated into the request URL) and CyberArk Conjur (interpolated into a
YAML policy body) when store_virtual_keys is enabled. This check must reject
path traversal and control characters unconditionally, since
enable_key_alias_format_validation defaults to False and its charset alone
(when enabled) still permits ".." sequences.
"""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_validate_key_alias_format,
)
with pytest.raises(ProxyException) as exc:
_validate_key_alias_format(unsafe_alias)
assert str(exc.value.code) == "400"
assert "Invalid key_alias" in str(exc.value.message)
def test_validate_key_alias_format_charset_alone_permits_dot_dot(self):
"""
Documents the pre-existing gap this PR closes: the opt-in charset regex allows
'.' and '/' individually, so it does not by itself catch '..' path traversal.
The unconditional security check (tested above) is what actually blocks it.
"""
from litellm.proxy.management_endpoints.key_management_endpoints import (
_KEY_ALIAS_PATTERN,
)
assert _KEY_ALIAS_PATTERN.match("a/../../etc/secret") is not None
def test_validate_key_alias_format_valid(self):
from litellm.proxy.management_endpoints.key_management_endpoints import (
_validate_key_alias_format,

View file

@ -0,0 +1,51 @@
"""
Test raise_if_unsafe_secret_name, the shared guard applied before any secret_name
(derived from user-controlled key_alias) reaches a secret manager backend.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath("../..")) # Adds the parent directory to the system path
from litellm.secret_managers.base_secret_manager import raise_if_unsafe_secret_name
@pytest.mark.parametrize(
"secret_name",
[
"..",
"../../../other-app/creds",
"litellm/../../secret",
"foo\nbar",
"foo\rbar",
"foo\x00bar",
"foo\x7fbar",
"foo\x85bar", # YAML NEL
"foobar", # YAML LINE SEPARATOR
"foobar", # YAML PARAGRAPH SEPARATOR
],
)
def test_raise_if_unsafe_secret_name_rejects_traversal_and_line_breaks(secret_name):
with pytest.raises(ValueError):
raise_if_unsafe_secret_name(secret_name)
@pytest.mark.parametrize(
"secret_name",
[
"plain-alias",
"my-key-123",
"prod/my-service-key",
"team/user@example.com",
"foo: bar",
"foo # bar",
"foo?evil=1",
"foo#bar",
"a" * 500,
],
)
def test_raise_if_unsafe_secret_name_allows_legitimate_aliases(secret_name):
raise_if_unsafe_secret_name(secret_name)