Dispatch PR previews to paired Web Simulator

.github/scripts/dispatch_browser_preview.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Validate a live Specter PR and dispatch its exact preview request."""
+from datetime import datetime, timezone
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+import json
+import os
+import re
+import sys
+
+
+API = "https://api.github.com"
+REPOSITORY_RE = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+\Z")
+SHA_RE = re.compile(r"[a-f0-9]{40}\Z")
+
+
+def gh(method, path, token, data=None):
+    headers = {"Accept": "application/vnd.github+json", "User-Agent": "specter-preview"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    raw = json.dumps(data).encode() if data is not None else None
+    if raw is not None:
+        headers["Content-Type"] = "application/json"
+    try:
+        with urlopen(Request(API + path, data=raw, headers=headers, method=method), timeout=25) as response:
+            result = response.read()
+            return json.loads(result) if result else None
+    except HTTPError as exc:
+        raise RuntimeError(f"GitHub API {method} {path.split('?', 1)[0]} returned HTTP {exc.code}") from exc
+    except (URLError, TimeoutError, json.JSONDecodeError) as exc:
+        raise RuntimeError("GitHub API request failed") from exc
+
+
+def parse_time(value):
+    if not isinstance(value, str) or len(value) > 40:
+        raise ValueError("source_updated_at must be an RFC3339 timestamp")
+    try:
+        result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError("source_updated_at must be an RFC3339 timestamp") from exc
+    if result.tzinfo is None:
+        raise ValueError("source_updated_at must include a timezone")
+    return result.astimezone(timezone.utc)
+
+
+def valid_ref(value):
+    if (not isinstance(value, str) or not 1 <= len(value) <= 255 or
+            value.startswith(("/", ".")) or value.endswith(("/", ".", ".lock"))):
+        return False
+    if any(ord(char) < 32 or ord(char) == 127 for char in value):
+        return False
+    if any(part in value for part in ("..", "//", "@{", "\\", " ", "~", "^", ":", "?", "*", "[")):
+        return False
+    return all(part and not part.startswith(".") and not part.endswith(".lock")
+               for part in value.split("/"))
+
+
+def request_from_environment(env):
+    base = env["BASE_REPOSITORY"]
+    simulator = env.get("WEB_SIMULATOR_REPOSITORY", "").strip()
+    if not simulator and REPOSITORY_RE.fullmatch(base):
+        simulator = f"{base.split('/', 1)[0]}/specter-diy-web-simulator"
+    action = env["ACTION"]
+    try:
+        number = int(env["PR_NUMBER"])
+        run_id = int(env["GITHUB_RUN_ID"])
+        run_attempt = int(env["GITHUB_RUN_ATTEMPT"])
+    except (KeyError, ValueError) as exc:
+        raise ValueError("invalid PR or workflow run identity") from exc
+
+    head_repository = env.get("HEAD_REPOSITORY", "")
+    head_ref = env.get("HEAD_REF", "")
+    head_sha = env["HEAD_SHA"]
+    base_sha = env["BASE_SHA"]
+    base_ref = env["BASE_REF"]
+    source_updated_at = env["SOURCE_UPDATED_AT"]
+
+    if not REPOSITORY_RE.fullmatch(base) or base.rsplit("/", 1)[1].lower() != "specter-diy":
+        raise ValueError("invalid Specter base repository")
+    if (not REPOSITORY_RE.fullmatch(simulator) or
+            simulator.split("/", 1)[0].lower() != base.split("/", 1)[0].lower()):
+        raise ValueError("invalid paired Web Simulator repository")
+    if not 1 <= number <= 9999999 or run_id <= 0 or run_attempt <= 0:
+        raise ValueError("invalid PR or workflow run identity")
+    if action not in ("build", "delete"):
+        raise ValueError("invalid preview action")
+    if not SHA_RE.fullmatch(head_sha) or not SHA_RE.fullmatch(base_sha):
+        raise ValueError("invalid full PR SHA")
+    if not valid_ref(base_ref):
+        raise ValueError("invalid PR base ref")
+    if action == "build":
+        if (not REPOSITORY_RE.fullmatch(head_repository) or
+                not valid_ref(head_ref)):
+            raise ValueError("invalid PR head repository or ref")
+    elif head_repository and not REPOSITORY_RE.fullmatch(head_repository):
+        raise ValueError("invalid PR head repository")
+
+    parse_time(source_updated_at)
+    request_id = f"specter-pr-{number}-{head_sha}-{run_id}-{run_attempt}"
+    return {
+        "request_id": request_id,
+        "action": action,
+        "base_repository": base,
+        "base_sha": base_sha,
+        "base_ref": base_ref,
+        "pr_number": number,
+        "head_repository": head_repository,
+        "head_sha": head_sha,
+        "head_ref": head_ref,
+        "source_updated_at": source_updated_at,
+    }
+
+
+def current(request, token, pull_fetcher=None):
+    fetch = pull_fetcher or (lambda repo, number: gh(
+        "GET", f"/repos/{repo}/pulls/{number}", token))
+    pr = fetch(request["base_repository"], request["pr_number"])
+    if int(pr.get("number", -1)) != request["pr_number"]:
+        return False
+    base, head = pr.get("base") or {}, pr.get("head") or {}
+    base_repo = (base.get("repo") or {}).get("full_name", "")
+    if base_repo.lower() != request["base_repository"].lower():
+        return False
+    if head.get("sha") != request["head_sha"]:
+        return False
+    if request["action"] == "build":
+        if (pr.get("state") != "open" or base.get("sha") != request["base_sha"] or
+                base.get("ref") != request["base_ref"]):
+            return False
+        if ((head.get("repo") or {}).get("full_name", "").lower() !=
+                request["head_repository"].lower() or head.get("ref") != request["head_ref"]):
+            return False
+    else:
+        if pr.get("state") != "closed":
+            return False
+        live_repository = (head.get("repo") or {}).get("full_name", "")
+        if live_repository and request["head_repository"] and (
+                live_repository.lower() != request["head_repository"].lower()):
+            return False
+    if parse_time(pr.get("updated_at", "")) < parse_time(request["source_updated_at"]):
+        return False
+    return True
+
+
+def dispatch(request, simulator, token, github=None):
+    github = github or gh
+    service = github("GET", f"/repos/{simulator}", token)
+    default_branch = service.get("default_branch") if isinstance(service, dict) else None
+    if not isinstance(default_branch, str) or not valid_ref(default_branch):
+        raise RuntimeError("Web Simulator repository has no valid default branch")
+    inputs = {key: str(value) for key, value in request.items()}
+    github("POST", f"/repos/{simulator}/actions/workflows/preview.yml/dispatches", token,
+           {"ref": default_branch, "inputs": inputs})
+
+
+def main():
+    request = request_from_environment(os.environ)
+    token = os.environ["GITHUB_TOKEN"]
+    if not current(request, token):
+        return
+    simulator = os.environ.get("WEB_SIMULATOR_REPOSITORY", "").strip()
+    if not simulator:
+        simulator = f"{request['base_repository'].split('/', 1)[0]}/specter-diy-web-simulator"
+    dispatch(request, simulator,
+             os.environ["WEB_SIMULATOR_DISPATCH_TOKEN"])
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(f"Browser preview dispatcher failed: {type(exc).__name__}: {exc}", file=sys.stderr)
+        sys.exit(1)

.github/scripts/test_dispatch_browser_preview.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""Tests for exact PR validation and short-lived preview dispatch."""
+from io import BytesIO
+from pathlib import Path
+from unittest.mock import patch
+from urllib.error import HTTPError
+import json
+import os
+import sys
+import unittest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import dispatch_browser_preview as dispatcher
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SHA = "a" * 40
+BASE_SHA = "b" * 40
+UPDATED = "2026-10-02T12:00:00Z"
+
+
+def environment():
+    return {
+        "BASE_REPOSITORY": "cryptoadvance/specter-diy",
+        "PR_NUMBER": "12",
+        "BASE_SHA": BASE_SHA,
+        "BASE_REF": "master",
+        "HEAD_REPOSITORY": "contributor/specter-diy",
+        "HEAD_SHA": SHA,
+        "HEAD_REF": "feature",
+        "SOURCE_UPDATED_AT": UPDATED,
+        "ACTION": "build",
+        "GITHUB_RUN_ID": "123",
+        "GITHUB_RUN_ATTEMPT": "2",
+        "GITHUB_TOKEN": "workflow-token",
+        "WEB_SIMULATOR_DISPATCH_TOKEN": "dispatch-token",
+        "WEB_SIMULATOR_REPOSITORY": "cryptoadvance/specter-diy-web-simulator",
+    }
+
+
+def live_pr(**changes):
+    result = {
+        "number": 12,
+        "state": "open",
+        "updated_at": UPDATED,
+        "base": {"repo": {"full_name": "cryptoadvance/specter-diy"},
+                 "sha": BASE_SHA, "ref": "master"},
+        "head": {"repo": {"full_name": "contributor/specter-diy"},
+                 "sha": SHA, "ref": "feature"},
+    }
+    for key, value in changes.items():
+        result[key] = value
+    return result
+
+
+class BrowserPreviewDispatcherTests(unittest.TestCase):
+    def test_dispatches_exact_fork_pr_metadata_and_returns_without_polling(self):
+        calls = []
+
+        def fake_gh(method, path, token="", data=None):
+            calls.append((method, path, token, data))
+            if path == "/repos/cryptoadvance/specter-diy/pulls/12":
+                return live_pr()
+            if path == "/repos/cryptoadvance/specter-diy-web-simulator":
+                return {"default_branch": "main"}
+            return None
+
+        with patch.dict(os.environ, environment()), patch.object(dispatcher, "gh", side_effect=fake_gh):
+            dispatcher.main()
+
+        posts = [call for call in calls if call[0] == "POST"]
+        self.assertEqual(len(posts), 1)
+        method, path, token, body = posts[0]
+        self.assertEqual(method, "POST")
+        self.assertEqual(path, "/repos/cryptoadvance/specter-diy-web-simulator/actions/workflows/preview.yml/dispatches")
+        self.assertEqual(token, "dispatch-token")
+        self.assertEqual(body["ref"], "main")
+        self.assertEqual(body["inputs"], {
+            "request_id": f"specter-pr-12-{SHA}-123-2",
+            "action": "build",
+            "base_repository": "cryptoadvance/specter-diy",
+            "base_sha": BASE_SHA,
+            "base_ref": "master",
+            "pr_number": "12",
+            "head_repository": "contributor/specter-diy",
+            "head_sha": SHA,
+            "head_ref": "feature",
+            "source_updated_at": UPDATED,
+        })
+        self.assertEqual([call[1] for call in calls].count("/repos/cryptoadvance/specter-diy/pulls/12"), 1)
+
+    def test_stale_live_pr_is_not_dispatched(self):
+        for pr in (live_pr(head={"repo": {"full_name": "other/specter-diy"},
+                             "sha": SHA, "ref": "feature"}),
+                   live_pr(head={"repo": {"full_name": "contributor/specter-diy"},
+                                 "sha": "c" * 40, "ref": "feature"}),
+                   live_pr(base={"repo": {"full_name": "cryptoadvance/specter-diy"},
+                                 "sha": "d" * 40, "ref": "master"}),
+                   live_pr(state="closed")):
+            with self.subTest(pr=pr), patch.dict(os.environ, environment()), \
+                    patch.object(dispatcher, "gh", side_effect=[pr]) as api:
+                dispatcher.main()
+                api.assert_called_once()
+
+    def test_request_metadata_requires_full_exact_identity(self):
+        for key, value in (("PR_NUMBER", "0"), ("HEAD_SHA", "A" * 40),
+                           ("BASE_SHA", "short"), ("BASE_REF", "../main"),
+                           ("HEAD_REF", "feature\\branch"),
+                           ("SOURCE_UPDATED_AT", "2026-10-02T12:00:00"),
+                           ("GITHUB_RUN_ATTEMPT", "0"),
+                           ("WEB_SIMULATOR_REPOSITORY", "attacker/example")):
+            with self.subTest(key=key), self.assertRaises(ValueError):
+                dispatcher.request_from_environment({**environment(), key: value})
+
+    def test_renamed_fork_is_allowed_when_it_matches_live_pr(self):
+        env = {**environment(), "HEAD_REPOSITORY": "contributor/renamed-specter-fork"}
+        pr = live_pr(head={"repo": {"full_name": env["HEAD_REPOSITORY"]},
+                           "sha": SHA, "ref": "feature"})
+        with patch.dict(os.environ, env), patch.object(dispatcher, "gh",
+                side_effect=[pr, {"default_branch": "main"}, None]) as api:
+            dispatcher.main()
+        self.assertEqual(api.call_args_list[-1].args[0], "POST")
+
+    def test_default_pairing_uses_the_base_repository_owner(self):
+        env = environment()
+        env.pop("WEB_SIMULATOR_REPOSITORY")
+        with patch.dict(os.environ, env), patch.object(dispatcher, "gh") as api:
+            api.side_effect = [live_pr(), {"default_branch": "main"}, None]
+            dispatcher.main()
+        self.assertEqual(api.call_args_list[-1].args[1],
+                         "/repos/cryptoadvance/specter-diy-web-simulator/actions/workflows/preview.yml/dispatches")
+
+    def test_close_dispatch_requires_live_closed_pr(self):
+        env = {**environment(), "ACTION": "delete", "HEAD_REPOSITORY": "",
+               "HEAD_REF": ""}
+        request = dispatcher.request_from_environment(env)
+        self.assertFalse(dispatcher.current(request, "token", lambda *_: live_pr()))
+        closed = live_pr(state="closed", head={"repo": None, "sha": SHA, "ref": "feature"})
+        self.assertTrue(dispatcher.current(request, "token", lambda *_: closed))
+
+    def test_edit_events_do_not_start_builds_and_closed_event_dispatches_cleanup(self):
+        workflow = (ROOT / ".github/workflows/browser-preview.yml").read_text(encoding="utf-8")
+        types = next(line for line in workflow.splitlines() if "types:" in line)
+        self.assertNotIn("edited", types)
+        self.assertIn("closed", types)
+        self.assertIn("github.event.action == 'closed' && 'delete'", workflow)
+
+    def test_http_error_does_not_expose_response_body(self):
+        error = HTTPError("https://api.github.com/private", 403, "denied", {},
+                          BytesIO(b"sensitive response content"))
+        with patch.object(dispatcher, "urlopen", side_effect=error):
+            with self.assertRaisesRegex(RuntimeError, "HTTP 403") as raised:
+                dispatcher.gh("POST", "/repos/example/repo/dispatches", "token", {})
+        self.assertNotIn("sensitive response content", str(raised.exception))
+
+    def test_workflow_is_short_trusted_and_has_no_polling_or_pr_comment_write(self):
+        workflow = (ROOT / ".github/workflows/browser-preview.yml").read_text(encoding="utf-8")
+        source = (ROOT / ".github/scripts/dispatch_browser_preview.py").read_text(encoding="utf-8")
+        self.assertIn("timeout-minutes: 10", workflow)
+        self.assertIn("pull-requests: read", workflow)
+        self.assertNotIn("pull-requests: write", workflow)
+        self.assertNotIn("issues: write", workflow)
+        self.assertNotIn("github.event.pull_request.head", workflow.split("run:", 1)[-1])
+        self.assertNotIn("sleep(", source)
+        self.assertNotIn("status/pr/", source)
+        self.assertNotIn("comments", source)
+        self.assertNotIn("poll", source.lower())
+        self.assertIn("ref: ${{ github.event.repository.default_branch }}", workflow)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/workflows/browser-preview.yml

