Build tagged releases and preserve publisher history

.github/scripts/build_payload.py

@@ -25,12 +25,13 @@
 import json
 import os
 from pathlib import Path
-import re
 import subprocess
 import sys
 import urllib.error
 import urllib.request
 
+from resolve_release import parse_release_tag
+
 
 MAX_CONTENT_BYTES = 256 * 1024
 
@@ -67,10 +68,10 @@ def find_package_dir(ext_id: str) -> str:
     return matches[0]
 
 
-def fetch_existing_releases(ext_id: str) -> list:
+def fetch_existing_releases(ext_id: str) -> list | None:
     """Fetch the extension's currently published releases for merge history.
 
-    Returns [] for a new extension (API 404) or when no API token is
+    Returns None for a new extension (API 404) or when no API token is
     configured (offline use). Any other failure is fatal: silently
     submitting without history would wipe the extension's changelog.
     Entries are reduced to the release fields the API schema
@@ -79,7 +80,7 @@ def fetch_existing_releases(ext_id: str) -> list:
     api_token = os.environ.get("API_TOKEN", "")
     api_base = os.environ.get("API_BASE_URL", "")
     if not api_token or not api_base:
-        return []
+        return None
 
     http_request = urllib.request.Request(
         f"{api_base.rstrip('/')}/extensions/{ext_id}",
@@ -90,7 +91,7 @@ def fetch_existing_releases(ext_id: str) -> list:
             body = json.load(response)
     except urllib.error.HTTPError as exc:
         if exc.code == 404:
-            return []
+            return None
         fail(f"Could not fetch existing releases for '{ext_id}': HTTP {exc.code}")
     except Exception as exc:
         fail(f"Could not fetch existing releases for '{ext_id}': {exc}")
@@ -130,7 +131,7 @@ def canonical_tag(tag: object, ext_id: str) -> object:
     return tag
 
 
-def merge_releases(new_release: dict, ext_id: str) -> list:
+def merge_releases(new_release: dict, ext_id: str, history: list) -> list:
     """Prepend the new release to the published history.
 
     Same-tag re-runs replace in place so a retried publish is idempotent;
@@ -144,16 +145,20 @@ def merge_releases(new_release: dict, ext_id: str) -> list:
     merged = [new_release]
     merged.extend(
         entry
-        for entry in fetch_existing_releases(ext_id)
+        for entry in history
         if canonical_tag(entry.get("tag"), ext_id) != new_release["tag"]
     )
     return merged[:MAX_RELEASES]
 
 
 def main() -> None:
     ext_id, version, tag = sys.argv[1], sys.argv[2], sys.argv[3]
-    if not re.fullmatch(r"[a-z0-9_-]+", ext_id) or len(ext_id) > 64:
-        fail(f"Extension id {ext_id!r} is not a valid extension id (1–64 lowercase letters, digits, underscores, or hyphens)")
+    try:
+        target = parse_release_tag(tag)
+    except ValueError as exc:
+        fail(str(exc))
+    if target != (ext_id, version):
+        fail("Extension id and version must match the release tag")
     ext_dir = find_package_dir(ext_id)
     manifest_path = os.path.join(ext_dir, "manifest.json")
     meta = json.loads(Path(manifest_path).read_text())
@@ -199,6 +204,10 @@ def main() -> None:
     if not readme.strip():
         fail(f"{manifest_path} has an empty description and no README.md to fall back on")
 
