Skip to content

Harden source validation in live bounty checker scripts (#1145) #1172

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yanyishuai wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Harden source validation in live bounty checker scripts (#1145) #1172

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 17 additions & 50 deletions scripts/check_bounty_issue_states.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,18 +4,19 @@
import json
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

if __package__ in {None, ""}:
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from scripts.api_host_args import public_api_host
from scripts.gh_cli import DEFAULT_GH_TIMEOUT_SECONDS as GH_TIMEOUT_SECONDS
from scripts.gh_cli import run_gh_json as _run_gh_json
from scripts.public_api_json import load_public_bounty_list
from scripts.source_args import validate_source_args

DEFAULT_API_HOST = "https://api.mrwk.online"
GH_TIMEOUT_SECONDS = 30


def _int_or_none(value: Any) -> int | None:
Expand Down Expand Up @@ -119,31 +120,8 @@ def format_text_report(report: dict[str, Any]) -> str:
return "\n".join(lines)


def _run_gh_json(args: list[str]) -> Any:
command = " ".join(args)
try:
completed = subprocess.run(
args,
check=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=GH_TIMEOUT_SECONDS,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") from exc
except subprocess.CalledProcessError as exc:
raise RuntimeError(
"gh command failed "
f"(exit {exc.returncode}): {command}\n"
f"stdout:\n{exc.stdout or exc.output or ''}\n"
f"stderr:\n{exc.stderr or ''}"
) from exc
return json.loads(completed.stdout)


def _run_gh(args: list[str]) -> None:
"""Run gh for maintainer --fix paths (may mutate GitHub state)."""
command = " ".join(args)
try:
subprocess.run(
Expand All @@ -166,23 +144,6 @@ def _run_gh(args: list[str]) -> None:
) from exc


def _fetch_json(url: str) -> Any:
request = urllib.request.Request(url, headers={"Accept": "application/json"})
try:
with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response:
return json.loads(response.read().decode("utf-8"))
except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc:
raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc


def _load_public_bounties(api_host: str) -> list[dict[str, Any]]:
url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200"
data = _fetch_json(url)
if not isinstance(data, list):
raise RuntimeError(f"expected a JSON list from {url}")
return [item for item in data if isinstance(item, dict)]


def _load_issue(repo: str, issue_number: int) -> dict[str, Any]:
return _run_gh_json(
[
Expand All @@ -199,7 +160,7 @@ def _load_issue(repo: str, issue_number: int) -> dict[str, Any]:


def load_live_data(repo: str, api_host: str) -> dict[str, Any]:
bounties = _load_public_bounties(api_host)
bounties = load_public_bounty_list(api_host, query="status=open&limit=200")
issue_numbers = sorted(
number for number in (_issue_number(bounty) for bounty in bounties) if number is not None
)
Expand Down Expand Up @@ -240,13 +201,19 @@ def main(argv: list[str] | None = None) -> int:
)
args = parser.parse_args(argv)

data = _load_input(args.input) if args.input else load_live_data(args.repo, args.api_host)
input_arg, repo_arg = validate_source_args(
parser,
input_value=args.input,
repo_value=args.repo,
fix=args.fix,
)
data = (
_load_input(input_arg) if input_arg is not None else load_live_data(repo_arg, args.api_host)
)
report = analyze_issue_states(data)
if args.fix:
if args.input:
raise SystemExit("--fix requires --repo, not --input")
reopen_violations(args.repo, report["violations"])
data = load_live_data(args.repo, args.api_host)
reopen_violations(repo_arg, report["violations"])
data = load_live_data(repo_arg, args.api_host)
report = analyze_issue_states(data)
if args.format == "json":
print(json.dumps(report, indent=2, sort_keys=True))
Expand Down
62 changes: 13 additions & 49 deletions scripts/check_live_bounty_closing_refs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,8 @@

import argparse
import json
import subprocess
import subprocess # noqa: F401 - monkeypatched in tests
import sys
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

Expand All @@ -14,9 +12,11 @@

from scripts.api_host_args import public_api_host
from scripts.bounty_refs import GITHUB_CLOSING_ISSUE_RE
from scripts.gh_cli import run_gh_json as _run_gh_json
from scripts.public_api_json import load_public_bounty_list
from scripts.source_args import validate_source_args

DEFAULT_API_HOST = "https://api.mrwk.online"
GH_TIMEOUT_SECONDS = 30
GH_PR_SAFETY_CAP = 200
MAX_BOUNTY_REF = 2**63 - 1

Expand Down Expand Up @@ -122,47 +122,6 @@ def format_text_report(report: dict[str, Any]) -> str:
return "\n".join(lines)


def _run_gh_json(args: list[str]) -> Any:
command = " ".join(args)
try:
completed = subprocess.run(
args,
check=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=GH_TIMEOUT_SECONDS,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"gh command timed out after {GH_TIMEOUT_SECONDS}s: {command}") from exc
except subprocess.CalledProcessError as exc:
raise RuntimeError(
"gh command failed "
f"(exit {exc.returncode}): {command}\n"
f"stdout:\n{exc.stdout or exc.output or ''}\n"
f"stderr:\n{exc.stderr or ''}"
) from exc
return json.loads(completed.stdout)


def _fetch_json(url: str) -> Any:
request = urllib.request.Request(url, headers={"Accept": "application/json"})
try:
with urllib.request.urlopen(request, timeout=GH_TIMEOUT_SECONDS) as response:
return json.loads(response.read().decode("utf-8"))
except (TimeoutError, urllib.error.URLError, json.JSONDecodeError) as exc:
raise RuntimeError(f"failed to fetch JSON from {url}: {exc}") from exc


def _load_public_bounties(api_host: str) -> list[dict[str, Any]]:
url = f"{api_host.rstrip('/')}/api/v1/bounties?status=open&limit=200"
data = _fetch_json(url)
if not isinstance(data, list):
raise RuntimeError(f"expected a JSON list from {url}")
return [item for item in data if isinstance(item, dict)]


def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[dict[str, Any]]:
if pr_numbers:
return [
Expand Down Expand Up @@ -205,7 +164,7 @@ def _load_pull_requests(repo: str, state: str, pr_numbers: list[int]) -> list[di

def load_live_data(repo: str, api_host: str, state: str, pr_numbers: list[int]) -> dict[str, Any]:
return {
"bounties": _load_public_bounties(api_host),
"bounties": load_public_bounty_list(api_host, query="status=open&limit=200"),
"pull_requests": _load_pull_requests(repo, state, pr_numbers),
}

Expand Down Expand Up @@ -235,10 +194,15 @@ def main(argv: list[str] | None = None) -> int:
parser.add_argument("--fail-on-issues", action="store_true")
args = parser.parse_args(argv)

input_arg, repo_arg = validate_source_args(
parser,
input_value=args.input,
repo_value=args.repo,
)
data = (
_load_input(args.input)
if args.input
else load_live_data(args.repo, args.api_host, args.state, args.pr)
_load_input(input_arg)
if input_arg is not None
else load_live_data(repo_arg, args.api_host, args.state, args.pr)
)
report = analyze_closing_refs(data)
if args.format == "json":
Expand Down
95 changes: 95 additions & 0 deletions scripts/gh_cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
from __future__ import annotations

import json
import subprocess
from typing import Any

DEFAULT_GH_TIMEOUT_SECONDS = 30

_NON_READ_ONLY_GH_API = frozenset({"POST", "PUT", "PATCH", "DELETE"})
_NON_READ_ONLY_GH_SUBCOMMANDS = frozenset(
{"comment", "edit", "close", "reopen", "merge", "review", "delete"}
)


def _command_text(args: list[str]) -> str:
return " ".join(args)


def assert_read_only_gh_command(args: list[str]) -> None:
"""Reject gh invocations that could mutate GitHub state."""
if args[:2] == ["gh", "api"]:
for flag in ("--method", "-X"):
if flag not in args:
continue
index = args.index(flag)
if index + 1 >= len(args):
continue
if args[index + 1].upper() in _NON_READ_ONLY_GH_API:
raise RuntimeError(f"refusing non-read-only gh api command: {_command_text(args)}")
if any(arg in {"issue", "pr"} for arg in args) and any(
arg in _NON_READ_ONLY_GH_SUBCOMMANDS for arg in args
):
raise RuntimeError(f"refusing non-read-only gh command: {_command_text(args)}")


def run_gh(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> str:
"""Run a read-only gh command and return stdout text."""
assert_read_only_gh_command(args)
command = _command_text(args)
try:
completed = subprocess.run(
args,
check=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=timeout_seconds,
)
except subprocess.TimeoutExpired as exc:
raise RuntimeError(f"gh command timed out after {timeout_seconds}s: {command}") from exc
except FileNotFoundError as exc:
raise RuntimeError(
"GitHub CLI executable 'gh' was not found; install gh and ensure it is on PATH "
"before using live --repo mode"
) from exc
except subprocess.CalledProcessError as exc:
raise RuntimeError(
"gh command failed "
f"(exit {exc.returncode}): {command}\n"
f"stdout:\n{exc.stdout or exc.output or ''}\n"
f"stderr:\n{exc.stderr or ''}"
) from exc
return completed.stdout


def run_gh_json(args: list[str], *, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS) -> Any:
"""Run a read-only gh command and parse JSON stdout."""
return json.loads(run_gh(args, timeout_seconds=timeout_seconds))


def ensure_json_list(data: Any, *, label: str) -> list[Any]:
if not isinstance(data, list):
raise RuntimeError(f"{label} returned non-list JSON ({type(data).__name__})")
return data


def ensure_json_object(data: Any, *, label: str) -> dict[str, Any]:
if not isinstance(data, dict):
raise RuntimeError(f"{label} returned non-object JSON ({type(data).__name__})")
return data


def run_gh_json_list(
args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS
) -> list[Any]:
context = label or _command_text(args)
return ensure_json_list(run_gh_json(args, timeout_seconds=timeout_seconds), label=context)


def run_gh_json_object(
args: list[str], *, label: str | None = None, timeout_seconds: int = DEFAULT_GH_TIMEOUT_SECONDS
) -> dict[str, Any]:
context = label or _command_text(args)
return ensure_json_object(run_gh_json(args, timeout_seconds=timeout_seconds), label=context)
Loading
Loading