fix(mcp): restrict catalog environment writes
This commit is contained in:
committed by
Teknium
parent
1a5547c5c5
commit
08cf4fea5d
@@ -215,6 +215,14 @@ _ENV_VAR_NAME_DENYLIST: frozenset[str] = frozenset({
|
||||
# NOT a HERMES_* blanket: integration credentials (HERMES_GEMINI_*,
|
||||
# HERMES_LANGFUSE_*, HERMES_SPOTIFY_*, ...) ARE allowed.
|
||||
"HERMES_HOME", "HERMES_PROFILE", "HERMES_CONFIG", "HERMES_ENV",
|
||||
"HERMES_CONFIG_PATH", "HERMES_ENV_PATH",
|
||||
# Hermes security policy / approval-routing context. These remain available
|
||||
# through their dedicated CLI/config/session controls, but a generic
|
||||
# credential writer must not persist them for the next process startup.
|
||||
"HERMES_YOLO_MODE", "HERMES_ACCEPT_HOOKS", "HERMES_REDACT_SECRETS",
|
||||
"HERMES_INTERACTIVE", "HERMES_EXEC_ASK", "HERMES_GATEWAY_SESSION",
|
||||
"HERMES_CRON_SESSION", "HERMES_SINGLE_QUERY_SESSION",
|
||||
"HERMES_SESSION_KEY", "HERMES_SESSION_PLATFORM",
|
||||
})
|
||||
|
||||
|
||||
@@ -229,11 +237,23 @@ def _reject_denylisted_env_var(key: str) -> None:
|
||||
f"Environment variable {key!r} is on the writer denylist. "
|
||||
"Names that influence subprocess execution (LD_PRELOAD, "
|
||||
"PYTHONPATH, PATH, EDITOR, ...) or Hermes runtime location "
|
||||
"(HERMES_HOME, HERMES_PROFILE, ...) cannot be persisted via "
|
||||
"and security policy (HERMES_HOME, HERMES_YOLO_MODE, ...) "
|
||||
"cannot be persisted via "
|
||||
"the env writer. If you really need this, edit "
|
||||
"~/.hermes/.env directly."
|
||||
)
|
||||
|
||||
|
||||
def validate_env_var_name_for_write(key: str) -> None:
|
||||
"""Validate an environment name before a generic persistence write.
|
||||
|
||||
Exposed separately from :func:`save_env_value` so batch-style callers can
|
||||
validate their complete request before writing the first value.
|
||||
"""
|
||||
if not _ENV_VAR_NAME_RE.match(key):
|
||||
raise ValueError(f"Invalid environment variable name: {key!r}")
|
||||
_reject_denylisted_env_var(key)
|
||||
|
||||
_LAST_EXPANDED_CONFIG_BY_PATH: Dict[str, Any] = {}
|
||||
# (path, mtime_ns, size) -> cached expanded config dict.
|
||||
# load_config() returns a deepcopy of the cached value when the file
|
||||
@@ -4243,9 +4263,7 @@ def save_env_value(key: str, value: str):
|
||||
file=sys.stderr,
|
||||
)
|
||||
return
|
||||
if not _ENV_VAR_NAME_RE.match(key):
|
||||
raise ValueError(f"Invalid environment variable name: {key!r}")
|
||||
_reject_denylisted_env_var(key)
|
||||
validate_env_var_name_for_write(key)
|
||||
value = value.replace("\n", "").replace("\r", "")
|
||||
# API keys / tokens must be ASCII — strip non-ASCII with a warning.
|
||||
value = _check_non_ascii_credential(key, value)
|
||||
|
||||
@@ -503,8 +503,31 @@ async def install_mcp_catalog_entry(body: MCPCatalogInstall, profile: Optional[s
|
||||
if entry is None:
|
||||
raise HTTPException(status_code=404, detail=f"No catalog entry '{name}'")
|
||||
|
||||
# Persist any supplied env vars first (catalog entries declare which names
|
||||
# they need; we only write the ones the user provided).
|
||||
# Catalog credentials are a closed schema: configuring one MCP must not
|
||||
# become a generic write primitive for unrelated process environment.
|
||||
declared_env = {spec.name for spec in (entry.auth.env or [])}
|
||||
undeclared_env = sorted(set(body.env) - declared_env)
|
||||
if undeclared_env:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=(
|
||||
f"Catalog entry '{name}' does not declare environment "
|
||||
f"variable(s): {', '.join(undeclared_env)}"
|
||||
),
|
||||
)
|
||||
|
||||
# Validate the complete map before the first write. This preserves the
|
||||
# existing writer/install flow while ensuring a mixed valid+invalid request
|
||||
# cannot partially persist credentials.
|
||||
from hermes_cli.config import validate_env_var_name_for_write
|
||||
|
||||
try:
|
||||
for key in body.env:
|
||||
validate_env_var_name_for_write(key)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
# Persist any supplied, declared env vars first.
|
||||
effective_profile = body.profile or profile
|
||||
if body.env:
|
||||
def _write_env():
|
||||
|
||||
@@ -1050,7 +1050,7 @@ class TestDiscordChannelPromptsConfig:
|
||||
class TestEnvWriteDenylist:
|
||||
"""``save_env_value`` refuses to persist env-var names that
|
||||
influence how subprocesses execute — ``LD_PRELOAD``, ``PYTHONPATH``,
|
||||
``PATH``, ``EDITOR``, etc. — or any ``HERMES_*`` runtime flag.
|
||||
``PATH``, ``EDITOR``, etc. — or selected Hermes runtime/security controls.
|
||||
|
||||
The dashboard exposes ``PUT /api/env`` to any authed caller (and
|
||||
the session token lives in the SPA's HTML where any future plugin
|
||||
@@ -1080,15 +1080,39 @@ class TestEnvWriteDenylist:
|
||||
],
|
||||
)
|
||||
def test_hermes_integration_keys_still_writable(self, allowed_key):
|
||||
"""``HERMES_*`` overall is NOT blocked — only the four runtime
|
||||
location names (HOME/PROFILE/CONFIG/ENV) are. Integration
|
||||
credentials following the ``HERMES_*`` convention must keep
|
||||
working or we'd regress every provider setup wizard that
|
||||
currently writes one of these (auth.py, Spotify, Langfuse, …)."""
|
||||
"""``HERMES_*`` overall is NOT blocked.
|
||||
|
||||
Integration credentials following that convention must keep working
|
||||
or we'd regress provider setup flows (auth.py, Spotify, Langfuse, …).
|
||||
"""
|
||||
save_env_value(allowed_key, "test-value-123")
|
||||
env = load_env()
|
||||
assert env[allowed_key] == "test-value-123"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"protected_key",
|
||||
[
|
||||
"HERMES_CONFIG_PATH",
|
||||
"HERMES_ENV_PATH",
|
||||
"HERMES_YOLO_MODE",
|
||||
"HERMES_ACCEPT_HOOKS",
|
||||
"HERMES_REDACT_SECRETS",
|
||||
"HERMES_INTERACTIVE",
|
||||
"HERMES_EXEC_ASK",
|
||||
"HERMES_GATEWAY_SESSION",
|
||||
"HERMES_CRON_SESSION",
|
||||
"HERMES_SINGLE_QUERY_SESSION",
|
||||
"HERMES_SESSION_KEY",
|
||||
"HERMES_SESSION_PLATFORM",
|
||||
],
|
||||
)
|
||||
def test_hermes_security_control_keys_are_not_writable(self, protected_key):
|
||||
"""Generic writers must not persist runtime or approval controls."""
|
||||
with pytest.raises(ValueError, match="denylist"):
|
||||
save_env_value(protected_key, "1")
|
||||
|
||||
assert protected_key not in load_env()
|
||||
|
||||
|
||||
|
||||
def test_save_env_value_secure_inherits_denylist(self):
|
||||
|
||||
181
tests/hermes_cli/test_mcp_catalog_env_boundary.py
Normal file
181
tests/hermes_cli/test_mcp_catalog_env_boundary.py
Normal file
@@ -0,0 +1,181 @@
|
||||
"""Security boundary tests for dashboard MCP catalog credential writes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from hermes_cli.web_server import _SESSION_TOKEN, app
|
||||
|
||||
|
||||
HEADERS = {"X-Hermes-Session-Token": _SESSION_TOKEN}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def catalog_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _isolate_hermes_home):
|
||||
"""Install one synthetic API-key catalog entry in the isolated test home."""
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.config import invalidate_env_cache
|
||||
|
||||
catalog = tmp_path / "optional-mcps"
|
||||
entry_dir = catalog / "demo"
|
||||
entry_dir.mkdir(parents=True)
|
||||
(entry_dir / "manifest.yaml").write_text(
|
||||
yaml.safe_dump(
|
||||
{
|
||||
"manifest_version": 1,
|
||||
"name": "demo",
|
||||
"description": "Synthetic dashboard boundary fixture",
|
||||
"source": "https://example.test/demo",
|
||||
"transport": {
|
||||
"type": "stdio",
|
||||
"command": "demo-mcp",
|
||||
},
|
||||
"auth": {
|
||||
"type": "api_key",
|
||||
"env": [
|
||||
{
|
||||
"name": "DEMO_API_KEY",
|
||||
"prompt": "Demo API key",
|
||||
"secret": True,
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("HERMES_OPTIONAL_MCPS", str(catalog))
|
||||
invalidate_env_cache()
|
||||
return get_hermes_home()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
def test_catalog_rejects_undeclared_key_before_any_write_or_install(
|
||||
client: TestClient,
|
||||
catalog_env: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
import hermes_cli.mcp_catalog as mcp_catalog
|
||||
|
||||
installs: list[str] = []
|
||||
monkeypatch.setattr(
|
||||
mcp_catalog,
|
||||
"install_entry",
|
||||
lambda entry, enable=True: installs.append(entry.name),
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mcp/catalog/install",
|
||||
headers=HEADERS,
|
||||
json={
|
||||
"name": "demo",
|
||||
"env": {
|
||||
"DEMO_API_KEY": "valid-demo-value",
|
||||
"UNRELATED_SETTING": "must-not-land",
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
detail = response.json()["detail"]
|
||||
assert "UNRELATED_SETTING" in detail
|
||||
assert "valid-demo-value" not in detail
|
||||
assert "must-not-land" not in detail
|
||||
assert installs == []
|
||||
env_path = catalog_env / ".env"
|
||||
assert not env_path.exists() or env_path.read_text(encoding="utf-8") == ""
|
||||
|
||||
|
||||
def test_catalog_cannot_declare_reserved_control_key(
|
||||
client: TestClient,
|
||||
catalog_env: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
import hermes_cli.mcp_catalog as mcp_catalog
|
||||
|
||||
catalog_root = Path(os.environ["HERMES_OPTIONAL_MCPS"])
|
||||
manifest_path = catalog_root / "demo" / "manifest.yaml"
|
||||
manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
|
||||
manifest["auth"]["env"].append(
|
||||
{
|
||||
"name": "HERMES_YOLO_MODE",
|
||||
"prompt": "Unsafe control",
|
||||
"secret": False,
|
||||
}
|
||||
)
|
||||
manifest_path.write_text(yaml.safe_dump(manifest), encoding="utf-8")
|
||||
|
||||
installs: list[str] = []
|
||||
monkeypatch.setattr(
|
||||
mcp_catalog,
|
||||
"install_entry",
|
||||
lambda entry, enable=True: installs.append(entry.name),
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mcp/catalog/install",
|
||||
headers=HEADERS,
|
||||
json={"name": "demo", "env": {"HERMES_YOLO_MODE": "1"}},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "denylist" in response.json()["detail"]
|
||||
assert installs == []
|
||||
env_path = catalog_env / ".env"
|
||||
assert not env_path.exists() or "HERMES_YOLO_MODE" not in env_path.read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def test_catalog_accepts_declared_credential(
|
||||
client: TestClient,
|
||||
catalog_env: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
):
|
||||
import hermes_cli.mcp_catalog as mcp_catalog
|
||||
|
||||
installs: list[str] = []
|
||||
monkeypatch.setattr(
|
||||
mcp_catalog,
|
||||
"install_entry",
|
||||
lambda entry, enable=True: installs.append(entry.name),
|
||||
)
|
||||
|
||||
response = client.post(
|
||||
"/api/mcp/catalog/install",
|
||||
headers=HEADERS,
|
||||
json={"name": "demo", "env": {"DEMO_API_KEY": "valid-demo-value"}},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert installs == ["demo"]
|
||||
assert "DEMO_API_KEY=valid-demo-value" in (
|
||||
catalog_env / ".env"
|
||||
).read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_generic_env_endpoint_rejects_yolo_control_key(
|
||||
client: TestClient,
|
||||
catalog_env: Path,
|
||||
):
|
||||
response = client.put(
|
||||
"/api/env",
|
||||
headers=HEADERS,
|
||||
json={"key": "HERMES_YOLO_MODE", "value": "1"},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
env_path = catalog_env / ".env"
|
||||
assert not env_path.exists() or "HERMES_YOLO_MODE" not in env_path.read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
Reference in New Issue
Block a user