+    # Zod measures JavaScript strings in UTF-16 code units, not code points.
+    if len(readme.encode("utf-16-le")) // 2 > 100_000:
+        fail(f"{readme_path} exceeds the API's 100,000-character README limit")
+
     release = json.loads(
         subprocess.run(
             ["gh", "api", f"repos/{os.environ['GITHUB_REPOSITORY']}/releases/tags/{tag}"],
@@ -239,14 +248,15 @@ def main() -> None:
     license_name = meta["license"]
     if not isinstance(license_name, str) or not license_name.strip():
         fail(f"{manifest_path} license must be a non-empty license expression")
+    history = fetch_existing_releases(ext_id)
     content: dict = {
         "manifest_version": 1,
         "type": meta["type"],
         "author": meta["author"],
         "requires": requires,
         "name": meta["name"],
         "description": meta["description"],
-        "releases": merge_releases(new_release, ext_id),
+        "releases": merge_releases(new_release, ext_id, history or []),
         "license": license_name,
         "readme": readme,
         "source": {"type": "github", "repo": os.environ.get("GITHUB_REPOSITORY", "FOSSBilling/modules")},
@@ -263,14 +273,16 @@ def main() -> None:
     create_content = dict(content)
     create_content["id"] = ext_id
 
-    size = len(json.dumps(create_content).encode())
+    size = len(json.dumps(create_content, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
     if size > MAX_CONTENT_BYTES:
         fail("Built payload exceeds the 256 KiB content limit")
 
-    with open("payload-create.json", "w") as handle:
-        json.dump(create_content, handle)
-    with open("payload-update.json", "w") as handle:
-        json.dump(content, handle)
+    with open("payload-create.json", "w", encoding="utf-8") as handle:
+        json.dump(create_content, handle, ensure_ascii=False, separators=(",", ":"))
+    with open("payload-update.json", "w", encoding="utf-8") as handle:
+        json.dump(content, handle, ensure_ascii=False, separators=(",", ":"))
+    with open("payload-target.json", "w") as handle:
+        json.dump({"exists": history is not None}, handle)
     print(f"Built payloads for {ext_id} {version} ({size} bytes)")
 
 

.github/scripts/resolve_release.py

@@ -0,0 +1,27 @@
+"""Resolve a release tag without interpolating user input into shell code."""
+import json
+import os
+from pathlib import Path
+import re
+
+
+SCHEMA = json.loads((Path(__file__).resolve().parents[2] / "manifest.schema.json").read_text())
+
+
+def parse_release_tag(tag: str) -> tuple[str, str]:
+    match = re.fullmatch(r"([a-z0-9_-]{1,64})-v(.+)", tag)
+    if not match or len(match[2]) > 100 or not re.fullmatch(SCHEMA["properties"]["version"]["pattern"], match[2]):
+        raise ValueError("Release tag must match <extension-id>-v<semver>")
+    return match[1], match[2]
+
+
+def main() -> None:
+    tag = os.environ["RELEASE_TAG"]
+    ext_id, version = parse_release_tag(tag)
+    with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+        output.write(f"tag={tag}\nextension_id={ext_id}\nversion={version}\n")
+    print(f"Publishing {ext_id} version {version}")
+
+
+if __name__ == "__main__":
+    main()

.github/scripts/test_build_payload.py

@@ -41,7 +41,7 @@ def setUp(self):
 
     def build(self, history=None):
         (self.package / "manifest.json").write_text(json.dumps(self.meta))
-        with patch.dict(os.environ, {"GITHUB_REPOSITORY": "FOSSBilling/modules"}), patch.object(publisher.sys, "argv", ["build_payload.py", "example_module", "1.2.3", "example_module-v1.2.3"]), patch.object(publisher.subprocess, "run", return_value=SimpleNamespace(stdout=json.dumps(self.release))), patch.object(publisher, "fetch_existing_releases", return_value=history or []):
+        with patch.dict(os.environ, {"GITHUB_REPOSITORY": "FOSSBilling/modules"}), patch.object(publisher.sys, "argv", ["build_payload.py", "example_module", "1.2.3", "example_module-v1.2.3"]), patch.object(publisher.subprocess, "run", return_value=SimpleNamespace(stdout=json.dumps(self.release))), patch.object(publisher, "fetch_existing_releases", return_value=history):
             publisher.main()
         return json.loads(Path("payload-create.json").read_text()), json.loads(Path("payload-update.json").read_text())
 
@@ -78,6 +78,22 @@ def test_requires_a_distributable_and_matching_version(self):
         with self.assertRaises(SystemExit):
             self.build()
 
+    def test_readme_limit_matches_api_utf16_length(self):
+        (self.package / "README.md").write_text("x" * 100_001)
+        with self.assertRaises(SystemExit):
+            self.build()
+        (self.package / "README.md").write_text("😀" * 50_001)
+        with self.assertRaises(SystemExit):
+            self.build()
+        (self.package / "README.md").write_text("😀" * 50_000)
+        self.build()
+
+    def test_target_comes_from_the_history_read(self):
+        self.build()
+        self.assertEqual(json.loads(Path("payload-target.json").read_text()), {"exists": False})
+        self.build([{"tag": "1.0.0", "requires": self.meta["requires"]}])
+        self.assertEqual(json.loads(Path("payload-target.json").read_text()), {"exists": True})
+
     def test_requires_php(self):
         del self.meta["requires"]["php"]
         with self.assertRaises(SystemExit):
@@ -87,6 +103,17 @@ def test_release_cap_is_preserved(self):
         history = [{"tag": f"0.0.{index}"} for index in range(101)]
         self.assertEqual(len(self.build(history)[0]["releases"]), 100)
 
+    def test_fetch_only_treats_404_as_a_new_extension(self):
+        for code in [404, 401, 429, 500]:
+            error = publisher.urllib.error.HTTPError("https://api.example.test", code, "error", {}, None)
+            self.addCleanup(error.close)
+            with self.subTest(code=code), patch.dict(os.environ, {"API_TOKEN": "test", "API_BASE_URL": "https://api.example.test"}), patch.object(publisher.urllib.request, "urlopen", side_effect=error):
+                if code == 404:
+                    self.assertIsNone(publisher.fetch_existing_releases("example"))
+                else:
+                    with self.assertRaises(SystemExit):
+                        publisher.fetch_existing_releases("example")
+
     def test_fetch_retains_requires_and_rejects_pre_migration_history(self):
         requirements = {"fossbilling": "^0.8 || ^1", "php": "^8", "modules": {"branding": "*"}}
         entry = {"tag": "1.0.0", "requires": requirements, "unknown": "ignored"}

.github/scripts/test_publish_contract.py

@@ -0,0 +1,143 @@
+"""Execute the workflow's submit shell offline against scripted API replies."""
+import json
+import os
+import shutil
+from pathlib import Path
+import subprocess
+import tempfile
+import textwrap
+import unittest
+
+from resolve_release import parse_release_tag
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def workflow_run(step):
+    source = (ROOT / ".github/workflows/publish.yml").read_text()
+    block = source.split(f"      - name: {step}\n", 1)[1].split("\n      - ", 1)[0]
+    command = block.split("        run: ", 1)[1]
+    return textwrap.dedent(command[2:]) if command.startswith("|\n") else command.splitlines()[0]
+
+
+class ReleaseTargetTests(unittest.TestCase):
+    def test_valid_tags_preserve_prereleases_and_build_metadata(self):
+        for version in ["1.2.3", "1.2.3-v1.0.0", "1.2.3-alpha.1+build.4"]:
+            self.assertEqual(parse_release_tag(f"example_module-v{version}"), ("example_module", version))
+
+    def test_invalid_tags_are_rejected_before_outputs_or_checkout(self):
+        for tag in ["example-v01.2.3", "example-v1.2.3\nother=value", "example-v1.2.3;echo bad", "EXAMPLE-v1.2.3", "../example-v1.2.3"]:
+            with self.subTest(tag=tag), self.assertRaises(ValueError):
+                parse_release_tag(tag)
+
+
+class PublishWorkflowTests(unittest.TestCase):
+    def run_submit(self, exists, replies, step="Submit revision (create or update)", hostile_module=None, after_api=False):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / "payload-target.json").write_text(json.dumps({"exists": exists}))
+            (root / "replies.json").write_text(json.dumps(replies))
+            (root / "payload-create.json").write_text('{}')
+            (root / "payload-update.json").write_text('{}')
+            hostile_source = "from pathlib import Path\nPath('hostile-imported').write_text('executed')\nraise RuntimeError('tag-controlled import')\n"
+            if hostile_module and not after_api:
+                module = root / hostile_module
+                module.parent.mkdir(parents=True, exist_ok=True)
+                module.write_text(hostile_source)
+            curl = root / "curl"
+            curl.write_text('''#!/usr/bin/env -S python3 -I
+import json, os, pathlib, sys
+args = sys.argv[1:]
+root = pathlib.Path('.')
+replies = json.loads((root / 'replies.json').read_text())
+response = replies.pop(0)
+(root / 'replies.json').write_text(json.dumps(replies))
+method = args[args.index('-X') + 1]
+with open('calls.txt', 'a') as calls:
+    calls.write(method + ' ' + args[-1] + '\\n')
+assert method == response['method'], (method, response)
+if os.environ.get('HOSTILE_MODULE'):
+    module = root / os.environ['HOSTILE_MODULE']
+    module.parent.mkdir(parents=True, exist_ok=True)
+    module.write_text(os.environ['HOSTILE_SOURCE'])
+if '-o' in args:
+    (root / args[args.index('-o') + 1]).write_text(json.dumps(response['body']))
+    (root / args[args.index('-D') + 1]).write_text('Retry-After: 0\\n')
+    print(response['code'], end='')
+else:
+    print(json.dumps(response['body']) + '\\n' + str(response['code']))
+''')
+            curl.chmod(0o755)
+            env = {**os.environ, "PATH": f"{root}:{os.environ['PATH']}", "API_BASE_URL": "https://api.example.test/extensions/v2", "API_TOKEN": "offline", "EXT_ID": "example", "REVISION_ID": "revision", "GITHUB_OUTPUT": str(root / "outputs")}
+            if hostile_module and after_api:
+                env.update({"HOSTILE_MODULE": hostile_module, "HOSTILE_SOURCE": hostile_source})
+            result = subprocess.run(["bash", "-e", "-o", "pipefail", "-c", workflow_run(step)], cwd=root, env=env, capture_output=True, text=True)
+            self.assertFalse((root / 'hostile-imported').exists(), 'Tagged content executed during JSON parsing')
+            calls = (root / 'calls.txt').read_text().splitlines()
+            outputs = (root / 'outputs').read_text() if (root / 'outputs').exists() else ''
+            return result, calls, outputs
+
+    def test_tagged_modules_cannot_hijack_either_submit_parser(self):
+        for module in ["json.py", "json/__init__.py"]:
+            for after_api in [False, True]:
+                with self.subTest(module=module, after_api=after_api):
+                    result, _, outputs = self.run_submit(False, [{"method": "POST", "code": 201, "body": {"result": {"revision_id": "revision", "status": "pending"}}}], hostile_module=module, after_api=after_api)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+                    self.assertEqual(outputs, 'revision_id=revision\n')
+
+    def test_build_uses_trusted_sibling_imports_despite_pythonpath(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            scripts = root / '.github/scripts'
+            scripts.mkdir(parents=True)
+            for name in ['build_payload.py', 'resolve_release.py']:
+                shutil.copyfile(ROOT / '.github/scripts' / name, scripts / name)
+            shutil.copyfile(ROOT / 'manifest.schema.json', root / 'manifest.schema.json')
+            release = root / 'release-source'
+            release.mkdir()
+            package = release / 'Branding'
+            package.mkdir()
+            for name in ['manifest.json', 'README.md']:
+                shutil.copyfile(ROOT / 'Branding' / name, package / name)
+            version = json.loads((package / 'manifest.json').read_text())['version']
+            (release / 'json.py').write_text("from pathlib import Path\nPath('hostile-imported').write_text('executed')\nraise RuntimeError('tag-controlled import')\n")
+            bin_dir = root / 'bin'
+            bin_dir.mkdir()
+            gh = bin_dir / 'gh'
+            gh.write_text('#!/usr/bin/env -S python3 -I\nprint(' + repr(json.dumps({"published_at": "2026-01-01", "assets": [{"name": "release.zip", "browser_download_url": "https://example.test/release.zip"}]})) + ')\n')
+            gh.chmod(0o755)
+            env = {**os.environ, 'PATH': f"{bin_dir}:{os.environ['PATH']}", 'PYTHONPATH': str(release), 'API_TOKEN': 'offline', 'API_BASE_URL': '', 'GITHUB_REPOSITORY': 'FOSSBilling/modules', 'EXT_ID': 'branding', 'VERSION': version, 'RELEASE_TAG': f'branding-v{version}'}
+            result = subprocess.run(['bash', '-e', '-c', workflow_run('Build extension payload')], cwd=release, env=env, capture_output=True, text=True)
+            self.assertFalse((release / 'hostile-imported').exists(), 'Tagged module imported via PYTHONPATH')
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertEqual(json.loads((release / 'payload-create.json').read_text())['id'], 'branding')
+
+    def test_create_and_update_accept_the_merged_api_statuses(self):
+        for exists, method, code in [(False, "POST", 201), (True, "PUT", 202)]:
+            with self.subTest(method=method):
+                result, calls, outputs = self.run_submit(exists, [{"method": method, "code": code, "body": {"result": {"revision_id": "revision", "status": "pending"}}}])
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(len(calls), 1)
+                self.assertEqual(outputs, 'revision_id=revision\n')
+
+    def test_conflict_without_history_never_switches_to_update(self):
+        result, calls, _ = self.run_submit(False, [{"method": "POST", "code": 409, "body": {"error": {"message": "An extension with this id already exists"}}}])
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(len(calls), 1)
+        self.assertIn("refusing to overwrite releases", result.stdout)
+
+    def test_explicit_machine_approval_matches_the_merged_api(self):
+        for code in [200, 403, 409]:
+            result, calls, _ = self.run_submit(True, [{"method": "POST", "code": code, "body": {"result": {"status": "approved"}}}], step="Approve revision")
+            self.assertEqual(result.returncode == 0, code == 200, result.stderr)
+            self.assertEqual(calls, ["POST https://api.example.test/extensions/v2/extensions/example/revisions/revision/approve?notify=false"])
+
+    def test_retries_preserve_update_method_and_payload(self):
+        result, calls, _ = self.run_submit(True, [{"method": "PUT", "code": 429, "body": {}}, {"method": "PUT", "code": 202, "body": {"result": {"revision_id": "revision", "status": "pending"}}}])
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(len(calls), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()