712 lines
29 KiB
Python
712 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
META_ROOT = Path(__file__).resolve().parents[1]
|
|
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
|
|
if str(RELEASE_TOOLS_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(RELEASE_TOOLS_ROOT))
|
|
|
|
from govoplan_release.publisher import ( # noqa: E402
|
|
FrozenPublicationRemote,
|
|
bind_publication_remote,
|
|
commit_publication_tree,
|
|
publication_mutation_trust_issues,
|
|
publish_catalog_candidate,
|
|
remote_publication_identity,
|
|
run_checked,
|
|
verify_committed_publication,
|
|
verify_remote_branch_head,
|
|
verify_remote_publication,
|
|
_trusted_npm_command,
|
|
)
|
|
|
|
|
|
class ReleaseCatalogPublicationTests(unittest.TestCase):
|
|
def test_build_command_uses_its_pinned_node_directory(self) -> None:
|
|
completed = subprocess.CompletedProcess(
|
|
["/trusted/node/bin/npm"],
|
|
0,
|
|
stdout=b"",
|
|
stderr=b"",
|
|
)
|
|
with patch(
|
|
"govoplan_release.publisher.subprocess.run",
|
|
return_value=completed,
|
|
) as runner:
|
|
run_checked(
|
|
["/trusted/node/bin/npm", "run", "build"],
|
|
cwd=Path("/trusted/website"),
|
|
)
|
|
|
|
environment = runner.call_args.kwargs["env"]
|
|
self.assertEqual(
|
|
"/trusted/node/bin:/usr/bin:/bin",
|
|
environment["PATH"],
|
|
)
|
|
|
|
def test_npm_command_requires_trusted_npm_and_sibling_node(self) -> None:
|
|
with (
|
|
patch(
|
|
"govoplan_release.publisher.shutil.which",
|
|
return_value="/trusted/node/bin/npm",
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher._trusted_runtime_executable_issue",
|
|
side_effect=(None, None),
|
|
) as trust_check,
|
|
):
|
|
self.assertEqual(
|
|
"/trusted/node/bin/npm",
|
|
_trusted_npm_command("npm"),
|
|
)
|
|
|
|
self.assertEqual(Path("/trusted/node/bin/npm"), trust_check.call_args_list[0].args[0])
|
|
self.assertEqual(Path("/trusted/node/bin/node"), trust_check.call_args_list[1].args[0])
|
|
|
|
def test_npm_command_rejects_caller_relative_path(self) -> None:
|
|
with (
|
|
patch("govoplan_release.publisher.shutil.which") as which,
|
|
self.assertRaisesRegex(RuntimeError, "absolute path or the bare name"),
|
|
):
|
|
_trusted_npm_command("./npm")
|
|
which.assert_not_called()
|
|
|
|
def test_committed_publication_must_match_validated_blobs_exactly(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
web_root = Path(tmp) / "website"
|
|
module_root = web_root / "public" / "catalogs" / "v1" / "modules"
|
|
module_root.mkdir(parents=True)
|
|
catalog = (
|
|
web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json"
|
|
)
|
|
keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
catalog.parent.mkdir(parents=True)
|
|
expected = {
|
|
catalog.relative_to(web_root).as_posix(): b'{"catalog":true}\n',
|
|
keyring.relative_to(web_root).as_posix(): b'{"keys":[]}\n',
|
|
(module_root / "index.json")
|
|
.relative_to(web_root)
|
|
.as_posix(): b'{"modules":[]}\n',
|
|
}
|
|
for relative, encoded in expected.items():
|
|
target = web_root / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(encoded)
|
|
self._git(web_root, "init", "--quiet")
|
|
self._git(web_root, "config", "user.email", "test@example.test")
|
|
self._git(web_root, "config", "user.name", "Test")
|
|
self._git(web_root, "add", ".")
|
|
self._git(web_root, "commit", "--quiet", "-m", "publication")
|
|
commit_sha = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
|
|
verify_committed_publication(
|
|
web_root=web_root,
|
|
commit_sha=commit_sha,
|
|
expected_blobs=expected,
|
|
module_root=module_root,
|
|
)
|
|
changed = dict(expected)
|
|
changed[catalog.relative_to(web_root).as_posix()] = b'{"catalog":false}\n'
|
|
with self.assertRaisesRegex(RuntimeError, "differs"):
|
|
verify_committed_publication(
|
|
web_root=web_root,
|
|
commit_sha=commit_sha,
|
|
expected_blobs=changed,
|
|
module_root=module_root,
|
|
)
|
|
catalog_path = catalog.relative_to(web_root).as_posix()
|
|
self._git(web_root, "update-index", "--chmod=+x", catalog_path)
|
|
self._git(web_root, "commit", "--quiet", "-m", "unsafe mode")
|
|
executable_commit = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
with self.assertRaisesRegex(RuntimeError, "regular blob"):
|
|
verify_committed_publication(
|
|
web_root=web_root,
|
|
commit_sha=executable_commit,
|
|
expected_blobs=expected,
|
|
module_root=module_root,
|
|
)
|
|
|
|
def test_private_index_commit_has_one_parent_and_only_computed_delta(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
web_root = root / "website"
|
|
module_root = web_root / "public" / "catalogs" / "v1" / "modules"
|
|
channel = (
|
|
web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json"
|
|
)
|
|
keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
old_module = module_root / "obsolete.json"
|
|
unrelated = web_root / "unrelated.txt"
|
|
for path, encoded in (
|
|
(channel, b'{"old":true}\n'),
|
|
(keyring, b'{"keys":[]}\n'),
|
|
(old_module, b'{"obsolete":true}\n'),
|
|
(unrelated, b"keep\n"),
|
|
):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(encoded)
|
|
self._git(web_root, "init", "--quiet")
|
|
self._git(web_root, "config", "user.email", "test@example.test")
|
|
self._git(web_root, "config", "user.name", "Test")
|
|
self._git(web_root, "add", ".")
|
|
self._git(web_root, "commit", "--quiet", "-m", "base")
|
|
frozen_head = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
branch = self._git(web_root, "branch", "--show-current").strip()
|
|
|
|
malicious = web_root / "not-in-publication.txt"
|
|
malicious.write_text("staged but excluded\n", encoding="utf-8")
|
|
self._git(web_root, "add", malicious.name)
|
|
marker = root / "reference-hook-ran"
|
|
hook = web_root / ".git" / "hooks" / "reference-transaction"
|
|
hook.write_text(f"#!/bin/sh\ntouch {marker}\n", encoding="utf-8")
|
|
hook.chmod(0o755)
|
|
expected = {
|
|
channel.relative_to(web_root).as_posix(): b'{"new":true}\n',
|
|
keyring.relative_to(web_root).as_posix(): b'{"keys":["release"]}\n',
|
|
(module_root / "index.json")
|
|
.relative_to(web_root)
|
|
.as_posix(): b'{"modules":[]}\n',
|
|
}
|
|
redirected = root / "attacker-index"
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"GIT_DIR": str(root / "attacker-git-dir"),
|
|
"GIT_INDEX_FILE": str(redirected),
|
|
"GIT_OBJECT_DIRECTORY": str(root / "attacker-objects"),
|
|
"GIT_CONFIG_GLOBAL": str(root / "attacker-config"),
|
|
},
|
|
):
|
|
commit_sha = commit_publication_tree(
|
|
web_root=web_root,
|
|
frozen_head=frozen_head,
|
|
branch=branch,
|
|
expected_blobs=expected,
|
|
module_root=module_root,
|
|
message="Exact publication",
|
|
)
|
|
|
|
parents = self._git(
|
|
web_root, "rev-list", "--parents", "-n", "1", commit_sha
|
|
).split()
|
|
changed = set(
|
|
filter(
|
|
None,
|
|
self._git(
|
|
web_root,
|
|
"diff-tree",
|
|
"--no-commit-id",
|
|
"--name-only",
|
|
"-r",
|
|
frozen_head,
|
|
commit_sha,
|
|
).splitlines(),
|
|
)
|
|
)
|
|
hook_ran = marker.exists()
|
|
inherited_index_used = redirected.exists()
|
|
commit_paths = self._git(
|
|
web_root, "ls-tree", "-r", "--name-only", commit_sha
|
|
)
|
|
|
|
self.assertEqual([commit_sha, frozen_head], parents)
|
|
self.assertEqual(
|
|
{
|
|
"public/catalogs/v1/channels/stable.json",
|
|
"public/catalogs/v1/keyring.json",
|
|
"public/catalogs/v1/modules/index.json",
|
|
"public/catalogs/v1/modules/obsolete.json",
|
|
},
|
|
changed,
|
|
)
|
|
self.assertFalse(hook_ran)
|
|
self.assertFalse(inherited_index_used)
|
|
self.assertNotIn("not-in-publication.txt", commit_paths)
|
|
|
|
def test_remote_binding_and_annotated_publication_identity_are_exact(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
remote_root = root / "website.git"
|
|
remote_root.mkdir()
|
|
self._git(remote_root, "init", "--bare", "--quiet")
|
|
web_root = root / "website"
|
|
web_root.mkdir()
|
|
self._git(web_root, "init", "--quiet")
|
|
self._git(web_root, "config", "user.email", "test@example.test")
|
|
self._git(web_root, "config", "user.name", "Test")
|
|
self._git(web_root, "branch", "-M", "main")
|
|
web_root.joinpath("base.txt").write_text("base\n", encoding="utf-8")
|
|
self._git(web_root, "add", ".")
|
|
self._git(web_root, "commit", "--quiet", "-m", "base")
|
|
self._git(web_root, "remote", "add", "origin", str(remote_root))
|
|
self._git(web_root, "push", "--quiet", "-u", "origin", "main")
|
|
base_commit = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
|
|
frozen = bind_publication_remote(
|
|
web_root=web_root,
|
|
remote="origin",
|
|
registered_remote=str(remote_root),
|
|
)
|
|
self.assertIsInstance(frozen, FrozenPublicationRemote)
|
|
verify_remote_branch_head(
|
|
web_root=web_root,
|
|
remote_url=frozen.url,
|
|
branch="main",
|
|
expected_commit=base_commit,
|
|
)
|
|
web_root.joinpath("catalog.json").write_text("{}\n", encoding="utf-8")
|
|
self._git(web_root, "add", "catalog.json")
|
|
self._git(web_root, "commit", "--quiet", "-m", "publication")
|
|
commit_sha = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
tag_name = "catalog-test"
|
|
self._git(web_root, "tag", "-a", tag_name, "-m", "publication")
|
|
tag_object = self._git(
|
|
web_root, "rev-parse", f"refs/tags/{tag_name}"
|
|
).strip()
|
|
self._git(
|
|
web_root,
|
|
"push",
|
|
"--quiet",
|
|
"--atomic",
|
|
"origin",
|
|
f"{commit_sha}:refs/heads/main",
|
|
f"{tag_object}:refs/tags/{tag_name}",
|
|
)
|
|
|
|
identity = remote_publication_identity(
|
|
web_root=web_root,
|
|
remote_url=frozen.url,
|
|
branch="main",
|
|
tag_name=tag_name,
|
|
)
|
|
verified = verify_remote_publication(
|
|
web_root=web_root,
|
|
remote_url=frozen.url,
|
|
branch="main",
|
|
tag_name=tag_name,
|
|
expected_commit=commit_sha,
|
|
expected_tag_object=tag_object,
|
|
)
|
|
self._git(
|
|
web_root,
|
|
"remote",
|
|
"set-url",
|
|
"--add",
|
|
"--push",
|
|
"origin",
|
|
str(root / "other.git"),
|
|
)
|
|
with self.assertRaisesRegex(RuntimeError, "do not match"):
|
|
bind_publication_remote(
|
|
web_root=web_root,
|
|
remote="origin",
|
|
registered_remote=str(remote_root),
|
|
)
|
|
|
|
self.assertEqual(identity, verified)
|
|
self.assertEqual(commit_sha, identity["publication_commit_sha"])
|
|
self.assertEqual(tag_object, identity["publication_tag_object_sha"])
|
|
self.assertEqual(commit_sha, identity["publication_tag_commit_sha"])
|
|
|
|
def test_mutation_rejects_writable_website_and_git_configuration(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
web_root = root / "website"
|
|
web_root.mkdir()
|
|
self._git(web_root, "init", "--quiet")
|
|
candidate = self._candidate(root, key="trusted-key")
|
|
target_root = web_root / "public" / "catalogs" / "v1"
|
|
target_catalog = target_root / "channels" / "stable.json"
|
|
target_keyring = target_root / "keyring.json"
|
|
target_keyring.parent.mkdir(parents=True)
|
|
target_keyring.write_text("{}\n", encoding="utf-8")
|
|
|
|
web_root.chmod(0o777)
|
|
root_issues = publication_mutation_trust_issues(
|
|
web_root=web_root,
|
|
candidate_catalog=candidate / "channels" / "stable.json",
|
|
candidate_keyring=candidate / "keyring.json",
|
|
target_catalog=target_catalog,
|
|
target_keyring=target_keyring,
|
|
target_modules=target_root / "modules",
|
|
)
|
|
web_root.chmod(0o755)
|
|
config = web_root / ".git" / "config"
|
|
config.chmod(0o666)
|
|
config_issues = publication_mutation_trust_issues(
|
|
web_root=web_root,
|
|
candidate_catalog=candidate / "channels" / "stable.json",
|
|
candidate_keyring=candidate / "keyring.json",
|
|
target_catalog=target_catalog,
|
|
target_keyring=target_keyring,
|
|
target_modules=target_root / "modules",
|
|
)
|
|
|
|
self.assertIn("website root is writable by another user", root_issues)
|
|
self.assertIn("website Git config is writable by another user", config_issues)
|
|
|
|
def test_push_returns_only_independently_verified_publication_receipt(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
candidate = self._candidate(root, key="trusted-key")
|
|
catalog_path = candidate / "channels" / "stable.json"
|
|
catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
catalog["sequence"] = 7
|
|
catalog["core_release"]["python_package"] = "govoplan-core"
|
|
catalog["release"] = {
|
|
"selected_units": [
|
|
{
|
|
"repo": "govoplan-core",
|
|
"version": "1.2.3",
|
|
"tag": "v1.2.3",
|
|
"commit_sha": "a" * 40,
|
|
"tag_object_sha": "b" * 40,
|
|
}
|
|
],
|
|
"artifacts": [
|
|
{
|
|
"artifact_kind": "python-wheel",
|
|
"package_name": "govoplan-core",
|
|
"package_version": "1.2.3",
|
|
"archive_sha256": "c" * 64,
|
|
"archive_size": 100,
|
|
"installed_payload": {
|
|
"algorithm": "govoplan-wheel-declared-payload-v1",
|
|
"sha256": "d" * 64,
|
|
"file_count": 1,
|
|
},
|
|
"requires_installer_receipt": True,
|
|
}
|
|
],
|
|
}
|
|
catalog["signatures"] = [{}]
|
|
catalog_path.write_text(json.dumps(catalog), encoding="utf-8")
|
|
|
|
remote_root = root / "website.git"
|
|
remote_root.mkdir()
|
|
self._git(remote_root, "init", "--bare", "--quiet")
|
|
web_root = root / "website"
|
|
target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
target_keyring.parent.mkdir(parents=True)
|
|
target_keyring.write_text(json.dumps(self._keyring("trusted-key")))
|
|
self._git(web_root, "init", "--quiet")
|
|
self._git(web_root, "config", "user.email", "test@example.test")
|
|
self._git(web_root, "config", "user.name", "Test")
|
|
self._git(web_root, "branch", "-M", "main")
|
|
self._git(web_root, "remote", "add", "origin", str(remote_root))
|
|
self._git(web_root, "add", ".")
|
|
self._git(web_root, "commit", "--quiet", "-m", "base")
|
|
self._git(web_root, "push", "--quiet", "-u", "origin", "main")
|
|
frozen_head = self._git(web_root, "rev-parse", "HEAD").strip()
|
|
frozen_remote = bind_publication_remote(
|
|
web_root=web_root,
|
|
remote="origin",
|
|
registered_remote=str(remote_root),
|
|
)
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_release.publisher.validate_module_package_catalog",
|
|
return_value={"valid": True, "warnings": [], "error": None},
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.source_tag_provenance_issues",
|
|
return_value=(),
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.publication_runtime_trust_issues",
|
|
return_value=(),
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.registered_website_remote",
|
|
return_value=str(remote_root),
|
|
),
|
|
):
|
|
result = publish_catalog_candidate(
|
|
candidate_dir=candidate,
|
|
web_root=web_root,
|
|
workspace_root=root,
|
|
apply=True,
|
|
push=True,
|
|
branch="main",
|
|
tag_name="catalog-test",
|
|
expected_website_head=frozen_head,
|
|
expected_website_branch="main",
|
|
expected_remote_sha256=frozen_remote.sha256,
|
|
)
|
|
|
|
remote_identity = remote_publication_identity(
|
|
web_root=web_root,
|
|
remote_url=str(remote_root),
|
|
branch="main",
|
|
tag_name="catalog-test",
|
|
)
|
|
|
|
self.assertEqual("published", result.status)
|
|
self.assertEqual("origin", result.remote)
|
|
self.assertEqual(
|
|
remote_identity["publication_commit_sha"], result.publication_commit_sha
|
|
)
|
|
self.assertEqual(
|
|
remote_identity["publication_tag_object_sha"],
|
|
result.publication_tag_object_sha,
|
|
)
|
|
self.assertEqual(
|
|
remote_identity["publication_tag_commit_sha"],
|
|
result.publication_tag_commit_sha,
|
|
)
|
|
|
|
def test_apply_writes_validated_objects_even_if_candidate_path_changes(
|
|
self,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
candidate = self._candidate(root, key="trusted-key")
|
|
catalog_path = candidate / "channels" / "stable.json"
|
|
catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
catalog.update({"channel": "stable", "sequence": 1})
|
|
catalog["core_release"]["python_package"] = "govoplan-core"
|
|
catalog["release"] = {
|
|
"selected_units": [
|
|
{
|
|
"repo": "govoplan-core",
|
|
"version": "1.2.3",
|
|
"tag": "v1.2.3",
|
|
"commit_sha": "a" * 40,
|
|
"tag_object_sha": "b" * 40,
|
|
}
|
|
],
|
|
"artifacts": [
|
|
{
|
|
"artifact_kind": "python-wheel",
|
|
"package_name": "govoplan-core",
|
|
"package_version": "1.2.3",
|
|
"archive_sha256": "c" * 64,
|
|
"archive_size": 100,
|
|
"installed_payload": {
|
|
"algorithm": "govoplan-wheel-declared-payload-v1",
|
|
"sha256": "d" * 64,
|
|
"file_count": 1,
|
|
},
|
|
"requires_installer_receipt": True,
|
|
}
|
|
],
|
|
}
|
|
catalog["signatures"] = [{}]
|
|
catalog_path.write_text(json.dumps(catalog), encoding="utf-8")
|
|
web_root = root / "website"
|
|
target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
target_keyring.parent.mkdir(parents=True)
|
|
target_keyring.write_text(json.dumps(self._keyring("trusted-key")))
|
|
registered_remote = "ssh://example.test/release/website.git"
|
|
self._git(web_root, "init", "--quiet")
|
|
self._git(web_root, "config", "user.email", "test@example.test")
|
|
self._git(web_root, "config", "user.name", "Test")
|
|
self._git(web_root, "remote", "add", "origin", registered_remote)
|
|
self._git(web_root, "add", ".")
|
|
self._git(web_root, "commit", "--quiet", "-m", "base")
|
|
|
|
def validate_and_swap(*args, **kwargs):
|
|
del args, kwargs
|
|
catalog_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"channel": "stable",
|
|
"sequence": 999,
|
|
"malicious": True,
|
|
"signatures": [{}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return {"valid": True, "warnings": [], "error": None}
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_release.publisher.validate_module_package_catalog",
|
|
side_effect=validate_and_swap,
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.source_tag_provenance_issues",
|
|
return_value=(),
|
|
),
|
|
patch("govoplan_release.publisher.website_dirty", return_value=False),
|
|
patch(
|
|
"govoplan_release.publisher.publication_runtime_trust_issues",
|
|
return_value=(),
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.registered_website_remote",
|
|
return_value=registered_remote,
|
|
),
|
|
):
|
|
result = publish_catalog_candidate(
|
|
candidate_dir=candidate,
|
|
web_root=web_root,
|
|
workspace_root=root,
|
|
apply=True,
|
|
allow_dirty_website=True,
|
|
)
|
|
|
|
published = json.loads(
|
|
(
|
|
web_root / "public" / "catalogs" / "v1" / "channels" / "stable.json"
|
|
).read_text(encoding="utf-8")
|
|
)
|
|
|
|
self.assertEqual("applied", result.status)
|
|
self.assertEqual(1, published["sequence"])
|
|
self.assertNotIn("malicious", published)
|
|
|
|
def test_apply_blocks_selected_python_release_without_built_identity(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
candidate = self._candidate(root, key="trusted-key")
|
|
catalog_path = candidate / "channels" / "stable.json"
|
|
catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
|
|
catalog["core_release"]["python_package"] = "govoplan-core"
|
|
catalog["release"] = {
|
|
"selected_units": [
|
|
{
|
|
"repo": "govoplan-core",
|
|
"version": "1.2.3",
|
|
"tag": "v1.2.3",
|
|
"commit_sha": "a" * 40,
|
|
"tag_object_sha": "b" * 40,
|
|
}
|
|
]
|
|
}
|
|
catalog_path.write_text(json.dumps(catalog), encoding="utf-8")
|
|
web_root = root / "website"
|
|
target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
target_keyring.parent.mkdir(parents=True)
|
|
target_keyring.write_text(json.dumps(self._keyring("trusted-key")))
|
|
(web_root / ".git").mkdir()
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_release.publisher.validate_module_package_catalog",
|
|
return_value={"valid": True, "warnings": [], "error": None},
|
|
),
|
|
patch(
|
|
"govoplan_release.publisher.source_tag_provenance_issues",
|
|
return_value=(),
|
|
),
|
|
patch("govoplan_release.publisher.website_dirty", return_value=False),
|
|
):
|
|
result = publish_catalog_candidate(
|
|
candidate_dir=candidate,
|
|
web_root=web_root,
|
|
workspace_root=root,
|
|
apply=True,
|
|
)
|
|
|
|
self.assertEqual("blocked", result.status)
|
|
self.assertIn(
|
|
"selected Python repository govoplan-core has no built artifact identity",
|
|
" ".join(result.notes),
|
|
)
|
|
|
|
def test_publication_requires_an_existing_trust_anchor(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
candidate = self._candidate(root, key="candidate-key")
|
|
web_root = root / "website"
|
|
web_root.mkdir()
|
|
|
|
with patch(
|
|
"govoplan_release.publisher.validate_module_package_catalog",
|
|
return_value={"valid": True, "warnings": [], "error": None},
|
|
):
|
|
result = publish_catalog_candidate(
|
|
candidate_dir=candidate,
|
|
web_root=web_root,
|
|
workspace_root=root,
|
|
)
|
|
|
|
self.assertEqual("blocked", result.status)
|
|
self.assertIn("publication trust anchor is missing", " ".join(result.notes))
|
|
|
|
def test_publication_rejects_rebinding_an_existing_key_id(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
candidate = self._candidate(root, key="replacement-key")
|
|
web_root = root / "website"
|
|
target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
|
|
target_keyring.parent.mkdir(parents=True)
|
|
target_keyring.write_text(json.dumps(self._keyring("trusted-key")))
|
|
|
|
with patch(
|
|
"govoplan_release.publisher.validate_module_package_catalog",
|
|
return_value={"valid": True, "warnings": [], "error": None},
|
|
):
|
|
result = publish_catalog_candidate(
|
|
candidate_dir=candidate,
|
|
web_root=web_root,
|
|
workspace_root=root,
|
|
)
|
|
|
|
self.assertEqual("blocked", result.status)
|
|
self.assertIn("changes the public key", " ".join(result.notes))
|
|
|
|
@staticmethod
|
|
def _candidate(root: Path, *, key: str) -> Path:
|
|
candidate = root / "candidate"
|
|
channel = candidate / "channels"
|
|
channel.mkdir(parents=True)
|
|
channel.joinpath("stable.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"channel": "stable",
|
|
"core_release": {
|
|
"version": "1.2.3",
|
|
"python_ref": (
|
|
"govoplan-core @ git+ssh://git@example.test/acme/"
|
|
"govoplan-core.git@v1.2.3"
|
|
),
|
|
},
|
|
"modules": [],
|
|
}
|
|
)
|
|
)
|
|
candidate.joinpath("keyring.json").write_text(
|
|
json.dumps(ReleaseCatalogPublicationTests._keyring(key))
|
|
)
|
|
return candidate
|
|
|
|
@staticmethod
|
|
def _keyring(key: str) -> dict[str, object]:
|
|
return {
|
|
"keys": [
|
|
{
|
|
"key_id": "release-key",
|
|
"status": "active",
|
|
"public_key": key,
|
|
}
|
|
]
|
|
}
|
|
|
|
@staticmethod
|
|
def _git(root: Path, *args: str) -> str:
|
|
return subprocess.run(
|
|
["git", *args],
|
|
cwd=root,
|
|
check=True,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
).stdout
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|