@@ -0,0 +1,43 @@
+name: Browser simulator preview
+
+on:
+  pull_request_target:
+    types: [opened, synchronize, reopened, ready_for_review, closed]
+
+permissions:
+  contents: read
+  pull-requests: read
+
+concurrency:
+  group: browser-preview-${{ github.repository }}-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  preview:
+    name: Dispatch browser preview
+    if: github.event.pull_request.base.repo.full_name == github.repository
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Check out trusted default-branch dispatcher
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 pinned commit
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+          persist-credentials: false
+      - name: Dispatch exact PR metadata to the paired Web Simulator
+        env:
+          BASE_REPOSITORY: ${{ github.event.repository.full_name }}
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          BASE_REF: ${{ github.event.pull_request.base.ref }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          HEAD_REPOSITORY: ${{ github.event.pull_request.head.repo.full_name || '' }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+          HEAD_REF: ${{ github.event.pull_request.head.ref || '' }}
+          SOURCE_UPDATED_AT: ${{ github.event.pull_request.updated_at }}
+          ACTION: ${{ github.event.action == 'closed' && 'delete' || 'build' }}
+          GITHUB_RUN_ID: ${{ github.run_id }}
+          GITHUB_RUN_ATTEMPT: ${{ github.run_attempt }}
+          GITHUB_TOKEN: ${{ github.token }}
+          WEB_SIMULATOR_DISPATCH_TOKEN: ${{ secrets.WEB_SIMULATOR_DISPATCH_TOKEN }}
+          WEB_SIMULATOR_REPOSITORY: ${{ vars.WEB_SIMULATOR_REPOSITORY }}
+        run: python3 .github/scripts/dispatch_browser_preview.py

.github/workflows/test.yml

@@ -46,3 +46,5 @@ jobs:
             python3
       - name: Run tests
         run: make test
+      - name: Test browser preview dispatcher
+        run: python3 .github/scripts/test_dispatch_browser_preview.py