111 lines
6.6 KiB
Python
111 lines
6.6 KiB
Python
"""审核、签名、撤回和七类受控包的真实 HTTP 测试。"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
|
|
from fastapi.testclient import TestClient
|
|
import pytest
|
|
|
|
from community.app import Registry, create_app
|
|
from community.package import Release, signed_payload
|
|
|
|
|
|
@pytest.fixture
|
|
def env(tmp_path):
|
|
registry = Registry(tmp_path / "catalog.db")
|
|
author = registry.add_principal("author", "author", "examples")
|
|
moderator = registry.add_principal("reviewer", "moderator")
|
|
private = Ed25519PrivateKey.generate()
|
|
registry.add_key("test-key", "examples", private.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw))
|
|
with TestClient(create_app(registry)) as client:
|
|
yield client, private, {"Authorization": "Bearer " + author}, {"Authorization": "Bearer " + moderator}
|
|
|
|
|
|
def package(private, kind="persona", files=None):
|
|
names = {"theme": ("theme.yaml", "theme_id: test-package\nversion: 1.0.0"), "plugin": ("plugin.yaml", "plugin_id: test-package\nversion: 1.0.0"),
|
|
"skill": ("skill.yaml", "skill_id: test-package\nversion: 1.0.0"), "mcp": ("mcp.json", json.dumps({"transport": "stdio", "args": []})),
|
|
"persona": ("persona.json", json.dumps({"system_prompt": "受控样例"})),
|
|
"template": ("template.json", json.dumps({"markdown": "# {{title}}"})),
|
|
"model": ("model.json", json.dumps({"source": "https://example.org/model", "revision": "fixed", "license": "MIT", "resources": {"ram_gb": 8}, "verified_platforms": ["test-only"]}))}
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as archive:
|
|
for name, value in (files or dict([names[kind]])).items(): archive.writestr(name, value)
|
|
blob = buffer.getvalue()
|
|
release = Release(namespace="examples", package_id="test-package", type=kind, version="1.0.0", name="受控示例", author_id="author", license="MIT",
|
|
description="用于隔离验收", sha256=hashlib.sha256(blob).hexdigest(), size=len(blob), platforms=["windows"], architectures=["x86_64"],
|
|
min_app_version="0.2.0", changelog="初始版本", published_at="2026-09-07T00:00:00Z", key_id="test-key", signature="")
|
|
release.signature = base64.b64encode(private.sign(signed_payload(release))).decode()
|
|
return {"release": release.model_dump(), "archive_base64": base64.b64encode(blob).decode()}
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["theme", "skill", "plugin", "mcp", "persona", "template", "model"])
|
|
def test_publish_review_read_withdraw(env, kind):
|
|
client, private, author, moderator = env
|
|
payload = package(private, kind)
|
|
submit = client.post("/catalog/v1/publish/submissions", headers=author, json=payload)
|
|
assert submit.status_code == 200, submit.text
|
|
submission = submit.json()["submission_id"]
|
|
assert client.get("/catalog/v1/packages").json()["total"] == 0
|
|
assert client.post("/catalog/v1/moderation/reviews", headers=author, json={"submission_id": submission, "approve": True, "reason": "自审"}).status_code == 403
|
|
assert client.post("/catalog/v1/moderation/reviews", headers=moderator, json={"submission_id": submission, "approve": True, "reason": "受控验收"}).status_code == 200
|
|
listing = client.get("/catalog/v1/packages", params={"type": kind, "q": "示例"})
|
|
assert listing.json()["total"] == 1
|
|
assert client.get("/catalog/v1/packages", params={"type": kind, "q": "示例"}, headers={"If-None-Match": listing.headers["ETag"]}).status_code == 304
|
|
path = listing.json()["items"][0]["download_path"]
|
|
assert hashlib.sha256(client.get(path).content).hexdigest() == payload["release"]["sha256"]
|
|
assert client.post("/catalog/v1/publish/submissions", headers=author, json=payload).status_code == 409
|
|
assert client.post(f"/catalog/v1/releases/{submission}/withdraw", headers=author, json={"reason": "撤回验收"}).status_code == 200
|
|
assert client.get(path).status_code == 410
|
|
assert client.get("/catalog/v1/packages").json()["items"][0]["withdrawn"]
|
|
|
|
|
|
def test_tampered_permissions_and_signer_revocation(env):
|
|
client, private, author, moderator = env
|
|
payload = package(private)
|
|
payload["release"]["permissions"] = ["network.request"]
|
|
assert client.post("/catalog/v1/publish/submissions", headers=author, json=payload).status_code == 422
|
|
payload = package(private)
|
|
submission = client.post("/catalog/v1/publish/submissions", headers=author, json=payload).json()["submission_id"]
|
|
assert client.post("/catalog/v1/keys/test-key/revoke", headers=moderator, json={"reason": "密钥轮换测试"}).status_code == 200
|
|
assert client.post("/catalog/v1/moderation/reviews", headers=moderator, json={"submission_id": submission, "approve": True, "reason": "签名已撤回"}).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize("files", [{"../persona.json": "{}"}, {"CON": ""}, {"a": "", "A": ""}, {"persona.json": '{"system_prompt":"x","api_key":"fixture"}'}])
|
|
def test_bad_archives_and_embedded_secret_rejected(env, files):
|
|
client, private, author, _ = env
|
|
assert client.post("/catalog/v1/publish/submissions", headers=author, json=package(private, files=files)).status_code == 422
|
|
|
|
|
|
def test_cross_namespace_and_missing_auth(env):
|
|
client, private, author, _ = env
|
|
payload = package(private)
|
|
assert client.post("/catalog/v1/publish/submissions", json=payload).status_code == 401
|
|
payload["release"]["namespace"] = "other"
|
|
assert client.post("/catalog/v1/publish/submissions", headers=author, json=payload).status_code == 403
|
|
|
|
|
|
@pytest.mark.parametrize("kind,identifier", [("plugin", "markdown-workbench"), ("skill", "note-reviewer")])
|
|
def test_repository_manifest_identity_aliases_and_conflicts(kind, identifier):
|
|
from pathlib import Path
|
|
import yaml
|
|
from community.package import inspect
|
|
source = Path(__file__).resolve().parents[2] / "backend" / "extensions" / "community" / (kind + "s") / identifier / (kind + ".yaml")
|
|
body = source.read_text(encoding="utf-8")
|
|
manifest = yaml.safe_load(body)
|
|
private = Ed25519PrivateKey.generate()
|
|
payload = package(private, kind, files={kind + ".yaml": body})
|
|
release = Release(**payload["release"])
|
|
release.package_id = identifier
|
|
release.permissions = manifest["permissions"]
|
|
inspect(release, base64.b64decode(payload["archive_base64"]))
|
|
bad = package(private, kind, files={kind + ".yaml": body + "\n" + kind + "_id: other-id\n"})
|
|
release.sha256 = bad["release"]["sha256"]
|
|
release.size = bad["release"]["size"]
|
|
with pytest.raises(ValueError):
|
|
inspect(release, base64.b64decode(bad["archive_base64"]))
|