validate_submission.py

"""Validate plugin catalogue submissions in a pull request."""

from __future__ import annotations

import argparse
import base64
import json
import os
import struct
import subprocess
import sys
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlparse
from urllib.request import Request, urlopen

CATALOGUE = Path("catalogue.json")
ICONS = Path("icons")

SECTION_NAMES = ("official", "native", "device", "stream-deck")
ALLOWED_SECTIONS = {"native", "device"}
MAINTAINER_SECTIONS = {"official", "stream-deck"}


class ValidationError(Exception):
	pass


def load_catalogue(path: Path) -> dict[str, dict]:
	try:
		with path.open(encoding="utf-8") as file:
			return json.load(file, object_pairs_hook=_unique_object)
	except (OSError, json.JSONDecodeError) as exc:
		raise ValidationError(f"{path} is not valid JSON: {exc}") from exc


def load_catalogue_text(contents: str) -> dict[str, dict]:
	try:
		return json.loads(contents, object_pairs_hook=_unique_object)
	except json.JSONDecodeError as exc:
		raise ValidationError(f"catalogue is not valid JSON: {exc}") from exc


def validate_catalogue_format(path: Path, catalogue: dict[str, dict]) -> None:
	try:
		contents = path.read_text(encoding="utf-8")
	except OSError as exc:
		raise ValidationError(f"could not read {path}: {exc}") from exc

	for line_number, line in enumerate(contents.splitlines(), start=1):
		if line.rstrip() != line:
			raise ValidationError(f"{path}:{line_number}: trailing whitespace is not allowed")
		if line and line[0] == " ":
			raise ValidationError(f"{path}:{line_number}: indentation must use tabs")
	if not contents.endswith("\n"):
		raise ValidationError(f"{path}: file must end with a newline")

	expected = json.dumps(catalogue, indent="\t", ensure_ascii=False) + "\n"
	if contents != expected:
		raise ValidationError(
			f"{path}: formatting must match tab-indented JSON generated by json.dumps"
		)


def _unique_object(pairs: list[tuple[str, object]]) -> dict:
	result: dict[str, object] = {}
	for key, value in pairs:
		if key in result:
			raise ValidationError(f"duplicate catalogue key: {key}")
		result[key] = value
	return result


def changed_files(base: str, head: str) -> set[str]:
	output = subprocess.check_output(
		["git", "diff", "--name-only", f"{base}...{head}", "--"],
		text=True,
	)
	return {line for line in output.splitlines() if line}


def section_for(plugin_id: str, catalogue: dict[str, dict]) -> str:
	ids = list(catalogue)
	try:
		index = ids.index(plugin_id)
	except ValueError as exc:
		raise ValidationError(f"{plugin_id} is not present in the catalogue") from exc

	section_starts = catalogue_section_starts(catalogue)
	for position, start in enumerate(section_starts):
		end = section_starts[position + 1] if position + 1 < len(section_starts) else len(ids)
		if start <= index < end:
			return SECTION_NAMES[position]
	raise ValidationError(f"could not determine the catalogue section for {plugin_id}")


def catalogue_section_starts(catalogue: dict[str, dict]) -> list[int]:
	"""Find section starts from the URL sort resets in the ordered catalogue."""
	ids = list(catalogue)
	repositories = [catalogue[plugin_id].get("repository", "").casefold() for plugin_id in ids]
	resets = [
		index for index in range(1, len(ids)) if repositories[index] < repositories[index - 1]
	]
	if len(resets) < len(SECTION_NAMES) - 1:
		raise ValidationError("could not determine all catalogue sections from repository ordering")
	return [0, *resets[: len(SECTION_NAMES) - 1]]


def validate_order(
	catalogue: dict[str, dict], relevant_ids: set[str], allowed_sections: set[str]
) -> None:
	ids = list(catalogue)
	section_starts = catalogue_section_starts(catalogue)
	for plugin_id in relevant_ids:
		section = section_for(plugin_id, catalogue)
		if section not in allowed_sections:
			continue
		index = ids.index(plugin_id)
		section_index = SECTION_NAMES.index(section)
		section_start = section_starts[section_index]
		section_end = (
			section_starts[section_index + 1]
			if section_index + 1 < len(section_starts)
			else len(ids)
		)
		repository = catalogue[plugin_id].get("repository", "")
		if (
			index > section_start
			and catalogue[ids[index - 1]].get("repository", "").casefold() > repository.casefold()
		):
			raise ValidationError(f"{plugin_id}: plugin is not sorted by repository URL")
		if (
			index + 1 < section_end
			and repository.casefold() > catalogue[ids[index + 1]].get("repository", "").casefold()
		):
			raise ValidationError(f"{plugin_id}: plugin is not sorted by repository URL")


def github_json(path: str) -> dict:
	url = f"https://api.github.com{path}"
	headers = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"}
	token = os.environ.get("GITHUB_TOKEN")
	if token:
		headers["Authorization"] = "Bearer " + token
	try:
		with urlopen(Request(url, headers=headers), timeout=20) as response:
			return json.load(response)
	except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc:
		raise ValidationError(f"GitHub API request failed for {path}: {exc}") from exc


def repository_parts(url: str) -> tuple[str, str]:
	parsed = urlparse(url)
	if parsed.scheme != "https" or parsed.netloc.lower() != "github.com":
		raise ValidationError(f"repository must be an HTTPS GitHub URL: {url}")
	parts = parsed.path.strip("/").split("/")
	if len(parts) != 2 or not all(parts):
		raise ValidationError(f"repository URL must identify an owner and repository: {url}")
	if parts[1].endswith(".git"):
		raise ValidationError(f"repository URL must not use a .git suffix: {url}")
	return parts[0], parts[1]


