327 lines
12 KiB
Python
327 lines
12 KiB
Python
"""Market validation uses synthetic packages/evidence, never a real release."""
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
from artemis.plugins.market import (
|
|
CHECKS,
|
|
REPORT_NAME,
|
|
content_fingerprint,
|
|
draft_submission,
|
|
package_files,
|
|
package_directory,
|
|
validate_submission,
|
|
)
|
|
from artemis.plugins.models import PluginManifest
|
|
from artemis.plugins.validation import check, main
|
|
from artemis.plugins import validation
|
|
|
|
|
|
@pytest.fixture
|
|
def candidate(tmp_path):
|
|
manifest = PluginManifest.model_validate(
|
|
{
|
|
"schema_version": 2,
|
|
"id": "synthetic-market-test",
|
|
"name": "Synthetic",
|
|
"version": "1.0.0",
|
|
"platforms": [sys.platform],
|
|
"skill": "SKILL.md",
|
|
"license": "Apache-2.0",
|
|
"capabilities": {"kind": "file", "tools": {}},
|
|
}
|
|
)
|
|
(tmp_path / "plugin.json").write_text(manifest.model_dump_json(), encoding="utf-8")
|
|
for name in ("SKILL.md", "LICENSE", "SOURCE.md", "VERIFICATION.md"):
|
|
(tmp_path / name).write_text(
|
|
"Synthetic test fixture, not a published claim.\n", encoding="utf-8"
|
|
)
|
|
return tmp_path, manifest
|
|
|
|
|
|
def submitted(candidate):
|
|
root, manifest = candidate
|
|
files = package_files(root)
|
|
report = draft_submission(manifest, files)
|
|
report["traits"] = {"uses_subprocesses": False, "file_inputs_for_writes": False}
|
|
report["components"] = [
|
|
{
|
|
"name": "synthetic fixture",
|
|
"paths": ["."],
|
|
"license": "Apache-2.0",
|
|
"permission": "permitted",
|
|
"license_file": "LICENSE",
|
|
"source_file": "SOURCE.md",
|
|
}
|
|
]
|
|
for record in report["checks"].values():
|
|
record.update(
|
|
status="passed",
|
|
evidence_file="VERIFICATION.md",
|
|
evidence_sha256=files["VERIFICATION.md"],
|
|
notes="Synthetic validation record for unit testing only.",
|
|
)
|
|
write_report(root, report)
|
|
return root, manifest, report
|
|
|
|
|
|
def write_report(root, report):
|
|
(root / REPORT_NAME).write_text(json.dumps(report), encoding="utf-8")
|
|
|
|
|
|
def validate(root, manifest):
|
|
return validate_submission(root, manifest, package_files(root))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_development_does_not_require_public_distribution(candidate):
|
|
root, _ = candidate
|
|
result = await check(root, False, {})
|
|
assert result["scope"] == "development"
|
|
assert not (root / REPORT_NAME).exists()
|
|
with pytest.raises(ValueError, match="MARKET.json"):
|
|
await check(root, False, {}, market=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generated_market_draft_is_unverified_and_never_overwrites(candidate):
|
|
root, _ = candidate
|
|
result = await check(root, False, {}, prepare_market=True)
|
|
assert result["status"] == "draft"
|
|
report = json.loads((root / REPORT_NAME).read_text("utf-8"))
|
|
assert all(record["status"] == "unverified" for record in report["checks"].values())
|
|
assert report["components"][0]["permission"] == "unknown"
|
|
assert report["traits"]["uses_subprocesses"] is None
|
|
before = (root / REPORT_NAME).read_bytes()
|
|
with pytest.raises(FileExistsError):
|
|
await check(root, False, {}, prepare_market=True)
|
|
assert (root / REPORT_NAME).read_bytes() == before
|
|
with pytest.raises(ValueError):
|
|
await check(root, False, {}, market=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_market_materials_report_and_zip_have_same_content_hash(
|
|
candidate, tmp_path_factory
|
|
):
|
|
root, manifest, report = submitted(candidate)
|
|
checked = await check(root, False, {}, market=True)
|
|
assert checked["scope"] == "market_materials"
|
|
archive = tmp_path_factory.mktemp("market-zip") / "candidate.zip"
|
|
with zipfile.ZipFile(archive, "w") as output:
|
|
for path in root.iterdir():
|
|
output.write(path, path.name)
|
|
zipped = await check(archive, False, {}, market=True)
|
|
assert zipped["content_sha256"] == checked["content_sha256"] == report["content_sha256"]
|
|
assert zipped["archive_sha256"] == hashlib.sha256(archive.read_bytes()).hexdigest()
|
|
|
|
|
|
@pytest.mark.parametrize("permission", ["unknown", "local_only"])
|
|
def test_market_rejects_components_without_public_distribution(candidate, permission):
|
|
root, manifest, report = submitted(candidate)
|
|
report["components"][0]["permission"] = permission
|
|
write_report(root, report)
|
|
with pytest.raises(ValueError, match="市场分发"):
|
|
validate(root, manifest)
|
|
|
|
|
|
def test_market_rejects_changed_payload_and_forged_evidence(candidate):
|
|
root, manifest, report = submitted(candidate)
|
|
report["checks"]["source_review"]["evidence_sha256"] = "0" * 64
|
|
write_report(root, report)
|
|
with pytest.raises(ValueError, match="记录 SHA256"):
|
|
validate(root, manifest)
|
|
(root / "SKILL.md").write_text("different bytes", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="旧的市场验证记录失效"):
|
|
validate(root, manifest)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
[
|
|
"version",
|
|
"missing_check",
|
|
"unverified",
|
|
"no_notes",
|
|
"uncovered",
|
|
"source_escape",
|
|
"license_missing",
|
|
"na_required",
|
|
"process_na",
|
|
"binding_na",
|
|
],
|
|
)
|
|
def test_market_blocks_incomplete_or_inapplicable_claims(candidate, change):
|
|
root, manifest, report = submitted(candidate)
|
|
if change == "version":
|
|
report["plugin_version"] = "2.0.0"
|
|
elif change == "missing_check":
|
|
del report["checks"]["real_task"]
|
|
elif change == "unverified":
|
|
report["checks"]["real_task"]["status"] = "unverified"
|
|
elif change == "no_notes":
|
|
report["checks"]["source_review"]["notes"] = " "
|
|
elif change == "uncovered":
|
|
report["components"][0]["paths"] = ["LICENSE"]
|
|
elif change == "source_escape":
|
|
report["components"][0]["source_file"] = "../outside"
|
|
elif change == "license_missing":
|
|
report["components"][0]["license_file"] = "missing.txt"
|
|
elif change == "na_required":
|
|
report["checks"]["real_task"]["status"] = "not_applicable"
|
|
elif change == "process_na":
|
|
report["traits"]["uses_subprocesses"] = True
|
|
report["checks"]["process_tree"]["status"] = "not_applicable"
|
|
elif change == "binding_na":
|
|
report["traits"]["file_inputs_for_writes"] = True
|
|
report["checks"]["file_binding"]["status"] = "not_applicable"
|
|
write_report(root, report)
|
|
with pytest.raises(ValueError):
|
|
validate(root, manifest)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"name", [".ruff_cache/entry", "__pycache__/x", "module.pyc", ".env", ".env.local"]
|
|
)
|
|
def test_market_blocks_cache_and_personal_configuration(candidate, name):
|
|
root, _ = candidate
|
|
path = root / name
|
|
path.parent.mkdir(exist_ok=True)
|
|
path.write_text("synthetic only", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="市场包"):
|
|
package_files(root)
|
|
|
|
|
|
def test_fingerprint_is_order_independent_but_binds_paths_and_bytes():
|
|
assert content_fingerprint({"b": "2", "a": "1"}) == content_fingerprint({"a": "1", "b": "2"})
|
|
assert content_fingerprint({"a": "1"}) != content_fingerprint({"a": "2"})
|
|
assert content_fingerprint({"a": "1"}) != content_fingerprint({"b": "1"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_zip_path_traversal_cannot_escape_inspection(candidate, tmp_path_factory):
|
|
root, _ = candidate
|
|
archive = tmp_path_factory.mktemp("bad-market-zip") / "bad.zip"
|
|
with zipfile.ZipFile(archive, "w") as output:
|
|
output.writestr("../outside.txt", "synthetic")
|
|
with pytest.raises(ValueError):
|
|
await check(archive, False, {})
|
|
assert not (archive.parent / "outside.txt").exists()
|
|
|
|
|
|
def test_json_output_must_not_modify_candidate_or_overwrite_zip(candidate, monkeypatch, capsys):
|
|
root, _ = candidate
|
|
target = root / "report.json"
|
|
monkeypatch.setattr(sys, "argv", ["check_plugin", str(root), "--json-output", str(target)])
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
assert not target.exists()
|
|
archive = root / "original.zip"
|
|
archive.write_bytes(b"original immutable zip placeholder")
|
|
monkeypatch.setattr(sys, "argv", ["check_plugin", str(archive), "--json-output", str(archive)])
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
assert archive.read_bytes() == b"original immutable zip placeholder"
|
|
|
|
|
|
def test_invalid_values_are_not_echoed(candidate, monkeypatch, capsys):
|
|
root, _ = candidate
|
|
private_marker = "SYNTHETIC-DO-NOT-ECHO"
|
|
monkeypatch.setattr(
|
|
sys, "argv", ["check_plugin", str(root), "--values", '{"key":"' + private_marker]
|
|
)
|
|
with pytest.raises(SystemExit):
|
|
main()
|
|
assert private_marker not in capsys.readouterr().out
|
|
|
|
|
|
def add_synthetic_server(candidate):
|
|
root, manifest = candidate
|
|
data = manifest.model_dump(mode="json")
|
|
data["server"] = {"command": sys.executable, "args": ["synthetic-unused-server.py"]}
|
|
manifest = PluginManifest.model_validate(data)
|
|
(root / "plugin.json").write_text(manifest.model_dump_json(), encoding="utf-8")
|
|
return root, manifest
|
|
|
|
|
|
def handshake_fixture(monkeypatch, action):
|
|
class FakeSession:
|
|
tools = []
|
|
|
|
def __init__(self, server, root):
|
|
self.root = root
|
|
|
|
async def start(self):
|
|
action(self.root)
|
|
return self
|
|
|
|
async def aclose(self):
|
|
pass
|
|
|
|
monkeypatch.setattr(validation, "PluginSession", FakeSession)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("changed", ["payload", "submission"])
|
|
async def test_market_rejects_changes_during_handshake(candidate, monkeypatch, changed):
|
|
root, manifest, report = submitted(add_synthetic_server(candidate))
|
|
|
|
def change(directory):
|
|
if changed == "payload":
|
|
(directory / "SKILL.md").write_text("changed during handshake", encoding="utf-8")
|
|
else:
|
|
report["checks"]["real_task"]["notes"] = "A different synthetic claim."
|
|
write_report(directory, report)
|
|
|
|
handshake_fixture(monkeypatch, change)
|
|
with pytest.raises(ValueError, match="发生变化"):
|
|
await check(root, False, {}, market=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_zip_digest_is_bound_to_bytes_actually_inspected(
|
|
candidate, tmp_path_factory, monkeypatch
|
|
):
|
|
root, _, report = submitted(add_synthetic_server(candidate))
|
|
archive = tmp_path_factory.mktemp("changing-zip") / "candidate.zip"
|
|
with zipfile.ZipFile(archive, "w") as output:
|
|
for path in root.iterdir():
|
|
output.write(path, path.name)
|
|
approved = hashlib.sha256(archive.read_bytes()).hexdigest()
|
|
handshake_fixture(monkeypatch, lambda root: archive.write_bytes(b"uninspected replacement"))
|
|
with pytest.raises(ValueError, match="ZIP 发生变化"):
|
|
await check(archive, False, {}, market=True)
|
|
assert approved != hashlib.sha256(archive.read_bytes()).hexdigest()
|
|
|
|
|
|
def test_archive_fingerprint_uses_extracted_snapshot(candidate, tmp_path_factory):
|
|
root, _ = candidate
|
|
archive = tmp_path_factory.mktemp("zip-snapshot") / "candidate.zip"
|
|
with zipfile.ZipFile(archive, "w") as output:
|
|
for path in root.iterdir():
|
|
output.write(path, path.name)
|
|
approved = hashlib.sha256(archive.read_bytes()).hexdigest()
|
|
with pytest.raises(ValueError, match="ZIP 发生变化"):
|
|
with package_directory(archive) as (extracted, digest):
|
|
archive.write_bytes(b"uninspected replacement")
|
|
assert (extracted / "plugin.json").read_bytes() == (root / "plugin.json").read_bytes()
|
|
assert digest == approved
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incomplete_materials_never_launch_server(candidate, monkeypatch):
|
|
root, _ = add_synthetic_server(candidate)
|
|
|
|
def never_start(root):
|
|
raise AssertionError("Incomplete market materials launched a plugin")
|
|
|
|
handshake_fixture(monkeypatch, never_start)
|
|
with pytest.raises(ValueError, match="MARKET.json"):
|
|
await check(root, False, {}, market=True)
|