"""Phase 3 skill-level authorization tests. Covers Layer 1 enforcement: - ``filter_available_skills_by_authorization`` — the lead-agent / subagent entry point that filters the skill-name allowlist by the provider's ``"skill"`` policy. - ``filter_resources_by_authorization`` — the generic batch filter. - Subagent executor integration: ``_load_skills`` respects authorization. """ from __future__ import annotations from pathlib import Path from types import SimpleNamespace import pytest from deerflow.authz.enforcement import filter_resources_by_authorization from deerflow.authz.principal import build_principal_from_context from deerflow.authz.rbac import RbacAuthorizationProvider from deerflow.authz.skill_filter import filter_available_skills_by_authorization from deerflow.config.app_config import AppConfig from deerflow.config.authorization_config import AuthorizationConfig from deerflow.config.model_config import ModelConfig from deerflow.config.sandbox_config import SandboxConfig # ── Helpers ──────────────────────────────────────────────────────────── def _make_app_config() -> AppConfig: """Build a minimal AppConfig for authorization tests.""" return AppConfig( models=[ModelConfig(name="gpt-4", model="gpt-4", use="langchain_openai:ChatOpenAI")], sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"), authorization=AuthorizationConfig(), ) def _context(**overrides): values = { "user_id": "user-123", "user_role": "user", "oauth_provider": "github", "oauth_id": "oauth-456", "is_internal": False, } values.update(overrides) return values def _rbac_provider(roles: dict) -> RbacAuthorizationProvider: return RbacAuthorizationProvider(roles=roles) def _enable_authz(app_config: AppConfig, *, fail_closed: bool = True, default_role: str = "user") -> None: app_config.authorization = AuthorizationConfig( enabled=True, fail_closed=fail_closed, default_role=default_role, ) # ── filter_available_skills_by_authorization ─────────────────────────── def test_filter_available_skills_disabled_is_noop(): """When authorization is disabled, the allowlist is returned unchanged.""" app_config = _make_app_config() assert filter_available_skills_by_authorization({"a", "b"}, context=_context(), app_config=app_config) == {"a", "b"} assert filter_available_skills_by_authorization(None, context=_context(), app_config=app_config) is None assert filter_available_skills_by_authorization(set(), context=_context(), app_config=app_config) == set() def test_filter_available_skills_rbac_allow_subset(monkeypatch): """Role allowlist intersects with the agent allowlist.""" provider = _rbac_provider({"user": {"skills": {"allow": ["skill-a", "skill-c"]}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) result = filter_available_skills_by_authorization( {"skill-a", "skill-b", "skill-c"}, context=_context(), app_config=app_config, ) assert result == {"skill-a", "skill-c"} def test_filter_available_skills_provider_injected_names_excluded(monkeypatch): """A buggy provider that returns names outside candidates can't expand the allowlist. Regression for zhfeng's P2 finding: filter_resources contract says results must be a subset of candidates, but only list[str] is validated. The filter must intersect the provider result with candidates so injected names are dropped. """ class _InjectingProvider: name = "injecting" def filter_resources(self, principal, resource_type, candidates): # Returns a name not in the candidate set (simulating a buggy provider). return ["allowed-skill", "injected-skill"] def authorize(self, request): return None async def aauthorize(self, request): return None app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: _InjectingProvider(), ) result = filter_available_skills_by_authorization( {"allowed-skill"}, context=_context(), app_config=app_config, ) # injected-skill was never a candidate, so it must not appear. assert result == {"allowed-skill"} assert "injected-skill" not in result def test_filter_available_skills_rbac_deny(monkeypatch): """Deny removes a skill even when allow is wildcard.""" provider = _rbac_provider({"user": {"skills": {"allow": "*", "deny": ["skill-b"]}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) result = filter_available_skills_by_authorization( {"skill-a", "skill-b", "skill-c"}, context=_context(), app_config=app_config, ) assert result == {"skill-a", "skill-c"} def test_filter_available_skills_wildcard_returns_all(monkeypatch): """Allow: '*' preserves the full allowlist.""" provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) result = filter_available_skills_by_authorization( {"skill-a", "skill-b"}, context=_context(), app_config=app_config, ) assert result == {"skill-a", "skill-b"} def test_filter_available_skills_empty_allowlist_preserved(monkeypatch): """Empty set (no skills for this agent) is preserved as-is.""" provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) result = filter_available_skills_by_authorization( set(), context=_context(), app_config=app_config, ) assert result == set() def test_filter_available_skills_provider_error_fail_closed(monkeypatch): """Provider error + fail_closed → empty set (deny all skills).""" class _ErrorProvider: name = "error" def authorize(self, request): raise RuntimeError("boom") async def aauthorize(self, request): raise RuntimeError("boom") def filter_resources(self, principal, resource_type, candidates): raise RuntimeError("boom") app_config = _make_app_config() _enable_authz(app_config, fail_closed=True) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: _ErrorProvider(), ) result = filter_available_skills_by_authorization( {"skill-a", "skill-b"}, context=_context(), app_config=app_config, ) assert result == set() def test_filter_available_skills_provider_error_fail_open(monkeypatch): """Provider error + fail_open → original allowlist preserved.""" class _ErrorProvider: name = "error" def authorize(self, request): raise RuntimeError("boom") async def aauthorize(self, request): raise RuntimeError("boom") def filter_resources(self, principal, resource_type, candidates): raise RuntimeError("boom") app_config = _make_app_config() _enable_authz(app_config, fail_closed=False) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: _ErrorProvider(), ) result = filter_available_skills_by_authorization( {"skill-a", "skill-b"}, context=_context(), app_config=app_config, ) assert result == {"skill-a", "skill-b"} def test_filter_available_skills_internal_caller_uses_default_role(monkeypatch): """Internal callers (system_role=internal) fall under default_role.""" provider = _rbac_provider( { "user": {"skills": {"allow": ["skill-a"]}}, "admin": {"skills": {"allow": "*"}}, } ) app_config = _make_app_config() _enable_authz(app_config, default_role="user") monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) # Internal caller with system_role=None → default_role="user" result = filter_available_skills_by_authorization( {"skill-a", "skill-b"}, context=_context(user_role=None, is_internal=True), app_config=app_config, ) assert result == {"skill-a"} def test_filter_available_skills_candidate_resolution_error_fail_closed(monkeypatch): """available_skills=None + candidate-set resolution error + fail_closed → empty set. Regression for willem-bd's fail-open bypass finding: when there is no agent-level allowlist (``available_skills=None``) and ``_all_configured_skill_names`` raises (storage I/O error), the result must NOT be ``None`` (which would bypass authorization — every skill becomes activatable/loadable via SkillActivationMiddleware and subagent _load_skills). With ``fail_closed=true`` it must return an empty set (deny all). """ provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config, fail_closed=True) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) monkeypatch.setattr( "deerflow.authz.skill_filter._all_configured_skill_names", lambda app_config, **kw: (_ for _ in ()).throw(RuntimeError("storage I/O failed")), ) result = filter_available_skills_by_authorization( None, context=_context(), app_config=app_config, ) # fail-closed → deny all skills, NOT None (which would bypass authorization). assert result == set() def test_filter_available_skills_candidate_resolution_error_fail_open(monkeypatch): """available_skills=None + candidate-set resolution error + fail_open → None. Fail-open preserves the legacy "no constraint" behavior (None = unrestricted), matching how provider errors are handled in the fail-open branch above. """ provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config, fail_closed=False) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) monkeypatch.setattr( "deerflow.authz.skill_filter._all_configured_skill_names", lambda app_config, **kw: (_ for _ in ()).throw(RuntimeError("storage I/O failed")), ) result = filter_available_skills_by_authorization( None, context=_context(), app_config=app_config, ) # fail-open → unrestricted (None), matching provider-error fail-open behavior. assert result is None def test_filter_available_skills_none_with_empty_config_no_bypass(monkeypatch): """available_skills=None + zero configured skills → None (no bypass, nothing to load). Distinct from the resolution-error case: when ``_all_configured_skill_names`` succeeds and returns an empty list (the user genuinely configured no skills), returning ``None`` is safe because there is nothing for SkillActivationMiddleware or subagent _load_skills to activate/load. """ provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config, fail_closed=True) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) # Resolution succeeds, but the configured-skill set is genuinely empty. monkeypatch.setattr( "deerflow.authz.skill_filter._all_configured_skill_names", lambda app_config, **kw: [], ) result = filter_available_skills_by_authorization( None, context=_context(), app_config=app_config, ) assert result is None def test_filter_available_skills_user_id_threads_into_candidate_resolution(monkeypatch): """user_id is forwarded to _all_configured_skill_names so per-user custom skills are candidates. Regression for willem-bd's Round 3 finding: when available_skills=None and authz is enabled, the candidate universe must include per-user custom skills (UserScopedSkillStorage), not just the process-global public skills. Otherwise a skills:{allow:"*"} wildcard can never reach per-user custom skills, and SkillActivationMiddleware blocks them. """ provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) captured = {} def _record(app_config, *, user_id=None): captured["user_id"] = user_id # Simulate a per-user custom skill that only user-scoped storage would return. return ["public-skill", "user-custom-skill"] monkeypatch.setattr( "deerflow.authz.skill_filter._all_configured_skill_names", _record, ) result = filter_available_skills_by_authorization( None, context=_context(user_id="user-123"), app_config=app_config, user_id="user-123", ) # The user_id was forwarded to the resolver. assert captured["user_id"] == "user-123" # The per-user custom skill is in the filtered set (wildcard allows all). assert result == {"public-skill", "user-custom-skill"} def test_filter_available_skills_omits_user_id_when_not_provided(monkeypatch): """When user_id is not passed, candidate resolution uses global storage (user_id=None).""" provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) captured = {} def _record(app_config, *, user_id=None): captured["user_id"] = user_id return ["public-skill"] monkeypatch.setattr( "deerflow.authz.skill_filter._all_configured_skill_names", _record, ) filter_available_skills_by_authorization( None, context=_context(), app_config=app_config, # user_id deliberately omitted (e.g. caller has no user context). ) assert captured["user_id"] is None # ── filter_resources_by_authorization (generic) ──────────────────────── def _named(name: str): return SimpleNamespace(name=name) def test_filter_resources_by_authorization_no_provider(): """No provider → original list returned.""" principal = build_principal_from_context(_context(), default_role="user") resources = [_named("a"), _named("b")] assert filter_resources_by_authorization(resources, resource_type="skill", provider=None, principal=principal, fail_closed=True) == resources def test_filter_resources_by_authorization_rbac(monkeypatch): """Generic filter works with any resource type.""" provider = _rbac_provider({"user": {"skills": {"allow": ["a"], "deny": ["c"]}}}) principal = build_principal_from_context(_context(), default_role="user") resources = [_named("a"), _named("b"), _named("c")] result = filter_resources_by_authorization( resources, resource_type="skill", provider=provider, principal=principal, fail_closed=True, ) assert [r.name for r in result] == ["a"] def test_filter_resources_by_authorization_provider_error_fail_closed(): """Provider error + fail_closed → empty list.""" class _ErrorProvider: name = "error" def authorize(self, request): raise RuntimeError("boom") async def aauthorize(self, request): raise RuntimeError("boom") def filter_resources(self, principal, resource_type, candidates): raise RuntimeError("boom") principal = build_principal_from_context(_context(), default_role="user") resources = [_named("a"), _named("b")] result = filter_resources_by_authorization( resources, resource_type="skill", provider=_ErrorProvider(), principal=principal, fail_closed=True, ) assert result == [] def test_filter_resources_by_authorization_preserves_order(): """Original order is preserved in the filtered result.""" provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) principal = build_principal_from_context(_context(), default_role="user") resources = [_named("z"), _named("a"), _named("m")] result = filter_resources_by_authorization( resources, resource_type="skill", provider=provider, principal=principal, fail_closed=True, ) assert [r.name for r in result] == ["z", "a", "m"] # ── DeerFlowClient._ensure_agent path ───────────────────────────────── # Regression for willem-bd's Round 2 coverage observation: the embedded # lead-agent construction path (DeerFlowClient._ensure_agent) must filter # skills through authorization too, mirroring _make_lead_agent. Otherwise a # caller building the agent via DeerFlowClient(available_skills=...) bypasses # the role's skills policy. def test_client_ensure_agent_filters_skills_by_authorization(monkeypatch): """_ensure_agent passes an authorization-filtered skill set to build_middlewares. Real-path test: a genuine RbacAuthorizationProvider denies 'denied-skill' for the 'user' role. The embedded _ensure_agent must filter it out before wiring SkillActivationMiddleware, so the denied skill can't be slash-activated. """ from langchain_core.runnables import RunnableConfig from deerflow.client import DeerFlowClient app_config = _make_app_config() _enable_authz(app_config) provider = _rbac_provider({"user": {"skills": {"allow": ["allowed-skill"], "deny": ["denied-skill"]}}}) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) monkeypatch.setattr( "deerflow.authz.tool_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"tools": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.agents.lead_agent.agent.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"models": {"allow": "*"}}}), ) captured = {} def _capture_build_middlewares(*args, **kwargs): captured["available_skills"] = kwargs.get("available_skills") return [] monkeypatch.setattr("deerflow.client.create_chat_model", lambda **kw: object()) monkeypatch.setattr("deerflow.client.create_agent", lambda **kw: object()) monkeypatch.setattr("deerflow.client.build_middlewares", _capture_build_middlewares) monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005 monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config, **kw: []) monkeypatch.setattr( "deerflow.client.build_skill_search_setup", lambda skills, *, enabled, container_base_path, skill_authorization=None: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()), ) monkeypatch.setattr( "deerflow.client.assemble_deferred_tools", lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())), ) monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *a, **kw: None) monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *a, **kw: "") monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kw: "") monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *a, **kw: object()) monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: []) monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123") client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = app_config client._agent_name = "default" client._available_skills = {"allowed-skill", "denied-skill"} client._checkpoint_channel_mode = "full" client._checkpoint_snapshot_frequency = None client._middlewares = [] client._agent = None client._agent_config_key = None client._checkpointer = object() config = RunnableConfig(configurable={"user_id": "user-123", "user_role": "user"}) client._ensure_agent(config) # The denied skill was filtered out; only 'allowed-skill' reaches the middleware. assert captured["available_skills"] == {"allowed-skill"} def test_client_ensure_agent_noop_when_authorization_disabled(monkeypatch): """When authorization is disabled, _ensure_agent leaves the skill set unchanged.""" from langchain_core.runnables import RunnableConfig from deerflow.client import DeerFlowClient app_config = _make_app_config() # AuthorizationConfig() defaults to enabled=False. captured = {} def _capture_build_middlewares(*args, **kwargs): captured["available_skills"] = kwargs.get("available_skills") return [] monkeypatch.setattr("deerflow.client.create_chat_model", lambda **kw: object()) monkeypatch.setattr("deerflow.client.create_agent", lambda **kw: object()) monkeypatch.setattr("deerflow.client.build_middlewares", _capture_build_middlewares) monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005 monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config, **kw: []) monkeypatch.setattr( "deerflow.client.build_skill_search_setup", lambda skills, *, enabled, container_base_path, skill_authorization=None: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()), ) monkeypatch.setattr( "deerflow.client.assemble_deferred_tools", lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())), ) monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *a, **kw: None) monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *a, **kw: "") monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kw: "") monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *a, **kw: object()) monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: []) monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123") client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = app_config client._agent_name = "default" client._available_skills = {"skill-a", "skill-b"} client._checkpoint_channel_mode = "full" client._checkpoint_snapshot_frequency = None client._middlewares = [] client._agent = None client._agent_config_key = None client._checkpointer = object() config = RunnableConfig(configurable={"user_id": "user-123", "user_role": "user"}) client._ensure_agent(config) # Disabled → no-op: original skill set passed through unchanged. assert captured["available_skills"] == {"skill-a", "skill-b"} # ── Runtime skill:activate enforcement (Layer 2) ────────────────────── class _ActionAwareProvider: """Minimal provider distinguishing visibility (filter_resources) from activation.""" def __init__(self, *, denied_activate: set[str]): self.denied_activate = denied_activate self.sync_calls: list[str] = [] self.async_calls: list[str] = [] def filter_resources(self, principal, resource, candidates): return list(candidates) def authorize(self, request): from deerflow.authz.provider import AuthzDecision if request.resource == "skill" and request.action == "activate": self.sync_calls.append(request.target) if request.resource == "skill" and request.action == "activate" and request.target in self.denied_activate: return AuthzDecision(allow=False) return AuthzDecision(allow=True) async def aauthorize(self, request): from deerflow.authz.provider import AuthzDecision if request.resource == "skill" and request.action == "activate": self.async_calls.append(request.target) if request.resource == "skill" and request.action == "activate" and request.target in self.denied_activate: return AuthzDecision(allow=False) return AuthzDecision(allow=True) class _RaisingAuthorizeProvider(_ActionAwareProvider): def authorize(self, request): raise RuntimeError("provider blew up") async def aauthorize(self, request): raise RuntimeError("provider blew up") class _AsyncOnlyProvider: """Loop-affine provider: the sync API is the wrong one and always fails. Models a provider whose clients are bound to the running event loop — ``authorize()`` from a worker thread or the loop itself is unsupported, while ``aauthorize()`` works. Async paths must use the async API; if they fall back to the sync call, fail-closed denies (or fail-open admits) with the wrong semantics. """ name = "async-only" def __init__(self, *, denied_activate: set[str] = set()): self.denied_activate = denied_activate self.async_calls: list[str] = [] def filter_resources(self, principal, resource, candidates): return list(candidates) def authorize(self, request): raise RuntimeError("sync authorize() is not supported by this provider") async def aauthorize(self, request): from deerflow.authz.provider import AuthzDecision if request.resource == "skill" and request.action == "activate": self.async_calls.append(request.target) if request.resource == "skill" and request.action == "activate" and request.target in self.denied_activate: return AuthzDecision(allow=False) return AuthzDecision(allow=True) def _resolved_skill_authorization(provider, *, fail_closed: bool): from deerflow.authz.skill_filter import resolve_skill_authorization app_config = _make_app_config() _enable_authz(app_config, fail_closed=fail_closed) import deerflow.authz.skill_filter as skill_filter_module original = skill_filter_module.resolve_authorization_provider skill_filter_module.resolve_authorization_provider = lambda config: provider try: resolved = resolve_skill_authorization(_context(), app_config) finally: skill_filter_module.resolve_authorization_provider = original assert resolved is not None return resolved def _middleware_for(tmp_path, monkeypatch, skills, **kwargs): from pathlib import Path from deerflow.agents.middlewares import skill_activation_middleware as middleware_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory def _make_skill(name: str) -> SkillObject: skill_dir = tmp_path / name skill_dir.mkdir(exist_ok=True) skill_file = skill_dir / "SKILL.md" skill_file.write_text(f"# {name}\nskill body", encoding="utf-8") return SkillObject( 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=True, ) made = [_make_skill(name) for name in skills] def _validate_skill_file_path(skill_file): resolved = skill_file.resolve() resolved.relative_to(tmp_path.resolve()) return resolved storage = SimpleNamespace( load_skills=lambda *, enabled_only: made, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, validate_skill_file_path=_validate_skill_file_path, ) monkeypatch.setattr(middleware_module, "get_or_new_skill_storage", lambda **kw: storage) return SkillActivationMiddleware(slash_source_owner_token="test-token", **kwargs) def test_activation_middleware_requires_shared_slash_source_token(): """SkillActivationMiddleware's run-context publications are token-bound so caller-forged context cannot impersonate a slash activation.""" from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware with pytest.raises(TypeError, match="slash_source_owner_token"): SkillActivationMiddleware() @pytest.mark.parametrize("invalid_token", [None, "", 7]) def test_activation_middleware_rejects_invalid_slash_source_tokens(invalid_token): from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware with pytest.raises(ValueError, match="non-empty string"): SkillActivationMiddleware(slash_source_owner_token=invalid_token) def test_activation_enforces_action_scoped_activate_decision(tmp_path, monkeypatch): """A skill visible in the Layer 1 set but denied by authorize(activate) must not slash-activate, while an allowed one still does (review repro).""" provider = _ActionAwareProvider(denied_activate={"demo-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill", "other-skill"], available_skills={"demo-skill", "other-skill"}, skill_authorization=resolved, ) denied = middleware._resolve_activation("/demo-skill analyze this") assert denied is not None assert denied.activation is None assert "not available" in denied.failure_message allowed = middleware._resolve_activation("/other-skill analyze this") assert allowed is not None assert allowed.activation is not None assert allowed.activation.skill_name == "other-skill" def test_activation_provider_error_fail_closed(tmp_path, monkeypatch): """authorize() raising under fail_closed=True denies activation.""" provider = _RaisingAuthorizeProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill"], available_skills={"demo-skill"}, skill_authorization=resolved, ) result = middleware._resolve_activation("/demo-skill hi") assert result is not None assert result.activation is None assert "not available" in result.failure_message def test_activation_provider_error_fail_open(tmp_path, monkeypatch): """authorize() raising under fail_closed=False allows (fail-open policy).""" provider = _RaisingAuthorizeProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=False) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill"], available_skills={"demo-skill"}, skill_authorization=resolved, ) result = middleware._resolve_activation("/demo-skill hi") assert result is not None assert result.activation is not None def test_activation_rbac_provider_membership_equivalent(tmp_path, monkeypatch): """Built-in RBAC (action-agnostic) keeps working: allowed skill activates.""" provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill"], available_skills={"demo-skill"}, skill_authorization=resolved, ) result = middleware._resolve_activation("/demo-skill hi") assert result is not None assert result.activation is not None def test_client_wires_skill_authorization_into_middleware(monkeypatch): """_ensure_agent resolves the skill authorization context once and passes it to build_middlewares so SkillActivationMiddleware enforces skill:activate.""" from langchain_core.runnables import RunnableConfig from deerflow.authz.skill_filter import ResolvedSkillAuthorization from deerflow.client import DeerFlowClient app_config = _make_app_config() _enable_authz(app_config) provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: provider, ) monkeypatch.setattr( "deerflow.authz.tool_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"tools": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.agents.lead_agent.agent.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"models": {"allow": "*"}}}), ) captured = {} def _capture_build_middlewares(*args, **kwargs): captured["skill_authorization"] = kwargs.get("skill_authorization") return [] monkeypatch.setattr("deerflow.client.create_chat_model", lambda **kw: object()) monkeypatch.setattr("deerflow.client.create_agent", lambda **kw: object()) monkeypatch.setattr("deerflow.client.build_middlewares", _capture_build_middlewares) monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005 monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", lambda app_config, **kw: []) monkeypatch.setattr( "deerflow.client.build_skill_search_setup", lambda skills, *, enabled, container_base_path, skill_authorization=None: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()), ) monkeypatch.setattr( "deerflow.client.assemble_deferred_tools", lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())), ) monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *a, **kw: None) monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *a, **kw: "") monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kw: "") monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *a, **kw: object()) monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: []) monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123") client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = app_config client._agent_name = "default" client._available_skills = None client._checkpoint_channel_mode = "full" client._checkpoint_snapshot_frequency = None client._middlewares = [] client._agent = None client._agent_config_key = None client._checkpointer = object() client._ensure_agent(RunnableConfig(configurable={"user_id": "user-123", "user_role": "user"})) wired = captured["skill_authorization"] assert isinstance(wired, ResolvedSkillAuthorization) assert wired.provider is provider assert wired.fail_closed is True def test_client_filter_candidates_reuse_catalog_loader(monkeypatch): """With no agent-level allowlist, the filter's candidates come from the cached catalog loader — the filter's uncached storage scan never runs.""" from langchain_core.runnables import RunnableConfig from deerflow.client import DeerFlowClient app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"skills": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.authz.tool_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"tools": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.agents.lead_agent.agent.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"models": {"allow": "*"}}}), ) catalog_calls = {"count": 0} def _catalog_loader(app_config, **kw): catalog_calls["count"] += 1 return [SimpleNamespace(name="catalog-skill")] def _must_not_scan(*a, **kw): raise AssertionError("filter must not rescan storage when candidates were provided") monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", _catalog_loader) monkeypatch.setattr("deerflow.authz.skill_filter._all_configured_skill_names", _must_not_scan) import deerflow.authz.skill_filter as skill_filter_module captured = {} _original_filter = skill_filter_module.filter_available_skills_by_authorization def _filter_spy(available_skills, **kwargs): captured["candidate_skill_names"] = kwargs.get("candidate_skill_names") return _original_filter(available_skills, **kwargs) # client._ensure_agent lazy-imports the filter from this module, so the # spy has to replace it at the source. monkeypatch.setattr(skill_filter_module, "filter_available_skills_by_authorization", _filter_spy) monkeypatch.setattr("deerflow.client.create_chat_model", lambda **kw: object()) monkeypatch.setattr("deerflow.client.create_agent", lambda **kw: object()) monkeypatch.setattr("deerflow.client.build_middlewares", lambda *a, **kw: []) monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005 monkeypatch.setattr( "deerflow.client.build_skill_search_setup", lambda skills, *, enabled, container_base_path, skill_authorization=None: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()), ) monkeypatch.setattr( "deerflow.client.assemble_deferred_tools", lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())), ) monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *a, **kw: None) monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *a, **kw: "") monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kw: "") monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *a, **kw: object()) monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: []) monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123") client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = app_config client._agent_name = "default" client._available_skills = None # no agent-level allowlist → candidates matter client._checkpoint_channel_mode = "full" client._checkpoint_snapshot_frequency = None client._middlewares = [] client._agent = None client._agent_config_key = None client._checkpointer = object() client._ensure_agent(RunnableConfig(configurable={"user_id": "user-123", "user_role": "user"})) assert captured["candidate_skill_names"] == ["catalog-skill"] assert catalog_calls["count"] >= 1 def test_subagent_chain_arms_skill_activate_check(monkeypatch): """build_subagent_runtime_middlewares forwards skill_authorization into the subagent chain's SkillActivationMiddleware (a delegated task is a plain HumanMessage, so /skill-name in task text reaches the activation path).""" import deerflow.authz.skill_filter as skill_filter_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares from deerflow.authz.skill_filter import resolve_skill_authorization app_config = _make_app_config() _enable_authz(app_config) provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) monkeypatch.setattr(skill_filter_module, "resolve_authorization_provider", lambda config: provider) resolved = resolve_skill_authorization(_context(), app_config) assert resolved is not None middlewares = build_subagent_runtime_middlewares( app_config=app_config, model_name=None, lazy_init=True, available_skills={"demo-skill"}, user_id="user-123", authorization_provider=provider, skill_authorization=resolved, ) activation = [m for m in middlewares if isinstance(m, SkillActivationMiddleware)] assert len(activation) == 1 assert activation[0]._skill_authorization is resolved # The skill-read stamp gate on the same chain shares the resolved context # so an autonomously read SKILL.md cannot activate a denied skill. from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware stampers = [m for m in middlewares if isinstance(m, ToolErrorHandlingMiddleware)] assert len(stampers) == 1 assert stampers[0]._skill_authorization is resolved def test_subagent_executor_resolves_skill_authorization_for_chain(monkeypatch): """SubagentExecutor._create_agent passes a resolved ResolvedSkillAuthorization into build_subagent_runtime_middlewares, built from the executor identity.""" import importlib import sys from types import SimpleNamespace import deerflow.authz.skill_filter as skill_filter_module from deerflow.authz.skill_filter import ResolvedSkillAuthorization # tests/conftest.py injects a MagicMock for deerflow.subagents.executor to # break a production circular import; load the real module for this test # (same pattern as tests/test_delegation_ledger_live.py). sys.modules.pop("deerflow.subagents.executor", None) executor_module = importlib.import_module("deerflow.subagents.executor") SubagentExecutor = executor_module.SubagentExecutor app_config = _make_app_config() _enable_authz(app_config) provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) monkeypatch.setattr(skill_filter_module, "resolve_authorization_provider", lambda config: provider) monkeypatch.setattr(executor_module, "create_chat_model", lambda **kw: object()) monkeypatch.setattr(executor_module, "resolve_subagent_model_name", lambda *a, **kw: "m") captured = {} def _capture_builder(**kwargs): captured["skill_authorization"] = kwargs.get("skill_authorization") return [] monkeypatch.setattr( "deerflow.agents.middlewares.tool_error_handling_middleware.build_subagent_runtime_middlewares", _capture_builder, ) executor = SubagentExecutor.__new__(SubagentExecutor) executor.config = SimpleNamespace(name="sub", skills=None, max_turns=25) executor.model_name = "m" executor.app_config = app_config executor._resolved_app_config = app_config executor._available_skill_names = None executor.extensions = None executor.user_id = "user-123" executor.trace_id = "test-trace" executor.user_role = "user" executor.oauth_provider = None executor.oauth_id = None executor.channel_user_id = None executor.is_internal = False executor.authz_attributes = None executor.tools = [] import asyncio asyncio.run(executor._create_agent(tools=[])) wired = captured["skill_authorization"] assert isinstance(wired, ResolvedSkillAuthorization) assert wired.provider is provider def test_client_skill_surface_uses_one_effective_user(monkeypatch): """The filter, the candidate pre-load, and the catalog all resolve the same effective user — a request whose configurable carries no user_id must not fall back to the process-global skill bucket while the middleware loads user-scoped storage (per-user custom skills would then be filtered out).""" from langchain_core.runnables import RunnableConfig from deerflow.client import DeerFlowClient app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.authz.skill_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"skills": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.authz.tool_filter.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"tools": {"allow": "*"}}}), ) monkeypatch.setattr( "deerflow.agents.lead_agent.agent.resolve_authorization_provider", lambda config: _rbac_provider({"user": {"models": {"allow": "*"}}}), ) catalog_calls = [] def _catalog_loader(app_config, **kw): catalog_calls.append(kw.get("user_id")) return [SimpleNamespace(name="catalog-skill")] monkeypatch.setattr("deerflow.client.get_enabled_skills_for_config", _catalog_loader) import deerflow.authz.skill_filter as skill_filter_module captured = {} _original_filter = skill_filter_module.filter_available_skills_by_authorization def _filter_spy(available_skills, **kwargs): captured["user_id"] = kwargs.get("user_id") return _original_filter(available_skills, **kwargs) monkeypatch.setattr(skill_filter_module, "filter_available_skills_by_authorization", _filter_spy) monkeypatch.setattr("deerflow.client.create_chat_model", lambda **kw: object()) monkeypatch.setattr("deerflow.client.create_agent", lambda **kw: object()) monkeypatch.setattr("deerflow.client.build_middlewares", lambda *a, **kw: []) monkeypatch.setattr("deerflow.client.DeerFlowClient._get_tools", staticmethod(lambda *, model_name, subagent_enabled, mcp_plugins=None: [])) # noqa: ARG005 monkeypatch.setattr( "deerflow.client.build_skill_search_setup", lambda skills, *, enabled, container_base_path, skill_authorization=None: SimpleNamespace(describe_skill_tool=None, skill_names=frozenset()), ) monkeypatch.setattr( "deerflow.client.assemble_deferred_tools", lambda tools, *, enabled: ([], SimpleNamespace(deferred_names=frozenset())), ) monkeypatch.setattr("deerflow.client.build_mcp_routing_middleware", lambda *a, **kw: None) monkeypatch.setattr("deerflow.client.get_mcp_routing_hints_prompt_section", lambda *a, **kw: "") monkeypatch.setattr("deerflow.client.apply_prompt_template", lambda **kw: "") monkeypatch.setattr("deerflow.client.get_thread_state_schema", lambda *a, **kw: object()) monkeypatch.setattr("deerflow.client.normalize_middleware_state_schemas", lambda schemas, mode, freq: []) # No user_id in configurable: the effective id must come from here. monkeypatch.setattr("deerflow.client.get_effective_user_id", lambda: "user-123") client = DeerFlowClient.__new__(DeerFlowClient) client._app_config = app_config client._agent_name = "default" client._available_skills = None client._checkpoint_channel_mode = "full" client._checkpoint_snapshot_frequency = None client._middlewares = [] client._agent = None client._agent_config_key = None client._checkpointer = object() client._ensure_agent(RunnableConfig(configurable={"user_role": "user"})) # The filter, the candidate pre-load, and the catalog all used the # context-scoped effective user, never the global bucket. assert captured["user_id"] == "user-123" assert catalog_calls and all(call == "user-123" for call in catalog_calls) # ── Autonomous path: describe_skill + skill-file-load gates (Layer 2) ── def _describe_setup_for(provider, *, fail_closed: bool = True, skills=("demo-skill", "other-skill")): """Build a real describe_skill tool over *skills* with the provider wired.""" from pathlib import Path from deerflow.skills.catalog import SkillCatalog from deerflow.skills.describe import build_describe_skill_tool from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory made = [] for name in skills: base = Path("/mnt/skills/public") / name made.append( SkillObject( name=name, description=f"Description for {name}", license=None, skill_dir=base, skill_file=base / "SKILL.md", relative_path=Path(name), category=SkillCategory.PUBLIC, enabled=True, ) ) resolved = _resolved_skill_authorization(provider, fail_closed=fail_closed) return build_describe_skill_tool(SkillCatalog(tuple(made)), skill_authorization=resolved) def _invoke_describe(tool, query: str) -> str: result = tool.invoke({"args": {"name": query}, "name": "describe_skill", "type": "tool_call", "id": "describe-call"}) return result.update["messages"][0].content def test_describe_skill_gates_on_activate_decision(): """describe_skill omits skills denied by authorize(skill, activate) even though the Layer 1 catalog lists them — the autonomous load path must not be able to discover and read_file a skill whose activation is denied.""" provider = _ActionAwareProvider(denied_activate={"demo-skill"}) tool = _describe_setup_for(provider) content = _invoke_describe(tool, "select:demo-skill,other-skill") assert "other-skill" in content assert "demo-skill" not in content def test_describe_skill_denied_only_reports_no_match(): """A describe whose every match is denied is indistinguishable from a non-match (the denial reason is not leaked).""" provider = _ActionAwareProvider(denied_activate={"demo-skill"}) tool = _describe_setup_for(provider, skills=("demo-skill",)) content = _invoke_describe(tool, "select:demo-skill") assert "No skills matched" in content def test_describe_skill_provider_error_fail_closed(): provider = _RaisingAuthorizeProvider(denied_activate=set()) tool = _describe_setup_for(provider, fail_closed=True) content = _invoke_describe(tool, "select:demo-skill") assert "No skills matched" in content def test_describe_skill_provider_error_fail_open(): provider = _RaisingAuthorizeProvider(denied_activate=set()) tool = _describe_setup_for(provider, fail_closed=False) content = _invoke_describe(tool, "select:demo-skill") assert "demo-skill" in content def test_describe_skill_disabled_authorization_describes_all(): """Without a resolved authorization context the tool describes everything in the catalog (pre-authorization behavior).""" from pathlib import Path from deerflow.skills.catalog import SkillCatalog from deerflow.skills.describe import build_describe_skill_tool from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory made = [] for name in ("demo-skill", "other-skill"): base = Path("/mnt/skills/public") / name made.append( SkillObject( name=name, description=f"Description for {name}", license=None, skill_dir=base, skill_file=base / "SKILL.md", relative_path=Path(name), category=SkillCategory.PUBLIC, enabled=True, ) ) tool = build_describe_skill_tool(SkillCatalog(tuple(made))) content = _invoke_describe(tool, "select:demo-skill,other-skill") assert "demo-skill" in content assert "other-skill" in content # ── Skill-file-load stamping gate ───────────────────────────────────── def _read_call_and_message(path: str): from langchain_core.messages import ToolMessage request = SimpleNamespace(tool_call={"name": "read_file", "id": "call-1", "args": {"path": path}}) message = ToolMessage(content="---\ndescription: Demo skill\n---\n# demo", tool_call_id="call-1", name="read_file") return request, message def test_skill_read_stamp_gates_on_activate_decision(): """A completed SKILL.md read only records a skill_context entry when the action-scoped skill:activate decision allows it; a denied read gets the denial marker instead (no durable context, tool policy, or secrets).""" from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware provider = _ActionAwareProvider(denied_activate={"demo-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/demo-skill/SKILL.md") stamped = middleware._stamp_skill_read_metadata(message, request, tool_name="read_file") assert SKILL_CONTEXT_ENTRY_KEY not in stamped.additional_kwargs assert stamped.additional_kwargs.get(SKILL_CONTEXT_DENIED_KEY) is True request_ok, message_ok = _read_call_and_message("/mnt/skills/public/other-skill/SKILL.md") stamped_ok = middleware._stamp_skill_read_metadata(message_ok, request_ok, tool_name="read_file") assert SKILL_CONTEXT_DENIED_KEY not in stamped_ok.additional_kwargs assert SKILL_CONTEXT_ENTRY_KEY in stamped_ok.additional_kwargs def test_extract_skills_skips_denied_reads_without_warning(caplog): """extract_skills treats the denial marker as intentional — no entry and no misleading 'missing skill read metadata' warning.""" from langchain_core.messages import AIMessage, ToolMessage from deerflow.agents.middlewares import skill_context as skill_context_module from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY messages = [ AIMessage(content="", tool_calls=[{"name": "read_file", "id": "call-1", "args": {"path": "/mnt/skills/public/demo-skill/SKILL.md"}}]), ToolMessage(content="# demo", tool_call_id="call-1", name="read_file", additional_kwargs={SKILL_CONTEXT_DENIED_KEY: True}), ] with caplog.at_level("WARNING", logger="deerflow.agents.middlewares.skill_context"): entries = skill_context_module.extract_skills(messages, skills_root="/mnt/skills", read_tool_names={"read_file"}) assert entries == [] assert not caplog.records # ── Runtime-chain wiring for the autonomous-path gates ──────────────── def test_lead_runtime_chain_forwards_skill_authorization_to_stamp_gate(): """build_lead_runtime_middlewares hands the resolved skill authorization to ToolErrorHandlingMiddleware so the skill-read stamp gate runs with the same provider instance the activation middleware uses — and the effective user id, so the stamp gate canonicalizes read paths against the same user-scoped registry the activation middleware resolves (a global registry would fall back to path-derived names for per-user custom skills, producing a wrong authorization target).""" from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware, build_lead_runtime_middlewares provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middlewares = build_lead_runtime_middlewares(app_config=_make_app_config(), skill_authorization=resolved, user_id="user-123") stampers = [m for m in middlewares if isinstance(m, ToolErrorHandlingMiddleware)] assert len(stampers) == 1 assert stampers[0]._skill_authorization is resolved assert stampers[0]._user_id == "user-123" def test_subagent_runtime_chain_forwards_user_id_to_stamp_gate(): """build_subagent_runtime_middlewares hands the effective user id to ToolErrorHandlingMiddleware: the subagent stamp gate must canonicalize skill-read paths against the user-scoped registry, matching the activation middleware constructed in the same builder.""" from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware, build_subagent_runtime_middlewares middlewares = build_subagent_runtime_middlewares(app_config=_make_app_config(), user_id="user-456") stampers = [m for m in middlewares if isinstance(m, ToolErrorHandlingMiddleware)] assert len(stampers) == 1 assert stampers[0]._user_id == "user-456" def test_subagent_executor_shares_one_skill_authorization_instance(monkeypatch): """[P2 regression] A provider factory returning a distinct instance per resolve must not split the subagent's layers: the Layer 1 filter in ``_load_skills`` and the runtime ``skill:activate`` checks in ``_create_agent`` see the same ResolvedSkillAuthorization (one resolve per executor assembly).""" import asyncio import importlib import sys import deerflow.authz.skill_filter as skill_filter_module from deerflow.authz.skill_filter import ResolvedSkillAuthorization sys.modules.pop("deerflow.subagents.executor", None) executor_module = importlib.import_module("deerflow.subagents.executor") SubagentExecutor = executor_module.SubagentExecutor resolved_providers: list = [] def _distinct_factory(config): provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) resolved_providers.append(provider) return provider monkeypatch.setattr(skill_filter_module, "resolve_authorization_provider", _distinct_factory) filter_captured: dict = {} _original_filter = skill_filter_module.filter_available_skills_by_authorization def _filter_spy(available_skills, **kwargs): filter_captured["authorization"] = kwargs.get("authorization") return _original_filter(available_skills, **kwargs) monkeypatch.setattr(skill_filter_module, "filter_available_skills_by_authorization", _filter_spy) app_config = _make_app_config() _enable_authz(app_config) monkeypatch.setattr( "deerflow.skills.storage.get_or_new_user_skill_storage", lambda user_id, **kw: SimpleNamespace(load_skills=lambda *, enabled_only: [SimpleNamespace(name="demo-skill")]), ) middleware_captured: dict = {} def _capture_builder(**kwargs): middleware_captured["skill_authorization"] = kwargs.get("skill_authorization") return [] monkeypatch.setattr( "deerflow.agents.middlewares.tool_error_handling_middleware.build_subagent_runtime_middlewares", _capture_builder, ) monkeypatch.setattr(executor_module, "create_chat_model", lambda **kw: object()) executor = SubagentExecutor.__new__(SubagentExecutor) executor.config = SimpleNamespace(name="sub", skills=None, max_turns=25) executor.model_name = "m" executor.app_config = app_config executor._resolved_app_config = app_config executor._available_skill_names = None executor._skill_authorization = None executor._stop_reason_middlewares = [] executor.extensions = None executor.trace_id = "test-trace" executor.user_id = "user-123" executor.trace_id = "test-trace" executor.user_role = "user" executor.oauth_provider = None executor.oauth_id = None executor.channel_user_id = None executor.is_internal = False executor.authz_attributes = None executor.tools = [] asyncio.run(executor._load_skills()) import asyncio asyncio.run(executor._create_agent(tools=[])) layer1 = filter_captured["authorization"] layer2 = middleware_captured["skill_authorization"] assert isinstance(layer1, ResolvedSkillAuthorization) assert layer2 is layer1 # One resolve per assembly even though the factory returns distinct objects. assert len(resolved_providers) == 1 # ── Re-authorization of persisted skill_context entries (review round 5) ── def test_in_context_secret_sources_reauthorize_persisted_entries(tmp_path, monkeypatch): """[P1 regression] An entry stamped while activation was allowed must stop binding secrets on the next run once the provider denies skill:activate — skill_context persists across runs; the decision does not.""" provider = _ActionAwareProvider(denied_activate={"demo-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill"], available_skills={"demo-skill"}, skill_authorization=resolved, ) def _make_skill_with_secrets(name: str): made = middleware._storage().load_skills(enabled_only=True) skill = next(s for s in made if s.name == name) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name="API_KEY", optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) return skill skill = _make_skill_with_secrets("demo-skill") container_root = middleware._storage().get_container_root() registry = {posixpath_normpath(skill.get_container_file_path(container_root)): skill} request = SimpleNamespace(state={"skill_context": [{"name": "demo-skill", "path": skill.get_container_file_path(container_root)}]}) # Run 1: activation allowed -> the persisted entry binds its secrets. provider.denied_activate = set() sources = middleware._in_context_secret_sources(request, registry) assert [name for name, _ in sources] == ["demo-skill"] # Run 2 (deny-next-run): same persisted entry, provider now denies. provider.denied_activate = {"demo-skill"} sources = middleware._in_context_secret_sources(request, registry) assert sources == [] def test_async_activation_path_uses_aauthorize(tmp_path, monkeypatch): """awrap_model_call resolves skill:activate with aauthorize() on the loop — a loop-affine provider whose sync API always fails still activates.""" import asyncio from langchain_core.messages import HumanMessage provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = _middleware_for( tmp_path, monkeypatch, ["demo-skill"], available_skills={"demo-skill"}, skill_authorization=resolved, ) class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) clone = _Request(**updates) return clone request = _Request( messages=[HumanMessage(content="/demo-skill hi")], state={}, runtime=None, ) captured = {} async def _handler(prepared): captured["messages"] = list(prepared.messages) return SimpleNamespace(ok=True) result = asyncio.run(middleware.awrap_model_call(request, _handler)) assert getattr(result, "ok", False) # aauthorize answered; the failing sync authorize() was never consulted. assert provider.async_calls == ["demo-skill"] assert len(captured["messages"]) == 2 # activation reminder + original assert any(is_slash_activation_reminder(m) for m in captured["messages"]) def test_async_stamp_path_uses_aauthorize(): """awrap_tool_call's skill-read stamp awaits aauthorize() — a loop-affine provider still yields an entry (not a fail-closed denial marker).""" import asyncio from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/demo-skill/SKILL.md") async def _run(): return await middleware._amaybe_stamp(message, request) stamped = asyncio.run(_run()) assert provider.async_calls == ["demo-skill"] assert SKILL_CONTEXT_ENTRY_KEY in stamped.additional_kwargs assert SKILL_CONTEXT_DENIED_KEY not in stamped.additional_kwargs def test_async_stamp_path_denies_via_aauthorize(): import asyncio from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware provider = _AsyncOnlyProvider(denied_activate={"demo-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/demo-skill/SKILL.md") async def _run(): return await middleware._amaybe_stamp(message, request) stamped = asyncio.run(_run()) assert provider.async_calls == ["demo-skill"] assert SKILL_CONTEXT_ENTRY_KEY not in stamped.additional_kwargs assert stamped.additional_kwargs.get(SKILL_CONTEXT_DENIED_KEY) is True def test_describe_async_uses_aauthorize(): """describe_skill's async invocation gates on aauthorize().""" import asyncio provider = _AsyncOnlyProvider(denied_activate={"demo-skill"}) tool = _describe_setup_for(provider, skills=("demo-skill", "other-skill")) async def _run(): return await tool.ainvoke({"args": {"name": "select:demo-skill,other-skill"}, "name": "describe_skill", "type": "tool_call", "id": "async-call"}) result = asyncio.run(_run()) content = result.update["messages"][0].content assert "other-skill" in content assert "demo-skill" not in content assert sorted(provider.async_calls) == ["demo-skill", "other-skill"] # (``skills/public/vercel-deploy-claimable/SKILL.md`` declares ``name: # vercel-deploy``). Layer 1, slash activation, and describe_skill authorize # the declared ``Skill.name``; every runtime gate that starts from a path # must resolve it through the same registry before asking the provider. def _mismatched_skill(tmp_path): """Bundled-skill shape: directory name differs from the declared name.""" from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "vercel-deploy-claimable" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: Deploy stuff\n---\n# vercel-deploy\nbody", encoding="utf-8") return SkillObject( name="vercel-deploy", description="Deploy stuff", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("vercel-deploy-claimable"), category=SkillCategory.PUBLIC, enabled=True, allowed_tools=("bash",), ) def _patch_registry_storage(monkeypatch, middleware_cls, skills, container_root="/mnt/skills"): monkeypatch.setattr( middleware_cls, "_storage", lambda self: SimpleNamespace( load_skills=lambda *, enabled_only: list(skills), get_container_root=lambda: container_root, ), ) def test_skill_read_stamp_authorizes_declared_name_not_directory(tmp_path, monkeypatch): """[P2 regression] The stamp gate resolves the read path through the registry and authorizes the declared Skill.name — never the directory basename the path would suggest.""" from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware skill = _mismatched_skill(tmp_path) provider = _ActionAwareProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=True) _patch_registry_storage(monkeypatch, ToolErrorHandlingMiddleware, [skill]) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/vercel-deploy-claimable/SKILL.md") stamped = middleware._stamp_skill_read_metadata(message, request, tool_name="read_file") # The provider saw the declared name, and the read activates the skill. assert provider.sync_calls == ["vercel-deploy"] assert SKILL_CONTEXT_DENIED_KEY not in stamped.additional_kwargs assert SKILL_CONTEXT_ENTRY_KEY in stamped.additional_kwargs def test_skill_read_stamp_rbac_allowed_read_activates_mismatched_skill(tmp_path, monkeypatch): """[P2 regression] RBAC allow: ["vercel-deploy"] — a read of the advertised container path activates; allow: ["vercel-deploy-claimable"] (directory name) does not, because the decision is keyed by the declared name.""" from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware skill = _mismatched_skill(tmp_path) def _stamped_with(provider): resolved = _resolved_skill_authorization(provider, fail_closed=True) _patch_registry_storage(monkeypatch, ToolErrorHandlingMiddleware, [skill]) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/vercel-deploy-claimable/SKILL.md") return middleware._stamp_skill_read_metadata(message, request, tool_name="read_file") allowed = _stamped_with(_rbac_provider({"user": {"skills": {"allow": ["vercel-deploy"]}}})) assert SKILL_CONTEXT_ENTRY_KEY in allowed.additional_kwargs assert SKILL_CONTEXT_DENIED_KEY not in allowed.additional_kwargs denied = _stamped_with(_rbac_provider({"user": {"skills": {"allow": ["vercel-deploy-claimable"]}}})) assert SKILL_CONTEXT_ENTRY_KEY not in denied.additional_kwargs assert denied.additional_kwargs.get(SKILL_CONTEXT_DENIED_KEY) is True def test_async_stamp_path_authorizes_declared_name(tmp_path, monkeypatch): """[P2 regression] The async stamp path canonicalizes off-loop and awaits aauthorize() for the declared name — a loop-affine provider sees exactly that target, and an allowed read still yields an entry (not a fail-closed denial marker from a synchronous authorize() fallback).""" import asyncio from deerflow.agents.middlewares.skill_context import SKILL_CONTEXT_DENIED_KEY, SKILL_CONTEXT_ENTRY_KEY from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware skill = _mismatched_skill(tmp_path) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) _patch_registry_storage(monkeypatch, ToolErrorHandlingMiddleware, [skill]) middleware = ToolErrorHandlingMiddleware(app_config=_make_app_config(), skill_authorization=resolved) request, message = _read_call_and_message("/mnt/skills/public/vercel-deploy-claimable/SKILL.md") async def _run(): return await middleware._amaybe_stamp(message, request) stamped = asyncio.run(_run()) assert provider.async_calls == ["vercel-deploy"] assert SKILL_CONTEXT_ENTRY_KEY in stamped.additional_kwargs assert SKILL_CONTEXT_DENIED_KEY not in stamped.additional_kwargs def test_async_secret_binding_canonicalizes_persisted_entry_paths(tmp_path, monkeypatch): """[P2 regression] A persisted skill_context entry whose stamped name is path-derived (directory != declared Skill.name) is canonicalized through the registry before the async decision map is built: the loop-affine provider answers for the declared name and the entry's declared secrets still bind. A path-derived map key would miss and fall back to the sync API from the worker thread, failing closed and dropping the binding.""" import asyncio from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY skill = _mismatched_skill(tmp_path) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name="API_KEY", optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) storage = SimpleNamespace( load_skills=lambda *, enabled_only: [skill], get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"vercel-deploy"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) skill_path = posixpath_normpath(skill.get_container_file_path("/mnt/skills")) # The stamped entry name is the directory name — the historical shape. run_context = {"secrets": {"API_KEY": "secret-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="deploy the thing")], state={"skill_context": [{"name": "vercel-deploy-claimable", "path": skill_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) assert provider.async_calls == ["vercel-deploy"] assert "vercel-deploy-claimable" not in provider.async_calls assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"API_KEY": "secret-value"} def test_async_secret_binding_preserves_prepass_failure(tmp_path, monkeypatch): """[P2 regression, 4th instance of the sync-fallback class] A transient prepass registry failure must not let the secret-binding resolution silently recover with a fresh load: entry names resolved from a recovered registry would miss the (empty) decision map and fall back to the synchronous provider API — for a loop-affine provider a denial becomes a fail-open allow. Entries bind nothing for that call. The slash source is a run-scoped commitment validated at activation and does NOT consult the decision map — it must survive the entry prepass failure unharmed.""" import asyncio from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY def _skill(name: str, secret: str): from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / name skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text(f"# {name}", encoding="utf-8") skill = SkillObject( name=name, description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path(name), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name=secret, optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) return skill slash_skill = _skill("slash-skill", "SLASH_KEY") entry_skill = _skill("entry-skill", "ENTRY_KEY") # Load order: #1 call-1 prepass snapshot, #2 call-1 slash activation, # #3 call-1 fresh slash-source lookup, #4 call-2 prepass (FAILS), # #5 call-2 fresh slash-source lookup (recovers; entries bind nothing # from the sentinel, the slash binding survives). load_calls = {"count": 0} def _flaky_load(*, enabled_only): load_calls["count"] += 1 if load_calls["count"] == 4: raise RuntimeError("transient registry failure") return [slash_skill, entry_skill] storage = SimpleNamespace( load_skills=_flaky_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider(denied_activate={"entry-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=False) middleware = SkillActivationMiddleware( available_skills={"slash-skill", "entry-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) entry_path = posixpath_normpath(entry_skill.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"SLASH_KEY": "slash-value", "ENTRY_KEY": "entry-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) async def _identity(prepared): return prepared # Call 1: user slash-activates /slash-skill with the entry already in # skill_context. Prepass succeeds; the aauthorize-DENIED entry binds # nothing, the slash source binds. first = _Request( messages=[HumanMessage(content="/slash-skill go")], state={"skill_context": [{"name": "entry-skill", "path": entry_path}]}, runtime=SimpleNamespace(context=run_context), ) asyncio.run(middleware.awrap_model_call(first, _identity)) assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"SLASH_KEY": "slash-value"} # Call 2 (tool loop): the entry prepass fails transiently (#4). The slash # source must keep binding; the entry must bind nothing — a fresh-load # recovery would fall back to the sync API and fail-open allow it. second = _Request( messages=[HumanMessage(content="continue")], state={"skill_context": [{"name": "entry-skill", "path": entry_path}]}, runtime=SimpleNamespace(context=run_context), ) asyncio.run(middleware.awrap_model_call(second, _identity)) assert load_calls["count"] == 5 assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"SLASH_KEY": "slash-value"}, "entry must bind nothing after the prepass failure; the slash binding must survive it" def test_async_secret_binding_resolves_entries_against_prepass_snapshot(tmp_path, monkeypatch): """[P2 regression] Entry sources must resolve against the SAME registry snapshot the decision map was keyed by. If the resolution re-loaded the registry instead, a mid-step storage change (rename/uninstall) could resolve the path to a name absent from the map — the sync-API fallback divergence again, now without any load failure at all.""" import asyncio import dataclasses from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "entry-skill" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("# entry", encoding="utf-8") skill = SkillObject( name="entry-skill", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("entry-skill"), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name="ENTRY_KEY", optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) # What a FRESH load would return instead: the same container path now # declaring a different name (mid-step rename). renamed = dataclasses.replace(skill, name="renamed-skill") load_calls = {"count": 0} def _mutating_load(*, enabled_only): load_calls["count"] += 1 return [skill if load_calls["count"] == 1 else renamed] storage = SimpleNamespace( load_skills=_mutating_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) # aauthorize allows "entry-skill" (the snapshot name) and denies # "renamed-skill" (the fresh name); fail_closed so a wrong-name fallback # fails closed and drops the binding. provider = _AsyncOnlyProvider(denied_activate={"renamed-skill"}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"entry-skill", "renamed-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) entry_path = posixpath_normpath(skill.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"ENTRY_KEY": "entry-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="plain task")], state={"skill_context": [{"name": "entry-skill", "path": entry_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) assert provider.async_calls == ["entry-skill"] assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"ENTRY_KEY": "entry-value"}, "the entry must resolve against the prepass snapshot (allowed name), not a fresh load (denied name)" def test_async_model_call_without_skill_refs_skips_registry_scan(tmp_path, monkeypatch): """[P2 regression] An ordinary authorization-enabled async model step (no slash reference, no persisted entries, no supplied secrets) must not pay a full skill-tree scan — the previously passive path needed no registry I/O, and large or NFS-backed catalogs would add an uncached load_skills before every LLM call.""" import asyncio from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware load_calls = {"count": 0} def _counting_load(*, enabled_only): load_calls["count"] += 1 return [] storage = SimpleNamespace( load_skills=_counting_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"demo-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request(messages=[HumanMessage(content="an ordinary task, no slash reference")], state={}, runtime=None) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) assert load_calls["count"] == 0 assert provider.async_calls == [] def test_slash_secret_binding_uses_post_activation_registry(tmp_path, monkeypatch): """[P2 regression] The slash source's secret binding resolves from a fresh, post-activation registry — never the prepass snapshot. The snapshot is taken before the aauthorize awaits; a skill whose declared secrets change in that window would otherwise activate NEW_KEY content while binding OLD_KEY (the just-activated skill misses the credential it declares and receives one it no longer does).""" import asyncio import dataclasses from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "foo-skill" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: d\n---\n# foo", encoding="utf-8") def _skill(secret_name: str) -> SkillObject: skill = SkillObject( name="foo-skill", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("foo-skill"), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name=secret_name, optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) return skill old_version = _skill("OLD_KEY") new_version = dataclasses.replace(old_version, required_secrets=[SimpleNamespace(name="NEW_KEY", optional=False)]) # A persisted (unrelated) entry makes the async prepass actually snapshot # the registry; load #1 (prepass) sees the OLD declaration, the # activation-era loads (#2 _resolve_activation, #3 slash binding) see NEW. load_calls = {"count": 0} def _mutating_load(*, enabled_only): load_calls["count"] += 1 return [old_version] if load_calls["count"] == 1 else [new_version] storage = SimpleNamespace( load_skills=_mutating_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"foo-skill", "other-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) other_path = "/mnt/skills/custom/other-skill/SKILL.md" run_context: dict = {"secrets": {"OLD_KEY": "old-value", "NEW_KEY": "new-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="/foo-skill go")], state={"skill_context": [{"name": "other-skill", "path": other_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) # The activation read NEW; the binding must have followed it, not the # prepass snapshot: NEW_KEY injected, OLD_KEY not. The slash decision # itself came from aauthorize (not bypassed). assert provider.async_calls == ["foo-skill"] assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"NEW_KEY": "new-value"} def test_snapshot_entries_bind_while_slash_fresh_load_fails(tmp_path, monkeypatch): """The two secret sources fail independently, each against its own reference point: a transient failure of the fresh post-activation load binds nothing for the slash source while the entry sources keep binding from the prepass snapshot their decisions were keyed by.""" import asyncio from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory def _skill(name: str, secret: str) -> SkillObject: skill_dir = tmp_path / name skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text(f"# {name}", encoding="utf-8") skill = SkillObject( name=name, description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path(name), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name=secret, optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) return skill slash_skill = _skill("slash-skill", "SLASH_KEY") entry_skill = _skill("entry-skill", "ENTRY_KEY") # Load #1: prepass snapshot (both skills). Load #2: activation-era # (_resolve_activation). Load #3: fresh slash lookup — FAILS transiently. load_calls = {"count": 0} def _flaky_fresh_load(*, enabled_only): load_calls["count"] += 1 if load_calls["count"] == 3: raise RuntimeError("transient fresh-load failure") return [slash_skill, entry_skill] storage = SimpleNamespace( load_skills=_flaky_fresh_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"slash-skill", "entry-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) entry_path = posixpath_normpath(entry_skill.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"SLASH_KEY": "slash-value", "ENTRY_KEY": "entry-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="/slash-skill go")], state={"skill_context": [{"name": "entry-skill", "path": entry_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) assert load_calls["count"] == 3 # Both candidates were authorized via aauthorize (prepass succeeded); the # fresh-load failure then gates only the slash source's resolution. assert provider.async_calls == ["entry-skill", "slash-skill"] assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"ENTRY_KEY": "entry-value"} def test_slash_era_dominates_entry_source_for_same_skill(tmp_path, monkeypatch): """[construction-audit regression] When the same skill is BOTH the slash-activated source (resolved from the fresh, post-activation registry) and a persisted skill_context entry (resolved from the older prepass snapshot), only the activation-era declaration binds. Unioning both sources would inject a secret the skill no longer declares across the prepass-to-activation window — the same staleness shape the slash-fresh fix closed, arriving via the entry path. Mirrors the tool-policy rule: explicit slash activation dominates for the run.""" import asyncio import dataclasses from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "foo" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: d\n---\n# foo", encoding="utf-8") def _skill(secret: str) -> SkillObject: base = SkillObject( name="foo", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("foo"), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(base, "required_secrets", [SimpleNamespace(name=secret, optional=False)]) object.__setattr__(base, "secrets_autonomous", True) return dataclasses.replace(base, required_secrets=[SimpleNamespace(name=secret, optional=False)]) old_version = _skill("OLD_KEY") new_version = _skill("NEW_KEY") # Load #1 (prepass snapshot) sees the OLD declaration; the activation-era # loads (#2 activation, #3 fresh slash lookup) see NEW. load_calls = {"count": 0} def _mutating_load(*, enabled_only): load_calls["count"] += 1 return [old_version] if load_calls["count"] == 1 else [new_version] storage = SimpleNamespace( load_skills=_mutating_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"foo"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) foo_path = posixpath_normpath(old_version.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"OLD_KEY": "old-value", "NEW_KEY": "new-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="/foo go")], state={"skill_context": [{"name": "foo", "path": foo_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) # Only the activation-era declaration binds — not the union of both eras; # the slash decision was authorized via aauthorize. assert provider.async_calls == ["foo"] assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"NEW_KEY": "new-value"} def test_sync_chain_secret_binding_uses_sync_api_and_canonical_names(tmp_path, monkeypatch): """[Round-1 audit gap] The SYNC chain (wrap_model_call) with authorization enabled resolves entry sources against a fresh registry and consults the synchronous authorize() — the correct API there — with the canonical declared name. The async snapshot path has regressions; this branch had none, so a regression in the entry_registry=None fallback would pass the whole suite.""" from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "vercel-deploy-claimable" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: d\n---\n# x", encoding="utf-8") skill = SkillObject( name="vercel-deploy", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("vercel-deploy-claimable"), category=SkillCategory.PUBLIC, enabled=True, ) object.__setattr__(skill, "required_secrets", [SimpleNamespace(name="API_KEY", optional=False)]) object.__setattr__(skill, "secrets_autonomous", True) load_calls = {"count": 0} def _load(*, enabled_only): load_calls["count"] += 1 return [skill] storage = SimpleNamespace( load_skills=_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _ActionAwareProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"vercel-deploy"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) skill_path = posixpath_normpath(skill.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"API_KEY": "v"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="plain task")], state={"skill_context": [{"name": "vercel-deploy-claimable", "path": skill_path}]}, runtime=SimpleNamespace(context=run_context), ) def _handler(prepared): return prepared middleware.wrap_model_call(request, _handler) # Two legitimate sync-API calls on the sync chain: the render-decision # publication (path-keyed entry decisions) and the binding re-authorization. assert provider.sync_calls == ["vercel-deploy", "vercel-deploy"] assert provider.async_calls == [] assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"API_KEY": "v"} # Two loads on the sync chain with active entries: the render-decision # publication scan and the binding's fresh registry. assert load_calls["count"] == 2 def _durable_render_setup(tmp_path, monkeypatch, *, provider): """Composed activation(outer) -> durable(inner) pair over two skills.""" from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory made = [] for name in ("demo-skill", "ok-skill"): skill_dir = tmp_path / name skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text(f"# {name}", encoding="utf-8") made.append( SkillObject( name=name, description=f"Description for {name}", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path(name), category=SkillCategory.CUSTOM, enabled=True, ) ) demo_skill, ok_skill = made load_calls = {"count": 0} def _load(*, enabled_only): load_calls["count"] += 1 return made storage = SimpleNamespace( load_skills=_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) resolved = _resolved_skill_authorization(provider, fail_closed=True) token = "test-token" activation = SkillActivationMiddleware( available_skills={"demo-skill", "ok-skill"}, skill_authorization=resolved, slash_source_owner_token=token, ) durable = DurableContextMiddleware( skills_container_path="/mnt/skills", skill_authorization=resolved, entry_decisions_owner_token=token, ) entries = [ {"name": "demo-skill", "path": posixpath_normpath(demo_skill.get_container_file_path("/mnt/skills")), "description": "d", "loaded_at": 0}, {"name": "ok-skill", "path": posixpath_normpath(ok_skill.get_container_file_path("/mnt/skills")), "description": "o", "loaded_at": 1}, ] return activation, durable, entries, load_calls def _rendered_skill_names(prepared) -> set[str]: text = "\n".join(str(getattr(m, "content", "")) for m in prepared.messages) return {name for name in ("demo-skill", "ok-skill") if f"- {name}" in text} def test_durable_render_hides_denied_entries_async(tmp_path, monkeypatch): """The model-visible 'Active skills' reminder stops advertising skills the provider denies (async chain): the activation middleware publishes the per-step path-keyed decisions from its prepass (zero extra registry scans) and the durable renderer consumes them.""" import asyncio from langchain_core.messages import HumanMessage provider = _ActionAwareProvider(denied_activate={"demo-skill"}) activation, durable, entries, load_calls = _durable_render_setup(tmp_path, monkeypatch, provider=provider) class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="plain task")], state={"skill_context": entries}, runtime=SimpleNamespace(context={}), ) async def _inner(prepared): return await durable.awrap_model_call(prepared, _async_identity) async def _async_identity(prepared): # noqa: ARG001 return prepared prepared = asyncio.run(activation.awrap_model_call(request, _inner)) assert _rendered_skill_names(prepared) == {"ok-skill"} assert load_calls["count"] == 1, "the renderer must reuse the prepass publication, not scan" def test_durable_render_hides_denied_entries_sync(tmp_path, monkeypatch): """Same filter on the sync chain: the activation middleware's sync hook publishes the decisions (one scan + sync authorize per entry name).""" from langchain_core.messages import HumanMessage provider = _ActionAwareProvider(denied_activate={"demo-skill"}) activation, durable, entries, load_calls = _durable_render_setup(tmp_path, monkeypatch, provider=provider) class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="plain task")], state={"skill_context": entries}, runtime=SimpleNamespace(context={}), ) def _inner(prepared): return durable.wrap_model_call(prepared, _sync_identity) def _sync_identity(prepared): # noqa: ARG001 return prepared prepared = activation.wrap_model_call(request, _inner) assert _rendered_skill_names(prepared) == {"ok-skill"} assert load_calls["count"] == 1 def test_durable_render_unfiltered_without_authorization(tmp_path, monkeypatch): """Authorization disabled: nothing is published and the reminder renders every persisted entry — the absent-is-permissive carrier contract.""" import asyncio from langchain_core.messages import HumanMessage from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware activation, durable, entries, load_calls = _durable_render_setup(tmp_path, monkeypatch, provider=_ActionAwareProvider(denied_activate=set())) # Rebuild both without authorization, sharing the fixture's storage/skills. activation = SkillActivationMiddleware(available_skills={"demo-skill", "ok-skill"}, slash_source_owner_token="test-token") durable = DurableContextMiddleware(skills_container_path="/mnt/skills") class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="plain task")], state={"skill_context": entries}, runtime=SimpleNamespace(context={}), ) async def _inner(prepared): return await durable.awrap_model_call(prepared, _identity) async def _identity(prepared): return prepared prepared = asyncio.run(activation.awrap_model_call(request, _inner)) assert _rendered_skill_names(prepared) == {"demo-skill", "ok-skill"} def test_lead_chain_wires_durable_render_filter(): """The lead chain's DurableContextMiddleware receives the resolved skill authorization AND the same owner token as the chain's activation middleware, and is constructed AFTER it (publication must precede consumption in the wrap order). Dropping either silently disables the rendered-reminder filter — nothing else fails.""" from deerflow.agents.lead_agent.agent import build_middlewares from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middlewares = build_middlewares({"configurable": {}}, app_config=_make_app_config(), model_name="gpt-4", skill_authorization=resolved, user_id="user-123") activations = [m for m in middlewares if isinstance(m, SkillActivationMiddleware)] durables = [m for m in middlewares if isinstance(m, DurableContextMiddleware)] assert len(activations) == 1 and len(durables) == 1 assert durables[0]._skill_authorization is resolved assert durables[0]._entry_decisions_owner_token == activations[0]._slash_source_owner_token assert middlewares.index(activations[0]) < middlewares.index(durables[0]), "activation must wrap outside durable so the per-step decisions are published before the renderer reads them" def test_subagent_chain_wires_durable_render_filter(): """Same wiring contract for the subagent builder's DurableContextMiddleware.""" from deerflow.agents.middlewares.durable_context_middleware import DurableContextMiddleware from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares provider = _rbac_provider({"user": {"skills": {"allow": "*"}}}) resolved = _resolved_skill_authorization(provider, fail_closed=True) middlewares = build_subagent_runtime_middlewares(app_config=_make_app_config(), user_id="user-456", skill_authorization=resolved) activations = [m for m in middlewares if isinstance(m, SkillActivationMiddleware)] durables = [m for m in middlewares if isinstance(m, DurableContextMiddleware)] assert len(activations) == 1 and len(durables) == 1 assert durables[0]._skill_authorization is resolved assert durables[0]._entry_decisions_owner_token == activations[0]._slash_source_owner_token assert middlewares.index(activations[0]) < middlewares.index(durables[0]) def test_entry_decisions_carrier_is_redaction_listed(): """The per-step entry-decisions run-context carrier is stripped by ``redact_secret_context_keys`` — the redaction allowlist must enumerate every run-context key the middlewares add (paths and decisions are not secrets, but the carrier must never leak onto observable surfaces).""" from deerflow.runtime.secret_context import redact_secret_context_keys, write_skill_entry_decisions context: dict = {"secrets": {"K": "v"}, "__skill_entry_activation_decisions": {"decisions": {}, "owner_token": "t"}, "keep": 1} write_skill_entry_decisions(context, {"/mnt/skills/custom/foo/SKILL.md": True}, owner_token="t") redacted = redact_secret_context_keys(context) assert "__skill_entry_activation_decisions" not in redacted assert "secrets" not in redacted assert redacted.get("keep") == 1 # redacted, not suppressed: the source carrier itself is untouched. assert context["__skill_entry_activation_decisions"]["decisions"] == {"/mnt/skills/custom/foo/SKILL.md": True} def test_slash_dominance_holds_when_activation_era_declares_no_secrets(tmp_path, monkeypatch): """[P2 regression, R10] The same-skill exclusion must be derived from the authenticated slash activation identity, not only from successfully bound slash sources. When the prepass snapshots the OLD declaration (required-secrets: [OLD_KEY]) and the operator edits the skill to declare no secrets during the await window, the slash source binds nothing (required_secrets gate) — the stale entry view must STILL be suppressed, or a secret the just-activated skill no longer declares keeps injecting.""" import asyncio import dataclasses from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "foo" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: d\n---\n# foo", encoding="utf-8") base = SkillObject( name="foo", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("foo"), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(base, "secrets_autonomous", True) old_version = dataclasses.replace(base, required_secrets=[SimpleNamespace(name="OLD_KEY", optional=False)]) # The activation-era version declares NO secrets at all. new_version = dataclasses.replace(base, required_secrets=[]) # Load #1 (prepass snapshot) sees OLD; #2 (activation) and #3 (fresh # slash lookup) see the no-secrets version. load_calls = {"count": 0} def _mutating_load(*, enabled_only): load_calls["count"] += 1 return [old_version] if load_calls["count"] == 1 else [new_version] storage = SimpleNamespace( load_skills=_mutating_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"foo"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) foo_path = posixpath_normpath(old_version.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"OLD_KEY": "old-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) request = _Request( messages=[HumanMessage(content="/foo go")], state={"skill_context": [{"name": "foo", "path": foo_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) # The slash-activated skill declares no secrets now; its stale entry view # must not inject OLD_KEY. assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) is None def test_slash_dominance_anchors_on_path_across_midrun_rename(tmp_path, monkeypatch): """[R10-fix audit regression] The same-skill exclusion must anchor on the authenticated identity PATH, not the declared name: when the operator renames the skill (same path, new declared name) during the await window and the user activates by the NEW name, a name-anchored exclusion set (holding only the new name) lets the stale entry — resolving the same path to the OLD name in the snapshot — bind two eras of declarations.""" import asyncio import dataclasses from langchain_core.messages import HumanMessage from deerflow.agents.middlewares import skill_activation_middleware as activation_module from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.runtime.secret_context import ACTIVE_SECRETS_CONTEXT_KEY from deerflow.skills.types import Skill as SkillObject from deerflow.skills.types import SkillCategory skill_dir = tmp_path / "foo" skill_dir.mkdir(exist_ok=True) (skill_dir / "SKILL.md").write_text("---\ndescription: d\n---\n# foo", encoding="utf-8") base = SkillObject( name="foo", description="d", license=None, skill_dir=skill_dir, skill_file=skill_dir / "SKILL.md", relative_path=Path("foo"), category=SkillCategory.CUSTOM, enabled=True, ) object.__setattr__(base, "secrets_autonomous", True) old_version = dataclasses.replace(base, required_secrets=[SimpleNamespace(name="OLD_KEY", optional=False)]) renamed_version = dataclasses.replace(base, name="bar", required_secrets=[SimpleNamespace(name="NEW_KEY", optional=False)]) load_calls = {"count": 0} def _mutating_load(*, enabled_only): load_calls["count"] += 1 return [old_version] if load_calls["count"] == 1 else [renamed_version] storage = SimpleNamespace( load_skills=_mutating_load, get_container_root=lambda: "/mnt/skills", get_skills_root_path=lambda: tmp_path, ) monkeypatch.setattr(activation_module, "get_or_new_skill_storage", lambda **kw: storage) provider = _AsyncOnlyProvider() resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"foo", "bar"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) identity_path = posixpath_normpath(old_version.get_container_file_path("/mnt/skills")) run_context: dict = {"secrets": {"OLD_KEY": "old-value", "NEW_KEY": "new-value"}} class _Request(SimpleNamespace): def override(self, **kwargs): updates = dict(self.__dict__) updates.update(kwargs) return _Request(**updates) # The user activates by the NEW name; the persisted entry is the OLD read # of the same path. request = _Request( messages=[HumanMessage(content="/bar go")], state={"skill_context": [{"name": "foo", "path": identity_path}]}, runtime=SimpleNamespace(context=run_context), ) async def _identity(prepared): return prepared asyncio.run(middleware.awrap_model_call(request, _identity)) # Only the renamed (activation-era) declaration binds — no union of eras. assert run_context.get(ACTIVE_SECRETS_CONTEXT_KEY) == {"NEW_KEY": "new-value"} def test_async_batch_miss_never_calls_synchronous_provider(): """[construction hardening] A name missing from an async batch resolves per the provider-error policy WITHOUT consulting the synchronous authorize() — the wrong API from a worker thread. Under the old dict contract the miss silently took that path; with a loop-affine provider and fail_closed=false it flipped denials into allows.""" from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.authz.activation_decisions import ActivationDecisions provider = _ActionAwareProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=False) middleware = SkillActivationMiddleware( available_skills={"demo-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) empty_batch = ActivationDecisions({}, fail_closed=False) # Miss under fail-open allows — and the sync API is never consulted # (the old fallback would have recorded a sync call). assert middleware._activation_allowed("uncovered", activation_decisions=empty_batch) is True assert middleware._activation_allowed("demo-skill", activation_decisions=empty_batch) is True assert provider.sync_calls == [] assert provider.async_calls == [] def test_async_batch_miss_fails_closed_under_default_policy(): """The production default (fail_closed=True) denies a miss — loudly, and still without touching the synchronous provider.""" from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware from deerflow.authz.activation_decisions import ActivationDecisions provider = _ActionAwareProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=True) middleware = SkillActivationMiddleware( available_skills={"demo-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) empty_batch = ActivationDecisions({}, fail_closed=True) assert middleware._activation_allowed("uncovered", activation_decisions=empty_batch) is False assert provider.sync_calls == [] def test_sync_chain_none_still_uses_synchronous_api(): """``None`` remains the sync-chain signal: the synchronous authorize() is the correct API there and stays in use.""" provider = _ActionAwareProvider(denied_activate=set()) resolved = _resolved_skill_authorization(provider, fail_closed=True) from deerflow.agents.middlewares.skill_activation_middleware import SkillActivationMiddleware middleware = SkillActivationMiddleware( available_skills={"demo-skill"}, skill_authorization=resolved, slash_source_owner_token="test-token", ) assert middleware._activation_allowed("demo-skill", activation_decisions=None) is True assert provider.sync_calls == ["demo-skill"] def posixpath_normpath(path: str) -> str: import posixpath return posixpath.normpath(path) def is_slash_activation_reminder(message) -> bool: from deerflow.agents.middlewares.skill_activation_middleware import is_slash_skill_activation_reminder return is_slash_skill_activation_reminder(message)