mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-09 16:08:31 +00:00
4d660b202a
* feat(skills): bind request-scoped secrets for in-context (autonomously invoked) skills Extends the #3861 binding point A (slash-activation only) to A+: the injection set is recomputed on every model call from two unioned sources — the run's most recent slash activation (persisted on the run context so the tool loop keeps the binding) and skills the model actually loaded in this thread (ThreadState.skill_context), re-validated against the live registry each call. Authorization stays three-gated regardless of activation style: skill enabled by the operator, values supplied per-request by the caller in context.secrets (never persisted server-side, never from the host env), names declared in the skill's required-secrets frontmatter. Because the set is replaced per call, eviction from skill_context or a caller that stops supplying a value revokes injection on the next call. New frontmatter field secrets-autonomous (default true) lets a skill restrict binding to explicit slash activation; malformed values fail closed to false. Binding changes are recorded as a middleware:skill_secrets journal event carrying names only. Design informed by a survey of peer systems (Claude Code, Codex CLI, opencode, pi, deepagents, hermes-agent, QwenPaw) and specs (agentskills.io, MCP 2025-11-25): the industry trust boundary is enable-time consent plus caller-scoped credentials, not per-invocation ceremony; no surveyed system scopes secrets to an activation turn. Part of #3914 * refactor(skills): centralize secret context keys, document intentional per-call reload Review follow-ups (no behavior change): move the two private binding keys (__slash_skill_secret_source, __skill_secrets_binding_audit) into secret_context.py and add them to REDACTED_CONTEXT_KEYS so the redaction allowlist stays a complete guard even though both keys hold names only. Document why _in_context_secret_sources reloads skills every call rather than caching: load_skills re-reads enabled state so an operator disabling a skill revokes its binding on the next model call — an mtime cache would miss enable/disable toggles and keep injecting after a disable. * fix(skills): match in-context secret bindings by path only, never by name Review finding (confused deputy): _in_context_secret_sources fell back to name matching when a skill_context path did not resolve. DeerFlow lets a custom skill shadow a same-named public/legacy one (load_skills de-dupes by name, custom wins), so a thread that read public/foo could bind the custom foo's declared secrets although the custom skill was never loaded in the thread. The recent user-isolation path changes make by-path misses (and thus the dangerous fallback) more likely. Drop the by-name fallback: match strictly by the exact container file path the model read; an unresolved path simply does not bind (the safe direction). Regression tests cover the shadowing case and a stale path. Part of #3914 * fix(skills): resolve secret-binding sources via registry; strip caller __-keys Security review (willem-bd, #3938): 1. Forged `__slash_skill_secret_source` bypassed the enabled/allowlist/ secrets-autonomous gates. runtime.context is caller-mergeable, and the slash source was trusted as authoritative (its stored requirements were injected directly). Now the slash source records only the activated skill's canonical container path, and BOTH the slash and in-context sources resolve the live registry skill by normalized path each call (_resolve_registry_skill) — binding only that real, enabled, allowlisted skill's own declared secrets. A forged path resolves to nothing. As defense in depth, build_run_config strips caller-supplied __-prefixed context keys at the gateway boundary. 2. Malformed caller requirements crashed the run (unguarded tuple unpack / DoS). The middleware no longer unpacks caller-provided requirement data at all — declarations come from the registry — so a malformed source fails closed instead of raising. 3. Path-normalization asymmetry silently disabled in-context binding on a trailing-slash container_path config. Both the registry keys and the lookup path are now posixpath.normpath'd. Regression tests: forged source rejected, forged-but-real path ignores caller requirements + allowlist, malformed source fails closed, trailing- slash config binds, gateway strips __-keys. Part of #3914 * docs(skills): correct _SLASH_SECRET_SOURCE_KEY comment and note fail-closed trade-off Post-review cleanup: the key now stores only the canonical container path (the comment still described the pre-fix skill-name+requirements shape), and document that a transient registry-load failure fails closed (drops the binding for that call) rather than trusting stale data. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
1088 lines
53 KiB
Python
1088 lines
53 KiB
Python
"""Tests for request-scoped secret injection into skills (issue #3861).
|
|
|
|
Covers the full feature surface:
|
|
- Slice 1: ``Sandbox.execute_command(command, env=...)`` per-call env injection
|
|
on both the local and AIO backends.
|
|
- Slice 2: ``SKILL.md`` ``requires-secrets`` frontmatter parsing.
|
|
- Slice 3: gateway carrier (``context.secrets``) and runtime-context passthrough.
|
|
- Slice 4: activation-turn binding + ``bash`` tool injection.
|
|
- Slice 5: the five leak surfaces (prompt / trace / checkpoint / audit / stdout).
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from langchain.agents.middleware.types import ModelRequest
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from deerflow.sandbox.local.local_sandbox import LocalSandbox
|
|
from deerflow.skills.types import SecretRequirement, Skill, SkillCategory
|
|
|
|
|
|
class TestLocalSandboxEnvInjection:
|
|
"""LocalSandbox.execute_command(env=...) injects per-call env into the subprocess."""
|
|
|
|
def test_injected_env_visible_to_command(self):
|
|
sandbox = LocalSandbox(id="local")
|
|
out = sandbox.execute_command(
|
|
"echo $DEERFLOW_TEST_SECRET",
|
|
env={"DEERFLOW_TEST_SECRET": "s3cret-value"},
|
|
)
|
|
assert "s3cret-value" in out
|
|
|
|
def test_env_none_keeps_inherited_environment(self, monkeypatch):
|
|
"""env=None preserves the legacy inherited-os.environ behaviour."""
|
|
monkeypatch.setenv("DEERFLOW_INHERITED_VAR", "inherited-value")
|
|
sandbox = LocalSandbox(id="local")
|
|
out = sandbox.execute_command("echo $DEERFLOW_INHERITED_VAR")
|
|
assert "inherited-value" in out
|
|
|
|
def test_injected_env_is_per_call_only(self):
|
|
"""Injected env must not leak into a subsequent call that does not pass it."""
|
|
sandbox = LocalSandbox(id="local")
|
|
sandbox.execute_command("true", env={"DEERFLOW_EPHEMERAL": "leaky"})
|
|
out = sandbox.execute_command("echo [$DEERFLOW_EPHEMERAL]")
|
|
assert "leaky" not in out
|
|
|
|
def test_platform_secret_scrubbed_from_inherited_env(self, monkeypatch):
|
|
"""A platform credential present in os.environ must NOT reach the sandbox
|
|
subprocess (the baseline-env leak surface). Without this, scoped injection
|
|
is security theatre — a skill script could simply read $OPENAI_API_KEY."""
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-platform-should-not-leak")
|
|
sandbox = LocalSandbox(id="local")
|
|
out = sandbox.execute_command("echo [$OPENAI_API_KEY]")
|
|
assert "sk-platform-should-not-leak" not in out
|
|
|
|
def test_benign_env_still_inherited_after_scrub(self, monkeypatch):
|
|
"""Scrubbing platform secrets must not strip harmless vars that skills rely on."""
|
|
monkeypatch.setenv("DEERFLOW_PLAIN_VAR", "harmless-value")
|
|
sandbox = LocalSandbox(id="local")
|
|
out = sandbox.execute_command("echo [$DEERFLOW_PLAIN_VAR]")
|
|
assert "harmless-value" in out
|
|
|
|
def test_injected_secret_survives_scrub(self, monkeypatch):
|
|
"""An explicitly injected secret must win even if its name matches a blocked
|
|
pattern — injection happens after scrubbing the inherited environment."""
|
|
sandbox = LocalSandbox(id="local")
|
|
out = sandbox.execute_command(
|
|
"echo [$INJECTED_API_KEY]",
|
|
env={"INJECTED_API_KEY": "scoped-value"},
|
|
)
|
|
assert "scoped-value" in out
|
|
|
|
|
|
class TestAioSandboxEnvInjection:
|
|
@pytest.fixture
|
|
def sandbox(self):
|
|
with patch("deerflow.community.aio_sandbox.aio_sandbox.AioSandboxClient"):
|
|
from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
|
|
|
|
return AioSandbox(id="test-sandbox", base_url="http://localhost:8080")
|
|
|
|
def test_env_none_uses_legacy_shell_path(self, sandbox):
|
|
"""No injected env → unchanged shell.exec_command path (backward compat)."""
|
|
sandbox._client.shell.exec_command = MagicMock(return_value=SimpleNamespace(data=SimpleNamespace(output="hello")))
|
|
sandbox._client.bash.exec = MagicMock()
|
|
out = sandbox.execute_command("echo hello")
|
|
sandbox._client.shell.exec_command.assert_called_once()
|
|
sandbox._client.bash.exec.assert_not_called()
|
|
assert "hello" in out
|
|
|
|
def test_injected_env_uses_bash_exec_with_env_dict(self, sandbox):
|
|
"""Injected env → bash.exec(env=...) carries the dict; secret stays out of the command string."""
|
|
sandbox._client.bash.exec = MagicMock(return_value=SimpleNamespace(data=SimpleNamespace(stdout="hello", stderr=None)))
|
|
sandbox._client.shell.exec_command = MagicMock()
|
|
out = sandbox.execute_command("echo $TOK", env={"TOK": "secret-v"})
|
|
sandbox._client.bash.exec.assert_called_once()
|
|
_, kwargs = sandbox._client.bash.exec.call_args
|
|
assert kwargs["env"] == {"TOK": "secret-v"}
|
|
# Secret must NOT be smuggled into the command string (audit / ps safety).
|
|
assert "secret-v" not in kwargs["command"]
|
|
sandbox._client.shell.exec_command.assert_not_called()
|
|
assert "hello" in out
|
|
|
|
def test_env_path_uses_hard_timeout_not_no_change_timeout(self, sandbox):
|
|
"""The env path routes through bash.exec which exposes no idle/no-change
|
|
timeout; it must use the dedicated wall-clock ``_DEFAULT_HARD_TIMEOUT``,
|
|
not the legacy idle constant (same numeric value today, but distinct
|
|
semantics so a future change to one does not silently alter the other)."""
|
|
from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox
|
|
|
|
sandbox._client.bash.exec = MagicMock(return_value=SimpleNamespace(data=SimpleNamespace(stdout="ok", stderr=None)))
|
|
sandbox.execute_command("echo hi", env={"X": "1"})
|
|
_, kwargs = sandbox._client.bash.exec.call_args
|
|
assert kwargs["hard_timeout"] == AioSandbox._DEFAULT_HARD_TIMEOUT
|
|
assert AioSandbox._DEFAULT_HARD_TIMEOUT != AioSandbox._DEFAULT_NO_CHANGE_TIMEOUT or (
|
|
# Same numeric value is fine today; the contract is that they are
|
|
# named independently so the two call sites evolve independently.
|
|
AioSandbox._DEFAULT_HARD_TIMEOUT == AioSandbox._DEFAULT_NO_CHANGE_TIMEOUT
|
|
)
|
|
|
|
def test_env_path_retries_on_error_observation_signature(self, sandbox):
|
|
"""The env path shares the legacy persistent-shell recovery contract: if
|
|
the (unlikely, fresh-session) corruption marker appears, the call is
|
|
retried rather than returned verbatim."""
|
|
from deerflow.community.aio_sandbox.aio_sandbox import _ERROR_OBSERVATION_SIGNATURE
|
|
|
|
corrupted = SimpleNamespace(data=SimpleNamespace(stdout=_ERROR_OBSERVATION_SIGNATURE, stderr=None))
|
|
clean = SimpleNamespace(data=SimpleNamespace(stdout="recovered", stderr=None))
|
|
sandbox._client.bash.exec = MagicMock(side_effect=[corrupted, clean])
|
|
out = sandbox.execute_command("script", env={"TOK": "v"})
|
|
assert sandbox._client.bash.exec.call_count == 2
|
|
assert "recovered" in out
|
|
assert _ERROR_OBSERVATION_SIGNATURE not in out
|
|
|
|
|
|
class TestEnvPolicy:
|
|
"""Platform-secret scrubbing policy for sandbox subprocesses (delta 1)."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"LANGFUSE_SECRET_KEY",
|
|
"GITHUB_TOKEN",
|
|
"AWS_SECRET_ACCESS_KEY",
|
|
"DB_PASSWORD",
|
|
"MY_SERVICE_CREDENTIAL",
|
|
"api_key",
|
|
"Some_Token_Here",
|
|
# Connection-string credentials (no KEY/SECRET/TOKEN substring) — these
|
|
# routinely embed a password, e.g. postgresql://user:pw@host/db.
|
|
"DATABASE_URL",
|
|
"REDIS_URL",
|
|
"MONGODB_URI",
|
|
"AMQP_URL",
|
|
"SENTRY_DSN",
|
|
"POSTGRES_DSN",
|
|
"CONN_STR",
|
|
"GH_PAT",
|
|
],
|
|
)
|
|
def test_secret_like_names_are_blocked(self, name):
|
|
from deerflow.sandbox.env_policy import is_blocked_env_name
|
|
|
|
assert is_blocked_env_name(name) is True
|
|
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"PATH",
|
|
"HOME",
|
|
"SHELL",
|
|
"USER",
|
|
"LANG",
|
|
"LC_ALL",
|
|
"PWD",
|
|
"TMPDIR",
|
|
"VIRTUAL_ENV",
|
|
"PYTHONPATH",
|
|
"DEERFLOW_PLAIN_VAR",
|
|
# Not a blanket *URL* block: a benign service URL a skill may legitimately
|
|
# read is not treated as a credential.
|
|
"NEXT_PUBLIC_BASE_URL",
|
|
"SERVICE_ENDPOINT",
|
|
],
|
|
)
|
|
def test_benign_names_are_allowed(self, name):
|
|
from deerflow.sandbox.env_policy import is_blocked_env_name
|
|
|
|
assert is_blocked_env_name(name) is False
|
|
|
|
def test_build_sandbox_env_scrubs_inherited_and_layers_injected(self, monkeypatch):
|
|
from deerflow.sandbox.env_policy import build_sandbox_env
|
|
|
|
monkeypatch.setenv("OPENAI_API_KEY", "platform-key-should-vanish")
|
|
monkeypatch.setenv("HARMLESS_PLAIN", "ok")
|
|
env = build_sandbox_env(injected={"SCOPED_TOKEN": "v"})
|
|
assert "OPENAI_API_KEY" not in env # platform secret scrubbed
|
|
assert env.get("HARMLESS_PLAIN") == "ok" # benign preserved
|
|
assert env.get("SCOPED_TOKEN") == "v" # injected layered on top
|
|
assert env.get("PATH") # core var preserved
|
|
|
|
def test_build_sandbox_env_none_injection_still_scrubs(self, monkeypatch):
|
|
from deerflow.sandbox.env_policy import build_sandbox_env
|
|
|
|
monkeypatch.setenv("ANTHROPIC_API_KEY", "leak")
|
|
env = build_sandbox_env()
|
|
assert "ANTHROPIC_API_KEY" not in env
|
|
|
|
|
|
class TestRequiredSecretsParsing:
|
|
"""SKILL.md ``required-secrets`` frontmatter parsing (Slice 2)."""
|
|
|
|
def _write_skill(self, tmp_path, frontmatter_body: str):
|
|
skill_dir = tmp_path / "erp-report"
|
|
skill_dir.mkdir()
|
|
skill_file = skill_dir / "SKILL.md"
|
|
skill_file.write_text(f"---\n{frontmatter_body}\n---\n# body\n", encoding="utf-8")
|
|
return skill_file
|
|
|
|
def test_absent_field_defaults_to_empty(self, tmp_path):
|
|
from deerflow.skills.parser import parse_skill_file
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
skill_file = self._write_skill(tmp_path, "name: erp-report\ndescription: Pull an ERP report")
|
|
skill = parse_skill_file(skill_file, SkillCategory.CUSTOM)
|
|
assert skill is not None
|
|
assert skill.required_secrets == ()
|
|
|
|
def test_string_list_form(self, tmp_path):
|
|
from deerflow.skills.parser import parse_skill_file
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
skill_file = self._write_skill(
|
|
tmp_path,
|
|
"name: erp-report\ndescription: d\nrequired-secrets:\n - ERP_TOKEN\n - OTHER_TOKEN",
|
|
)
|
|
skill = parse_skill_file(skill_file, SkillCategory.CUSTOM)
|
|
assert [s.name for s in skill.required_secrets] == ["ERP_TOKEN", "OTHER_TOKEN"]
|
|
assert all(s.optional is False for s in skill.required_secrets)
|
|
|
|
def test_object_list_with_optional(self, tmp_path):
|
|
from deerflow.skills.parser import parse_skill_file
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
skill_file = self._write_skill(
|
|
tmp_path,
|
|
"name: erp-report\ndescription: d\nrequired-secrets:\n - name: ERP_TOKEN\n optional: true\n - name: REQUIRED_ONE",
|
|
)
|
|
skill = parse_skill_file(skill_file, SkillCategory.CUSTOM)
|
|
by_name = {s.name: s for s in skill.required_secrets}
|
|
assert by_name["ERP_TOKEN"].optional is True
|
|
assert by_name["REQUIRED_ONE"].optional is False
|
|
|
|
def test_invalid_env_name_entry_is_dropped(self, tmp_path):
|
|
from deerflow.skills.parser import parse_skill_file
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
skill_file = self._write_skill(
|
|
tmp_path,
|
|
'name: erp-report\ndescription: d\nrequired-secrets:\n - "bad name!"\n - GOOD_TOKEN',
|
|
)
|
|
skill = parse_skill_file(skill_file, SkillCategory.CUSTOM)
|
|
# The malformed entry is dropped; the valid one survives — one bad
|
|
# declaration must not nuke the whole skill.
|
|
assert [s.name for s in skill.required_secrets] == ["GOOD_TOKEN"]
|
|
|
|
|
|
class TestSecretCarrier:
|
|
"""Request-scoped secret carrier: context.secrets → runtime.context (Slice 3)."""
|
|
|
|
def test_build_run_config_keeps_secrets_in_context_not_configurable(self):
|
|
from app.gateway.services import build_run_config
|
|
|
|
config = build_run_config("thread-1", {"context": {"secrets": {"ERP_TOKEN": "v"}}}, None)
|
|
assert config["context"]["secrets"] == {"ERP_TOKEN": "v"}
|
|
# Secrets must never be mirrored into configurable (which legacy readers
|
|
# and some trace backends surface).
|
|
assert "secrets" not in config.get("configurable", {})
|
|
|
|
def test_runtime_context_carries_secrets(self):
|
|
from deerflow.runtime.runs.worker import _build_runtime_context
|
|
|
|
ctx = _build_runtime_context("t", "r", {"secrets": {"ERP_TOKEN": "v"}})
|
|
assert ctx["secrets"] == {"ERP_TOKEN": "v"}
|
|
|
|
def test_build_run_config_strips_caller_dunder_context_keys(self):
|
|
"""Security (#3938): the harness writes private ``__``-prefixed keys into
|
|
``runtime.context`` (binding sources, active-secret set, run journal). A
|
|
caller must not be able to seed them via ``config.context`` and forge
|
|
internal state — they are stripped at the gateway boundary."""
|
|
from app.gateway.services import build_run_config
|
|
|
|
config = build_run_config(
|
|
"thread-1",
|
|
{"context": {"secrets": {"ERP_TOKEN": "v"}, "__slash_skill_secret_source": {"path": "x"}, "__active_skill_secrets": {"ADMIN": "stolen"}}},
|
|
None,
|
|
)
|
|
assert config["context"]["secrets"] == {"ERP_TOKEN": "v"}
|
|
assert "__slash_skill_secret_source" not in config["context"]
|
|
assert "__active_skill_secrets" not in config["context"]
|
|
|
|
def test_extract_request_secrets_filters_non_string_pairs(self):
|
|
from deerflow.runtime.secret_context import extract_request_secrets
|
|
|
|
assert extract_request_secrets({"secrets": {"A": "x", "B": 123, 4: "y"}}) == {"A": "x"}
|
|
|
|
def test_extract_request_secrets_missing_or_malformed(self):
|
|
from deerflow.runtime.secret_context import extract_request_secrets
|
|
|
|
assert extract_request_secrets({}) == {}
|
|
assert extract_request_secrets({"secrets": "not-a-dict"}) == {}
|
|
assert extract_request_secrets(None) == {}
|
|
|
|
|
|
def _make_secret_skill(tmp_path: Path, name: str, required_secrets, *, enabled: bool = True, secrets_autonomous: bool = True):
|
|
skill_dir = tmp_path / name
|
|
skill_dir.mkdir()
|
|
skill_file = skill_dir / "SKILL.md"
|
|
skill_file.write_text(f"# {name}\n", encoding="utf-8")
|
|
return Skill(
|
|
name=name,
|
|
description=f"Description for {name}",
|
|
license="MIT",
|
|
skill_dir=skill_dir,
|
|
skill_file=skill_file,
|
|
relative_path=Path(name),
|
|
category=SkillCategory.CUSTOM,
|
|
enabled=enabled,
|
|
required_secrets=tuple(required_secrets),
|
|
secrets_autonomous=secrets_autonomous,
|
|
)
|
|
|
|
|
|
class TestActivationBindsSecrets:
|
|
"""Binding point A: activation turn resolves declared secrets into the per-run injection set."""
|
|
|
|
def _activate(self, tmp_path, monkeypatch, skill, context):
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: [skill],
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
middleware = SkillActivationMiddleware()
|
|
request = ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content=f"/{skill.name} do it", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
)
|
|
middleware.wrap_model_call(request, lambda r: AIMessage(content="ok"))
|
|
|
|
def test_declared_secret_resolved_into_active_set(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123", "UNUSED": "x"}}
|
|
self._activate(tmp_path, monkeypatch, skill, context)
|
|
# Only the declared secret is injected — not the whole secrets bag.
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
def test_skill_without_declaration_gets_no_injection(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "plain", [])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
self._activate(tmp_path, monkeypatch, skill, context)
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_missing_required_secret_not_injected(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {}} # caller provided none
|
|
self._activate(tmp_path, monkeypatch, skill, context)
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_caller_secret_wins_over_host_value_of_same_name(self, tmp_path, monkeypatch):
|
|
"""A skill may declare a name that also exists in the host env (e.g. a
|
|
per-user key overriding a shared platform key — the #3861 use case). The
|
|
skill receives the CALLER's value (from context.secrets), never the host's:
|
|
the inherited host value is scrubbed and the caller's value is injected on
|
|
top. There is therefore no host-credential harvest to guard against."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
from deerflow.sandbox.env_policy import build_sandbox_env
|
|
|
|
monkeypatch.setenv("MEMOS_API_KEY", "host-shared-key-MUST-NOT-LEAK")
|
|
skill = _make_secret_skill(tmp_path, "memos", [SecretRequirement("MEMOS_API_KEY")])
|
|
context = {"secrets": {"MEMOS_API_KEY": "caller-per-user-key"}}
|
|
self._activate(tmp_path, monkeypatch, skill, context)
|
|
|
|
injected = read_active_secrets(context)
|
|
assert injected == {"MEMOS_API_KEY": "caller-per-user-key"} # caller's value injected
|
|
|
|
# The subprocess env gets the caller's value; the host's value is scrubbed.
|
|
env = build_sandbox_env(injected)
|
|
assert env["MEMOS_API_KEY"] == "caller-per-user-key"
|
|
assert "host-shared-key-MUST-NOT-LEAK" not in str(env.values())
|
|
|
|
def test_undeclared_host_secret_is_scrubbed_not_harvested(self, tmp_path, monkeypatch):
|
|
"""If a skill does NOT declare a host credential, the inherited value is
|
|
scrubbed — a skill can never read a platform credential it wasn't given."""
|
|
from deerflow.sandbox.env_policy import build_sandbox_env
|
|
|
|
monkeypatch.setenv("OPENAI_API_KEY", "host-key-do-not-harvest")
|
|
env = build_sandbox_env(None)
|
|
assert "OPENAI_API_KEY" not in env
|
|
|
|
def test_activation_fires_after_input_sanitization_wrapping(self, tmp_path, monkeypatch):
|
|
"""Integration: in the real chain InputSanitizationMiddleware wraps the user
|
|
message in ``--- BEGIN USER INPUT ---`` markers before SkillActivationMiddleware
|
|
sees it. Slash activation (and therefore secret resolution) must still fire — it
|
|
relies on the original content being recoverable. Regression for the gateway
|
|
path where no upload preserved it."""
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.input_sanitization_middleware import InputSanitizationMiddleware
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: [skill],
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
|
|
context = {"secrets": {"ERP_TOKEN": "tok-xyz"}}
|
|
request = ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/erp-report pull it", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
)
|
|
# The sanitizer loads enabled skills during wrap, so keep a stub app config
|
|
# in place for the whole composed call.
|
|
set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
|
|
try:
|
|
sanitizer = InputSanitizationMiddleware()
|
|
skill_mw = SkillActivationMiddleware()
|
|
|
|
# Compose in real order: sanitizer (outer) -> skill activation (inner) -> model.
|
|
def skill_layer(req):
|
|
return skill_mw.wrap_model_call(req, lambda r: AIMessage(content="ok"))
|
|
|
|
sanitizer.wrap_model_call(request, skill_layer)
|
|
finally:
|
|
reset_app_config()
|
|
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-xyz"}
|
|
|
|
def test_prior_activation_secrets_cleared_when_next_skill_declares_none(self, tmp_path, monkeypatch):
|
|
"""A later skill in the same run never inherits an earlier skill's secrets.
|
|
Turn 1 activates /skill-a (declares A_TOKEN, caller supplies it) → injected.
|
|
Turn 2 activates /skill-b (declares nothing) → A_TOKEN must be cleared so
|
|
bash in skill-b's turn cannot receive a value it never declared."""
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill_a = _make_secret_skill(tmp_path, "skill-a", [SecretRequirement("A_TOKEN")])
|
|
skill_b = _make_secret_skill(tmp_path, "skill-b", [])
|
|
|
|
def _storage(skills):
|
|
return SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: skills,
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
|
|
context = {"secrets": {"A_TOKEN": "v-a"}}
|
|
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: _storage([skill_a]))
|
|
SkillActivationMiddleware().wrap_model_call(
|
|
ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/skill-a go", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
),
|
|
lambda r: AIMessage(content="ok"),
|
|
)
|
|
assert read_active_secrets(context) == {"A_TOKEN": "v-a"}
|
|
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: _storage([skill_b]))
|
|
SkillActivationMiddleware().wrap_model_call(
|
|
ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/skill-b go", id="m2")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
),
|
|
lambda r: AIMessage(content="ok"),
|
|
)
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_prior_activation_secrets_cleared_when_caller_omits_required(self, tmp_path, monkeypatch):
|
|
"""Even when the next skill DOES declare a required secret, if the caller
|
|
omits it the prior skill's value must not linger — the injection set ends
|
|
up empty, not stale."""
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp", [SecretRequirement("ERP_TOKEN")])
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: [skill],
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
|
|
# Turn 1: caller supplies ERP_TOKEN → injected.
|
|
context = {"secrets": {"ERP_TOKEN": "tok-1"}}
|
|
mw_inst = SkillActivationMiddleware()
|
|
mw_inst.wrap_model_call(
|
|
ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/erp go", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
),
|
|
lambda r: AIMessage(content="ok"),
|
|
)
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-1"}
|
|
|
|
# Turn 2: caller omits ERP_TOKEN → prior value cleared, set empty (not stale).
|
|
context2 = {"secrets": {}}
|
|
mw_inst.wrap_model_call(
|
|
ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/erp again", id="m2")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context2),
|
|
),
|
|
lambda r: AIMessage(content="ok"),
|
|
)
|
|
assert read_active_secrets(context2) == {}
|
|
|
|
|
|
def _skill_context_entry(skill) -> dict:
|
|
return {
|
|
"name": skill.name,
|
|
"path": f"/mnt/skills/{skill.category}/{skill.name}/SKILL.md",
|
|
"description": skill.description,
|
|
"loaded_at": 0,
|
|
}
|
|
|
|
|
|
class TestInContextBindsSecrets:
|
|
"""Binding point A+ (issue #3914 gap 1): a skill the model loaded earlier in
|
|
the thread (tracked by ``ThreadState.skill_context``) keeps receiving its
|
|
declared secrets on later turns — without a fresh ``/slash`` — as long as
|
|
the caller supplies the values on the current request. Authorization stays
|
|
three-gated regardless of activation style: skill enabled by the operator,
|
|
values supplied per-request by the caller, names declared in frontmatter.
|
|
"""
|
|
|
|
def _run_call(self, tmp_path, monkeypatch, skills, *, context, skill_context=None, message="continue the report", available_skills=None, middleware=None, container_root="/mnt/skills"):
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: list(skills),
|
|
get_container_root=lambda: container_root,
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
mw_inst = middleware or SkillActivationMiddleware(available_skills=available_skills)
|
|
mw_inst.wrap_model_call(
|
|
ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content=message, id="m1")],
|
|
state={"messages": [], "skill_context": skill_context or []},
|
|
runtime=SimpleNamespace(context=context),
|
|
),
|
|
lambda r: AIMessage(content="ok"),
|
|
)
|
|
return mw_inst
|
|
|
|
def test_in_context_skill_binds_secrets_without_slash(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123", "UNRELATED": "x"}}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
def test_binding_clears_when_skill_evicted_from_context(self, tmp_path, monkeypatch):
|
|
"""Long-lived binding follows skill_context membership exactly: once the
|
|
entry is evicted (capacity) the injection disappears on the next call."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[], middleware=mw_inst)
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_disabled_skill_in_context_not_bound(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")], enabled=False)
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_skill_outside_agent_allowlist_not_bound(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
self._run_call(
|
|
tmp_path,
|
|
monkeypatch,
|
|
[skill],
|
|
context=context,
|
|
skill_context=[_skill_context_entry(skill)],
|
|
available_skills={"some-other-skill"},
|
|
)
|
|
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_secrets_autonomous_false_blocks_in_context_but_not_slash(self, tmp_path, monkeypatch):
|
|
"""The per-skill opt-out keeps explicit-activation ceremony available for
|
|
high-sensitivity skills: in-context binding is refused, slash still works."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")], secrets_autonomous=False)
|
|
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
assert read_active_secrets(context) == {}
|
|
|
|
slash_context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=slash_context, message="/erp-report go")
|
|
assert read_active_secrets(slash_context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
def test_slash_and_in_context_sources_merge(self, tmp_path, monkeypatch):
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
loaded = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
slashed = _make_secret_skill(tmp_path, "crm-sync", [SecretRequirement("CRM_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-erp", "CRM_TOKEN": "tok-crm"}}
|
|
self._run_call(
|
|
tmp_path,
|
|
monkeypatch,
|
|
[loaded, slashed],
|
|
context=context,
|
|
skill_context=[_skill_context_entry(loaded)],
|
|
message="/crm-sync push the numbers",
|
|
)
|
|
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-erp", "CRM_TOKEN": "tok-crm"}
|
|
|
|
def test_forged_slash_source_cannot_bypass_gates(self, tmp_path, monkeypatch):
|
|
"""Security (#3938): `runtime.context` is caller-mergeable, so a client can
|
|
forge `__slash_skill_secret_source`. The slash source is re-validated
|
|
against the live registry (enabled + allowlist), so a forged source naming
|
|
a non-existent skill binds nothing — no gate bypass."""
|
|
from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
|
|
|
|
context = {
|
|
"secrets": {"ADMIN_TOKEN": "stolen"},
|
|
_SLASH_SECRET_SOURCE_KEY: {"path": "/mnt/skills/custom/attacker/SKILL.md", "skill_name": "attacker", "requirements": [["ADMIN_TOKEN", False]]},
|
|
}
|
|
self._run_call(tmp_path, monkeypatch, [], context=context, message="no slash here")
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_forged_slash_source_ignores_caller_requirements_and_allowlist(self, tmp_path, monkeypatch):
|
|
"""Even if a forged path resolves to a real skill, the caller's forged
|
|
requirements are ignored (only the registry skill's own declared secrets
|
|
bind) and the allowlist still applies."""
|
|
from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {
|
|
"secrets": {"ADMIN_TOKEN": "stolen", "ERP_TOKEN": "ok"},
|
|
_SLASH_SECRET_SOURCE_KEY: {"path": "/mnt/skills/custom/erp-report/SKILL.md", "requirements": [["ADMIN_TOKEN", False]]},
|
|
}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, available_skills={"other"})
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_malformed_slash_source_does_not_crash(self, tmp_path, monkeypatch):
|
|
"""Robustness (#3938): a forged malformed slash source must fail closed
|
|
(bind nothing), never raise and 500 the run."""
|
|
from deerflow.runtime.secret_context import _SLASH_SECRET_SOURCE_KEY, read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
for bad in ({"requirements": [["X"]]}, {"requirements": "abc"}, {"path": 123}, "not-a-dict", {"path": ["a"]}, {}):
|
|
context = {"secrets": {"ERP_TOKEN": "v"}, _SLASH_SECRET_SOURCE_KEY: bad}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, message="x")
|
|
assert read_active_secrets(context) == {}, f"bad={bad!r}"
|
|
|
|
def test_trailing_slash_container_root_still_binds(self, tmp_path, monkeypatch):
|
|
"""Latent bug (#3938): a non-canonical container_path (trailing slash) must
|
|
not silently disable in-context binding — paths are normalized both sides."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
entry = {"name": "erp-report", "path": "/mnt/skills/custom/erp-report/SKILL.md", "description": "d", "loaded_at": 0}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[entry], container_root="/mnt/skills/")
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
def test_shadowing_name_does_not_bind_unread_skill(self, tmp_path, monkeypatch):
|
|
"""Confused-deputy guard: a custom skill may shadow a same-named public
|
|
one (load_skills de-dupes by name, custom wins). A thread that read the
|
|
PUBLIC foo (no declared secrets) must NOT bind the CUSTOM foo's declared
|
|
secret — matching is by exact container path, never by name."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
# Registry exposes only the custom foo (name de-dup, custom wins); the
|
|
# model read the public foo, whose path differs.
|
|
custom_foo = _make_secret_skill(tmp_path, "foo", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
entry = {
|
|
"name": "foo",
|
|
"path": "/mnt/skills/public/foo/SKILL.md", # the PUBLIC one the model actually read
|
|
"description": "d",
|
|
"loaded_at": 0,
|
|
}
|
|
self._run_call(tmp_path, monkeypatch, [custom_foo], context=context, skill_context=[entry])
|
|
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_stale_path_does_not_fall_back_to_name(self, tmp_path, monkeypatch):
|
|
"""A skill_context path that no longer resolves must not degrade to a
|
|
name match — it simply does not bind."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
entry = {
|
|
"name": "erp-report",
|
|
"path": "/mnt/skills/custom/erp-report-OLD-PATH/SKILL.md",
|
|
"description": "d",
|
|
"loaded_at": 0,
|
|
}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[entry])
|
|
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_no_caller_secrets_means_no_binding(self, tmp_path, monkeypatch):
|
|
"""The supply gate: without caller-provided values on THIS request there
|
|
is nothing to inject, no matter what is in skill_context."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {}}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
|
|
assert read_active_secrets(context) == {}
|
|
|
|
def test_binding_change_recorded_in_audit_journal_names_only(self, tmp_path, monkeypatch):
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
journal = MagicMock()
|
|
context = {"secrets": {"ERP_TOKEN": "tok-secret-value"}, "__run_journal": journal}
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
|
|
bind_calls = [call for call in journal.record_middleware.call_args_list if call.kwargs.get("action") == "bind_secrets"]
|
|
assert len(bind_calls) == 1
|
|
changes = bind_calls[0].kwargs["changes"]
|
|
assert changes["skills"] == ["erp-report"]
|
|
assert changes["secrets"] == ["ERP_TOKEN"]
|
|
# Values must never reach the audit journal.
|
|
assert "tok-secret-value" not in str(bind_calls[0])
|
|
|
|
def test_slash_binding_persists_across_model_calls_in_same_run(self, tmp_path, monkeypatch):
|
|
"""#3861 semantics preserved under per-call recompute: after the single
|
|
activation call, the tool loop issues more model calls without a fresh
|
|
slash — the binding must survive on the shared run context."""
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
context = {"secrets": {"ERP_TOKEN": "tok-123"}}
|
|
mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, message="/erp-report go")
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
# Later model call in the SAME run (same context object): no slash in the
|
|
# latest message, skill never entered skill_context (slash injects the
|
|
# body directly, no read_file happens).
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, message="tool loop continues", middleware=mw_inst)
|
|
assert read_active_secrets(context) == {"ERP_TOKEN": "tok-123"}
|
|
|
|
def test_unchanged_binding_not_re_recorded(self, tmp_path, monkeypatch):
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
journal = MagicMock()
|
|
context = {"secrets": {"ERP_TOKEN": "tok-1"}, "__run_journal": journal}
|
|
mw_inst = self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)])
|
|
self._run_call(tmp_path, monkeypatch, [skill], context=context, skill_context=[_skill_context_entry(skill)], middleware=mw_inst)
|
|
|
|
bind_calls = [call for call in journal.record_middleware.call_args_list if call.kwargs.get("action") == "bind_secrets"]
|
|
assert len(bind_calls) == 1
|
|
|
|
|
|
class TestSecretsAutonomousParsing:
|
|
"""Frontmatter ``secrets-autonomous`` controls in-context (autonomous) binding."""
|
|
|
|
def _parse(self, tmp_path, frontmatter_extra: str):
|
|
from deerflow.skills.parser import parse_skill_file
|
|
from deerflow.skills.types import SkillCategory
|
|
|
|
skill_dir = tmp_path / "erp-report"
|
|
skill_dir.mkdir()
|
|
skill_file = skill_dir / "SKILL.md"
|
|
skill_file.write_text(
|
|
f"""---
|
|
name: erp-report
|
|
description: Pull an ERP report.
|
|
required-secrets:
|
|
- ERP_TOKEN
|
|
{frontmatter_extra}---
|
|
|
|
Body.
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return parse_skill_file(skill_file, SkillCategory.CUSTOM)
|
|
|
|
def test_defaults_to_true(self, tmp_path):
|
|
skill = self._parse(tmp_path, "")
|
|
assert skill is not None
|
|
assert skill.secrets_autonomous is True
|
|
|
|
def test_explicit_false(self, tmp_path):
|
|
skill = self._parse(tmp_path, "secrets-autonomous: false\n")
|
|
assert skill is not None
|
|
assert skill.secrets_autonomous is False
|
|
|
|
def test_malformed_value_fails_closed(self, tmp_path, caplog):
|
|
"""A non-boolean value disables autonomous binding (the safer direction)
|
|
instead of silently enabling it."""
|
|
skill = self._parse(tmp_path, 'secrets-autonomous: "yes please"\n')
|
|
assert skill is not None
|
|
assert skill.secrets_autonomous is False
|
|
|
|
|
|
class TestBashToolInjectsActiveSecrets:
|
|
"""The bash tool forwards the per-run injection set to execute_command(env=...)."""
|
|
|
|
def _run_bash(self, context):
|
|
from deerflow.sandbox import tools as tools_mod
|
|
|
|
captured = {}
|
|
|
|
class FakeSandbox:
|
|
def execute_command(self, command, env=None, timeout=None):
|
|
captured["env"] = env
|
|
captured["timeout"] = timeout
|
|
return "done"
|
|
|
|
runtime = SimpleNamespace(context=context, state={"sandbox": {"sandbox_id": "aio:1"}})
|
|
with (
|
|
patch.object(tools_mod, "ensure_sandbox_initialized", return_value=FakeSandbox()),
|
|
patch.object(tools_mod, "is_local_sandbox", return_value=False),
|
|
patch.object(tools_mod, "ensure_thread_directories_exist", return_value=None),
|
|
):
|
|
out = tools_mod.bash_tool.func(runtime, "run skill", "echo hi")
|
|
return out, captured
|
|
|
|
def test_active_secret_forwarded_as_env(self):
|
|
out, captured = self._run_bash({"__active_skill_secrets": {"ERP_TOKEN": "tok-123"}})
|
|
assert captured["env"] == {"ERP_TOKEN": "tok-123"}
|
|
assert "done" in out
|
|
|
|
def test_no_active_secret_forwards_no_env(self):
|
|
out, captured = self._run_bash({})
|
|
assert captured["env"] in (None, {})
|
|
|
|
def test_local_bash_forwards_env_and_timeout(self, monkeypatch):
|
|
from deerflow.sandbox import tools as tools_mod
|
|
|
|
captured = {}
|
|
|
|
class FakeSandbox:
|
|
def execute_command(self, command, env=None, timeout=None):
|
|
captured["command"] = command
|
|
captured["env"] = env
|
|
captured["timeout"] = timeout
|
|
return "done"
|
|
|
|
runtime = SimpleNamespace(
|
|
context={"__active_skill_secrets": {"ERP_TOKEN": "tok-456"}},
|
|
state={"sandbox": {"sandbox_id": "local:1"}},
|
|
)
|
|
thread_data = {"workspace_path": "/tmp/ws", "cwd": "/mnt/user-data/workspace"}
|
|
fake_cfg = SimpleNamespace(sandbox=SimpleNamespace(bash_output_max_chars=321, bash_command_timeout=42))
|
|
with (
|
|
patch.object(tools_mod, "ensure_sandbox_initialized", return_value=FakeSandbox()),
|
|
patch.object(tools_mod, "is_local_sandbox", return_value=True),
|
|
patch.object(tools_mod, "is_host_bash_allowed", return_value=True),
|
|
patch.object(tools_mod, "ensure_thread_directories_exist", return_value=None),
|
|
patch.object(tools_mod, "get_thread_data", return_value=thread_data),
|
|
patch.object(tools_mod, "validate_local_bash_command_paths", return_value=None),
|
|
patch.object(tools_mod, "replace_virtual_paths_in_command", side_effect=lambda command, td: command),
|
|
patch.object(tools_mod, "_apply_cwd_prefix", side_effect=lambda command, td: command),
|
|
patch("deerflow.config.app_config.get_app_config", return_value=fake_cfg),
|
|
):
|
|
out = tools_mod.bash_tool.func(runtime, "run local skill", "echo hi")
|
|
|
|
assert out == "done"
|
|
assert captured["command"] == "echo hi"
|
|
assert captured["env"] == {"ERP_TOKEN": "tok-456"}
|
|
assert captured["timeout"] == 42
|
|
|
|
|
|
_SECRET = "sk-erp-9f3c-DO-NOT-LEAK"
|
|
|
|
|
|
class TestLeakSurfaces:
|
|
"""Assert the secret value is absent from all five leak surfaces (#3861)."""
|
|
|
|
def _activate_with_secret(self, tmp_path, monkeypatch):
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: [skill],
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
|
|
journal_records: list[dict] = []
|
|
journal = SimpleNamespace(record_middleware=lambda *a, **k: journal_records.append({"a": a, "k": k}))
|
|
context = {"secrets": {"ERP_TOKEN": _SECRET}, "__run_journal": journal}
|
|
request = ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/erp-report pull report", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
)
|
|
captured = {}
|
|
SkillActivationMiddleware().wrap_model_call(request, lambda r: captured.setdefault("messages", r.messages) or AIMessage(content="ok"))
|
|
return context, captured["messages"], journal_records
|
|
|
|
def test_prompt_surface_has_no_secret(self, tmp_path, monkeypatch):
|
|
# The injected activation message (the only thing added to the prompt /
|
|
# checkpointed messages) must not contain the secret value.
|
|
_, messages, _ = self._activate_with_secret(tmp_path, monkeypatch)
|
|
for m in messages:
|
|
assert _SECRET not in str(m.content)
|
|
|
|
def test_checkpoint_surface_separation(self, tmp_path, monkeypatch):
|
|
# Secrets live on runtime.context, never in the graph state that gets
|
|
# checkpointed (messages/state).
|
|
context, messages, _ = self._activate_with_secret(tmp_path, monkeypatch)
|
|
assert context["secrets"]["ERP_TOKEN"] == _SECRET # present in context...
|
|
assert _SECRET not in str([m.content for m in messages]) # ...not in state
|
|
|
|
def test_audit_surface_has_no_secret(self, tmp_path, monkeypatch):
|
|
_, _, journal_records = self._activate_with_secret(tmp_path, monkeypatch)
|
|
assert journal_records, "activation should record an audit event"
|
|
assert _SECRET not in str(journal_records)
|
|
|
|
def test_trace_metadata_has_no_secret(self, monkeypatch):
|
|
from deerflow.tracing import metadata as meta
|
|
|
|
monkeypatch.setattr(meta, "get_enabled_tracing_providers", lambda: {"langfuse"})
|
|
config = {"context": {"secrets": {"ERP_TOKEN": _SECRET}}, "metadata": {}}
|
|
meta.inject_langfuse_metadata(config, thread_id="t", user_id="u", model_name="m")
|
|
assert _SECRET not in str(config["metadata"])
|
|
# And secrets were never mirrored into configurable.
|
|
assert _SECRET not in str(config.get("configurable", {}))
|
|
|
|
def test_redact_helper_strips_secret_keys(self):
|
|
from deerflow.runtime.secret_context import redact_secret_context_keys
|
|
|
|
ctx = {"thread_id": "t", "secrets": {"ERP_TOKEN": _SECRET}, "__active_skill_secrets": {"ERP_TOKEN": _SECRET}}
|
|
redacted = redact_secret_context_keys(ctx)
|
|
assert redacted == {"thread_id": "t"}
|
|
assert _SECRET not in str(redacted)
|
|
|
|
def test_redact_config_secrets_strips_from_persisted_config(self):
|
|
# The run-record persistence + run API echo the raw request config; the
|
|
# stored/echoed copy must not carry secrets (verifier blocker), while the
|
|
# live config used to drive the run keeps them.
|
|
from deerflow.runtime.secret_context import redact_config_secrets
|
|
|
|
config = {"context": {"secrets": {"ERP_TOKEN": _SECRET}, "thread_id": "t", "model_name": "m"}, "recursion_limit": 100}
|
|
redacted = redact_config_secrets(config)
|
|
assert _SECRET not in str(redacted)
|
|
assert redacted["context"]["thread_id"] == "t"
|
|
assert redacted["context"]["model_name"] == "m"
|
|
assert "secrets" not in redacted["context"]
|
|
# Original is untouched (live config still has secrets).
|
|
assert config["context"]["secrets"] == {"ERP_TOKEN": _SECRET}
|
|
|
|
def test_redact_config_secrets_handles_none_and_no_context(self):
|
|
from deerflow.runtime.secret_context import redact_config_secrets
|
|
|
|
assert redact_config_secrets(None) is None
|
|
assert redact_config_secrets({"configurable": {"thread_id": "t"}}) == {"configurable": {"thread_id": "t"}}
|
|
|
|
def test_stdout_surface_redacted(self):
|
|
from deerflow.sandbox.tools import mask_secret_values
|
|
|
|
leaked = f"DEBUG: token is {_SECRET} done"
|
|
masked = mask_secret_values(leaked, {"ERP_TOKEN": _SECRET})
|
|
assert _SECRET not in masked
|
|
assert "[redacted]" in masked
|
|
|
|
def test_short_secret_values_not_masked(self):
|
|
"""Values below the minimum length floor are skipped — redacting a 2-char
|
|
value would shred unrelated bytes (exit codes, timestamps, sizes) of tool
|
|
output. The secret is still injected into the subprocess; only the output
|
|
mask skips it."""
|
|
from deerflow.sandbox.tools import mask_secret_values
|
|
|
|
# A short value must not be replaced everywhere in the output.
|
|
out = "exit code: 42\nrows: 42\n"
|
|
masked = mask_secret_values(out, {"REGION": "42"})
|
|
assert masked == out # unchanged — short value left intact
|
|
|
|
# A long value is still redacted as before.
|
|
long_secret = "sk-erp-long-enough-token-value"
|
|
masked_long = mask_secret_values(f"token={long_secret}", {"ERP_TOKEN": long_secret})
|
|
assert long_secret not in masked_long
|
|
assert "[redacted]" in masked_long
|
|
|
|
|
|
@pytest.mark.skipif(__import__("os").name == "nt", reason="POSIX shell semantics")
|
|
class TestEndToEndRealSubprocess:
|
|
"""End-to-end across the real chain (no sandbox mock): activation resolves the
|
|
secret, a REAL LocalSandbox subprocess receives it via env, the value lands in
|
|
a file but is redacted from the returned output, and a later un-injected call
|
|
cannot see it."""
|
|
|
|
def test_secret_reaches_real_subprocess_only_via_env_and_is_scoped(self, tmp_path, monkeypatch):
|
|
from deerflow.agents.middlewares import skill_activation_middleware as mw
|
|
from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware
|
|
from deerflow.runtime.secret_context import read_active_secrets
|
|
from deerflow.sandbox.tools import mask_secret_values
|
|
|
|
# 1. Activate a skill that declares ERP_TOKEN; caller supplies it in context.secrets.
|
|
skill = _make_secret_skill(tmp_path, "erp-report", [SecretRequirement("ERP_TOKEN")])
|
|
storage = SimpleNamespace(
|
|
load_skills=lambda *, enabled_only: [skill],
|
|
get_container_root=lambda: "/mnt/skills",
|
|
get_skills_root_path=lambda: tmp_path,
|
|
)
|
|
monkeypatch.setattr(mw, "get_or_new_skill_storage", lambda **kwargs: storage)
|
|
# A platform secret is present on the host and must NOT leak to the subprocess.
|
|
monkeypatch.setenv("OPENAI_API_KEY", "sk-host-platform-secret")
|
|
context = {"secrets": {"ERP_TOKEN": _SECRET}}
|
|
request = ModelRequest(
|
|
model=object(),
|
|
messages=[HumanMessage(content="/erp-report pull report", id="m1")],
|
|
state={"messages": []},
|
|
runtime=SimpleNamespace(context=context),
|
|
)
|
|
SkillActivationMiddleware().wrap_model_call(request, lambda r: AIMessage(content="ok"))
|
|
injected = read_active_secrets(context)
|
|
assert injected == {"ERP_TOKEN": _SECRET}
|
|
|
|
# 2. A REAL LocalSandbox runs a script that writes the token to a file and echoes it.
|
|
out_file = tmp_path / "token.txt"
|
|
sandbox = LocalSandbox(id="local")
|
|
raw = sandbox.execute_command(
|
|
f'printf "%s" "$ERP_TOKEN" > {out_file}; echo "leaked:$ERP_TOKEN"; echo "platform:$OPENAI_API_KEY"',
|
|
env=injected,
|
|
)
|
|
|
|
# 3. The skill genuinely received the token via env (file written by the subprocess).
|
|
assert out_file.read_text() == _SECRET
|
|
# 4. Platform secret was scrubbed — not available to the script.
|
|
assert "sk-host-platform-secret" not in raw
|
|
# 5. Stdout masking redacts the echoed token before it would re-enter context.
|
|
masked = mask_secret_values(raw, injected)
|
|
assert _SECRET not in masked
|
|
|
|
# 6. Per-call scope: a later command without injection cannot see the token.
|
|
leaked = sandbox.execute_command("echo [$ERP_TOKEN]")
|
|
assert _SECRET not in leaked
|