Kronecker/NotesAgentic
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 5 Wiki Activity

fix(extension): 收紧插件命令运行时契约

Browse Source
This commit is contained in:
admin
2026-09-02 20:03:39 +08:00
parent eb3464b522
commit d39ae727c1
10 changed files with 466 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_extension_core.py
+119 -3
View File
@@ -17,7 +17,10 @@ from app.contracts import (
)
from app.extensions import ExtensionError
from app.extensions.mcp import McpStdioClient
from app.extensions.runtime import _arguments_model_from_schema
from app.extensions.runtime import (
_arguments_model_from_schema,
_validate_mcp_command_target_schema,
)
from app.services import note_service
from app.config import BACKEND_DIR, get_settings
@@ -428,10 +431,9 @@ def test_mcp_command_target_receives_scoped_context_and_declared_secret(
)
assert result.effect.type == "notification"
assert result.effect.payload == {
assert result.effect.payload.model_dump() == {
"level": "success",
"message": "Fixture: 来自选区",
"secret_configured": True,
}
assert "mcp-command-secret" not in repr(
mcp_container.plugins.commands.audit_events()
@@ -440,6 +442,120 @@ def test_mcp_command_target_receives_scoped_context_and_declared_secret(
run(scenario())
def test_mcp_command_target_rejects_incompatible_envelope_schema(tmp_path) -> None:
package = tmp_path / "mcp-bad-command"
shutil.copytree(MCP_FIXTURE, package)
for filename in ("plugin.yaml", "commands.yaml", "settings.yaml"):
path = package / filename
path.write_text(
path.read_text(encoding="utf-8").replace(
"mcp-fixture", "mcp-bad-command"
),
encoding="utf-8",
)
server_path = package / "server.py"
server_path.write_text(
server_path.read_text(encoding="utf-8").replace(
'{"_notesagent": {"type": "object"}}',
'{"unexpected": {"type": "string"}}',
),
encoding="utf-8",
)
container = build_container()
container.plugins.install(package)
container.plugins.set_permissions(
"mcp-bad-command", ["notes.read", "secrets.use"]
)
try:
with pytest.raises(ExtensionError) as exc:
container.plugins.enable("mcp-bad-command")
assert exc.value.code == "PLUGIN_CONTRIBUTION_INVALID"
assert container.plugins.get("mcp-bad-command").status == "error"
finally:
container.plugins.shutdown()
def test_mcp_command_target_enable_check_only_requires_protocol_marker() -> None:
# `not`/`oneOf` 等完整语义由实际调用前的官方 Validator 处理；启用检查
# 只确认不可被引用或组合隐藏的稳定宿主入口，避免维护不完整的求解器。
_validate_mcp_command_target_schema(
{
"type": "object",
"properties": {
"_notesagent": {
"type": "object",
"not": {"type": "object"},
}
},
},
"marker.run",
)
invalid_markers = [
{
"$defs": {"envelope": {"type": "object"}},
"properties": {"_notesagent": {"$ref": "#/$defs/envelope"}},
},
{
"allOf": [
{"properties": {"_notesagent": {"type": "object"}}},
]
},
]
for schema in invalid_markers:
with pytest.raises(ExtensionError) as exc:
_validate_mcp_command_target_schema(schema, "marker.run")
assert exc.value.code == "PLUGIN_CONTRIBUTION_INVALID"
def test_mcp_command_validates_actual_envelope_before_call(tmp_path) -> None:
package = tmp_path / "mcp-runtime-schema"
shutil.copytree(MCP_FIXTURE, package)
for filename in ("plugin.yaml", "commands.yaml", "settings.yaml"):
path = package / filename
path.write_text(
path.read_text(encoding="utf-8").replace(
"mcp-fixture", "mcp-runtime-schema"
),
encoding="utf-8",
)
server_path = package / "server.py"
server_path.write_text(
server_path.read_text(encoding="utf-8").replace(
'{"_notesagent": {"type": "object"}}',
'{"_notesagent": {"type": "object", "properties": '
'{"arguments": {"type": "object", "maxProperties": 0}, '
'"context": {"type": "object", "properties": '
'{"selection": {"type": "string"}}, "required": ["selection"]}}, '
'"required": ["arguments", "context"]}}',
),
encoding="utf-8",
)
container = build_container()
container.plugins.install(package)
container.plugins.set_permissions(
"mcp-runtime-schema", ["notes.read", "secrets.use"]
)
try:
# context.selection 是 Command 的 when/context 契约保证的真实字段；
# 启用期结构检查不得因没有伪造该业务值而拒绝目标 Schema。
container.plugins.enable("mcp-runtime-schema")
container.plugins.put_setting_secret(
"mcp-runtime-schema", "api_key", "configured"
)
with pytest.raises(ExtensionError) as exc:
run(
container.plugins.execute_command(
"mcp-runtime-schema.notify",
{"message": "must be rejected locally"},
PluginCommandContext(selection="visible"),
)
)
assert exc.value.code == "PLUGIN_COMMAND_TARGET_SCHEMA_MISMATCH"
finally:
container.plugins.shutdown()
def test_agent_calls_mcp_tool_through_registry_and_writes_trace(mcp_container) -> None:
async def scenario() -> None:
mcp_container.plugins.enable("mcp-fixture")
backend/tests/test_plugin_contributions.py
+89 -3
View File
@@ -3,6 +3,7 @@ import json
from pathlib import Path
import pytest
from pydantic import TypeAdapter, ValidationError
from app.agent import ToolRegistry
from app.config import BACKEND_DIR, get_settings
@@ -10,6 +11,7 @@ from app.container import build_container
from app.contracts import (
PluginCommandContext,
PluginCommandEffect,
PluginNoEffect,
PluginSettingType,
)
from app.extensions import ExtensionError, PluginRuntime
@@ -70,7 +72,40 @@ def test_command_executes_with_scoped_context_and_settings() -> None:
assert result.status == "completed"
assert result.effect.type == "notification"
assert result.effect.payload == {"level": "success", "message": "ABCD"}
assert result.effect.payload.model_dump() == {
"level": "success",
"message": "ABCD",
}
def test_echo_command_returns_none_for_empty_message() -> None:
host = DeclarativePluginHost()
empty = run(host.execute_command("echo", {}, {}, {}, lambda _: None))
populated = run(
host.execute_command("echo", {"message": "hello"}, {}, {}, lambda _: None)
)
assert isinstance(empty, PluginNoEffect)
assert populated.type == "notification"
assert populated.payload.message == "hello"
@pytest.mark.parametrize(
("effect_type", "payload"),
[
("none", {"unexpected": True}),
("notification", {"level": "debug", "message": "invalid"}),
("navigate", {"route": "https://example.com"}),
("refresh", {"scope": "everything"}),
("job", {"job_id": "invalid job id"}),
],
)
def test_command_effect_rejects_untrusted_payloads(effect_type, payload) -> None:
with pytest.raises(ValidationError):
TypeAdapter(PluginCommandEffect).validate_python(
{"type": effect_type, "payload": payload}
)
def test_command_rejects_missing_context_and_invalid_arguments() -> None:
@@ -112,7 +147,7 @@ def test_command_only_receives_declared_context() -> None:
self, handler, arguments, context, settings, resolve_secret
):
self.context = context
return PluginCommandEffect(type="none")
return PluginNoEffect()
host = CapturingHost()
runtime = PluginRuntime(ToolRegistry(), host=host)
@@ -146,7 +181,7 @@ def test_command_resolves_only_declared_plugin_secrets(tmp_path: Path) -> None:
resolve_secret("undeclared")
except ExtensionError as exc:
self.denied_code = exc.code
return PluginCommandEffect(type="none")
return PluginNoEffect()
host = SecretHost()
runtime = PluginRuntime(ToolRegistry(), host=host)
@@ -248,6 +283,57 @@ def test_settings_update_validates_version_type_bounds_and_secret_boundary() ->
assert exc.value.code == code
def test_required_plain_setting_blocks_enable_until_configured(tmp_path: Path) -> None:
package = tmp_path / "required-setting"
package.mkdir()
(package / "plugin.yaml").write_text(
"""
id: required-setting
name: Required Setting
version: 1.0.0
contributes:
commands: [required-setting.run]
settings_sections: [required-setting.general]
backend:
type: internal_rpc
transport: none
""".strip(),
encoding="utf-8",
)
(package / "commands.yaml").write_text(
"""
commands:
- command_id: required-setting.run
title: Required Setting
locations: [command_palette]
handler: echo
""".strip(),
encoding="utf-8",
)
(package / "settings.yaml").write_text(
"""
section_id: required-setting.general
schema_version: 1
fields:
- key: endpoint
label: Endpoint
type: string
required: true
""".strip(),
encoding="utf-8",
)
runtime = PluginRuntime(ToolRegistry())
runtime.install(package)
with pytest.raises(ExtensionError) as exc:
runtime.enable("required-setting")
assert exc.value.code == "PLUGIN_SETTINGS_REQUIRED"
assert runtime.get("required-setting").status == "installed"
runtime.update_settings("required-setting", 1, {"endpoint": "local"})
assert runtime.enable("required-setting").status == "ready"
def test_secret_roundtrip_never_enters_plain_settings_storage() -> None:
container = build_container()
plaintext = "stage-d-secret-value"