From a40aa9fb1bea76309a2edc8bf0c8ee200ec2e4c2 Mon Sep 17 00:00:00 2001 From: 7fpb5574ck-maker <369284625@qq.com> Date: Thu, 15 Oct 2026 17:36:50 +0800 Subject: [PATCH] ci: add opt-in PR source evidence advisory --- .../workflows/pr-source-evidence-advisory.yml | 38 ++++ docs/GITHUB_PR_SOURCE_EVIDENCE_ADVISORY.md | 40 ++++ scripts/github_pr_source_evidence_advisory.py | 210 ++++++++++++++++++ ...test_github_pr_source_evidence_advisory.py | 120 ++++++++++ 4 files changed, 408 insertions(+) create mode 100644 .github/workflows/pr-source-evidence-advisory.yml create mode 100644 docs/GITHUB_PR_SOURCE_EVIDENCE_ADVISORY.md create mode 100644 scripts/github_pr_source_evidence_advisory.py create mode 100644 tests/test_github_pr_source_evidence_advisory.py diff --git a/.github/workflows/pr-source-evidence-advisory.yml b/.github/workflows/pr-source-evidence-advisory.yml new file mode 100644 index 0000000..f69ad24 --- /dev/null +++ b/.github/workflows/pr-source-evidence-advisory.yml @@ -0,0 +1,38 @@ +name: PR Source Evidence (advisory) + +# No secrets, no GitHub App and no privileged pull_request_target event. +# An untrusted PR can alter the check itself: advisory only, never release-gate truth. +on: + pull_request: + types: [opened, edited, synchronize, reopened] + +permissions: + contents: read + +jobs: + evidence: + name: Source evidence (advisory, not a fact-check) + if: contains(github.event.pull_request.body, '') + runs-on: ubuntu-latest + timeout-minutes: 4 + steps: + - name: Checkout without persisted credentials + uses: actions/checkout@v4 + with: + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install existing x-reader + run: python -m pip install -e . + + - name: Evaluate opted-in public URL sources + shell: bash + run: | + set -euo pipefail + python -B scripts/github_pr_source_evidence_advisory.py \ + --event "$GITHUB_EVENT_PATH" \ + --summary "$GITHUB_STEP_SUMMARY" diff --git a/docs/GITHUB_PR_SOURCE_EVIDENCE_ADVISORY.md b/docs/GITHUB_PR_SOURCE_EVIDENCE_ADVISORY.md new file mode 100644 index 0000000..65e4028 --- /dev/null +++ b/docs/GITHUB_PR_SOURCE_EVIDENCE_ADVISORY.md @@ -0,0 +1,40 @@ +# GitHub PR Source Evidence (advisory) + +This optional, read-only source-coverage report uses the EXISTING GitHub +Actions in runesleo/x-reader, and the canonical x_reader.evidence.build_receipt +without modifying that engine or repository branch rules. + +## Opt in + +Add this exact block to a GitHub PR description: + + + - https://x.com/cursor_ai/status/2110075611633074295 + + +The pull_request event runs the job only when the opt-in marker is present. +The report appears in the GitHub Actions job summary, NOT as a PR comment. +No GitHub App, API key, secret, hosted service, queue, scheduler, or notification +provider is added. It is NOT a required check. + +Only the explicit links in the PR description (up to three unique HTTPS URLs) +are assessed. Changed Markdown files, other sources in the diff, and factual +correctness are NOT verified. A PASS is an evidence-coverage claim only. +Malformed markers, too many URLs, userinfo credentials and insecure schemes +are rejected before any source read. + +Security: the workflow uses pull_request (not pull_request_target), grants +contents:read only, and checkout does NOT persist GitHub credentials. +The existing x-reader subprocess uses a temporary HOME, with no account +session/cookie inheritance. Job output is bounded host labels and reason +codes; full URLs, query strings and fetched content are not echoed. +The PR author can change the check code in their own branch; therefore this +advisory MUST NOT be treated as trustworthy branch-protection evidence. + +## Verification + +Run: python -m unittest discover -s tests -p test_github_pr_source_evidence_advisory.py -v + +A real opt-in PR must be tested on GitHub-hosted Actions before calling this +activated, and its actual job summary must be inspected. The presence of a +GitHub Action check alone is not proof of accurate source reading. diff --git a/scripts/github_pr_source_evidence_advisory.py b/scripts/github_pr_source_evidence_advisory.py new file mode 100644 index 0000000..193df00 --- /dev/null +++ b/scripts/github_pr_source_evidence_advisory.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +"""Opt-in GitHub PR source evidence report; read-only, advisory, never a fact-check.""" +from __future__ import annotations +import argparse +from collections import Counter +import hashlib +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +from typing import Callable +from urllib.parse import urlsplit + +from x_reader.evidence import build_receipt, failure_receipt +# URL discovery only. Canonical x-reader handles fetching, SSRF validation +# and evidence classification, including coverage of linked media. +_REFS = re.compile(r"https?://[^\s<>\[\]\"'\x60]+", re.I) +_FENCES = re.compile(r"\x60{3}[\s\S]*?\x60{3}") + + +def extract_pr_body_urls(markdown: str) -> list[str]: + if not isinstance(markdown, str) or len(markdown.encode("utf-8")) > MAX_BODY_BYTES: + raise ValueError("SOURCE_SECTION_TOO_LARGE") + refs = [] + for match in _REFS.findall(_FENCES.sub("", markdown)): + url = match.rstrip(".,;:!?)]}") + if len(url.encode("utf-8")) > 2048: + raise ValueError("SOURCE_URL_TOO_LARGE") + try: + parts = urlsplit(url) + if (parts.scheme not in ("http", "https") or not parts.hostname + or parts.username or parts.password): + raise ValueError("INVALID_REFERENCE_URL") + if parts.port is not None and not 1 <= parts.port <= 65535: + raise ValueError("INVALID_REFERENCE_PORT") + except ValueError: + raise ValueError("INVALID_REFERENCE_URL") from None + if url not in refs: + refs.append(url) + if len(refs) > MAX_SOURCES: + raise ValueError("SOURCE_BUDGET_EXCEEDED") + return refs + +OPEN = "" +CLOSE = "" +MAX_SOURCES = 3 +MAX_BODY_BYTES = 32768 +MAX_OUTPUT_BYTES = 6 * 1024 * 1024 +SAFE_REASON = re.compile(r"[A-Za-z0-9_-]{1,85}\Z") +HEAD_SHA = re.compile(r"(?:[0-9a-f]{40}|[0-9a-f]{64})\Z") +ROOT = Path(__file__).resolve().parents[1] + + +def source_urls(body: str | None) -> list[str] | None: + if body is None: + return None + if not isinstance(body, str) or len(body.encode()) > MAX_BODY_BYTES: + raise ValueError("PR_BODY_OVER_BUDGET") + if OPEN not in body and CLOSE not in body: + return None + if body.count(OPEN) != 1 or body.count(CLOSE) != 1: + raise ValueError("SOURCE_MARKERS_NOT_UNIQUE") + start, end = body.index(OPEN) + len(OPEN), body.index(CLOSE) + if end <= start: + raise ValueError("SOURCE_MARKERS_OUT_OF_ORDER") + urls = extract_pr_body_urls(body[start:end]) + if not urls: + raise ValueError("EMPTY_SOURCE_SECTION") + if len(urls) > MAX_SOURCES: + raise ValueError("SOURCE_BUDGET_EXCEEDED") + if any(urlsplit(url).scheme != "https" for url in urls): + raise ValueError("HTTPS_SOURCES_ONLY") + return urls + + +def read_source(url: str) -> dict: + """Invoke the existing x-reader once, with no inherited credentials or user HOME.""" + with tempfile.TemporaryDirectory(prefix="xreader-gh-pr-") as home: + env = { + "HOME": home, "PATH": os.environ.get("PATH", "/usr/bin:/bin"), + "LANG": "C.UTF-8", "PYTHONIOENCODING": "utf-8", + "PYTHONDONTWRITEBYTECODE": "1", + "X_READER_ALLOW_EXTERNAL_SESSION_COOKIES": "0", + } + try: + proc = subprocess.run( + [sys.executable, "-B", "-m", "x_reader.cli", url, "--json"], + cwd=ROOT, env=env, capture_output=True, text=True, + timeout=40, check=False, + ) + except subprocess.TimeoutExpired: + return failure_receipt(url, "reader_timeout", "Source read exceeded time budget") + except OSError: + return failure_receipt(url, "reader_unavailable", "Reader process unavailable") + if proc.returncode or len(proc.stdout.encode()) > MAX_OUTPUT_BYTES: + return failure_receipt(url, "reader_failed", "Reader failed or exceeded output budget") + try: + payload = json.loads(proc.stdout) + if not isinstance(payload, dict) or not payload.get("source_type"): + raise ValueError("INVALID_SOURCE") + result = build_receipt(payload, requested_url=url) + if result.get("status") not in {"PASS", "PARTIAL", "UNKNOWN", "FAIL"}: + raise ValueError("INVALID_RECEIPT") + return result + except (TypeError, ValueError, KeyError): + return failure_receipt(url, "reader_unverified", "Source receipt could not be verified") + + +def _source_tag(url: str, index: int) -> str: + host = (urlsplit(url).hostname or "").lower() + if not re.fullmatch(r"[a-z0-9.:-]{1,85}", host): + host = "public-source" + return f"{index}. {host} [{hashlib.sha256(url.encode()).hexdigest()[:8]}]" + + +def evaluate_pr(event: dict, read: Callable[[str], dict] = read_source) -> dict: + pr = event.get("pull_request") if isinstance(event, dict) else None + if not isinstance(pr, dict): + raise ValueError("GITHUB_PR_EVENT_REQUIRED") + head = pr.get("head") + sha = head.get("sha") if isinstance(head, dict) else None + if not isinstance(sha, str) or not HEAD_SHA.fullmatch(sha): + raise ValueError("INVALID_PR_HEAD_SHA") + if type(pr.get("number")) is not int or pr["number"] < 1: + raise ValueError("INVALID_PR_NUMBER") + if event.get("action") not in ("opened", "edited", "synchronize", "reopened"): + raise ValueError("UNSUPPORTED_PR_EVENT_ACTION") + rows = [] + try: + urls = source_urls(pr.get("body")) + if urls is None: + status, reason, urls = "NOT_REQUESTED", "opt_in_marker_absent", [] + else: + status, reason = "UNKNOWN", "not_evaluated" + except ValueError as exc: + status, reason, urls = "UNKNOWN", str(exc), [] + for i, url in enumerate(urls, 1): + try: + receipt = read(url) + except Exception: + receipt = failure_receipt(url, "reader_exception", "Reader raised an error") + state = str(receipt.get("status") or "UNKNOWN") + if state not in ("PASS", "PARTIAL", "FAIL", "UNKNOWN"): + state = "UNKNOWN" + reason_code = str(receipt.get("reason_code") or "reason_unknown") + if not SAFE_REASON.fullmatch(reason_code): + reason_code = "reason_untrusted" + rows.append({"source": _source_tag(url, i), "status": state, "reason": reason_code}) + if rows: + counts = Counter(row["status"] for row in rows) + status = ("FAIL" if counts["FAIL"] else "PARTIAL" if counts["PARTIAL"] + else "UNKNOWN" if counts["UNKNOWN"] else "PASS") + reason = "source_evidence_assessed" + return { + "head_sha": sha, "pr_number": pr["number"], + "status": status, + "reason": reason if SAFE_REASON.fullmatch(reason) else "source_input_invalid", + "sources": rows, "source_count": len(urls), + "advisory": True, "changed_files_checked": False, + "factual_truth_verified": False, "github_write_performed": False, + } + + +def summary_markdown(report: dict) -> str: + lines = [ + "## x-reader Source Evidence (advisory)", + f"**Coverage: {report['status']}** · PR #{report['pr_number']} · head {report['head_sha'][:12]}", + "", + "**Scope:** Opted-in PR description sources only. No changed files were checked.", + "**Not a fact-check.** This job does not block a merge or change branch protection.", + "", + ] + if not report["sources"]: + lines.append("No coverage claim: " + report["reason"] + ".") + else: + lines += ["| Source | Status | Reason |", "| --- | --- | --- |"] + for item in report["sources"]: + lines.append("| " + item["source"] + " | " + item["status"] + " | " + + item["reason"] + " |") + lines += ["", "Add an opt-in block to the PR description:", "", OPEN, + "- https://x.com/example/status/123", CLOSE, "", + "PASS only means coverage of the explicitly supplied source layers.", ""] + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--event", type=Path, required=True) + parser.add_argument("--summary", type=Path, required=True) + args = parser.parse_args() + try: + event = json.loads(args.event.read_text(encoding="utf-8")) + report = evaluate_pr(event) + except (OSError, ValueError, TypeError): + print("SOURCE_EVIDENCE=INFRA_ERROR", file=sys.stderr) + return 2 + args.summary.parent.mkdir(parents=True, exist_ok=True) + with args.summary.open("a", encoding="utf-8") as file: + file.write(summary_markdown(report)) + print("SOURCE_EVIDENCE=" + report["status"] + + " HEAD=" + report["head_sha"][:12] + + " COUNT=" + str(report["source_count"]) + " ADVISORY=true") + return 0 # Evidence gaps do not silently become mandatory branch rules. + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_github_pr_source_evidence_advisory.py b/tests/test_github_pr_source_evidence_advisory.py new file mode 100644 index 0000000..55a8f49 --- /dev/null +++ b/tests/test_github_pr_source_evidence_advisory.py @@ -0,0 +1,120 @@ +"""Advisory GitHub PR source-coverage contract, with no platform calls.""" +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +from github_pr_source_evidence_advisory import ( + OPEN, CLOSE, source_urls, evaluate_pr, summary_markdown +) +from x_reader.evidence import build_receipt + +URL = "https://x.com/example/status/123?token=not-to-echo" +BLOCK = OPEN + "\n- " + URL + "\n" + CLOSE + + +def event(body, sha="a"*40, action="opened"): + return {"action": action, "pull_request": { + "number": 42, "head": {"sha": sha}, "body": body + }} + + +def source(url, media="none", method="oembed"): + return {"source_type": "twitter", "url": url, "content": "Original X text", + "extra": {"fetch_method": method, "post_id_verified": True, + "media_status": media, "structure_status": "verified", + "quote_status": "none", "article_status": "none", + "poll_status": "none", "thread_marker": False}} + + +class GitHubAdvisoryTests(unittest.TestCase): + def test_unmarked_pr_skips_all_reads(self): + out = evaluate_pr(event("A PR body"), lambda _: self.fail("must not fetch")) + self.assertEqual(out["status"], "NOT_REQUESTED") + self.assertFalse(out["github_write_performed"]) + + def test_pass_scope_not_factual_truth(self): + got = evaluate_pr(event(BLOCK), + lambda u: build_receipt(source(u), requested_url=u)) + self.assertEqual(got["status"], "PASS") + self.assertFalse(got["factual_truth_verified"]) + self.assertFalse(got["changed_files_checked"]) + summary = summary_markdown(got) + self.assertIn("Not a fact-check", summary) + self.assertIn("No changed files were checked", summary) + self.assertNotIn("not-to-echo", summary) + self.assertNotIn("Original X text", summary) + + def test_media_unread_stays_partial(self): + out = evaluate_pr(event(BLOCK), + lambda u: build_receipt(source(u, "present"), requested_url=u)) + self.assertEqual(out["status"], "PARTIAL") + self.assertIn("attached_media_unread", summary_markdown(out)) + + def test_unknown_method_cannot_pass(self): + out = evaluate_pr(event(BLOCK), + lambda u: build_receipt(source(u, method="browser_page"), requested_url=u)) + self.assertEqual(out["status"], "UNKNOWN") + + def test_missing_or_malformed_markers_do_not_fetch(self): + for body in (OPEN + "\n" + URL, CLOSE + "\n" + URL, + CLOSE + "\n" + OPEN + URL, OPEN + OPEN + URL + CLOSE, + OPEN + CLOSE): + out = evaluate_pr(event(body), lambda _: self.fail("must not fetch")) + self.assertEqual(out["status"], "UNKNOWN") + self.assertEqual(out["source_count"], 0) + + def test_over_budget_and_unsafe_urls_do_not_fetch(self): + for url in ("http://example.org", "https://user:pass@example.org/private"): + out = evaluate_pr(event(OPEN + url + CLOSE), lambda _: self.fail("fetch")) + self.assertEqual(out["status"], "UNKNOWN") + too_many = "\n".join("https://example.org/" + str(i) for i in range(4)) + self.assertEqual(evaluate_pr(event(OPEN + too_many + CLOSE), + lambda _: self.fail("fetch"))["reason"], + "SOURCE_BUDGET_EXCEEDED") + + def test_deduplicate_and_ignore_fenced_markdown(self): + markdown = (OPEN + "\n" + URL + " " + URL + "\n" + + "https://example.org/safe\n" + CLOSE) + self.assertEqual(len(source_urls(markdown)), 2) + fenced = OPEN + "\n" + chr(96)*3 + "text\nhttps://evil.test/1\n" + chr(96)*3 + "\n" + URL + CLOSE + self.assertEqual(source_urls(fenced), [URL]) + + def test_reader_exception_and_unknown_status_not_success(self): + def exception(_): + raise RuntimeError("internal") + self.assertEqual(evaluate_pr(event(BLOCK), exception)["status"], "FAIL") + self.assertEqual(evaluate_pr(event(BLOCK), + lambda _: {"status":"INVENTED","reason_code":"source_complete"})["status"], "UNKNOWN") + + def test_event_validation_and_head_binding(self): + with self.assertRaisesRegex(ValueError, "INVALID_PR_HEAD_SHA"): + evaluate_pr(event(BLOCK, sha="bad")) + with self.assertRaisesRegex(ValueError, "UNSUPPORTED_PR_EVENT_ACTION"): + evaluate_pr(event(BLOCK, action="closed")) + reader = lambda _: {"status":"PARTIAL","reason_code":"attached_media_unread"} + a, b = evaluate_pr(event(BLOCK), reader), evaluate_pr(event(BLOCK), reader) + self.assertEqual(a,b) + self.assertNotEqual(a["head_sha"], evaluate_pr(event(BLOCK, sha="b"*40), reader)["head_sha"]) + + def test_cli_job_summary_nonpublishing(self): + with tempfile.TemporaryDirectory() as temp: + ev = Path(temp)/"event.json" + summ = Path(temp)/"job-summary.md" + ev.write_text(json.dumps(event("not opted in")), encoding="utf-8") + env = dict(os.environ, PYTHONPATH=str(ROOT), PYTHONDONTWRITEBYTECODE="1") + p = subprocess.run([sys.executable, "-B", + str(ROOT/"scripts/github_pr_source_evidence_advisory.py"), + "--event", str(ev), "--summary", str(summ)], + cwd=ROOT, capture_output=True, text=True, timeout=15, env=env) + self.assertEqual(p.returncode, 0, p.stderr) + self.assertIn("ADVISORY=true", p.stdout) + self.assertIn("NOT_REQUESTED", summ.read_text(encoding="utf-8")) + + +if __name__ == "__main__": + unittest.main()