yorickr/plezy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(supply-chain): verify CI and production inputs

Browse Source 
Pin external actions, images, toolchains, native archives, and tvOS engine artifacts; enforce fail-closed CI checks and keep website privacy disclosures aligned with shipped behavior.
This commit is contained in:
edde746
2026-07-24 03:56:40 +02:00
parent b41fb4fe75
commit 09656fa4d3
63 changed files with 5146 additions and 560 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/check_build_workflow.py
+66 -10
View File
@@ -6,7 +6,10 @@ import re
import sys
WORKFLOW = Path(__file__).resolve().parents[1] / ".github/workflows/build.yml"
DEFAULT_WORKFLOW = Path(__file__).resolve().parents[1] / ".github/workflows/build.yml"
if len(sys.argv) > 2:
raise SystemExit(f"Usage: {Path(sys.argv[0]).name} [workflow-path]")
WORKFLOW = Path(sys.argv[1]).resolve() if len(sys.argv) == 2 else DEFAULT_WORKFLOW
text = WORKFLOW.read_text(encoding="utf-8")
errors: list[str] = []
@@ -24,6 +27,60 @@ def job(name: str) -> str:
return match.group(0) if match else ""
def named_step(block: str, name: str) -> str:
match = re.search(
rf"(?ms)^ - name: {re.escape(name)}\n.*?(?=^ - |\Z)",
block,
)
require(match is not None, f"missing '{name}' step")
return match.group(0) if match else ""
def validate_windows_signing(block: str) -> None:
install = named_step(block, "Install dependencies")
signing = named_step(block, "Sign installer for WinSparkle (EdDSA)")
require(
block.find(" - name: Install dependencies")
< block.find(" - name: Sign installer for WinSparkle (EdDSA)"),
"locked root dependencies must be installed before Windows signing",
)
require(
"flutter pub get --enforce-lockfile --no-example" in install,
"Windows signing must use the enforced root dependency lock",
)
require(
"dart run auto_updater:sign_update plezy-windows-installer.exe $keyPath"
in signing,
"Windows signing must execute the locked auto_updater package",
)
require(
"$env:RUNNER_TEMP" in signing,
"Windows signing key must live under RUNNER_TEMP",
)
require(
"try {" in signing and "} finally {" in signing,
"Windows signing key cleanup must run from a finally block",
)
require(
"Remove-Item -Path $keyPath -Force -ErrorAction SilentlyContinue"
in signing,
"Windows signing must remove its temporary key",
)
lowered = signing.lower()
for forbidden in (
"raw.githubusercontent.com",
"invoke-webrequest",
"git clone",
"_signer",
"pubspec.yaml",
"dart pub get",
):
require(
forbidden not in lowered,
f"Windows signing step contains mutable or ad-hoc input: {forbidden}",
)
def require_explicit_shells(name: str, block: str, shell: str) -> None:
steps = re.findall(r"(?ms)^ - .*?(?=^ - |\Z)", block)
run_steps = [step for step in steps if re.search(r"(?m)^ run:", step)]
@@ -147,8 +204,16 @@ require(
"Linux build attestation permissions changed",
)
require_explicit_shells("build-linux", linux, "bash")
libmpv_cache = named_step(linux, "Cache libmpv build")
require(
"hashFiles('linux/packaging/build-libmpv.sh', 'linux/packaging/native-inputs.json')"
in libmpv_cache,
"libmpv cache identity must include its build script and native input manifest",
)
package_windows = job("package-windows")
validate_windows_signing(package_windows)
require("needs: build-windows" in package_windows, "Windows packaging must fan in the matrix")
for artifact in (
"windows-x64-build",
@@ -251,15 +316,6 @@ require(
text.count("persist-credentials: false") == checkout_count,
"every build checkout must discard GitHub credentials",
)
require(
"raw.githubusercontent.com/edde746/auto_updater/9e150f71e17495b7361aedbe6df22e89ad52c254/"
in text,
"Windows signing helper must remain pinned to the locked auto_updater commit",
)
require(
'dependencies=@{cryptography="2.9.0"}' in text,
"Windows signing dependency must remain exact",
)
if errors:
for error in errors:
scripts/check_bun_audit.py
Executable
+362
View File
@@ -0,0 +1,362 @@
#!/usr/bin/env python3
"""Enforce an exact, expiring baseline for the website's Bun audit results."""
from __future__ import annotations
import argparse
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date, timedelta
import json
from pathlib import Path
import re
import subprocess
import sys
from typing import Any
MAX_ACCEPTANCE_DAYS = 90
MAX_AUDIT_OUTPUT_BYTES = 5 * 1024 * 1024
MAX_DIAGNOSTICS = 20
MAX_ID_LENGTH = 128
MAX_PACKAGE_LENGTH = 214
MAX_RANGE_LENGTH = 256
MAX_RATIONALE_LENGTH = 500
MAX_SCANNER_DIAGNOSTIC_LENGTH = 300
SEVERITIES = frozenset({"low", "moderate", "high", "critical"})
ISO_DATE = re.compile(r"\d{4}-\d{2}-\d{2}")
PACKAGE_NAME = re.compile(
r"(?:@[a-z0-9][a-z0-9._~-]*/)?[a-z0-9][a-z0-9._~-]*",
re.IGNORECASE,
)
ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m")
EXPECTED_BUN_BANNER = re.compile(r"bun audit v1\.3\.14 \([0-9a-f]{8}\)")
@dataclass(frozen=True)
class Advisory:
advisory_id: str
package: str
severity: str
vulnerable_range: str
@property
def identity(self) -> tuple[str, str]:
return (self.advisory_id, self.package)
@property
def label(self) -> str:
return f"{self.advisory_id} ({self.package})"
@dataclass(frozen=True)
class Acceptance:
advisory: Advisory
expires_on: date
rationale: str
def _load_json(path: Path) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise ValueError(f"cannot read valid JSON from {path}: {error}") from error
def _parse_iso_date(value: Any, field: str) -> date:
if not isinstance(value, str) or ISO_DATE.fullmatch(value) is None:
raise ValueError(f"{field} must be an ISO date in YYYY-MM-DD form")
try:
parsed = date.fromisoformat(value)
except ValueError as error:
raise ValueError(f"{field} must be an ISO date in YYYY-MM-DD form") from error
if parsed.isoformat() != value:
raise ValueError(f"{field} must be an ISO date in YYYY-MM-DD form")
return parsed
def _bounded_string(
value: Any,
*,
field: str,
maximum: int,
pattern: re.Pattern[str] | None = None,
) -> str:
if (
not isinstance(value, str)
or not value
or value != value.strip()
or len(value) > maximum
or any(ord(character) < 32 for character in value)
or (pattern is not None and pattern.fullmatch(value) is None)
):
raise ValueError(f"{field} is invalid")
return value
def load_baseline(path: Path) -> tuple[date, dict[tuple[str, str], Acceptance]]:
payload = _load_json(path)
if not isinstance(payload, dict) or set(payload) != {
"schemaVersion",
"reviewedOn",
"accepted",
}:
raise ValueError("baseline must contain schemaVersion, reviewedOn, and accepted")
if payload["schemaVersion"] != 1:
raise ValueError("unsupported baseline schemaVersion")
reviewed_on = _parse_iso_date(payload["reviewedOn"], "reviewedOn")
entries = payload["accepted"]
if not isinstance(entries, list):
raise ValueError("baseline accepted must be a list")
accepted: dict[tuple[str, str], Acceptance] = {}
required = {
"id",
"package",
"severity",
"vulnerableRange",
"expiresOn",
"rationale",
}
for index, entry in enumerate(entries):
if not isinstance(entry, dict) or set(entry) != required:
raise ValueError(f"baseline accepted[{index}] has an unknown schema")
if not isinstance(entry["id"], (str, int)) or isinstance(entry["id"], bool):
raise ValueError(f"baseline accepted[{index}] has an invalid advisory id")
advisory_id = _bounded_string(
str(entry["id"]),
field=f"baseline accepted[{index}].id",
maximum=MAX_ID_LENGTH,
pattern=re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]*"),
)
package = _bounded_string(
entry["package"],
field=f"baseline accepted[{index}].package",
maximum=MAX_PACKAGE_LENGTH,
pattern=PACKAGE_NAME,
)
severity = _bounded_string(
entry["severity"],
field=f"baseline accepted[{index}].severity",
maximum=16,
)
if severity not in SEVERITIES:
raise ValueError(f"baseline accepted[{index}].severity is invalid")
vulnerable_range = _bounded_string(
entry["vulnerableRange"],
field=f"baseline accepted[{index}].vulnerableRange",
maximum=MAX_RANGE_LENGTH,
)
rationale = _bounded_string(
entry["rationale"],
field=f"baseline accepted[{index}].rationale",
maximum=MAX_RATIONALE_LENGTH,
)
advisory = Advisory(
advisory_id=advisory_id,
package=package,
severity=severity,
vulnerable_range=vulnerable_range,
)
expires_on = _parse_iso_date(entry["expiresOn"], f"accepted[{index}].expiresOn")
if expires_on < reviewed_on:
raise ValueError(f"baseline {advisory.label} expires before its review date")
if expires_on > reviewed_on + timedelta(days=MAX_ACCEPTANCE_DAYS):
raise ValueError(
f"baseline {advisory.label} expires more than {MAX_ACCEPTANCE_DAYS} days after review"
)
if advisory.identity in accepted:
raise ValueError(f"duplicate baseline advisory {advisory.label}")
accepted[advisory.identity] = Acceptance(
advisory=advisory,
expires_on=expires_on,
rationale=rationale,
)
return reviewed_on, accepted
def parse_audit_json(output: str) -> dict[tuple[str, str], Advisory]:
if len(output.encode("utf-8")) > MAX_AUDIT_OUTPUT_BYTES:
raise ValueError("bun audit result exceeds the policy size limit")
try:
payload = json.loads(output)
except json.JSONDecodeError as error:
raise ValueError(f"bun audit returned malformed JSON: {error.msg}") from error
if not isinstance(payload, dict):
raise ValueError("bun audit result must be a package object")
advisories: dict[tuple[str, str], Advisory] = {}
for package_value, package_entries in payload.items():
package = _bounded_string(
package_value,
field="bun audit package name",
maximum=MAX_PACKAGE_LENGTH,
pattern=PACKAGE_NAME,
)
if not isinstance(package_entries, list) or not package_entries:
raise ValueError(f"bun audit entries for {package} must be a non-empty list")
for index, entry in enumerate(package_entries):
if not isinstance(entry, dict):
raise ValueError(f"bun audit entry {package}[{index}] must be an object")
required = ("id", "severity", "vulnerable_versions")
if any(field not in entry for field in required):
raise ValueError(f"bun audit entry {package}[{index}] has an unknown schema")
if not isinstance(entry["id"], (str, int)) or isinstance(entry["id"], bool):
raise ValueError(f"bun audit entry {package}[{index}] has an invalid id")
advisory_id = _bounded_string(
str(entry["id"]),
field=f"bun audit entry {package}[{index}].id",
maximum=MAX_ID_LENGTH,
pattern=re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]*"),
)
severity = _bounded_string(
entry["severity"],
field=f"bun audit entry {package}[{index}].severity",
maximum=16,
)
if severity not in SEVERITIES:
raise ValueError(f"bun audit entry {package}[{index}].severity is invalid")
vulnerable_range = _bounded_string(
entry["vulnerable_versions"],
field=f"bun audit entry {package}[{index}].vulnerable_versions",
maximum=MAX_RANGE_LENGTH,
)
advisory = Advisory(
advisory_id=advisory_id,
package=package,
severity=severity,
vulnerable_range=vulnerable_range,
)
if advisory.identity in advisories:
raise ValueError(f"bun audit returned duplicate advisory {advisory.label}")
advisories[advisory.identity] = advisory
return advisories
def _has_scanner_diagnostic(stderr: str) -> bool:
rendered = ANSI_ESCAPE.sub("", stderr).strip()
return bool(rendered) and EXPECTED_BUN_BANNER.fullmatch(rendered) is None
def evaluate(
*,
exit_code: int,
output: str,
accepted: dict[tuple[str, str], Acceptance],
today: date,
stderr: str = "",
) -> list[str]:
if exit_code not in (0, 1):
return [f"bun audit execution failed with exit code {exit_code}"]
if _has_scanner_diagnostic(stderr):
return ["bun audit reported a scanner or network diagnostic"]
try:
current = parse_audit_json(output)
except (UnicodeError, ValueError) as error:
return [str(error)]
if exit_code == 0 and current:
return ["bun audit exited cleanly but returned advisories"]
if exit_code == 1 and not current:
return ["bun audit exited with advisories but returned an empty result"]
errors = []
for identity, advisory in sorted(current.items()):
acceptance = accepted.get(identity)
if acceptance is None:
errors.append(f"unaccepted advisory {advisory.label}")
continue
if acceptance.expires_on < today:
errors.append(
f"expired acceptance {advisory.label} ({acceptance.expires_on.isoformat()})"
)
if acceptance.advisory.severity != advisory.severity:
errors.append(
f"severity changed for {advisory.label}: "
f"{acceptance.advisory.severity} -> {advisory.severity}"
)
if acceptance.advisory.vulnerable_range != advisory.vulnerable_range:
errors.append(
f"vulnerable range changed for {advisory.label}: "
f"{acceptance.advisory.vulnerable_range} -> {advisory.vulnerable_range}"
)
for identity, acceptance in sorted(accepted.items()):
if identity not in current:
errors.append(f"stale baseline advisory {acceptance.advisory.label}")
if len(errors) > MAX_DIAGNOSTICS:
omitted = len(errors) - MAX_DIAGNOSTICS
return errors[:MAX_DIAGNOSTICS] + [
f"{omitted} additional policy error(s) omitted"
]
return errors
def _run_bun_audit(project: Path) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["bun", "audit", "--json"],
cwd=project,
check=False,
capture_output=True,
text=True,
)
def _scanner_diagnostic(value: str) -> str:
compact = " ".join(value.split())
if len(compact) > MAX_SCANNER_DIAGNOSTIC_LENGTH:
return f"{compact[:MAX_SCANNER_DIAGNOSTIC_LENGTH]}..."
return compact
def main(
argv: list[str] | None = None,
*,
run_audit: Callable[[Path], subprocess.CompletedProcess[str]] = _run_bun_audit,
today: date | None = None,
) -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--project", type=Path, required=True)
parser.add_argument("--baseline", type=Path, required=True)
args = parser.parse_args(argv)
evaluation_date = date.today() if today is None else today
project = args.project.resolve()
baseline_path = args.baseline
if not baseline_path.is_absolute():
baseline_path = project / baseline_path
if not (project / "bun.lock").is_file():
print(f"ERROR: missing Bun lockfile in {project}", file=sys.stderr)
return 1
try:
reviewed_on, accepted = load_baseline(baseline_path)
if reviewed_on > evaluation_date:
raise ValueError("baseline reviewedOn cannot be in the future")
except ValueError as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
try:
result = run_audit(project)
except (OSError, subprocess.SubprocessError) as error:
print(f"ERROR: cannot execute bun audit: {_scanner_diagnostic(str(error))}", file=sys.stderr)
return 1
errors = evaluate(
exit_code=result.returncode,
output=result.stdout,
stderr=result.stderr,
accepted=accepted,
today=evaluation_date,
)
if errors:
for error in errors:
print(f"ERROR: {error}", file=sys.stderr)
if result.stderr.strip():
print(
f"ERROR: bun audit diagnostic: {_scanner_diagnostic(result.stderr)}",
file=sys.stderr,
)
return 1
print(f"Bun advisory policy passed ({len(accepted)} reviewed acceptance(s)).")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/check_codegen.py
+3 -1
View File
@@ -81,7 +81,9 @@ def _copy_dependency_state(source_root: Path, target_root: Path) -> None:
def _create_isolated_checkout(root: Path, destination: Path) -> None:
_run(root, "git", "worktree", "add", "--detach", "--quiet", str(destination), "HEAD")
changed = _nul_paths(_run(root, "git", "diff", "--name-only", "-z", "HEAD", "--").stdout)
changed = _nul_paths(
_run(root, "git", "diff", "--no-renames", "--name-only", "-z", "HEAD", "--").stdout
)
untracked = _nul_paths(_run(root, "git", "ls-files", "--others", "--exclude-standard", "-z").stdout)
for relative in sorted(set(changed + untracked)):
_copy_overlay_path(root, destination, relative)
scripts/check_container_image_pins.py
Executable
+155
View File
@@ -0,0 +1,155 @@
#!/usr/bin/env python3
"""Require immutable, architecture-declared production container images."""
from __future__ import annotations
from dataclasses import dataclass
import re
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
PRODUCTION_DOCKERFILES = (ROOT / "server" / "Dockerfile",)
PRODUCTION_COMPOSE_FILES = (ROOT / "server" / "docker-compose.yml",)
SUPPORTED_PLATFORMS = frozenset({"linux/amd64", "linux/arm64"})
FROM_RE = re.compile(
r"^\s*FROM(?:\s+--platform=(?:\S+))?\s+(?P<reference>\S+)",
re.IGNORECASE,
)
IMAGE_RE = re.compile(r"^\s*image\s*:\s*(?P<reference>[^\s#]+)\s*(?:#.*)?$")
PLATFORMS_RE = re.compile(r"^\s*#\s*Platforms\s*:\s*(?P<platforms>.+?)\s*$", re.IGNORECASE)
PINNED_REFERENCE_RE = re.compile(
r"^(?P<name>[a-z0-9]+(?:[._-][a-z0-9]+)*(?::[0-9]+)?"
r"(?:/[a-z0-9]+(?:[._-][a-z0-9]+)*)*)"
r":(?P<tag>[A-Za-z0-9_][A-Za-z0-9_.-]{0,127})"
r"@sha256:(?P<digest>[0-9a-f]{64})$"
)
@dataclass(frozen=True)
class ImageReference:
path: Path
line_number: int
reference: str
platforms: frozenset[str] | None
def _adjacent_platforms(lines: list[str], reference_index: int) -> frozenset[str] | None:
for index in range(reference_index - 1, -1, -1):
stripped = lines[index].strip()
if not stripped:
break
if not stripped.startswith("#"):
break
match = PLATFORMS_RE.match(lines[index])
if match:
values = (value.strip() for value in match.group("platforms").split(","))
return frozenset(value for value in values if value)
return None
def iter_dockerfile_references(path: Path):
lines = path.read_text(encoding="utf-8").splitlines()
for index, line in enumerate(lines):
match = FROM_RE.match(line)
if not match:
continue
reference = match.group("reference")
if reference.lower() == "scratch":
continue
yield ImageReference(
path=path,
line_number=index + 1,
reference=reference,
platforms=_adjacent_platforms(lines, index),
)
def iter_compose_references(path: Path):
lines = path.read_text(encoding="utf-8").splitlines()
for index, line in enumerate(lines):
match = IMAGE_RE.match(line)
if not match:
continue
yield ImageReference(
path=path,
line_number=index + 1,
reference=match.group("reference").strip("'\""),
platforms=_adjacent_platforms(lines, index),
)
def validate_image(image: ImageReference) -> list[str]:
violations = []
match = PINNED_REFERENCE_RE.fullmatch(image.reference)
if not match:
violations.append(
"external images must use a readable tag and full lowercase sha256 digest"
)
elif match.group("tag").lower() == "latest":
violations.append("the readable image tag must not be latest")
if image.platforms is None:
violations.append(
"an adjacent '# Platforms:' declaration is required for each external image"
)
elif image.platforms != SUPPORTED_PLATFORMS:
expected = ", ".join(sorted(SUPPORTED_PLATFORMS))
actual = ", ".join(sorted(image.platforms)) or "none"
violations.append(f"platforms must be exactly {expected}; found {actual}")
return violations
def _display_path(path: Path) -> Path:
try:
return path.resolve().relative_to(ROOT)
except ValueError:
return path
def check_paths(dockerfiles: list[Path], compose_files: list[Path]) -> list[str]:
images = []
for path in dockerfiles:
images.extend(iter_dockerfile_references(path))
for path in compose_files:
images.extend(iter_compose_references(path))
violations = []
for image in images:
for reason in validate_image(image):
violations.append(
f"{_display_path(image.path)}:{image.line_number}: "
f"{image.reference!r}: {reason}"
)
return violations
def main(argv: list[str] | None = None) -> int:
args = list(sys.argv[1:] if argv is None else argv)
if args:
dockerfiles = [Path(value) for value in args if Path(value).name == "Dockerfile"]
compose_files = [Path(value) for value in args if Path(value).name != "Dockerfile"]
else:
dockerfiles = list(PRODUCTION_DOCKERFILES)
compose_files = list(PRODUCTION_COMPOSE_FILES)
violations = check_paths(dockerfiles, compose_files)
if violations:
print("Mutable or malformed production container references:", file=sys.stderr)
for violation in violations:
print(f" {violation}", file=sys.stderr)
return 1
image_count = sum(
1 for path in dockerfiles for _ in iter_dockerfile_references(path)
) + sum(1 for path in compose_files for _ in iter_compose_references(path))
print(
f"Production container pins verified ({image_count} external images; "
f"platforms: {', '.join(sorted(SUPPORTED_PLATFORMS))})."
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/check_workflow_security.py
+119 -19
View File
@@ -8,8 +8,9 @@ import sys
ROOT = Path(__file__).resolve().parents[1]
WORKFLOWS = ROOT / ".github" / "workflows"
CI_WORKFLOW = Path(".github/workflows/ci.yml")
FULL_SHA = re.compile(r"[0-9a-f]{40}")
PR_TRIGGER = re.compile(r"(?m)^ pull_request:\s*$")
USES_LINE = re.compile(r"^\s*(?:-\s+)?(?:uses|'uses'|\"uses\")\s*:\s*(.*?)\s*$")
def _active_text(text: str) -> str:
@@ -18,29 +19,129 @@ def _active_text(text: str) -> str:
)
def _scalar(value: str) -> str:
"""Remove an inline YAML comment and matching scalar quotes."""
quote: str | None = None
escaped = False
end = len(value)
for index, character in enumerate(value):
if escaped:
escaped = False
continue
if quote == '"' and character == "\\":
escaped = True
continue
if character in ("'", '"'):
if quote is None:
quote = character
elif quote == character:
quote = None
elif character == "#" and quote is None and (
index == 0 or value[index - 1].isspace()
):
end = index
break
result = value[:end].strip()
if len(result) >= 2 and result[0] == result[-1] and result[0] in ("'", '"'):
return result[1:-1]
return result
def _has_trigger(text: str, event: str) -> bool:
lines = text.splitlines()
on_key = r"""(?:on|'on'|"on")"""
event_key = rf"""(?:{re.escape(event)}|'{re.escape(event)}'|"{re.escape(event)}")"""
for index, line in enumerate(lines):
if re.fullmatch(rf"{on_key}:\s*", line):
for child in lines[index + 1 :]:
if child.strip() and not child.startswith((" ", "\t")):
break
if re.match(rf"^\s+{event_key}\s*:", child):
return True
match = re.fullmatch(rf"{on_key}:\s*(.+?)\s*", line)
if match is not None and re.search(
rf"""(?:^|[\[{{,\s])['"]?{re.escape(event)}['"]?(?:$|[\]}},\s:])""",
_scalar(match.group(1)),
):
return True
return False
def _step_block(lines: list[str], line_index: int) -> str:
uses_indent = len(lines[line_index]) - len(lines[line_index].lstrip())
start = line_index
for index in range(line_index, -1, -1):
line = lines[index]
indent = len(line) - len(line.lstrip())
if re.match(r"^\s*-\s+", line) and indent <= uses_indent:
start = index
break
if line.strip() and indent < uses_indent:
break
start_indent = len(lines[start]) - len(lines[start].lstrip())
end = len(lines)
for index in range(start + 1, len(lines)):
line = lines[index]
indent = len(line) - len(line.lstrip())
if re.match(r"^\s*-\s+", line) and indent <= start_indent:
end = index
break
if line.strip() and not line.lstrip().startswith("#") and indent < start_indent:
end = index
break
return "\n".join(lines[start:end])
def _check_fail_open(path: Path, text: str) -> list[str]:
"""CI quality gates must not silently convert failures to successes."""
if path.as_posix() != CI_WORKFLOW.as_posix():
return []
errors: list[str] = []
for line_number, line in enumerate(text.splitlines(), start=1):
match = re.match(
r"""^\s*(?:-\s+)?(?:continue-on-error|'continue-on-error'|"continue-on-error")\s*:\s*(.*?)\s*$""",
line,
)
if match is not None and _scalar(match.group(1)).lower() != "false":
errors.append(f"{path}:{line_number}: continue-on-error must remain false")
if re.search(r"\|\|\s*true(?:\s|$)", line):
errors.append(f"{path}:{line_number}: command must not suppress failure with || true")
return errors
def check_workflow(path: Path, text: str) -> list[str]:
errors: list[str] = []
active = _active_text(text)
for dangerous_trigger in ("pull_request_target:", "workflow_run:"):
if dangerous_trigger in active:
errors.append(f"{path}: unaudited privileged trigger {dangerous_trigger[:-1]}")
for dangerous_trigger in ("pull_request_target", "workflow_run"):
if _has_trigger(active, dangerous_trigger):
errors.append(f"{path}: unaudited privileged trigger {dangerous_trigger}")
checkout_count = 0
for line_number, line in enumerate(active.splitlines(), start=1):
match = re.match(r"^\s*(?:-\s+)?uses:\s+(.+?)\s*$", line)
pull_request = _has_trigger(active, "pull_request")
lines = active.splitlines()
for line_index, line in enumerate(lines):
match = USES_LINE.match(line)
if match is None:
continue
reference = match.group(1).split(" #", maxsplit=1)[0].strip()
reference = _scalar(match.group(1))
if reference.startswith("./"):
continue
action, separator, ref = reference.rpartition("@")
if not separator or not action or FULL_SHA.fullmatch(ref) is None:
errors.append(
f"{path}:{line_number}: external action must use a full commit SHA: {reference}"
f"{path}:{line_index + 1}: external action must use a full commit SHA: {reference}"
)
if action == "actions/checkout":
checkout_count += 1
if pull_request and action == "actions/checkout":
step = _step_block(lines, line_index)
if re.search(
r"""(?mi)^\s+(?:persist-credentials|'persist-credentials'|"persist-credentials")\s*:\s*['"]?false['"]?\s*(?:#.*)?$""",
step,
) is None:
errors.append(
f"{path}:{line_index + 1}: pull-request checkout must discard GitHub credentials"
)
if re.search(
r"https://raw\.githubusercontent\.com/[^/\s]+/[^/\s]+/(?:main|master)/",
@@ -48,22 +149,21 @@ def check_workflow(path: Path, text: str) -> list[str]:
):
errors.append(f"{path}: raw GitHub downloads must use an immutable commit")
if PR_TRIGGER.search(active):
if "secrets." in active:
if pull_request:
if re.search(r"\bsecrets\s*(?:\.|\[)", active):
errors.append(f"{path}: pull-request workflow must not reference repository secrets")
if re.search(r"(?m)^\s+[a-zA-Z0-9_-]+:\s+write\s*$", active):
if re.search(r"(?m)^\s+[a-zA-Z0-9_-]+:\s*write\s*(?:#.*)?$", active) or re.search(
r"(?m)^\s*permissions:\s*\{[^}\n]*:\s*write(?:\s*[,}])", active
):
errors.append(f"{path}: pull-request workflow must not request write permissions")
if active.count("persist-credentials: false") != checkout_count:
errors.append(
f"{path}: every pull-request checkout must discard GitHub credentials"
)
errors.extend(_check_fail_open(path, active))
return errors
def main() -> int:
errors: list[str] = []
for path in sorted(WORKFLOWS.glob("*.yml")):
for path in sorted((*WORKFLOWS.glob("*.yml"), *WORKFLOWS.glob("*.yaml"))):
errors.extend(check_workflow(path.relative_to(ROOT), path.read_text(encoding="utf-8")))
if errors:
scripts/ci_checks.sh
+8
View File
@@ -83,12 +83,18 @@ fi
# 4. Workflow and script regression guards
section "workflow and script guards"
if python3 scripts/check_build_workflow.py &&
python3 scripts/test_check_build_workflow.py &&
python3 scripts/check_apple_spm_locks.py &&
python3 scripts/test_check_apple_spm_locks.py &&
python3 scripts/check_workflow_security.py &&
python3 scripts/test_check_workflow_security.py &&
python3 scripts/check_workflow_action_pins.py &&
python3 scripts/test_check_workflow_action_pins.py &&
python3 scripts/check_container_image_pins.py &&
python3 scripts/test_check_container_image_pins.py &&
python3 scripts/verify_runtime_inputs.py &&
python3 scripts/test_verify_runtime_inputs.py &&
python3 scripts/test_fetch_tvos_engine.py &&
python3 scripts/test_check_codegen.py &&
python3 scripts/test_generate_relay_protocol.py &&
python3 scripts/test_format_native.py &&
@@ -96,6 +102,8 @@ if python3 scripts/check_build_workflow.py &&
python3 scripts/test_pubspec_version.py &&
python3 scripts/test_clean_translations.py &&
python3 scripts/test_run_maestro.py &&
python3 scripts/test_maestro_flow_contracts.py &&
python3 scripts/test_maestro_jellyfin_proxy.py &&
python3 scripts/test_check_icon_consistency.py; then
ok "workflow and script guards passed"
else
scripts/ci_server_checks.sh
Executable
+13
View File
@@ -0,0 +1,13 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT_DIR/server"
go mod download
go mod verify
git diff --exit-code -- go.mod go.sum
go vet -mod=readonly ./...
go test -mod=readonly -race -count=1 ./...
go install golang.org/x/vuln/cmd/govulncheck@v1.1.4
govulncheck ./...
scripts/ci_website_checks.sh
Executable
+12
View File
@@ -0,0 +1,12 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT_DIR/website"
bun install --frozen-lockfile
bun test
bun run check
bun run build
python3 "$ROOT_DIR/scripts/test_check_bun_audit.py"
bun run audit
scripts/test_check_build_workflow.py
Executable
+114
View File
@@ -0,0 +1,114 @@
#!/usr/bin/env python3
"""Behavior tests for the privileged build-workflow guard."""
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[1]
CHECKER = ROOT / "scripts/check_build_workflow.py"
WORKFLOW = ROOT / ".github/workflows/build.yml"
class BuildWorkflowGuardTest(unittest.TestCase):
def _run(self, workflow: str) -> subprocess.CompletedProcess[str]:
with tempfile.TemporaryDirectory(prefix="plezy-build-workflow-test-") as directory:
fixture = Path(directory) / "build.yml"
fixture.write_text(workflow, encoding="utf-8")
return subprocess.run(
[sys.executable, str(CHECKER), str(fixture)],
cwd=ROOT,
check=False,
capture_output=True,
text=True,
)
def _workflow(self) -> str:
return WORKFLOW.read_text(encoding="utf-8")
def test_locked_root_signer_passes(self) -> None:
result = self._run(self._workflow())
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("architecture matrix checks passed", result.stdout)
def test_mutable_download_in_signing_step_is_rejected(self) -> None:
workflow = self._workflow().replace(
" try {\n",
" Invoke-WebRequest -Uri https://raw.githubusercontent.com/example/main/sign.dart -OutFile sign.dart\n"
" try {\n",
1,
)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("mutable or ad-hoc input: raw.githubusercontent.com", result.stderr)
self.assertIn("mutable or ad-hoc input: invoke-webrequest", result.stderr)
def test_inline_dependency_resolution_in_signing_step_is_rejected(self) -> None:
workflow = self._workflow().replace(
" try {\n",
" Set-Content -Path pubspec.yaml -Value 'dependencies: {}'\n"
" dart pub get\n"
" try {\n",
1,
)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("mutable or ad-hoc input: pubspec.yaml", result.stderr)
self.assertIn("mutable or ad-hoc input: dart pub get", result.stderr)
def test_downloaded_signer_execution_is_rejected(self) -> None:
workflow = self._workflow().replace(
"dart run auto_updater:sign_update plezy-windows-installer.exe $keyPath",
"dart run sign.dart plezy-windows-installer.exe $keyPath",
1,
)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("must execute the locked auto_updater package", result.stderr)
def test_unlocked_install_is_rejected(self) -> None:
prefix, package_and_after = self._workflow().split(" package-windows:\n", 1)
workflow = prefix + " package-windows:\n" + package_and_after.replace(
"flutter pub get --enforce-lockfile --no-example",
"flutter pub get",
1,
)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("enforced root dependency lock", result.stderr)
def test_missing_finally_cleanup_is_rejected(self) -> None:
workflow = self._workflow().replace(" } finally {\n", " }\n", 1)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("cleanup must run from a finally block", result.stderr)
def test_libmpv_cache_without_native_manifest_is_rejected(self) -> None:
workflow = self._workflow().replace(
"hashFiles('linux/packaging/build-libmpv.sh', 'linux/packaging/native-inputs.json')",
"hashFiles('linux/packaging/build-libmpv.sh')",
1,
)
result = self._run(workflow)
self.assertNotEqual(result.returncode, 0)
self.assertIn("native input manifest", result.stderr)
if __name__ == "__main__":
unittest.main()
scripts/test_check_bun_audit.py
Executable
+415
View File
@@ -0,0 +1,415 @@
#!/usr/bin/env python3
"""Deterministic tests for the exact Bun advisory policy."""
import contextlib
from datetime import date
import io
import json
from pathlib import Path
import sys
import subprocess
import tempfile
import unittest
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
from check_bun_audit import Acceptance, Advisory, evaluate, load_baseline, main
TODAY = date(2026, 7, 21)
ADVISORY = Advisory("111", "fixture-package", "high", "<2.0.0")
ACCEPTANCE = Acceptance(ADVISORY, date(2026, 8, 1), "Not reachable in static output.")
def audit_json(
*,
advisory_id: int = 111,
severity: str = "high",
vulnerable_range: str = "<2.0.0",
) -> str:
return json.dumps(
{
"fixture-package": [
{
"id": advisory_id,
"severity": severity,
"vulnerable_versions": vulnerable_range,
"url": "https://example.invalid/advisory",
}
]
}
)
class BunAuditPolicyTest(unittest.TestCase):
def test_clean_audit_with_empty_baseline_passes(self) -> None:
self.assertEqual(evaluate(exit_code=0, output="{}", accepted={}, today=TODAY), [])
def test_exact_reviewed_advisory_passes(self) -> None:
self.assertEqual(
evaluate(
exit_code=1,
output=audit_json(),
accepted={ADVISORY.identity: ACCEPTANCE},
today=TODAY,
stderr="\x1b[1mbun audit \x1b[0m\x1b[2mv1.3.14 (d1632b29)\x1b[0m\n",
),
[],
)
def test_new_advisory_is_rejected(self) -> None:
errors = evaluate(exit_code=1, output=audit_json(), accepted={}, today=TODAY)
self.assertEqual(errors, ["unaccepted advisory 111 (fixture-package)"])
def test_changed_severity_and_range_are_rejected(self) -> None:
errors = evaluate(
exit_code=1,
output=audit_json(severity="critical", vulnerable_range="<3.0.0"),
accepted={ADVISORY.identity: ACCEPTANCE},
today=TODAY,
)
self.assertTrue(any("severity changed" in error for error in errors))
self.assertTrue(any("vulnerable range changed" in error for error in errors))
def test_expired_acceptance_is_rejected(self) -> None:
expired = Acceptance(ADVISORY, date(2026, 7, 20), "Reviewed fixture debt.")
errors = evaluate(
exit_code=1,
output=audit_json(),
accepted={ADVISORY.identity: expired},
today=TODAY,
)
self.assertEqual(errors, ["expired acceptance 111 (fixture-package) (2026-07-20)"])
def test_stale_baseline_entry_is_rejected(self) -> None:
errors = evaluate(
exit_code=0,
output="{}",
accepted={ADVISORY.identity: ACCEPTANCE},
today=TODAY,
)
self.assertEqual(errors, ["stale baseline advisory 111 (fixture-package)"])
def test_duplicate_and_overlong_acceptances_are_rejected(self) -> None:
entry = {
"id": 111,
"package": "fixture-package",
"severity": "high",
"vulnerableRange": "<2.0.0",
"expiresOn": "2026-08-01",
"rationale": "Not reachable in static output.",
}
with tempfile.TemporaryDirectory() as directory:
baseline = Path(directory) / "baseline.json"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry, entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "duplicate baseline advisory"):
load_baseline(baseline)
entry["id"] = 112
entry["expiresOn"] = "2026-11-01"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "more than 90 days"):
load_baseline(baseline)
entry["expiresOn"] = "2026-08-01"
entry["unexpected"] = True
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "unknown schema"):
load_baseline(baseline)
def test_malformed_json_and_schema_are_rejected(self) -> None:
malformed = evaluate(exit_code=1, output="not-json", accepted={}, today=TODAY)
self.assertTrue(any("malformed JSON" in error for error in malformed))
unknown = evaluate(
exit_code=1,
output=json.dumps({"fixture-package": [{"id": 111}]}),
accepted={},
today=TODAY,
)
self.assertTrue(any("unknown schema" in error for error in unknown))
empty_package = evaluate(
exit_code=0,
output=json.dumps({"fixture-package": []}),
accepted={},
today=TODAY,
)
self.assertTrue(any("non-empty list" in error for error in empty_package))
def test_scanner_failure_and_inconsistent_exit_are_rejected(self) -> None:
self.assertEqual(
evaluate(exit_code=2, output="", accepted={}, today=TODAY),
["bun audit execution failed with exit code 2"],
)
self.assertEqual(
evaluate(exit_code=1, output="{}", accepted={}, today=TODAY),
["bun audit exited with advisories but returned an empty result"],
)
def test_missing_lockfile_fails_before_scanner_execution(self) -> None:
with tempfile.TemporaryDirectory() as directory:
project = Path(directory)
baseline = project / "baseline.json"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [],
}
),
encoding="utf-8",
)
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
status = main(["--project", str(project), "--baseline", str(baseline)])
self.assertEqual(status, 1)
self.assertIn("missing Bun lockfile", stderr.getvalue())
(project / "bun.lock").write_text("", encoding="utf-8")
def execution_failure(_: Path) -> subprocess.CompletedProcess[str]:
raise OSError("registry unavailable")
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
status = main(
["--project", str(project), "--baseline", str(baseline)],
run_audit=execution_failure,
today=TODAY,
)
self.assertEqual(status, 1)
self.assertIn("cannot execute bun audit", stderr.getvalue())
def test_noncanonical_dates_and_invalid_acceptance_fields_are_rejected(self) -> None:
entry = {
"id": 111,
"package": "*",
"severity": "high",
"vulnerableRange": "<2.0.0",
"expiresOn": "20260801",
"rationale": "Not reachable in static output.",
}
with tempfile.TemporaryDirectory() as directory:
baseline = Path(directory) / "baseline.json"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "package is invalid"):
load_baseline(baseline)
entry["package"] = "fixture-package"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "YYYY-MM-DD"):
load_baseline(baseline)
entry["expiresOn"] = "2026-07-20"
baseline.write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [entry],
}
),
encoding="utf-8",
)
with self.assertRaisesRegex(ValueError, "expires before"):
load_baseline(baseline)
def test_duplicate_audit_advisory_and_unknown_severity_are_rejected(self) -> None:
duplicate = json.dumps(
{
"fixture-package": [
{
"id": 111,
"severity": "high",
"vulnerable_versions": "<2.0.0",
},
{
"id": 111,
"severity": "high",
"vulnerable_versions": "<2.0.0",
},
]
}
)
self.assertTrue(
any(
"duplicate advisory" in error
for error in evaluate(
exit_code=1, output=duplicate, accepted={}, today=TODAY
)
)
)
invalid_severity = audit_json(severity="unknown")
self.assertTrue(
any(
"severity is invalid" in error
for error in evaluate(
exit_code=1,
output=invalid_severity,
accepted={},
today=TODAY,
)
)
)
def test_network_diagnostic_fails_closed_without_echoing_payload(self) -> None:
diagnostic = "registry timeout " + ("secret-response " * 100)
self.assertEqual(
evaluate(
exit_code=1,
output=audit_json(),
stderr=diagnostic,
accepted={ADVISORY.identity: ACCEPTANCE},
today=TODAY,
),
["bun audit reported a scanner or network diagnostic"],
)
with tempfile.TemporaryDirectory() as directory:
project = Path(directory)
(project / "bun.lock").write_text("", encoding="utf-8")
(project / "baseline.json").write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [],
}
),
encoding="utf-8",
)
stderr = io.StringIO()
result = subprocess.CompletedProcess(
args=["bun", "audit", "--json"],
returncode=2,
stdout="",
stderr=diagnostic,
)
with contextlib.redirect_stderr(stderr):
status = main(
["--project", str(project), "--baseline", "baseline.json"],
run_audit=lambda _: result,
today=TODAY,
)
rendered = stderr.getvalue()
self.assertEqual(status, 1)
self.assertIn("execution failed with exit code 2", rendered)
self.assertLess(len(rendered), 500)
self.assertNotIn(diagnostic, rendered)
def test_main_uses_injected_audit_result_and_exact_baseline(self) -> None:
with tempfile.TemporaryDirectory() as directory:
project = Path(directory)
(project / "bun.lock").write_text("", encoding="utf-8")
(project / "baseline.json").write_text(
json.dumps(
{
"schemaVersion": 1,
"reviewedOn": "2026-07-21",
"accepted": [
{
"id": 111,
"package": "fixture-package",
"severity": "high",
"vulnerableRange": "<2.0.0",
"expiresOn": "2026-08-01",
"rationale": "Not reachable in static output.",
}
],
}
),
encoding="utf-8",
)
seen = []
def run_audit(path: Path) -> subprocess.CompletedProcess[str]:
seen.append(path)
return subprocess.CompletedProcess(
args=["bun", "audit", "--json"],
returncode=1,
stdout=audit_json(),
stderr="",
)
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
status = main(
["--project", str(project), "--baseline", "baseline.json"],
run_audit=run_audit,
today=TODAY,
)
self.assertEqual(status, 0)
self.assertEqual(seen, [project.resolve()])
self.assertIn("1 reviewed acceptance", stdout.getvalue())
def test_diagnostics_are_count_bounded(self) -> None:
payload = {
f"fixture-{index}": [
{
"id": index,
"severity": "high",
"vulnerable_versions": "<2.0.0",
}
]
for index in range(30)
}
errors = evaluate(
exit_code=1,
output=json.dumps(payload),
accepted={},
today=TODAY,
)
self.assertEqual(len(errors), 21)
self.assertEqual(errors[-1], "10 additional policy error(s) omitted")
if __name__ == "__main__":
unittest.main()
scripts/test_check_codegen.py
+19
View File
@@ -193,6 +193,25 @@ esac
self.assertEqual(incorrect.read_text(encoding="utf-8"), "incorrect staged output\n")
self.assertEqual(self.git_status(), status_before)
def test_rename_overlay_removes_source_before_running_generators(self) -> None:
renamed = self.root / "renamed-source.txt"
subprocess.run(["git", "mv", "source.txt", renamed.name], cwd=self.root, check=True)
for command in (self.bin / "python3", self.bin / "dart"):
contents = command.read_text(encoding="utf-8").replace(
"source.txt", renamed.name
)
contents = contents.replace(
"\n", "\nif [ -e source.txt ]; then exit 23; fi\n", 1
)
executable(command, contents)
result = self.run_codegen("--check")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse((self.root / "source.txt").exists())
self.assertEqual(renamed.read_text(encoding="utf-8"), "version one\n")
self.assert_isolation_cleaned_up()
def test_deleted_and_untracked_outputs_are_reported_without_repair(self) -> None:
deleted = self.root / "lib" / "models" / "model.g.dart"
deleted.unlink()
scripts/test_check_container_image_pins.py
+133
View File
@@ -0,0 +1,133 @@
import contextlib
import io
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
from check_container_image_pins import (
ImageReference,
SUPPORTED_PLATFORMS,
check_paths,
iter_compose_references,
iter_dockerfile_references,
main,
validate_image,
)
DIGEST = "0123456789abcdef" * 4
PLATFORMS = frozenset({"linux/amd64", "linux/arm64"})
class ContainerImagePinsTest(unittest.TestCase):
def test_accepts_readable_digest_pins_for_supported_platforms(self) -> None:
references = [
f"golang:1.22.12-alpine3.21@sha256:{DIGEST}",
f"ghcr.io/owner/image:sha-0123456@sha256:{DIGEST}",
]
for reference in references:
with self.subTest(reference=reference):
image = ImageReference(Path("fixture"), 1, reference, PLATFORMS)
self.assertEqual(validate_image(image), [])
def test_rejects_mutable_or_malformed_external_images(self) -> None:
references = [
"golang:1.22-alpine",
"golang:latest",
"golang@sha256:" + DIGEST,
"${BUILDER_IMAGE}",
f"golang:1.22@sha256:{DIGEST.upper()}",
"golang:1.22@sha256:0123456",
]
for reference in references:
with self.subTest(reference=reference):
image = ImageReference(Path("fixture"), 1, reference, PLATFORMS)
self.assertTrue(validate_image(image))
latest = ImageReference(
Path("fixture"),
1,
f"ghcr.io/owner/image:latest@sha256:{DIGEST}",
PLATFORMS,
)
self.assertIn("must not be latest", " ".join(validate_image(latest)))
def test_requires_exact_supported_platform_declaration(self) -> None:
reference = f"registry.example/image:v1@sha256:{DIGEST}"
cases = [
(None, "declaration is required"),
(frozenset({"linux/amd64"}), "linux/arm64"),
(
frozenset({"linux/amd64", "linux/arm64", "linux/s390x"}),
"linux/s390x",
),
]
for platforms, expected in cases:
with self.subTest(platforms=platforms):
image = ImageReference(Path("fixture"), 1, reference, platforms)
self.assertIn(expected, " ".join(validate_image(image)))
def test_parses_production_sources_and_ignores_scratch_and_comments(self) -> None:
dockerfile = f'''\
# FROM alpine:latest
# Review update details.
# Platforms: linux/amd64, linux/arm64
FROM --platform=$BUILDPLATFORM golang:1.22.12-alpine3.21@sha256:{DIGEST} AS build
FROM scratch
'''
compose = f'''\
services:
service:
# image: alpine:latest
# Review update details.
# Platforms: linux/amd64, linux/arm64
image: "ghcr.io/owner/service:sha-0123456@sha256:{DIGEST}"
'''
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
docker_path = root / "Dockerfile"
compose_path = root / "docker-compose.yml"
docker_path.write_text(dockerfile, encoding="utf-8")
compose_path.write_text(compose, encoding="utf-8")
docker_images = list(iter_dockerfile_references(docker_path))
compose_images = list(iter_compose_references(compose_path))
self.assertEqual(len(docker_images), 1)
self.assertEqual(len(compose_images), 1)
self.assertEqual(docker_images[0].platforms, SUPPORTED_PLATFORMS)
self.assertEqual(compose_images[0].platforms, SUPPORTED_PLATFORMS)
self.assertEqual(validate_image(docker_images[0]), [])
self.assertEqual(validate_image(compose_images[0]), [])
def test_checker_reports_each_source_location(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
docker_path = root / "Dockerfile"
compose_path = root / "docker-compose.yml"
docker_path.write_text("FROM golang:1.22-alpine AS build\n", encoding="utf-8")
compose_path.write_text(
"services:\n bugs:\n image: ghcr.io/owner/bugs:latest\n",
encoding="utf-8",
)
violations = check_paths([docker_path], [compose_path])
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
status = main([str(docker_path), str(compose_path)])
self.assertEqual(status, 1)
self.assertGreaterEqual(len(violations), 4)
output = stderr.getvalue()
self.assertIn("Dockerfile:1", output)
self.assertIn("docker-compose.yml:3", output)
self.assertIn("golang:1.22-alpine", output)
self.assertIn("ghcr.io/owner/bugs:latest", output)
def test_repository_production_references_pass(self) -> None:
self.assertEqual(main([]), 0)
if __name__ == "__main__":
unittest.main()
scripts/test_check_workflow_security.py
+83 -13
View File
@@ -7,48 +7,87 @@ from check_workflow_security import check_workflow
SAFE_SHA = "a" * 40
ROOT = Path(__file__).resolve().parents[1]
CI_PATH = Path(".github/workflows/ci.yml")
class WorkflowSecurityTests(unittest.TestCase):
def check(self, text: str) -> list[str]:
return check_workflow(Path(".github/workflows/test.yml"), text)
def check(self, text: str, path: Path | None = None) -> list[str]:
return check_workflow(path or Path(".github/workflows/test.yml"), text)
def test_accepts_read_only_pull_request_workflow_with_pinned_action(self) -> None:
errors = self.check(
f"""name: Test
on:
pull_request:
pull_request: {{}}
jobs:
test:
permissions:
contents: read
permissions: {{contents: read}}
steps:
- uses: actions/checkout@{SAFE_SHA} # v7
- name: Checkout
uses: "actions/checkout@{SAFE_SHA}" # reviewed pin
with:
persist-credentials: false
persist-credentials: "false"
"""
)
self.assertEqual(errors, [])
def test_accepts_benign_ci_names_runners_matrices_and_commands(self) -> None:
workflow = (ROOT / CI_PATH).read_text(encoding="utf-8")
changed = (
workflow.replace("name: CI - Sanity Checks", "name: Continuous integration")
.replace(" analyze:\n", " static-analysis:\n", 1)
.replace("name: Code Analysis", "name: Repository checks", 1)
.replace("runs-on: ubuntu-latest", "runs-on: internal-linux", 1)
.replace("- sanitizer: address", "- sanitizer: memory", 1)
.replace("dart run scripts/check_analyzer.dart", "dart run tool/check.dart", 1)
)
self.assertEqual(self.check(changed, CI_PATH), [])
def test_accepts_a_different_immutable_action_pin(self) -> None:
workflow = f"jobs:\n test:\n steps:\n - uses: actions/setup-go@{'b' * 40}\n"
self.assertEqual(self.check(workflow), [])
def test_rejects_mutable_action_reference(self) -> None:
errors = self.check("jobs:\n test:\n steps:\n - uses: actions/checkout@v7\n")
self.assertTrue(any("full commit SHA" in error for error in errors))
def test_rejects_missing_checkout_credential_guard_on_pull_requests(self) -> None:
errors = self.check(
f"""on: [pull_request]
jobs:
test:
steps:
- uses: actions/checkout@{SAFE_SHA}
with:
fetch-depth: 1
- run: echo 'persist-credentials: false elsewhere is not enough'
"""
)
self.assertTrue(any("discard GitHub credentials" in error for error in errors))
def test_rejects_secrets_in_pull_request_workflow(self) -> None:
errors = self.check(
"on:\n pull_request:\njobs:\n test:\n env:\n TOKEN: ${{ secrets.TOKEN }}\n"
)
self.assertTrue(any("must not reference repository secrets" in error for error in errors))
def test_rejects_write_permission_in_pull_request_workflow(self) -> None:
errors = self.check(
def test_rejects_block_or_flow_write_permission_on_pull_requests(self) -> None:
block_errors = self.check(
"on:\n pull_request:\njobs:\n test:\n permissions:\n contents: write\n"
)
self.assertTrue(any("must not request write permissions" in error for error in errors))
flow_errors = self.check(
"on: {pull_request: {}}\npermissions: {contents: write}\n"
)
self.assertTrue(any("must not request write permissions" in error for error in block_errors))
self.assertTrue(any("must not request write permissions" in error for error in flow_errors))
def test_rejects_privileged_untrusted_trigger(self) -> None:
errors = self.check("on:\n pull_request_target:\n")
self.assertTrue(any("unaudited privileged trigger" in error for error in errors))
def test_rejects_privileged_untrusted_triggers(self) -> None:
target_errors = self.check("on:\n pull_request_target:\n")
run_errors = self.check("on: [push, workflow_run]\n")
self.assertTrue(any("pull_request_target" in error for error in target_errors))
self.assertTrue(any("workflow_run" in error for error in run_errors))
def test_rejects_mutable_raw_github_download(self) -> None:
errors = self.check(
@@ -56,6 +95,37 @@ jobs:
)
self.assertTrue(any("immutable commit" in error for error in errors))
def test_rejects_ci_fail_open_constructs(self) -> None:
continued = self.check(
"jobs:\n test:\n steps:\n - continue-on-error: ${{ github.event_name == 'push' }}\n run: ./check\n",
CI_PATH,
)
suppressed = self.check(
"jobs:\n test:\n steps:\n - run: ./check || true\n",
CI_PATH,
)
explicit_false = self.check(
"jobs:\n test:\n steps:\n - continue-on-error: false\n run: ./check\n",
CI_PATH,
)
self.assertTrue(any("continue-on-error" in error for error in continued))
self.assertTrue(any("suppress failure" in error for error in suppressed))
self.assertEqual(explicit_false, [])
def test_comments_do_not_create_security_findings(self) -> None:
errors = self.check(
"""on:
push:
# pull_request_target:
jobs:
test:
steps:
# uses: actions/checkout@main
- run: echo safe
"""
)
self.assertEqual(errors, [])
if __name__ == "__main__":
unittest.main()
scripts/test_fetch_tvos_engine.py
Executable
+131
View File
@@ -0,0 +1,131 @@
#!/usr/bin/env python3
"""Regression tests for verified tvOS engine provisioning."""
from __future__ import annotations
import hashlib
import os
import shutil
import stat
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
FETCH_ENGINE = ROOT / "tvos/scripts/fetch_engine.sh"
class FetchTvosEngineTest(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory(prefix="plezy-tvos-engine-test-")
self.root = Path(self.temporary.name)
self.tvos = self.root / "tvos"
(self.tvos / "scripts").mkdir(parents=True)
shutil.copy2(FETCH_ENGINE, self.tvos / "scripts/fetch_engine.sh")
(self.tvos / "engine.version").write_text("fixture-1\n", encoding="utf-8")
(self.root / "pubspec.yaml").write_text("version: 1.2.3+4\n", encoding="utf-8")
self.cache = self.root / "cache"
self.archive = self.root / "engine.tar.gz"
self.bin = self.root / "bin"
self.bin.mkdir()
curl = self.bin / "curl"
curl.write_text(
"""#!/usr/bin/env bash
set -euo pipefail
output=
while (( $# )); do
case "$1" in
-o) output="$2"; shift 2 ;;
*) shift ;;
esac
done
cp "$FIXTURE_ARCHIVE" "$output"
""",
encoding="utf-8",
)
curl.chmod(curl.stat().st_mode | stat.S_IXUSR)
self.env = os.environ | {
"PATH": f"{self.bin}:{os.environ['PATH']}",
"FIXTURE_ARCHIVE": str(self.archive),
"FLUTTER_ROOT": str(self.root / "flutter"),
"FLUTTER_TVOS_ENGINE_CACHE": str(self.cache),
"FLUTTER_TVOS_RELEASES_URL": "https://example.invalid/flutter-tvos",
}
def tearDown(self) -> None:
self.temporary.cleanup()
def _write_archive(self, marker: str) -> str:
source = self.root / "marker.txt"
source.write_text(marker, encoding="utf-8")
with tarfile.open(self.archive, "w:gz") as archive:
archive.add(source, arcname="out/tvos_debug_sim_unopt_arm64/marker.txt")
return hashlib.sha256(self.archive.read_bytes()).hexdigest()
def _run(self) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["bash", "tvos/scripts/fetch_engine.sh"],
cwd=self.root,
env=self.env,
check=False,
capture_output=True,
text=True,
)
def test_verified_archive_installs_and_reuses_matching_cache(self) -> None:
digest = self._write_archive("reviewed engine")
(self.tvos / "engine.sha256").write_text(f"{digest}\n", encoding="utf-8")
first = self._run()
self.assertEqual(first.returncode, 0, first.stderr)
engine = self.cache / "vfixture-1"
self.assertEqual(
(engine / "out/tvos_debug_sim_unopt_arm64/marker.txt").read_text(encoding="utf-8"),
"reviewed engine",
)
self.archive.unlink()
second = self._run()
self.assertEqual(second.returncode, 0, second.stderr)
self.assertIn("using verified cached engine", second.stdout)
def test_checksum_mismatch_leaves_no_partial_engine(self) -> None:
self._write_archive("unreviewed engine")
(self.tvos / "engine.sha256").write_text(f"{'0' * 64}\n", encoding="utf-8")
result = self._run()
self.assertNotEqual(result.returncode, 0)
self.assertFalse((self.cache / "vfixture-1").exists())
self.assertEqual(list(self.cache.glob(".engine.*")), [])
def test_checksum_change_replaces_same_version_without_stale_files(self) -> None:
first_digest = self._write_archive("first engine")
checksum = self.tvos / "engine.sha256"
checksum.write_text(f"{first_digest}\n", encoding="utf-8")
self.assertEqual(self._run().returncode, 0)
stale = self.cache / "vfixture-1/stale-from-previous-archive"
stale.write_text("stale", encoding="utf-8")
second_digest = self._write_archive("second engine")
checksum.write_text(f"{second_digest}\n", encoding="utf-8")
result = self._run()
self.assertEqual(result.returncode, 0, result.stderr)
marker = self.cache / "vfixture-1/out/tvos_debug_sim_unopt_arm64/marker.txt"
self.assertEqual(marker.read_text(encoding="utf-8"), "second engine")
self.assertFalse(stale.exists())
self.assertEqual(
(self.cache / "vfixture-1/.installed").read_text(encoding="utf-8").strip(),
f"fixture-1 {second_digest}",
)
if __name__ == "__main__":
unittest.main()
scripts/test_verify_runtime_inputs.py
Executable
+193
View File
@@ -0,0 +1,193 @@
#!/usr/bin/env python3
import importlib.util
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
SCRIPT = Path(__file__).with_name("verify_runtime_inputs.py")
SPEC = importlib.util.spec_from_file_location("verify_runtime_inputs", SCRIPT)
CHECKER = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
SPEC.loader.exec_module(CHECKER)
REPOSITORY = Path(__file__).resolve().parents[1]
FIXTURES = (
"pubspec.lock",
"linux/CMakeLists.txt",
"linux/packaging/build-libmpv.sh",
"linux/packaging/native-inputs.json",
"packages/wakelock_plus/pubspec.yaml",
"packages/wakelock_plus/pubspec.lock",
"packages/wakelock_plus/provenance.json",
"packages/wakelock_plus/pigeons/messages.dart",
"packages/wakelock_plus/android/src/main/kotlin/dev/fluttercommunity/plus/wakelock/WakelockPlusMessages.g.kt",
"packages/wakelock_plus/ios/wakelock_plus/Sources/wakelock_plus/include/wakelock_plus/messages.g.h",
"packages/wakelock_plus/ios/wakelock_plus/Sources/wakelock_plus/messages.g.m",
)
class RuntimeInputVerifierTest(unittest.TestCase):
def setUp(self) -> None:
self.temporary = tempfile.TemporaryDirectory()
self.root = Path(self.temporary.name)
for relative in FIXTURES:
source = REPOSITORY / relative
destination = self.root / relative
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, destination)
def tearDown(self) -> None:
self.temporary.cleanup()
def _json(self, relative: str) -> dict:
return json.loads((self.root / relative).read_text(encoding="utf-8"))
def _write_json(self, relative: str, payload: dict) -> None:
(self.root / relative).write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def test_reviewed_inputs_pass_library_and_offline_cli(self) -> None:
self.assertEqual([], CHECKER.validate(self.root))
completed = subprocess.run(
[sys.executable, str(SCRIPT), "--root", str(self.root)],
check=False,
capture_output=True,
text=True,
)
self.assertEqual(0, completed.returncode, completed.stdout + completed.stderr)
self.assertIn("verified offline", completed.stdout)
def test_rejects_linux_cmake_checksum_drift(self) -> None:
path = self.root / "linux/CMakeLists.txt"
path.write_text(path.read_text(encoding="utf-8").replace("9fe4d6f5", "0fe4d6f5"), encoding="utf-8")
errors = CHECKER.validate(self.root)
self.assertTrue(any("simdutf SHA-256 differs" in error for error in errors))
def test_rejects_malformed_native_pin_and_version_url_drift(self) -> None:
manifest = self._json("linux/packaging/native-inputs.json")
manifest["inputs"]["ffmpeg"]["sha256"] = "not-a-digest"
manifest["inputs"]["mpv"]["url"] = "https://example.invalid/mpv-current.tar.gz"
self._write_json("linux/packaging/native-inputs.json", manifest)
errors = CHECKER.validate(self.root)
self.assertTrue(any("ffmpeg.sha256" in error for error in errors))
self.assertTrue(any("mpv.url" in error and "declared version" in error for error in errors))
def test_reports_missing_simdutf_fields_without_crashing(self) -> None:
manifest = self._json("linux/packaging/native-inputs.json")
simdutf = manifest["inputs"]["simdutf"]
simdutf.pop("url")
simdutf.pop("sha256")
self._write_json("linux/packaging/native-inputs.json", manifest)
errors = CHECKER.validate(self.root)
self.assertTrue(any("simdutf.url" in error and "non-empty text" in error for error in errors))
self.assertTrue(any("simdutf.sha256" in error and "lowercase full SHA-256" in error for error in errors))
def test_rejects_disconnected_production_acquisition(self) -> None:
path = self.root / "linux/packaging/build-libmpv.sh"
path.write_text(
path.read_text(encoding="utf-8").replace(
'download_verified "$MPV_URL" "$MPV_SHA256"',
'curl "$MPV_URL"',
),
encoding="utf-8",
)
errors = CHECKER.validate(self.root)
self.assertTrue(any("MPV_URL" in error and "manifest-backed" in error for error in errors))
def test_rejects_binding_source_or_output_drift(self) -> None:
schema = self.root / "packages/wakelock_plus/pigeons/messages.dart"
schema.write_text(schema.read_text(encoding="utf-8") + "// changed\n", encoding="utf-8")
kotlin = self.root / (
"packages/wakelock_plus/android/src/main/kotlin/"
"dev/fluttercommunity/plus/wakelock/WakelockPlusMessages.g.kt"
)
kotlin.write_bytes(kotlin.read_bytes() + b"\n")
errors = CHECKER.validate(self.root)
self.assertGreaterEqual(sum("SHA-256 drift" in error for error in errors), 2)
def test_rejects_generator_and_external_client_lock_drift(self) -> None:
pubspec = self.root / "packages/wakelock_plus/pubspec.yaml"
pubspec.write_text(pubspec.read_text(encoding="utf-8").replace("pigeon: 26.2.3", "pigeon: ^26.2.3"), encoding="utf-8")
lock = self.root / "packages/wakelock_plus/pubspec.lock"
lock.write_text(
lock.read_text(encoding="utf-8").replace(
"24b84143787220a403491c2e5de0877fbbb87baf3f0b18a2a988973863db4b03",
"04b84143787220a403491c2e5de0877fbbb87baf3f0b18a2a988973863db4b03",
),
encoding="utf-8",
)
root_lock = self.root / "pubspec.lock"
root_lock.write_text(
root_lock.read_text(encoding="utf-8").replace(
"24b84143787220a403491c2e5de0877fbbb87baf3f0b18a2a988973863db4b03",
"14b84143787220a403491c2e5de0877fbbb87baf3f0b18a2a988973863db4b03",
),
encoding="utf-8",
)
errors = CHECKER.validate(self.root)
self.assertTrue(any("Pigeon must be pinned exactly" in error for error in errors))
self.assertTrue(any("platform-interface version/checksum differs" in error for error in errors))
self.assertTrue(any("runtime platform-interface" in error for error in errors))
def test_missing_binding_reports_error_without_discarding_earlier_errors(self) -> None:
manifest = self._json("linux/packaging/native-inputs.json")
manifest["inputs"]["simdutf"].pop("url")
self._write_json("linux/packaging/native-inputs.json", manifest)
kotlin = self.root / (
"packages/wakelock_plus/android/src/main/kotlin/"
"dev/fluttercommunity/plus/wakelock/WakelockPlusMessages.g.kt"
)
kotlin.unlink()
errors = CHECKER.validate(self.root)
self.assertTrue(any("simdutf.url" in error for error in errors))
self.assertTrue(any(str(kotlin) in error and "cannot read generated binding" in error for error in errors))
def test_accepts_benign_prose_contract_edits(self) -> None:
native = self._json("linux/packaging/native-inputs.json")
native["refreshContract"] = {"rules": ["Reworded maintainer guidance."]}
native["inputs"]["ffmpeg"]["provenance"] = "Reviewed release evidence."
self._write_json("linux/packaging/native-inputs.json", native)
provenance = self._json("packages/wakelock_plus/provenance.json")
provenance["plezyDeltas"] = ["Reworded local-change notes."]
provenance["refreshContract"] = ["Reworded refresh guidance."]
provenance["externalDartClient"]["contract"] = "Reworded client guidance."
self._write_json("packages/wakelock_plus/provenance.json", provenance)
self.assertEqual([], CHECKER.validate(self.root))
def test_rejects_dart_output_from_host_only_schema(self) -> None:
schema = self.root / "packages/wakelock_plus/pigeons/messages.dart"
schema.write_text(
schema.read_text(encoding="utf-8").replace(
"PigeonOptions(",
"PigeonOptions(\n dartOut: '../other/lib/messages.g.dart',",
),
encoding="utf-8",
)
errors = CHECKER.validate(self.root)
self.assertTrue(any("must not generate Dart outputs" in error for error in errors))
if __name__ == "__main__":
unittest.main()
scripts/verify_runtime_inputs.py
Executable
+256
View File
@@ -0,0 +1,256 @@
#!/usr/bin/env python3
"""Offline verification for reviewed Linux native and vendored binding inputs."""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
from pathlib import Path
from typing import Any
HEX_256 = re.compile(r"^[0-9a-f]{64}$")
HEX_COMMIT = re.compile(r"^[0-9a-f]{40}$")
NATIVE_NAMES = {"ffmpeg", "shaderc", "libplacebo", "mpv", "simdutf"}
BINDING_ARTIFACTS = {
"pigeons/messages.dart",
"android/src/main/kotlin/dev/fluttercommunity/plus/wakelock/WakelockPlusMessages.g.kt",
"ios/wakelock_plus/Sources/wakelock_plus/include/wakelock_plus/messages.g.h",
"ios/wakelock_plus/Sources/wakelock_plus/messages.g.m",
}
def _load_json(path: Path, errors: list[str]) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
errors.append(f"{path}: cannot load JSON: {error}")
return {}
if not isinstance(value, dict):
errors.append(f"{path}: top-level value must be an object")
return {}
return value
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _require_text(value: Any, label: str, errors: list[str]) -> str:
if not isinstance(value, str) or not value.strip():
errors.append(f"{label}: must be non-empty text")
return ""
return value
def _locked_package(lock_text: str, name: str) -> tuple[str, str] | None:
pattern = re.compile(
rf"^ {re.escape(name)}:\n(?P<body>(?: .*\n| .*\n)+?)(?=^ [a-zA-Z0-9_]+:|\Z)",
re.MULTILINE,
)
match = pattern.search(lock_text)
if match is None:
return None
body = match.group("body")
version = re.search(r'^ version: "([^\"]+)"$', body, re.MULTILINE)
checksum = re.search(r'^ sha256: "?([0-9a-f]{64})"?$', body, re.MULTILINE)
if version is None or checksum is None:
return None
return version.group(1), checksum.group(1)
def _validate_native(root: Path, errors: list[str]) -> None:
manifest_path = root / "linux/packaging/native-inputs.json"
manifest = _load_json(manifest_path, errors)
if manifest.get("formatVersion") != 1:
errors.append(f"{manifest_path}: formatVersion must be 1")
inputs = manifest.get("inputs")
if not isinstance(inputs, dict) or set(inputs) != NATIVE_NAMES:
errors.append(f"{manifest_path}: inputs must be exactly {sorted(NATIVE_NAMES)}")
return
for name, value in inputs.items():
label = f"{manifest_path}: inputs.{name}"
if not isinstance(value, dict):
errors.append(f"{label}: must be an object")
continue
kind = value.get("kind")
version = _require_text(value.get("version"), f"{label}.version", errors)
url = _require_text(value.get("url"), f"{label}.url", errors)
_require_text(value.get("provenance"), f"{label}.provenance", errors)
if url and not url.startswith("https://"):
errors.append(f"{label}.url: production source must use HTTPS")
if version and url and name in {"ffmpeg", "mpv", "simdutf"} and version not in url:
errors.append(f"{label}.url: must identify declared version {version}")
if kind == "archive":
checksum = value.get("sha256")
if not isinstance(checksum, str) or HEX_256.fullmatch(checksum) is None:
errors.append(f"{label}.sha256: must be a lowercase full SHA-256")
elif kind == "git":
ref = value.get("ref")
commit = value.get("commit")
if not isinstance(ref, str) or ref != f"v{version}":
errors.append(f"{label}.ref: must be v{version}")
if not isinstance(commit, str) or HEX_COMMIT.fullmatch(commit) is None:
errors.append(f"{label}.commit: must be a lowercase full Git commit")
else:
errors.append(f"{label}.kind: must be archive or git")
cmake_path = root / "linux/CMakeLists.txt"
try:
cmake = cmake_path.read_text(encoding="utf-8")
except OSError as error:
errors.append(f"{cmake_path}: cannot read: {error}")
cmake = ""
simdutf = inputs.get("simdutf")
if isinstance(simdutf, dict):
simdutf_url = simdutf.get("url")
simdutf_sha256 = simdutf.get("sha256")
if isinstance(simdutf_url, str) and simdutf_url and f"URL {simdutf_url}" not in cmake:
errors.append(f"{cmake_path}: simdutf URL differs from native-inputs.json")
if (
isinstance(simdutf_sha256, str)
and HEX_256.fullmatch(simdutf_sha256) is not None
and f"URL_HASH SHA256={simdutf_sha256}" not in cmake
):
errors.append(f"{cmake_path}: simdutf SHA-256 differs from native-inputs.json")
builder_path = root / "linux/packaging/build-libmpv.sh"
try:
builder = builder_path.read_text(encoding="utf-8")
except OSError as error:
errors.append(f"{builder_path}: cannot read: {error}")
return
required_builder_contracts = (
"native-inputs.json",
'download_verified "$FFMPEG_URL" "$FFMPEG_SHA256"',
'download_verified "$MPV_URL" "$MPV_SHA256"',
'"$SHADERC_URL" "$SHADERC_REF" "$SHADERC_COMMIT"',
'"$LIBPLACEBO_URL" "$LIBPLACEBO_REF" "$LIBPLACEBO_COMMIT"',
'git submodule update --init --recursive',
)
for contract_text in required_builder_contracts:
if contract_text not in builder:
errors.append(f"{builder_path}: missing manifest-backed acquisition contract {contract_text!r}")
if re.search(r"curl[^\n]*\|[^\n]*tar", builder):
errors.append(f"{builder_path}: archive extraction must not consume a curl stream")
def _validate_wakelock(root: Path, errors: list[str]) -> None:
package = root / "packages/wakelock_plus"
provenance_path = package / "provenance.json"
provenance = _load_json(provenance_path, errors)
if provenance.get("formatVersion") != 1:
errors.append(f"{provenance_path}: formatVersion must be 1")
upstream = provenance.get("upstream")
if not isinstance(upstream, dict) or HEX_COMMIT.fullmatch(str(upstream.get("commit", ""))) is None:
errors.append(f"{provenance_path}: upstream.commit must be a full Git commit")
artifacts = provenance.get("artifacts")
if not isinstance(artifacts, dict) or set(artifacts) != BINDING_ARTIFACTS:
errors.append(f"{provenance_path}: artifacts must be exactly the schema and three host outputs")
else:
for relative, expected in artifacts.items():
path = package / relative
if not isinstance(expected, str) or HEX_256.fullmatch(expected) is None:
errors.append(f"{provenance_path}: invalid artifact SHA-256 for {relative}")
elif not path.is_file():
errors.append(f"{path}: required binding artifact is missing")
else:
actual = _sha256(path)
if actual != expected:
errors.append(f"{path}: SHA-256 drift (expected {expected}, got {actual})")
try:
pubspec = (package / "pubspec.yaml").read_text(encoding="utf-8")
lock = (package / "pubspec.lock").read_text(encoding="utf-8")
schema = (package / "pigeons/messages.dart").read_text(encoding="utf-8")
root_lock = (root / "pubspec.lock").read_text(encoding="utf-8")
except OSError as error:
errors.append(f"{package}: cannot read package provenance input: {error}")
return
generator = provenance.get("generator") if isinstance(provenance.get("generator"), dict) else {}
client = provenance.get("externalDartClient") if isinstance(provenance.get("externalDartClient"), dict) else {}
expected_pigeon = (str(generator.get("version", "")), str(generator.get("archiveSha256", "")))
expected_client = (str(client.get("version", "")), str(client.get("archiveSha256", "")))
if not re.search(rf"^ pigeon: {re.escape(expected_pigeon[0])}$", pubspec, re.MULTILINE):
errors.append(f"{package / 'pubspec.yaml'}: Pigeon must be pinned exactly to {expected_pigeon[0]}")
if not re.search(
rf"^ wakelock_plus_platform_interface: {re.escape(expected_client[0])}$", pubspec, re.MULTILINE
):
errors.append(
f"{package / 'pubspec.yaml'}: wakelock_plus_platform_interface must be pinned exactly to {expected_client[0]}"
)
if _locked_package(lock, "pigeon") != expected_pigeon:
errors.append(f"{package / 'pubspec.lock'}: Pigeon version/checksum differs from provenance.json")
if _locked_package(lock, "wakelock_plus_platform_interface") != expected_client:
errors.append(
f"{package / 'pubspec.lock'}: platform-interface version/checksum differs from provenance.json"
)
if _locked_package(root_lock, "wakelock_plus_platform_interface") != expected_client:
errors.append(
f"{root / 'pubspec.lock'}: runtime platform-interface version/checksum differs from provenance.json"
)
if "dartPackageName: 'wakelock_plus_platform_interface'" not in schema:
errors.append(f"{package / 'pigeons/messages.dart'}: external Dart package name is not explicit")
if re.search(r"\bdart(?:Test)?Out\s*:", schema):
errors.append(f"{package / 'pigeons/messages.dart'}: host-only schema must not generate Dart outputs")
for relative in BINDING_ARTIFACTS - {"pigeons/messages.dart"}:
if relative not in schema:
errors.append(f"{package / 'pigeons/messages.dart'}: missing owned output {relative}")
binding_sources = (
("Kotlin", package / "android/src/main/kotlin/dev/fluttercommunity/plus/wakelock/WakelockPlusMessages.g.kt"),
("Objective-C", package / "ios/wakelock_plus/Sources/wakelock_plus/messages.g.m"),
)
for generated_name, generated_path in binding_sources:
try:
generated = generated_path.read_text(encoding="utf-8")
except OSError as error:
errors.append(f"{generated_path}: cannot read generated binding: {error}")
continue
if "26.2.3" not in generated:
errors.append(f"{generated_name} binding was not generated by Pigeon 26.2.3")
for method in ("WakelockPlusApi.toggle", "WakelockPlusApi.isEnabled"):
if method not in generated:
errors.append(f"{generated_name} binding is missing channel suffix {method}")
for tag in ("129", "130"):
if tag not in generated:
errors.append(f"{generated_name} binding is missing codec tag {tag}")
def validate(root: Path) -> list[str]:
errors: list[str] = []
_validate_native(root, errors)
_validate_wakelock(root, errors)
return errors
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
arguments = parser.parse_args()
errors = validate(arguments.root.resolve())
if errors:
print("Runtime input provenance verification failed:", file=sys.stderr)
for error in errors:
print(f"- {error}", file=sys.stderr)
return 1
print("Runtime input provenance verified offline")
return 0
if __name__ == "__main__":
raise SystemExit(main())