feat(release): publish verified source tags

tests/test_release_source_provenance.py

@@ -0,0 +1,335 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from fastapi.testclient import TestClient
+
+
+META_ROOT = Path(__file__).resolve().parents[1]
+RELEASE_ROOT = META_ROOT / "tools" / "release"
+if str(RELEASE_ROOT) not in sys.path:
+    sys.path.insert(0, str(RELEASE_ROOT))
+
+from govoplan_release.publisher import publish_catalog_candidate  # noqa: E402
+from govoplan_release.selective_catalog import (  # noqa: E402
+    build_selective_catalog_candidate,
+    enforce_selected_source_provenance,
+)
+from govoplan_release.source_provenance import (  # noqa: E402
+    catalog_source_selection,
+    read_only_https_remote,
+    source_tag_provenance_issues,
+    tagged_source_version_issues,
+)
+from server.app import create_app  # noqa: E402
+
+
+class ReleaseSourceProvenanceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        self.workspace = self.root / "workspace"
+        self.workspace.mkdir()
+
+    def tearDown(self) -> None:
+        self.temporary.cleanup()
+
+    def test_selected_source_requires_local_and_remote_annotated_tag_at_clean_head(self) -> None:
+        repo, remote = make_repo(self.workspace, self.root, "govoplan-core", "1.2.3")
+
+        missing = source_tag_provenance_issues(
+            repo_versions={"govoplan-core": "1.2.3"},
+            workspace=self.workspace,
+            require_head_repos=("govoplan-core",),
+        )
+        self.assertIn("missing from the local source repository", messages(missing))
+        self.assertIn("missing from remote 'origin'", messages(missing))
+
+        git(repo, "tag", "-a", "v1.2.3", "-m", "Core 1.2.3")
+        git(repo, "push", "origin", "refs/tags/v1.2.3")
+        self.assertEqual(
+            (),
+            source_tag_provenance_issues(
+                repo_versions={"govoplan-core": "1.2.3"},
+                workspace=self.workspace,
+                require_head_repos=("govoplan-core",),
+            ),
+        )
+
+        git(remote, "config", "user.name", "Different Release Actor")
+        git(remote, "config", "user.email", "different@example.invalid")
+        git(remote, "tag", "-f", "-a", "v1.2.3", git_text(repo, "rev-parse", "HEAD"), "-m", "Different annotation")
+        inconsistent = source_tag_provenance_issues(
+            repo_versions={"govoplan-core": "1.2.3"},
+            workspace=self.workspace,
+            require_head_repos=("govoplan-core",),
+        )
+        self.assertIn("local and remote annotated tag objects differ", messages(inconsistent))
+
+        (repo / "dirty.txt").write_text("not reviewed\n", encoding="utf-8")
+        dirty = source_tag_provenance_issues(
+            repo_versions={"govoplan-core": "1.2.3"},
+            workspace=self.workspace,
+            require_head_repos=("govoplan-core",),
+        )
+        self.assertIn("selected source worktree is not clean", messages(dirty))
+        self.assertEqual(git_text(remote, "rev-parse", "refs/tags/v1.2.3^{commit}"), git_text(repo, "rev-parse", "HEAD"))
+
+    def test_selected_source_gate_is_fail_closed_before_signing(self) -> None:
+        make_repo(self.workspace, self.root, "govoplan-core", "1.2.3")
+
+        with self.assertRaisesRegex(ValueError, "Source tag provenance gate failed") as error:
+            enforce_selected_source_provenance(
+                repo_versions={"govoplan-core": "1.2.3"},
+                workspace=self.workspace,
+            )
+
+        self.assertIn("v1.2.3", str(error.exception))
+        self.assertIn("missing", str(error.exception))
+
+    def test_historical_tag_uses_installable_and_manifest_version_metadata(self) -> None:
+        repo, _ = make_repo(self.workspace, self.root, "govoplan-core", "1.2.3")
+        package = repo / "src" / "govoplan_core"
+        package.mkdir(parents=True)
+        (package / "__init__.py").write_text('__version__ = "1.2.2"\n', encoding="utf-8")
+        git(repo, "add", "src/govoplan_core/__init__.py")
+        git(repo, "commit", "-m", "Keep legacy runtime version string")
+        git(repo, "tag", "-a", "v1.2.3", "-m", "Core 1.2.3")
+
+        issues = tagged_source_version_issues(
+            path=repo,
+            repo="govoplan-core",
+            tag="v1.2.3",
+            expected_version="1.2.3",
+        )
+
+        self.assertEqual((), issues)
+
+    def test_signed_candidate_records_selected_commit_and_tag_object(self) -> None:
+        core, _ = make_repo(self.workspace, self.root, "govoplan-core", "1.2.3")
+        git(core, "tag", "-a", "v1.2.3", "-m", "Core 1.2.3")
+        git(core, "push", "origin", "refs/tags/v1.2.3")
+        base_catalog = self.root / "base.json"
+        base_catalog.write_text(
+            json.dumps(
+                {
+                    "channel": "stable",
+                    "sequence": 1,
+                    "core_release": {
+                        "version": "1.2.2",
+                        "python_ref": "govoplan-core @ git+ssh://git@example.test/acme/govoplan-core.git@v1.2.2",
+                    },
+                    "modules": [],
+                    "release": {"version": "1.2.2", "tag": "v1.2.2"},
+                }
+            ),
+            encoding="utf-8",
+        )
+        private_key = Ed25519PrivateKey.generate()
+        key_path = self.root / "release.pem"
+        key_path.write_bytes(
+            private_key.private_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PrivateFormat.PKCS8,
+                encryption_algorithm=serialization.NoEncryption(),
+            )
+        )
+        output = self.root / "candidate"
+
+        with patch(
+            "govoplan_release.selective_catalog.validate_module_package_catalog",
+            return_value={"valid": True, "warnings": [], "error": None},
+        ):
+            result = build_selective_catalog_candidate(
+                repo_versions={"govoplan-core": "1.2.3"},
+                workspace_root=self.workspace,
+                output_dir=output,
+                base_catalog=base_catalog,
+                signing_keys=(f"test-key={key_path}",),
+                check_public=False,
+            )
+
+        payload = json.loads(Path(result.catalog_path).read_text(encoding="utf-8"))
+        selected = payload["release"]["selected_units"][0]
+        self.assertEqual(git_text(core, "rev-parse", "refs/tags/v1.2.3^{commit}"), selected["commit_sha"])
+        self.assertEqual(git_text(core, "rev-parse", "refs/tags/v1.2.3"), selected["tag_object_sha"])
+        self.assertEqual("test-key", payload["signatures"][0]["key_id"])
+
+    def test_catalog_publication_checks_every_preserved_source_ref(self) -> None:
+        core, _ = make_repo(self.workspace, self.root, "govoplan-core", "1.2.3")
+        git(core, "tag", "-a", "v1.2.3", "-m", "Core 1.2.3")
+        git(core, "push", "origin", "refs/tags/v1.2.3")
+        make_repo(self.workspace, self.root, "govoplan-access", "1.1.0")
+
+        candidate = self.root / "candidate"
+        (candidate / "channels").mkdir(parents=True)
+        payload = catalog_payload(
+            commit_sha="f" * 40,
+            tag_object_sha=git_text(core, "rev-parse", "refs/tags/v1.2.3"),
+        )
+        (candidate / "channels" / "stable.json").write_text(json.dumps(payload), encoding="utf-8")
+        keyring = {"keys": [{"key_id": "release-key", "status": "active", "public_key": "trusted"}]}
+        (candidate / "keyring.json").write_text(json.dumps(keyring), encoding="utf-8")
+        web_root = self.root / "website"
+        target_keyring = web_root / "public" / "catalogs" / "v1" / "keyring.json"
+        target_keyring.parent.mkdir(parents=True)
+        target_keyring.write_text(json.dumps(keyring), encoding="utf-8")
+
+        with patch(
+            "govoplan_release.publisher.validate_module_package_catalog",
+            return_value={"valid": True, "warnings": [], "error": None},
+        ):
+            result = publish_catalog_candidate(
+                candidate_dir=candidate,
+                workspace_root=self.workspace,
+                web_root=web_root,
+            )
+
+        detail = " ".join(result.notes)
+        self.assertEqual("blocked", result.status)
+        self.assertIn("govoplan-access v1.1.0", detail)
+        self.assertIn("missing from the local source repository", detail)
+        self.assertIn("missing from remote 'origin'", detail)
+        self.assertIn("no longer matches signed commit_sha", detail)
+        self.assertNotIn("govoplan-core v1.2.3: annotated tag is missing", detail)
+
+    def test_catalog_selection_includes_core_and_all_modules(self) -> None:
+        selection = catalog_source_selection(catalog_payload())
+
+        self.assertEqual(
+            {"govoplan-core": "1.2.3", "govoplan-access": "1.1.0"},
+            selection.all_versions,
+        )
+        self.assertEqual({"govoplan-core": "1.2.3"}, selection.selected_versions)
+        self.assertEqual((), selection.issues)
+
+    def test_console_renders_publication_provenance_notes(self) -> None:
+        ui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8")
+
+        self.assertIn("Array.isArray(result.notes)", ui)
+        self.assertIn("Validation and provenance details", ui)
+
+    def test_console_returns_actionable_conflict_for_candidate_provenance_gate(self) -> None:
+        with patch(
+            "server.app.build_selective_catalog_candidate",
+            side_effect=ValueError("Complete catalog source provenance gate failed: govoplan-core v1.2.3: missing"),
+        ):
+            with TestClient(create_app(workspace_root=self.workspace)) as client:
+                response = client.post(
+                    "/api/catalog-candidates",
+                    json={
+                        "repo_versions": {"govoplan-core": "1.2.3"},
+                        "signing_keys": ["test=/unused/key.pem"],
+                    },
+                )
+
+        self.assertEqual(409, response.status_code)
+        self.assertIn("govoplan-core v1.2.3: missing", response.json()["detail"])
+
+    def test_read_only_ref_checks_can_reuse_the_same_gitea_https_endpoint(self) -> None:
+        self.assertEqual(
+            "https://git.example.test/acme/govoplan-core.git",
+            read_only_https_remote("git@git.example.test:acme/govoplan-core.git"),
+        )
+        self.assertEqual(
+            "https://git.example.test/acme/govoplan-core.git",
+            read_only_https_remote("ssh://git@git.example.test/acme/govoplan-core.git"),
+        )
+        self.assertIsNone(read_only_https_remote("ssh://git@git.example.test:2222/acme/govoplan-core.git"))
+        self.assertIsNone(read_only_https_remote("/srv/git/govoplan-core.git"))
+
+
+def catalog_payload(
+    *,
+    commit_sha: str = "0" * 40,
+    tag_object_sha: str = "1" * 40,
+) -> dict[str, object]:
+    return {
+        "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": [
+            {
+                "module_id": "access",
+                "version": "1.1.0",
+                "python_package": "govoplan-access",
+                "python_ref": "govoplan-access @ git+ssh://git@example.test/acme/govoplan-access.git@v1.1.0",
+            }
+        ],
+        "release": {
+            "version": "1.2.3",
+            "tag": "v1.2.3",
+            "selected_units": [
+                {
+                    "repo": "govoplan-core",
+                    "version": "1.2.3",
+                    "tag": "v1.2.3",
+                    "commit_sha": commit_sha,
+                    "tag_object_sha": tag_object_sha,
+                },
+            ],
+        },
+    }
+
+
+def make_repo(workspace: Path, root: Path, name: str, version: str) -> tuple[Path, Path]:
+    remote = root / f"{name}.git"
+    repo = workspace / name
+    git(root, "init", "--bare", str(remote))
+    git(workspace, "init", "-b", "main", str(repo))
+    git(repo, "config", "user.name", "GovOPlaN Release Test")
+    git(repo, "config", "user.email", "release-test@example.invalid")
+    (repo / "pyproject.toml").write_text(
+        f'[project]\nname = "{name}"\nversion = "{version}"\n',
+        encoding="utf-8",
+    )
+    git(repo, "add", "pyproject.toml")
+    git(repo, "commit", "-m", f"{name} {version}")
+    git(repo, "remote", "add", "origin", str(remote))
+    git(repo, "push", "-u", "origin", "main")
+    return repo, remote
+
+
+def messages(issues: object) -> str:
+    return " ".join(issue.message for issue in issues)  # type: ignore[attr-defined]
+
+
+def git(cwd: Path, *args: str) -> None:
+    result = subprocess.run(
+        ("git", *args),
+        cwd=cwd,
+        check=False,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    if result.returncode != 0:
+        raise AssertionError(f"git {' '.join(args)} failed: {result.stderr or result.stdout}")
+
+
+def git_text(cwd: Path, *args: str) -> str:
+    result = subprocess.run(
+        ("git", *args),
+        cwd=cwd,
+        check=False,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    if result.returncode != 0:
+        raise AssertionError(f"git {' '.join(args)} failed: {result.stderr or result.stdout}")
+    return result.stdout.strip()
+
+
+if __name__ == "__main__":
+    unittest.main()