mirror of
https://github.com/qelectrotech/qelectrotech-source-mirror.git
synced 2026-09-29 05:44:14 +02:00
Merge pull request #1129 from ispyisail/fix/qet-mcp-pin-binary
Fix the MCP server running any program an assistant names
This commit is contained in:
+28
-11
@@ -69,6 +69,7 @@ Register it with an MCP client, for example:
|
||||
"args": ["/path/to/qelectrotech/misc/qet-mcp/qet_mcp.py"],
|
||||
"env": {
|
||||
"QET_MCP_WORKSPACE": "/home/you/drawings",
|
||||
"QET_BINARY": "/usr/bin/qelectrotech",
|
||||
"QET_ENABLE_SCRIPTING": "1"
|
||||
}
|
||||
}
|
||||
@@ -76,6 +77,11 @@ Register it with an MCP client, for example:
|
||||
}
|
||||
```
|
||||
|
||||
`QET_BINARY` is the QElectroTech the tools launch. Leave it out when
|
||||
`qelectrotech` is on your `PATH`, or when the server is installed with
|
||||
QElectroTech (as `<prefix>/share/qelectrotech/mcp/qet_mcp.py`, which also
|
||||
finds the installed element collection).
|
||||
|
||||
## Using it from the Claude app
|
||||
|
||||
A web chat in a browser cannot start a program on your computer, so it
|
||||
@@ -106,9 +112,9 @@ and it uses the same account as the website.
|
||||
Windows path is written `\\`.
|
||||
4. Quit the app completely and start it again. The tools appear under the
|
||||
chat box's tools menu.
|
||||
5. In the chat, say where your `qelectrotech` executable is. `qet_edit` and
|
||||
`qet_export` take it as an argument on every call; the other tools do not
|
||||
need it.
|
||||
5. If `qelectrotech` is not on your `PATH`, add `"QET_BINARY"` to the
|
||||
`env` block with the full path to the executable
|
||||
(`C:\\Program Files\\...\\qelectrotech.exe`, written with `\\`).
|
||||
|
||||
Only files under `QET_MCP_WORKSPACE` can be read or written (see
|
||||
[What the server is allowed to touch](#what-the-server-is-allowed-to-touch)).
|
||||
@@ -179,11 +185,22 @@ Set the workspace to the folder your drawings live in. A path outside it is
|
||||
refused with an error naming what was allowed; symlinks are resolved first,
|
||||
so a link planted inside the workspace is judged by where it points.
|
||||
|
||||
Two arguments are deliberately **not** confined: `binary` (the
|
||||
`qelectrotech` executable) and `elements_dir` (the element collection).
|
||||
Those are configuration, chosen once by whoever runs the server, and both
|
||||
normally live in `/usr` or a build tree — outside any sensible workspace.
|
||||
Confining them would reject the ordinary case while stopping nothing.
|
||||
**The client does not choose what program runs.** The tools that launch
|
||||
QElectroTech use the one the server found (`QET_BINARY`, the install it
|
||||
ships in, or `PATH`). A call may still name `binary`, but only as that same
|
||||
file or one listed by whoever configured the server:
|
||||
|
||||
| | |
|
||||
|---|---|
|
||||
| `QET_BINARY` | the QElectroTech to launch |
|
||||
| `QET_MCP_BINARIES` | other executables a call may name, separated like `QET_MCP_WORKSPACE` (for comparing two builds) |
|
||||
| `QET_MCP_ALLOW_ANY_BINARY=1` | turns the check off: a call can then run any program |
|
||||
| `QET_MCP_ELEMENTS` | element collections a call may name as `elements_dir` besides the workspace and the installed one |
|
||||
|
||||
Anything else is refused, even a file inside the workspace: being there
|
||||
makes it readable, not runnable. Before this rule any executable a call
|
||||
named was run, with the call's own paths as arguments, so text inside a
|
||||
project could steer an assistant into starting another program.
|
||||
|
||||
`QET_MCP_ALLOW_ANY_PATH=1` is equivalent to granting the client local
|
||||
filesystem access with this process's privileges. It exists so that is a
|
||||
@@ -221,7 +238,6 @@ the answer a screenshot gave wrongly.
|
||||
|
||||
```json
|
||||
{"name": "qet_edit", "arguments": {
|
||||
"binary": "/path/to/qelectrotech",
|
||||
"project": "in.qet", "output": "out.qet",
|
||||
"elements_dir": "/path/to/qelectrotech/elements",
|
||||
"operations": [
|
||||
@@ -277,7 +293,7 @@ to emit.
|
||||
|
||||
```json
|
||||
{"name": "qet_query", "arguments": {
|
||||
"binary": "/path/to/qelectrotech", "project": "industrial.qet",
|
||||
"project": "industrial.qet",
|
||||
"sql": "SELECT label, COUNT(*) AS n FROM element_nomenclature_view WHERE label <> '' GROUP BY label HAVING n > 1 ORDER BY n DESC"}}
|
||||
```
|
||||
|
||||
@@ -420,7 +436,8 @@ Python, plus the hang guard on `addConductor` and the database refresh in
|
||||
changes nothing. `addElement` and the move/delete verbs shipped with the
|
||||
scripting API; `addConductor`, `rotateElement`, `setElementLabel`,
|
||||
`setElementInfo` and `setFolioTitle` are newer.
|
||||
- **`elements_dir` is not optional for `common://` paths.** The sandboxed
|
||||
- **`elements_dir` is not optional for `common://` paths** unless the
|
||||
server is installed with QElectroTech, which fills it in. The sandboxed
|
||||
run has its own empty HOME, so QElectroTech falls back to the compiled-in
|
||||
collection path, which on a machine that never ran `make install` does not
|
||||
exist. The only symptom is `addElement` reporting that a file plainly
|
||||
|
||||
+138
-22
@@ -2622,7 +2622,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"project": {"type": "string"},
|
||||
"format": {"type": "string", "enum": sorted(EXPORT_FORMATS)},
|
||||
"output": {"type": "string"},
|
||||
@@ -2631,7 +2631,7 @@ TOOLS = [
|
||||
"without this an existing file is never clobbered"},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "project", "format", "output"],
|
||||
"required": ["project", "format", "output"],
|
||||
},
|
||||
"handler": lambda a: tool_export(a["binary"], a["project"], a["format"],
|
||||
a["output"], a.get("timeout", 180)),
|
||||
@@ -2649,7 +2649,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"project": {"type": "string", "description": "the .qet to start from; not modified"},
|
||||
"output": {"type": "string", "description": "where to write the edited project"},
|
||||
"overwrite": {"type": "boolean", "default": False,
|
||||
@@ -2867,7 +2867,7 @@ TOOLS = [
|
||||
},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "project", "output", "operations"],
|
||||
"required": ["project", "output", "operations"],
|
||||
},
|
||||
"handler": lambda a: tool_edit(a["binary"], a["project"], a["operations"],
|
||||
a["output"], a.get("elements_dir"),
|
||||
@@ -2885,7 +2885,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"project": {"type": "string", "description": "the .qet to query; never modified"},
|
||||
"sql": {"type": "string",
|
||||
"description": "a single SELECT or WITH...SELECT. "
|
||||
@@ -2893,7 +2893,7 @@ TOOLS = [
|
||||
"elements_dir": {"type": "string"},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "project"],
|
||||
"required": ["project"],
|
||||
},
|
||||
"handler": lambda a: tool_query(a["binary"], a["project"], a.get("sql", ""),
|
||||
a.get("elements_dir"), a.get("timeout", 180)),
|
||||
@@ -2920,7 +2920,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"project": {"type": "string", "description": "the .qet to check; never modified"},
|
||||
"folio": {"type": "integer", "description":
|
||||
"check one folio only; omit for the whole project. An index "
|
||||
@@ -2931,7 +2931,7 @@ TOOLS = [
|
||||
"elements_dir": {"type": "string"},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "project"],
|
||||
"required": ["project"],
|
||||
},
|
||||
"handler": lambda a: tool_continuity(a["binary"], a["project"], a.get("folio"),
|
||||
a.get("elements_dir"), a.get("timeout", 180)),
|
||||
@@ -2947,7 +2947,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"output": {"type": "string", "description": "where to write the new .qet"},
|
||||
"title": {"type": "string", "description": "the project title"},
|
||||
"folios": {"description": "how many empty folios, or a list of folio titles",
|
||||
@@ -2959,7 +2959,7 @@ TOOLS = [
|
||||
"elements_dir": {"type": "string"},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "output", "title"],
|
||||
"required": ["output", "title"],
|
||||
},
|
||||
"handler": lambda a: tool_project_new(
|
||||
a["binary"], a["output"], a["title"], a.get("folios", 1), a.get("author", ""),
|
||||
@@ -3008,7 +3008,7 @@ TOOLS = [
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"binary": {"type": "string", "description": "path to the qelectrotech executable"},
|
||||
"binary": {"type": "string", "description": "the qelectrotech executable; leave it out to use the one this server is configured with. Any other is refused unless its configuration allows it"},
|
||||
"project": {"type": "string"},
|
||||
"checks": {"type": "array", "items": {"type": "string", "enum": sorted(CHECKS)},
|
||||
"description": "which checks to run; omit for all"},
|
||||
@@ -3017,7 +3017,7 @@ TOOLS = [
|
||||
"elements_dir": {"type": "string"},
|
||||
"timeout": {"type": "integer", "default": 180},
|
||||
},
|
||||
"required": ["binary", "project"],
|
||||
"required": ["project"],
|
||||
},
|
||||
"handler": lambda a: tool_check(a["binary"], a["project"], a.get("checks"),
|
||||
a.get("sample", 10), a.get("elements_dir"),
|
||||
@@ -3097,17 +3097,21 @@ _BY_NAME = {t["name"]: t for t in TOOLS}
|
||||
# sandboxed HOME each QElectroTech launch gets isolates *settings*, not the
|
||||
# filesystem.
|
||||
#
|
||||
# So data paths are confined to a workspace. Two kinds of path are treated
|
||||
# differently, deliberately:
|
||||
# So data paths are confined to a workspace, and the program the server
|
||||
# launches is not the client's to choose:
|
||||
#
|
||||
# data chosen by the client per call -- the projects, directories,
|
||||
# images and outputs below. Confined.
|
||||
# configuration chosen once by whoever runs the server -- "binary" (the
|
||||
# qelectrotech executable) and "elements_dir" (the element
|
||||
# collection). Both normally live in /usr or a build tree,
|
||||
# i.e. outside any sane workspace, so confining them would
|
||||
# reject the ordinary case while stopping nothing: they are
|
||||
# not where a model gets to point the server at /etc.
|
||||
# images and outputs below. Confined to the workspace.
|
||||
# executable "binary". Resolved by the server itself (resolve_binary());
|
||||
# a client may name it only when it is that same file or one
|
||||
# whoever configured the server listed in QET_MCP_BINARIES.
|
||||
# It once counted as configuration and went unchecked, but it
|
||||
# is a per-call argument: a model steered by text in a
|
||||
# project could run any program on the machine with it.
|
||||
# collection "elements_dir". Normally outside the workspace (in /usr or
|
||||
# a build tree), so allowed there, in the collection of the
|
||||
# resolved install, or in a directory listed in
|
||||
# QET_MCP_ELEMENTS.
|
||||
#
|
||||
# Enforced here, at the dispatcher, because this is the trust boundary --
|
||||
# the point where model-supplied arguments enter. Calling the tool_* helpers
|
||||
@@ -3131,6 +3135,10 @@ _DATA_PATHS = {
|
||||
"qet_element_build": {"write": ("output",)},
|
||||
}
|
||||
|
||||
# Tools that launch QElectroTech, and so take "binary" and "elements_dir".
|
||||
_LAUNCHES_QET = {"qet_export", "qet_edit", "qet_query", "qet_continuity",
|
||||
"qet_check", "qet_project_new"}
|
||||
|
||||
# qet_edit operations that name a file of their own.
|
||||
_DATA_PATH_OPS = {"add_image": "file", "add_pdf_page": "file"}
|
||||
|
||||
@@ -3169,6 +3177,106 @@ def _within_workspace(path: Path, roots: list) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _env_paths(name: str) -> list:
|
||||
"""An os.pathsep-separated list of paths from the environment, resolved."""
|
||||
out = []
|
||||
for part in os.environ.get(name, "").split(os.pathsep):
|
||||
if part.strip():
|
||||
try:
|
||||
out.append(Path(part).expanduser().resolve())
|
||||
except OSError:
|
||||
continue
|
||||
return out
|
||||
|
||||
|
||||
def _installed_prefix() -> Path | None:
|
||||
"""The install prefix when this script is <prefix>/share/qelectrotech/mcp/."""
|
||||
here = Path(__file__).resolve().parent
|
||||
if here.name == "mcp" and here.parent.name == "qelectrotech" \
|
||||
and here.parent.parent.name == "share":
|
||||
return here.parent.parent.parent
|
||||
return None
|
||||
|
||||
|
||||
def resolve_binary() -> Path | None:
|
||||
"""The QElectroTech this server launches, found without asking the client.
|
||||
|
||||
QET_BINARY first, then the install this script ships in, then
|
||||
qelectrotech on PATH. None when there is none; the tools that launch
|
||||
QElectroTech then say how to set it.
|
||||
"""
|
||||
env = os.environ.get("QET_BINARY", "").strip()
|
||||
if env:
|
||||
return Path(env).expanduser().resolve()
|
||||
prefix = _installed_prefix()
|
||||
if prefix is not None:
|
||||
for name in ("qelectrotech", "qelectrotech.exe"):
|
||||
cand = prefix / "bin" / name
|
||||
if cand.is_file():
|
||||
return cand.resolve()
|
||||
found = shutil.which("qelectrotech")
|
||||
return Path(found).resolve() if found else None
|
||||
|
||||
|
||||
def default_elements_dir() -> Path | None:
|
||||
"""The element collection of the install this script ships in, if any."""
|
||||
prefix = _installed_prefix()
|
||||
if prefix is not None:
|
||||
coll = prefix / "share" / "qelectrotech" / "elements"
|
||||
if coll.is_dir():
|
||||
return coll.resolve()
|
||||
return None
|
||||
|
||||
|
||||
def _check_binary(arguments: dict) -> None:
|
||||
"""Fill in "binary", or refuse one that is not the server's own choice."""
|
||||
if os.environ.get("QET_MCP_ALLOW_ANY_BINARY") == "1" and arguments.get("binary"):
|
||||
return
|
||||
default = resolve_binary()
|
||||
raw = arguments.get("binary")
|
||||
if not raw:
|
||||
if default is None:
|
||||
raise ValueError(
|
||||
"no QElectroTech found: set QET_BINARY to the qelectrotech "
|
||||
"executable in the environment this server is started in")
|
||||
arguments["binary"] = str(default)
|
||||
return
|
||||
if not isinstance(raw, str):
|
||||
raise ValueError("'binary' must be a path")
|
||||
given = Path(raw).expanduser().resolve()
|
||||
allowed = ([default] if default else []) + _env_paths("QET_MCP_BINARIES")
|
||||
if given not in allowed:
|
||||
raise ValueError(
|
||||
f"'binary' is not an allowed QElectroTech: {given}. Leave it out "
|
||||
"to use " + (str(default) if default else "QET_BINARY")
|
||||
+ "; whoever configured this server can list others in "
|
||||
"QET_MCP_BINARIES, or set QET_MCP_ALLOW_ANY_BINARY=1 to "
|
||||
"disable this check (which lets the client run any program).")
|
||||
arguments["binary"] = str(given)
|
||||
|
||||
|
||||
def _check_elements_dir(arguments: dict, roots: list) -> None:
|
||||
"""Fill in "elements_dir" from the install, or confine a given one."""
|
||||
raw = arguments.get("elements_dir")
|
||||
if not raw:
|
||||
default = default_elements_dir()
|
||||
if default is not None:
|
||||
arguments["elements_dir"] = str(default)
|
||||
return
|
||||
if not isinstance(raw, str):
|
||||
raise ValueError("'elements_dir' must be a path")
|
||||
given = Path(raw).expanduser().resolve()
|
||||
extra = _env_paths("QET_MCP_ELEMENTS")
|
||||
default = default_elements_dir()
|
||||
if default is not None:
|
||||
extra.append(default)
|
||||
if roots and not _within_workspace(given, roots + extra):
|
||||
raise ValueError(
|
||||
f"'elements_dir' is outside the workspace: {given}. Leave it out "
|
||||
"to use the installed collection, or list the directory in "
|
||||
"QET_MCP_ELEMENTS.")
|
||||
|
||||
|
||||
def _check_path(raw, arg: str, mode: str, roots: list) -> Path:
|
||||
"""Resolve one path and refuse it if it leaves the workspace.
|
||||
|
||||
@@ -3192,12 +3300,20 @@ def _check_path(raw, arg: str, mode: str, roots: list) -> Path:
|
||||
|
||||
|
||||
def enforce_path_policy(tool_name: str, arguments: dict) -> None:
|
||||
"""Apply the workspace and overwrite policy to one tool call."""
|
||||
"""Apply the workspace, executable and overwrite policy to one tool call.
|
||||
|
||||
For a tool that launches QElectroTech this also fills in "binary" and,
|
||||
when the install has one, "elements_dir", so a client need not know them.
|
||||
"""
|
||||
spec = _DATA_PATHS.get(tool_name)
|
||||
if spec is None:
|
||||
return
|
||||
roots = workspace_roots()
|
||||
|
||||
if tool_name in _LAUNCHES_QET:
|
||||
_check_binary(arguments)
|
||||
_check_elements_dir(arguments, roots)
|
||||
|
||||
for arg in spec.get("read", ()):
|
||||
if arg in arguments:
|
||||
_check_path(arguments[arg], arg, "read", roots)
|
||||
|
||||
+155
-17
@@ -74,6 +74,16 @@ PLC_MASTER = "common://plc_master_test.elmt"
|
||||
PLC_SLAVE = "common://plc_slave_test.elmt"
|
||||
|
||||
|
||||
def fake_qet(bindir: Path, name: str = "qelectrotech") -> Path:
|
||||
"""An executable file standing in for QElectroTech. The policy checks
|
||||
which file it is, never runs it."""
|
||||
bindir.mkdir(parents=True, exist_ok=True)
|
||||
exe = bindir / name
|
||||
exe.write_text("#!/bin/sh\nexit 0\n")
|
||||
exe.chmod(0o755)
|
||||
return exe
|
||||
|
||||
|
||||
def png(path: Path) -> None:
|
||||
"""A real 64x32 PNG from the standard library, so no imaging dependency."""
|
||||
import struct
|
||||
@@ -1767,6 +1777,8 @@ class PathPolicy(unittest.TestCase):
|
||||
self._saved = dict(os.environ)
|
||||
os.environ["QET_MCP_WORKSPACE"] = str(self.root)
|
||||
os.environ.pop("QET_MCP_ALLOW_ANY_PATH", None)
|
||||
self.qet = fake_qet(Path(self.tmp.name) / "bin")
|
||||
os.environ["QET_BINARY"] = str(self.qet)
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.clear()
|
||||
@@ -1800,7 +1812,7 @@ class PathPolicy(unittest.TestCase):
|
||||
def test_write_outside_the_workspace_is_refused(self):
|
||||
with self.assertRaisesRegex(ValueError, "outside the workspace"):
|
||||
m.enforce_path_policy("qet_export", {
|
||||
"binary": "/usr/bin/qelectrotech",
|
||||
"binary": str(self.qet),
|
||||
"project": str(self.root / "ok.qet"),
|
||||
"format": "pdf",
|
||||
"output": str(self.outside / "exfiltrated.pdf")})
|
||||
@@ -1808,7 +1820,7 @@ class PathPolicy(unittest.TestCase):
|
||||
def test_existing_output_is_not_clobbered_without_overwrite(self):
|
||||
target = self.root / "existing.qet"
|
||||
target.write_text("precious")
|
||||
args = {"binary": "/usr/bin/qelectrotech", "output": str(target), "title": "T"}
|
||||
args = {"binary": str(self.qet), "output": str(target), "title": "T"}
|
||||
with self.assertRaisesRegex(ValueError, "already exists"):
|
||||
m.enforce_path_policy("qet_project_new", args)
|
||||
# ... and goes through once the caller says so explicitly
|
||||
@@ -1817,12 +1829,12 @@ class PathPolicy(unittest.TestCase):
|
||||
|
||||
def test_new_output_needs_no_overwrite_flag(self):
|
||||
m.enforce_path_policy("qet_project_new", {
|
||||
"binary": "/usr/bin/qelectrotech",
|
||||
"binary": str(self.qet),
|
||||
"output": str(self.root / "brand_new.qet"), "title": "T"})
|
||||
|
||||
def test_operation_level_file_paths_are_checked(self):
|
||||
"""add_image/add_pdf_page carry their own path, one level down."""
|
||||
base = {"binary": "/usr/bin/qelectrotech",
|
||||
base = {"binary": str(self.qet),
|
||||
"project": str(self.root / "ok.qet"),
|
||||
"output": str(self.root / "out.qet")}
|
||||
outside_png = str(self.outside / "anything.png")
|
||||
@@ -1833,17 +1845,6 @@ class PathPolicy(unittest.TestCase):
|
||||
with self.assertRaisesRegex(ValueError, "outside the workspace"):
|
||||
m.enforce_path_policy("qet_edit", dict(base, operations=[op]))
|
||||
|
||||
def test_configuration_paths_are_exempt(self):
|
||||
"""binary and elements_dir are the operator's choice, not the model's.
|
||||
|
||||
Both normally live in /usr or a build tree, so confining them would
|
||||
reject the ordinary case while stopping nothing.
|
||||
"""
|
||||
m.enforce_path_policy("qet_query", {
|
||||
"binary": "/usr/bin/qelectrotech",
|
||||
"project": str(self.root / "ok.qet"),
|
||||
"elements_dir": "/usr/share/qelectrotech/elements"})
|
||||
|
||||
def test_several_roots_may_be_allowed(self):
|
||||
os.environ["QET_MCP_WORKSPACE"] = os.pathsep.join(
|
||||
[str(self.root), str(self.outside)])
|
||||
@@ -1880,8 +1881,8 @@ class PathPolicy(unittest.TestCase):
|
||||
def test_every_data_path_argument_is_guarded(self):
|
||||
"""The other direction: a tool whose schema takes a data path must be
|
||||
in the policy, or that path is read or written with no workspace
|
||||
check at all -- and nothing fails. binary and elements_dir are
|
||||
configuration, deliberately not confined (see the README)."""
|
||||
check at all -- and nothing fails. binary and elements_dir have
|
||||
their own rule (BinaryPolicy)."""
|
||||
pathish = {"path", "project", "before", "after", "output", "directory"}
|
||||
for t in m.TOOLS:
|
||||
with self.subTest(tool=t["name"]):
|
||||
@@ -1902,6 +1903,125 @@ class PathPolicy(unittest.TestCase):
|
||||
f"{name} has no {arg!r} argument to guard")
|
||||
|
||||
|
||||
class BinaryPolicy(unittest.TestCase):
|
||||
"""The program the server launches is not the client's to choose (F063).
|
||||
|
||||
"binary" used to be exempt from every check as "configuration", but it
|
||||
is a per-call argument: any executable file it named was run, with the
|
||||
client's own paths as arguments.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
base = Path(self.tmp.name)
|
||||
self.root = base / "workspace"
|
||||
self.root.mkdir()
|
||||
(self.root / "ok.qet").write_text("<project/>")
|
||||
self.qet = fake_qet(base / "bin")
|
||||
self.other = fake_qet(base / "other")
|
||||
self._saved = dict(os.environ)
|
||||
os.environ["QET_MCP_WORKSPACE"] = str(self.root)
|
||||
os.environ["QET_BINARY"] = str(self.qet)
|
||||
# A PATH with no qelectrotech on it, so only what a test sets counts.
|
||||
os.environ["PATH"] = str(base / "empty")
|
||||
for var in ("QET_MCP_ALLOW_ANY_PATH", "QET_MCP_ALLOW_ANY_BINARY",
|
||||
"QET_MCP_BINARIES", "QET_MCP_ELEMENTS"):
|
||||
os.environ.pop(var, None)
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.clear()
|
||||
os.environ.update(self._saved)
|
||||
self.tmp.cleanup()
|
||||
|
||||
def call(self, **extra):
|
||||
args = dict({"project": str(self.root / "ok.qet")}, **extra)
|
||||
m.enforce_path_policy("qet_query", args)
|
||||
return args
|
||||
|
||||
def test_left_out_it_is_filled_in(self):
|
||||
self.assertEqual(self.call()["binary"], str(self.qet.resolve()))
|
||||
|
||||
def test_the_configured_one_is_accepted(self):
|
||||
self.assertEqual(self.call(binary=str(self.qet))["binary"], str(self.qet.resolve()))
|
||||
|
||||
def test_any_other_program_is_refused(self):
|
||||
with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"):
|
||||
self.call(binary=str(self.other))
|
||||
|
||||
def test_a_program_inside_the_workspace_is_refused_too(self):
|
||||
"""Being inside the workspace makes a file readable, not runnable."""
|
||||
planted = fake_qet(self.root, "run.sh")
|
||||
with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"):
|
||||
self.call(binary=str(planted))
|
||||
|
||||
def test_a_symlink_is_judged_by_its_target(self):
|
||||
link = self.root / "qelectrotech"
|
||||
link.symlink_to(self.other)
|
||||
with self.assertRaisesRegex(ValueError, "not an allowed QElectroTech"):
|
||||
self.call(binary=str(link))
|
||||
good = self.root / "also-qet"
|
||||
good.symlink_to(self.qet)
|
||||
self.call(binary=str(good))
|
||||
|
||||
def test_others_can_be_listed(self):
|
||||
os.environ["QET_MCP_BINARIES"] = str(self.other)
|
||||
self.assertEqual(self.call(binary=str(self.other))["binary"], str(self.other.resolve()))
|
||||
|
||||
def test_the_check_can_be_switched_off(self):
|
||||
os.environ["QET_MCP_ALLOW_ANY_BINARY"] = "1"
|
||||
self.assertEqual(self.call(binary=str(self.other))["binary"], str(self.other))
|
||||
|
||||
def test_found_on_path(self):
|
||||
del os.environ["QET_BINARY"]
|
||||
os.environ["PATH"] = str(self.qet.parent)
|
||||
self.assertEqual(self.call()["binary"], str(self.qet.resolve()))
|
||||
|
||||
def test_none_found_says_what_to_set(self):
|
||||
del os.environ["QET_BINARY"]
|
||||
with self.assertRaisesRegex(ValueError, "set QET_BINARY"):
|
||||
self.call()
|
||||
|
||||
def test_every_tool_that_takes_binary_is_checked(self):
|
||||
"""A tool whose schema offers "binary" but that the policy skips
|
||||
would run whatever it was given."""
|
||||
takes = {t["name"] for t in m.TOOLS
|
||||
if "binary" in t["inputSchema"].get("properties", {})}
|
||||
self.assertEqual(takes, m._LAUNCHES_QET)
|
||||
for name in takes:
|
||||
self.assertNotIn("binary", m._BY_NAME[name]["inputSchema"].get("required", []),
|
||||
f"{name} still requires the client to name a binary")
|
||||
|
||||
def test_elements_dir_outside_the_workspace_is_refused(self):
|
||||
with self.assertRaisesRegex(ValueError, "'elements_dir' is outside"):
|
||||
self.call(elements_dir=self.tmp.name)
|
||||
|
||||
def test_elements_dir_can_be_listed(self):
|
||||
coll = Path(self.tmp.name) / "collection"
|
||||
coll.mkdir()
|
||||
os.environ["QET_MCP_ELEMENTS"] = str(coll)
|
||||
self.call(elements_dir=str(coll / "10_electric"))
|
||||
|
||||
def test_an_install_finds_its_own_qet_and_collection(self):
|
||||
"""Installed as <prefix>/share/qelectrotech/mcp/qet_mcp.py, the
|
||||
server needs no configuration at all."""
|
||||
import importlib.util
|
||||
prefix = Path(self.tmp.name) / "prefix"
|
||||
mcp = prefix / "share" / "qelectrotech" / "mcp"
|
||||
mcp.mkdir(parents=True)
|
||||
(prefix / "share" / "qelectrotech" / "elements").mkdir()
|
||||
shutil.copy2(HERE / "qet_mcp.py", mcp / "qet_mcp.py")
|
||||
exe = fake_qet(prefix / "bin")
|
||||
del os.environ["QET_BINARY"]
|
||||
spec = importlib.util.spec_from_file_location("qet_mcp_installed", mcp / "qet_mcp.py")
|
||||
inst = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(inst)
|
||||
args = {"project": str(self.root / "ok.qet")}
|
||||
inst.enforce_path_policy("qet_query", args)
|
||||
self.assertEqual(args["binary"], str(exe.resolve()))
|
||||
self.assertEqual(args["elements_dir"],
|
||||
str((prefix / "share" / "qelectrotech" / "elements").resolve()))
|
||||
|
||||
|
||||
class ScriptingDisabledHint(unittest.TestCase):
|
||||
"""QElectroTech may refuse to run scripts at all, and says so in French.
|
||||
|
||||
@@ -2036,6 +2156,24 @@ class PathPolicyOverStdio(unittest.TestCase):
|
||||
self.assertTrue(result.get("isError"), result)
|
||||
self.assertIn("outside the workspace", result["content"][0]["text"])
|
||||
|
||||
def test_a_tool_call_cannot_choose_the_program_to_run(self):
|
||||
"""F063's reproduction: a script in the workspace, named as binary."""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
(root / "a.qet").write_text('<project title="x"><diagram title="D"/></project>')
|
||||
planted = fake_qet(root, "argv.py")
|
||||
reply = self.rpc({"jsonrpc": "2.0", "id": 9, "method": "tools/call",
|
||||
"params": {"name": "qet_export",
|
||||
"arguments": {"binary": str(planted),
|
||||
"project": str(root / "a.qet"),
|
||||
"format": "pdf",
|
||||
"output": str(root / "o.pdf")}}},
|
||||
{"QET_MCP_WORKSPACE": str(root),
|
||||
"QET_BINARY": str(fake_qet(root / "bin"))})
|
||||
result = reply["result"]
|
||||
self.assertTrue(result.get("isError"), result)
|
||||
self.assertIn("not an allowed QElectroTech", result["content"][0]["text"])
|
||||
|
||||
def test_the_same_call_succeeds_inside_the_workspace(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp) / "ws"
|
||||
|
||||
Reference in New Issue
Block a user