[codex] Ensure PGP public key is available before Central release

.github/scripts/ensure_pgp_public_key_available.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Sequence
+
+
+DEFAULT_KEYSERVERS = ("keyserver.ubuntu.com",)
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+class GpgCommandError(RuntimeError):
+    pass
+
+
+def parse_primary_secret_key_fingerprints(colon_output: str) -> list[str]:
+    fingerprints: list[str] = []
+    waiting_for_primary_fingerprint = False
+
+    for line in colon_output.splitlines():
+        if not line:
+            continue
+
+        fields = line.split(":")
+        record_type = fields[0]
+        if record_type.startswith("sec"):
+            waiting_for_primary_fingerprint = True
+            continue
+        if record_type == "fpr" and waiting_for_primary_fingerprint:
+            fingerprint = fields[9].strip().upper() if len(fields) > 9 else ""
+            if fingerprint and fingerprint not in fingerprints:
+                fingerprints.append(fingerprint)
+            waiting_for_primary_fingerprint = False
+            continue
+        if record_type in {"pub", "sub", "ssb", "uid"}:
+            waiting_for_primary_fingerprint = False
+
+    return fingerprints
+
+
+def run_command(command: Sequence[str], *, gnupg_home: Path, input_text: str | None = None) -> CommandResult:
+    env = os.environ.copy()
+    env["GNUPGHOME"] = str(gnupg_home)
+    try:
+        completed = subprocess.run(
+            list(command),
+            input=input_text,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            env=env,
+            timeout=120,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as error:
+        return CommandResult(
+            124,
+            error.stdout if isinstance(error.stdout, str) else "",
+            error.stderr if isinstance(error.stderr, str) else "gpg command timed out",
+        )
+
+    return CommandResult(completed.returncode, completed.stdout, completed.stderr)
+
+
+def require_success(result: CommandResult, action: str) -> None:
+    if result.returncode == 0:
+        return
+
+    message = (result.stderr or result.stdout or "no output").strip()
+    raise GpgCommandError(f"{action} failed: {message}")
+
+
+def make_gnupg_home(path: str) -> Path:
+    home = Path(path)
+    home.chmod(0o700)
+    return home
+
+
+def verify_keyserver_has_key(
+    *,
+    fingerprint: str,
+    keyserver: str,
+    gpg: str,
+    runner: Callable[[Sequence[str]], CommandResult],
+    retries: int,
+    retry_delay_seconds: int,
+    sleep: Callable[[int], None],
+) -> bool:
+    for attempt in range(1, retries + 1):
+        result = runner(
+            [
+                gpg,
+                "--batch",
+                "--keyserver",
+                keyserver,
+                "--recv-keys",
+                fingerprint,
+            ]
+        )
+        if result.returncode == 0:
+            return True
+        if attempt < retries:
+            sleep(retry_delay_seconds)
+
+    return False
+
+
+def ensure_public_keys_available(
+    *,
+    key_contents: str,
+    keyservers: Sequence[str] | None = None,
+    gpg: str = "gpg",
+    runner: Callable[[Sequence[str]], CommandResult] | None = None,
+    retries: int = 6,
+    retry_delay_seconds: int = 10,
+    sleep: Callable[[int], None] = time.sleep,
+) -> list[str]:
+    if not key_contents.strip():
+        raise ValueError("GPG_KEY_CONTENTS is required")
+    if retries < 1:
+        raise ValueError("retries must be at least 1")
+
+    selected_keyservers = tuple(keyservers or DEFAULT_KEYSERVERS)
+    if not selected_keyservers:
+        raise ValueError("at least one keyserver is required")
+
+    with tempfile.TemporaryDirectory() as import_dir, tempfile.TemporaryDirectory() as verify_dir:
+        import_home = make_gnupg_home(import_dir)
+        verify_home = make_gnupg_home(verify_dir)
+
+        def run_in_import_home(command: Sequence[str], input_text: str | None = None) -> CommandResult:
+            active_runner = runner or run_command
+            return active_runner(command, gnupg_home=import_home, input_text=input_text)
+
+        def run_in_verify_home(command: Sequence[str]) -> CommandResult:
+            active_runner = runner or run_command
+            return active_runner(command, gnupg_home=verify_home, input_text=None)
+
+        require_success(
+            run_in_import_home([gpg, "--batch", "--import"], input_text=key_contents),
+            "Importing GPG private key",
+        )
+        listing = run_in_import_home([gpg, "--batch", "--with-colons", "--list-secret-keys", "--fingerprint"])
+        require_success(listing, "Listing GPG secret keys")
+
+        fingerprints = parse_primary_secret_key_fingerprints(listing.stdout)
+        if not fingerprints:
+            raise GpgCommandError("No primary secret key fingerprint found in GPG_KEY_CONTENTS")
+
+        for fingerprint in fingerprints:
+            publish_errors: list[str] = []
+            for keyserver in selected_keyservers:
+                send_result = run_in_import_home(
+                    [gpg, "--batch", "--keyserver", keyserver, "--send-keys", fingerprint]
+                )
+                if send_result.returncode != 0:
+                    publish_errors.append((send_result.stderr or send_result.stdout or "").strip())
+                    continue
+
+                if verify_keyserver_has_key(
+                    fingerprint=fingerprint,
+                    keyserver=keyserver,
+                    gpg=gpg,
+                    runner=run_in_verify_home,
+                    retries=retries,
+                    retry_delay_seconds=retry_delay_seconds,
+                    sleep=sleep,
+                ):
+                    break
+                publish_errors.append(f"{keyserver} did not return key {fingerprint} after upload")
+            else:
+                detail = "; ".join(error for error in publish_errors if error) or "no keyserver accepted the key"
+                raise GpgCommandError(f"Could not publish PGP public key {fingerprint}: {detail}")
+
+        return fingerprints
+
+
+def parse_args(argv: Sequence[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Publish and verify a PGP public key for Maven Central.")
+    parser.add_argument("--key-env", default="GPG_KEY_CONTENTS", help="Environment variable containing the private key")
+    parser.add_argument("--keyserver", action="append", dest="keyservers", help="GPG keyserver to publish to")
+    parser.add_argument("--gpg", default="gpg", help="gpg executable")
+    parser.add_argument("--retries", type=int, default=6, help="Keyserver lookup attempts after upload")
+    parser.add_argument("--retry-delay-seconds", type=int, default=10, help="Delay between lookup attempts")
+    return parser.parse_args(argv)
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = parse_args(sys.argv[1:] if argv is None else argv)
+    key_contents = os.environ.get(args.key_env, "")
+
+    try:
+        fingerprints = ensure_public_keys_available(
+            key_contents=key_contents,
+            keyservers=args.keyservers or DEFAULT_KEYSERVERS,
+            gpg=args.gpg,
+            retries=args.retries,
+            retry_delay_seconds=args.retry_delay_seconds,
+        )
+    except (GpgCommandError, ValueError) as error:
+        print(f"::error::{error}", file=sys.stderr)
+        return 1
+
+    for fingerprint in fingerprints:
+        print(f"PGP public key {fingerprint} is available on a Maven Central supported keyserver.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/ensure_pgp_public_key_available_test.py

@@ -0,0 +1,130 @@
+import unittest
+
+from ensure_pgp_public_key_available import (
+    CommandResult,
+    ensure_public_keys_available,
+    parse_primary_secret_key_fingerprints,
+)
+
+
+SECRET_KEY_LISTING = """\
+sec:u:3072:1:8190C4130ABA0F98:1624478814:1782245214::u:::scESC:::+:::23::0:
+fpr:::::::::CA925CD6C9E8D064FF05B4728190C4130ABA0F98:
+grp:::::::::0123456789ABCDEF:
+uid:u::::1624478814::1234567890ABCDEF::Central Repo Test <[email protected]>::::::::::0:
+ssb:u:3072:1:4656B4857C17C93B:1624478814:1782245214:::::e:::+:::23:
+fpr:::::::::515D4F4A1EDCCA2E7C904FA84656B4857C17C93B:
+"""
+
+
+class FakeGpgRunner:
+    def __init__(self):
+        self.calls = []
+        self.recv_attempts = 0
+
+    def __call__(self, command, *, gnupg_home, input_text=None):
+        self.calls.append((command, gnupg_home, input_text))
+        if "--list-secret-keys" in command:
+            return CommandResult(0, SECRET_KEY_LISTING, "")
+        return CommandResult(0, "", "")
+
+
+class EnsurePgpPublicKeyAvailableTest(unittest.TestCase):
+    def test_parses_primary_secret_key_fingerprint(self):
+        self.assertEqual(
+            parse_primary_secret_key_fingerprints(SECRET_KEY_LISTING),
+            ["CA925CD6C9E8D064FF05B4728190C4130ABA0F98"],
+        )
+
+    def test_parses_primary_fingerprint_from_secret_key_record_variants(self):
+        listing = SECRET_KEY_LISTING.replace("sec:u:", "sec#:", 1)
+
+        self.assertEqual(
+            parse_primary_secret_key_fingerprints(listing),
+            ["CA925CD6C9E8D064FF05B4728190C4130ABA0F98"],
+        )
+
+    def test_imports_private_key_then_publishes_and_verifies_public_key(self):
+        runner = FakeGpgRunner()
+
+        fingerprints = ensure_public_keys_available(
+            key_contents="secret-key",
+            keyservers=["keyserver.ubuntu.com"],
+            runner=runner,
+            retries=1,
+            sleep=lambda _: None,
+        )
+
+        self.assertEqual(fingerprints, ["CA925CD6C9E8D064FF05B4728190C4130ABA0F98"])
+        commands = [call[0] for call in runner.calls]
+        self.assertEqual(commands[0], ["gpg", "--batch", "--import"])
+        self.assertEqual(commands[1], ["gpg", "--batch", "--with-colons", "--list-secret-keys", "--fingerprint"])
+        self.assertEqual(
+            commands[2],
+            [
+                "gpg",
+                "--batch",
+                "--keyserver",
+                "keyserver.ubuntu.com",
+                "--send-keys",
+                "CA925CD6C9E8D064FF05B4728190C4130ABA0F98",
+            ],
+        )
+        self.assertEqual(
+            commands[3],
+            [
+                "gpg",
+                "--batch",
+                "--keyserver",
+                "keyserver.ubuntu.com",
+                "--recv-keys",
+                "CA925CD6C9E8D064FF05B4728190C4130ABA0F98",
+            ],
+        )
+        self.assertEqual(runner.calls[0][2], "secret-key")
+        self.assertEqual(runner.calls[0][1], runner.calls[1][1])
+        self.assertNotEqual(runner.calls[0][1], runner.calls[3][1])
+
+    def test_retries_public_key_lookup_until_keyserver_has_the_key(self):
+        class RetryRunner(FakeGpgRunner):
+            def __call__(self, command, *, gnupg_home, input_text=None):
+                if "--recv-keys" in command:
+                    self.recv_attempts += 1
+                    self.calls.append((command, gnupg_home, input_text))
+                    if self.recv_attempts == 1:
+                        return CommandResult(2, "", "not found")
+                    return CommandResult(0, "", "")
+                return super().__call__(command, gnupg_home=gnupg_home, input_text=input_text)
+
+        runner = RetryRunner()
+        sleeps = []
+
+        ensure_public_keys_available(
+            key_contents="secret-key",
+            keyservers=["keyserver.ubuntu.com"],
+            runner=runner,
+            retries=2,
+            retry_delay_seconds=3,
+            sleep=sleeps.append,
+        )
+
+        self.assertEqual(runner.recv_attempts, 2)
+        self.assertEqual(sleeps, [3])
+
+    def test_uses_custom_gpg_executable_for_all_commands(self):
+        runner = FakeGpgRunner()
+
+        ensure_public_keys_available(
+            key_contents="secret-key",
+            keyservers=["keyserver.ubuntu.com"],
+            gpg="gpg2",
+            runner=runner,
+            retries=1,
+            sleep=lambda _: None,
+        )
+
+        self.assertTrue(all(call[0][0] == "gpg2" for call in runner.calls))
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/publish_plugin_central_workflow_test.py

@@ -59,6 +59,21 @@ def test_central_publish_falls_back_when_signing_key_id_is_invalid(self):
         self.assertIn("falling back to infer it from GPG_KEY_CONTENTS", central_publish)
         self.assertIn("normalized to a Gradle-compatible long key id", central_publish)
 
+    def test_central_publish_makes_pgp_public_key_available_before_upload(self):
+        text = workflow_text()
+        public_key = step_block("Ensure PGP public key is available")
+
+        self.assertLess(
+            text.index("- name: Ensure PGP public key is available"),
+            text.index("- name: Publish to Central staging"),
+        )
+        self.assertIn("steps.release.outputs.publish_required == 'true'", public_key)
+        self.assertIn("GPG_KEY_CONTENTS:", public_key)
+        self.assertIn("ensure_pgp_public_key_available.py", public_key)
+        self.assertIn("keyserver.ubuntu.com", public_key)
+        self.assertNotIn("CENTRAL_PASSWORD:", public_key)
+        self.assertNotIn("SIGNING_PASSWORD:", public_key)
+
     def test_central_publish_is_gated_by_pre_publish_vs_main_version(self):
         release = step_block("Resolve release version")
         check_tag = step_block("Check release tag")

.github/workflows/publish-plugin-central.yml

@@ -221,6 +221,14 @@ jobs:
           ./gradlew :plugin_base:publishToMavenLocal --stacktrace
           python3 .github/scripts/validate_publish_plugin_publications.py
 
+      - name: Ensure PGP public key is available
+        if: steps.release.outputs.publish_required == 'true' && steps.release_tag.outputs.exists != 'true'
+        env:
+          GPG_KEY_CONTENTS: ${{ secrets.GPG_KEY_CONTENTS }}
+        run: |
+          python3 .github/scripts/ensure_pgp_public_key_available.py \
+            --keyserver keyserver.ubuntu.com
+
       - name: Publish to Central staging
         if: steps.release.outputs.publish_required == 'true' && steps.release_tag.outputs.exists != 'true'
         env: