bvandeusenandClaude Opus 5.5
d188a673a6
#5032 "Plugin install takes its URL and checksum from the catalog, requires the checksum, and extracts only inside the plugin folder". The install form carried the download URL and checksum as hidden fields, so any admin POST could make Steward fetch and import code from anywhere, unchecked, and a zip member like ../x.py was written wherever it pointed. - The route takes only the plugin name and reads download_url and checksum_sha256 from the configured catalogs; a name no repository lists is refused. Installs are audited (plugin.install / plugin.install_failed). - The checksum is required; the name must be a plain identifier. - extract_plugin refuses any member that resolves outside the plugin folder, caps the download at 50 MB and the unpacked size at 200 MB. - The zip unpacks into a staging folder and is swapped in whole: a refused or failed install leaves the installed copy untouched, and an update no longer leaves the old version's deleted files behind. - The catalog shows "Can't install: the catalog lists no checksum" in place of the button for an entry without one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
117 lines
5.0 KiB
Python
117 lines
5.0 KiB
Python
"""Plugin installs fetch only what the catalog names, verify it, and unpack it
|
|
only inside the plugin's own folder."""
|
|
import hashlib
|
|
import io
|
|
import zipfile
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from steward.core import plugin_manager as pm
|
|
|
|
|
|
def make_zip(files: dict[str, str]) -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
for name, text in files.items():
|
|
zf.writestr(name, text)
|
|
return buf.getvalue()
|
|
|
|
|
|
def sha(content: bytes) -> str:
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
|
|
@pytest.mark.parametrize("name", ["..", "a/b", "evil-name", "", "x.y", "ümlaut"])
|
|
def test_name_must_be_an_identifier(name):
|
|
with pytest.raises(pm.PluginInstallError):
|
|
pm.check_plugin_name(name)
|
|
|
|
|
|
def test_checksum_is_required_and_compared():
|
|
content = b"zip"
|
|
with pytest.raises(pm.PluginInstallError, match="no checksum"):
|
|
pm.verify_checksum(content, "")
|
|
with pytest.raises(pm.PluginInstallError, match="mismatch"):
|
|
pm.verify_checksum(content, "0" * 64)
|
|
pm.verify_checksum(content, sha(content).upper())
|
|
|
|
|
|
def test_extract_strips_the_plugin_folder(tmp_path):
|
|
content = make_zip({"demo-v1.2/__init__.py": "x = 1", "demo-v1.2/sub/a.txt": "a"})
|
|
pm.extract_plugin(content, "demo", tmp_path / "demo")
|
|
assert (tmp_path / "demo" / "__init__.py").read_text() == "x = 1"
|
|
assert (tmp_path / "demo" / "sub" / "a.txt").read_text() == "a"
|
|
|
|
|
|
@pytest.mark.parametrize("member", ["../escape.py", "demo/../../escape.py", "/abs.py"])
|
|
def test_extract_refuses_members_outside_the_folder(tmp_path, member):
|
|
content = make_zip({"__init__.py": "", member: "boom"})
|
|
with pytest.raises(pm.PluginInstallError, match="outside"):
|
|
pm.extract_plugin(content, "demo", tmp_path / "plugins" / "demo")
|
|
assert not (tmp_path / "escape.py").exists()
|
|
assert not (tmp_path / "plugins" / "escape.py").exists()
|
|
assert not (tmp_path / "plugins" / "demo").exists() # nothing half-written
|
|
|
|
|
|
def test_extract_refuses_an_oversized_unpack(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(pm, "MAX_PLUGIN_UNPACKED_BYTES", 10)
|
|
with pytest.raises(pm.PluginInstallError, match="Unpacks to more than"):
|
|
pm.extract_plugin(make_zip({"__init__.py": "x" * 11}), "demo", tmp_path / "demo")
|
|
|
|
|
|
def _app(tmp_path):
|
|
return SimpleNamespace(config={"PLUGIN_INSTALL_DIR": str(tmp_path), "PLUGIN_DIRS": [str(tmp_path)],
|
|
"DATABASE_URL": ""})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_replaces_the_folder_whole(tmp_path):
|
|
old = tmp_path / "demo"
|
|
old.mkdir()
|
|
(old / "gone.py").write_text("old")
|
|
content = make_zip({"demo/__init__.py": "new"})
|
|
with patch.object(pm, "_download", new=AsyncMock(return_value=content)):
|
|
ok, msg = await pm.download_and_install_plugin(_app(tmp_path), "demo", "https://x/demo.zip", sha(content))
|
|
assert ok, msg
|
|
assert (old / "__init__.py").read_text() == "new"
|
|
assert not (old / "gone.py").exists()
|
|
assert sorted(p.name for p in tmp_path.iterdir()) == ["demo"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_refused_download_leaves_the_installed_copy(tmp_path):
|
|
old = tmp_path / "demo"
|
|
old.mkdir()
|
|
(old / "__init__.py").write_text("old")
|
|
content = make_zip({"demo/__init__.py": "new"})
|
|
with patch.object(pm, "_download", new=AsyncMock(return_value=content)):
|
|
ok, msg = await pm.download_and_install_plugin(_app(tmp_path), "demo", "https://x/demo.zip", "")
|
|
assert not ok and "no checksum" in msg
|
|
assert (old / "__init__.py").read_text() == "old"
|
|
assert sorted(p.name for p in tmp_path.iterdir()) == ["demo"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_takes_url_and_checksum_from_the_catalog(client):
|
|
async with client.session_transaction() as sess:
|
|
sess.update(user_id="u", user_role="admin", username="admin", _csrf="t")
|
|
entry = SimpleNamespace(name="demo", version="1.0", download_url="https://catalog/demo.zip",
|
|
checksum_sha256="abc")
|
|
install = AsyncMock(return_value=(False, "stopped here"))
|
|
with patch("steward.settings.routes.get_all_settings", new=AsyncMock(return_value={})), \
|
|
patch("steward.settings.routes._get_plugin_repos", return_value=[{"url": "https://catalog/"}]), \
|
|
patch("steward.core.plugin_index.fetch_catalog", new=AsyncMock(return_value=[entry])), \
|
|
patch("steward.core.plugin_manager.download_and_install_plugin", new=install), \
|
|
patch("steward.settings.routes.log_audit", new=AsyncMock()):
|
|
resp = await client.post("/settings/plugins/install/demo", headers={"X-CSRF-Token": "t"}, form={
|
|
"download_url": "https://attacker/evil.zip", "checksum_sha256": "dead"})
|
|
assert resp.status_code == 200
|
|
assert install.await_args.args[1:] == ("demo", "https://catalog/demo.zip", "abc")
|
|
|
|
install.reset_mock()
|
|
resp = await client.post("/settings/plugins/install/other", headers={"X-CSRF-Token": "t"})
|
|
assert "No configured plugin repository lists this plugin" in await resp.get_data(as_text=True)
|
|
install.assert_not_awaited()
|