def is_maintainer(repository: str, username: str) -> bool:
	owner, repo = repository_parts(f"https://github.com/{repository}")
	permission = github_json(
		f"/repos/{owner}/{repo}/collaborators/{quote(username, safe='')}/permission"
	).get("permission")
	return permission in {"admin", "maintain"}


def validate_repository(plugin_id: str, entry: dict, changed: set[str]) -> None:
	for field in ("name", "author", "repository", "description"):
		if field not in entry:
			raise ValidationError(f"{plugin_id}: missing catalogue field '{field}'")

	owner, repo = repository_parts(entry["repository"])
	repo_info = github_json(f"/repos/{owner}/{repo}")
	release = github_json(f"/repos/{owner}/{repo}/releases/latest")
	if not any(
		asset.get("name", "").lower().endswith((".zip", ".streamdeckplugin"))
		for asset in release.get("assets", [])
	):
		raise ValidationError(
			f"{plugin_id}: latest release must include a .zip or .streamDeckPlugin artifact"
		)
	if "openaction" not in github_json(f"/repos/{owner}/{repo}/topics").get("names", []):
		raise ValidationError(f"{plugin_id}: repository is missing the openaction topic")
	if entry["description"] != repo_info.get("description"):
		raise ValidationError(
			f"{plugin_id}: description does not match the repository sidebar description"
		)

	tree = github_json(
		f"/repos/{owner}/{repo}/git/trees/{repo_info['default_branch']}?recursive=1"
	).get("tree", [])
	manifest_paths = [
		item["path"]
		for item in tree
		if item.get("type") == "blob" and item["path"].endswith("manifest.json")
	]
	if not manifest_paths:
		raise ValidationError(f"{plugin_id}: could not find a manifest.json in the repository")
	manifest_path = (
		"assets/manifest.json" if "assets/manifest.json" in manifest_paths else manifest_paths[0]
	)
	manifest_response = github_json(f"/repos/{owner}/{repo}/contents/{manifest_path}")
	try:
		manifest = json.loads(base64.b64decode(manifest_response["content"]).decode("utf-8"))
	except (KeyError, ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc:
		raise ValidationError(f"{plugin_id}: manifest.json is not valid JSON") from exc
	if manifest.get("Name") != entry["name"] or manifest.get("Author") != entry["author"]:
		raise ValidationError(f"{plugin_id}: name and author must match manifest.json")
	action_ids = [action.get("UUID", "") for action in manifest.get("Actions", [])]
	if action_ids and not any(action_id.startswith(f"{plugin_id}.") for action_id in action_ids):
		raise ValidationError(f"{plugin_id}: catalogue key is not the manifest bundle ID")

	icon_path = ICONS / f"{plugin_id}.png"
	if not icon_path.is_file():
		raise ValidationError(f"{plugin_id}: missing icon {icon_path}")
	if plugin_id not in {path.removeprefix("icons/").removesuffix(".png") for path in changed}:
		raise ValidationError(f"{plugin_id}: submission must add or update its icon")
	try:
		with icon_path.open("rb") as icon:
			if icon.read(8) != b"\x89PNG\r\n\x1a\n":
				raise ValueError("not a PNG")
			length = struct.unpack(">I", icon.read(4))[0]
			if icon.read(4) != b"IHDR" or length < 8:
				raise ValueError("missing PNG dimensions")
			width, height = struct.unpack(">II", icon.read(8))
	except (OSError, ValueError, struct.error) as exc:
		raise ValidationError(f"{plugin_id}: icon is not a valid PNG") from exc
	if width < 144 or height < 144:
		raise ValidationError(f"{plugin_id}: icon must be at least 144x144 pixels")


def main() -> int:
	parser = argparse.ArgumentParser()
	parser.add_argument("--base", required=True)
	parser.add_argument("--head", default="HEAD")
	parser.add_argument("--author")
	parser.add_argument("--repository")
	args = parser.parse_args()
	try:
		base_catalogue = load_catalogue_text(
			subprocess.check_output(["git", "show", f"{args.base}:catalogue.json"], text=True)
		)
		head_catalogue = load_catalogue(CATALOGUE)
		validate_catalogue_format(CATALOGUE, head_catalogue)
		changed = changed_files(args.base, args.head)
		changed_ids = {
			plugin_id
			for plugin_id in set(base_catalogue) | set(head_catalogue)
			if base_catalogue.get(plugin_id) != head_catalogue.get(plugin_id)
		}
		if not changed_ids:
			print("No catalogue entries changed; plugin submission checks skipped.")
			return 0
		allowed_sections = ALLOWED_SECTIONS
		if args.author and args.repository and is_maintainer(args.repository, args.author):
			allowed_sections = allowed_sections | MAINTAINER_SECTIONS
		validate_order(head_catalogue, changed_ids, allowed_sections)
		list(head_catalogue)
		for plugin_id in changed_ids:
			if plugin_id not in head_catalogue:
				raise ValidationError(
					f"{plugin_id}: catalogue entries may not be removed in a submission PR"
				)
			section = section_for(plugin_id, head_catalogue)
			if section not in allowed_sections:
				raise ValidationError(
					f"{plugin_id}: submissions are only allowed in Native or Device sections "
					"unless the PR author is a repository maintainer"
				)
			validate_repository(plugin_id, head_catalogue[plugin_id], changed)
		print(
			f"Validated {len(changed_ids)} plugin catalogue entr{'y' if len(changed_ids) == 1 else 'ies'}."
		)
		return 0
	except (ValidationError, subprocess.CalledProcessError) as exc:
		print(f"::error::{exc}", file=sys.stderr)
		return 1


if __name__ == "__main__":
	sys.exit(main())