Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 15 additions & 8 deletions .github/pluginctl/src/pluginctl/publish/manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,8 +192,13 @@ def write_manifest_if_changed(dest: str, payload: dict, generated_at: str) -> bo
return True


def _signing_payload(manifest: dict) -> str:
"""Return the canonical bytes reconstructed by Dispatcharr for verification."""
return json.dumps(manifest, separators=(",", ":"), ensure_ascii=True) + "\n"


def sig_is_current(file: str, gpg_key_id: str) -> bool:
"""True when the embedded ``.signature`` was issued by ``gpg_key_id``."""
"""True when the embedded signature is current and verifies for its payload."""
try:
with open(file, encoding="utf-8") as fh:
data = json.load(fh)
Expand All @@ -208,12 +213,16 @@ def sig_is_current(file: str, gpg_key_id: str) -> bool:
try:
out = subprocess.run(["gpg", "--list-packets", tmp_path],
capture_output=True, text=True).stdout
m = re.search(r"issuer key ID ([A-Fa-f0-9]{16})", out)
if not m or m.group(1).upper() != (gpg_key_id or "").upper():
return False
verify = subprocess.run(
["gpg", "--batch", "--verify", tmp_path, "-"],
input=_signing_payload(data["manifest"]), capture_output=True, text=True,
)
return verify.returncode == 0
finally:
os.unlink(tmp_path)
m = re.search(r"issuer key ID ([A-Fa-f0-9]{16})", out)
if not m:
return False
return m.group(1).upper() == (gpg_key_id or "").upper()


def sign_manifest(file: str, gpg_key_id: str, passphrase: Optional[str]) -> bool:
Expand All @@ -222,9 +231,7 @@ def sign_manifest(file: str, gpg_key_id: str, passphrase: Optional[str]) -> bool
return True
with open(file, encoding="utf-8") as fh:
data = json.load(fh)
# jq -c writes a terminating newline. Dispatcharr reconstructs this exact
# historical payload before verification, so it is part of the signature.
payload = jsonio.dumps(data["manifest"]) + "\n"
payload = _signing_payload(data["manifest"])
args = ["gpg", "--batch", "--yes", "--armor", "--detach-sign",
"--local-user", gpg_key_id, "--output", "-"]
if passphrase:
Expand Down
29 changes: 26 additions & 3 deletions .github/pluginctl/tests/test_manifest.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,9 +146,9 @@ def test_write_manifest_uses_formatted_json_and_ignores_existing_whitespace(tmp_
assert dest.read_text(encoding="utf-8") == json.dumps({"manifest": payload})


def test_sign_manifest_signs_compact_payload_with_jq_newline(tmp_path, monkeypatch):
def test_sign_manifest_uses_dispatcharr_verification_payload(tmp_path, monkeypatch):
path = tmp_path / "manifest.json"
manifest = {"registry_name": "demo", "plugins": []}
manifest = {"registry_name": "démo", "plugins": []}
path.write_text(json.dumps({"manifest": manifest}), encoding="utf-8")
captured = {}

Expand All @@ -164,10 +164,33 @@ def run(args, **kwargs):
monkeypatch.setattr(m.subprocess, "run", run)

assert m.sign_manifest(str(path), "0123456789ABCDEF", None)
assert captured["input"] == jsonio.dumps(manifest) + "\n"
assert captured["input"] == json.dumps(manifest, separators=(",", ":"), ensure_ascii=True) + "\n"
assert json.loads(path.read_text(encoding="utf-8"))["signature"] == "signature"


def test_sig_is_current_rejects_signature_for_an_old_payload_format(tmp_path, monkeypatch):
path = tmp_path / "manifest.json"
manifest = {"registry_name": "démo", "plugins": []}
path.write_text(json.dumps({"manifest": manifest, "signature": "signature"}), encoding="utf-8")
captured = {}

class Result:
def __init__(self, returncode=0, stdout=""):
self.returncode = returncode
self.stdout = stdout

def run(args, **kwargs):
if "--list-packets" in args:
return Result(stdout="issuer key ID 0123456789ABCDEF")
captured["input"] = kwargs["input"]
return Result(returncode=1)

monkeypatch.setattr(m.subprocess, "run", run)

assert not m.sig_is_current(str(path), "0123456789ABCDEF")
assert captured["input"] == json.dumps(manifest, separators=(",", ":"), ensure_ascii=True) + "\n"


def test_pages_url_returns_empty_string_when_pages_is_unavailable(monkeypatch):
monkeypatch.setattr(gh, "api", lambda *args, **kwargs: None)
assert gh.pages_url("org/repo") == ""
Expand Down
Loading