Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
164 changes: 45 additions & 119 deletions .github/scripts/cargo-release-quarantine.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Enforce a minimum crates.io release age on Cargo update proposals."""
"""Fail if a Cargo.lock change introduces crates.io releases younger than the quarantine."""

from __future__ import annotations

Expand All @@ -11,156 +11,82 @@
import tomllib
import urllib.parse
import urllib.request
from pathlib import Path

DEP_TABLES = {"dependencies", "dev-dependencies", "build-dependencies"}
CRATES_IO_SOURCE = "registry+https://github.com/rust-lang/crates.io-index"
_cache: dict[str, dict] = {}
CRATES_IO_MARKERS = ("crates.io-index", "index.crates.io")
_cache: dict[str, dict[str, dict]] = {}


def git_show(path: str) -> bytes:
return subprocess.check_output(["git", "show", f"HEAD:{path}"])
def git_show(ref: str, path: str) -> bytes:
return subprocess.check_output(["git", "show", f"{ref}:{path}"])


def load_lock(raw: bytes) -> list[dict]:
return tomllib.loads(raw.decode()).get("package", [])


def package_key(pkg: dict) -> tuple[str, str, str | None]:
def key(pkg: dict) -> tuple[str, str, str | None]:
return pkg["name"], pkg["version"], pkg.get("source")


def crate_versions(name: str) -> dict:
def is_crates_io(pkg: dict) -> bool:
source = pkg.get("source") or ""
return source.startswith("registry+") and any(marker in source for marker in CRATES_IO_MARKERS)


def crate_versions(name: str) -> dict[str, dict]:
if name in _cache:
return _cache[name]
url = "https://crates.io/api/v1/crates/" + urllib.parse.quote(name, safe="")
req = urllib.request.Request(
url,
headers={
"Accept": "application/json",
"User-Agent": "codeandsolder-maintenance-detector/1.0",
},
"https://crates.io/api/v1/crates/" + urllib.parse.quote(name, safe=""),
headers={"Accept": "application/json", "User-Agent": "codeandsolder-renovate-quarantine/1.0"},
)
with urllib.request.urlopen(req, timeout=30) as response:
payload = json.load(response)
versions = {v["num"]: v for v in payload.get("versions", [])}
versions = {item["num"]: item for item in payload.get("versions", [])}
_cache[name] = versions
return versions


def created_at(pkg: dict) -> dt.datetime | None:
if pkg.get("source") != CRATES_IO_SOURCE:
return None
def publication_time(pkg: dict) -> dt.datetime | None:
info = crate_versions(pkg["name"]).get(pkg["version"])
if not info:
return None
value = info.get("created_at")
if not value:
return None
return dt.datetime.fromisoformat(value.replace("Z", "+00:00"))


def ineligible(pkg: dict, cutoff: dt.datetime) -> str | None:
when = created_at(pkg)
if when is None:
return (
f"{pkg['name']} {pkg['version']}: no authoritative crates.io "
f"publication timestamp for source {pkg.get('source')!r}"
)
if when > cutoff:
return (
f"{pkg['name']} {pkg['version']}: published {when.isoformat()} "
f"(cutoff {cutoff.isoformat()})"
)
return None


def declaration_name(alias: str, spec) -> str:
if isinstance(spec, dict):
return spec.get("package", alias)
return alias


def collect_dependencies(obj, prefix=()) -> dict[tuple[str, str], object]:
out = {}
if not isinstance(obj, dict):
return out
for key, value in obj.items():
if key in DEP_TABLES and isinstance(value, dict):
table = ".".join((*prefix, key))
for alias, spec in value.items():
out[(table, alias)] = spec
if isinstance(value, dict):
out.update(collect_dependencies(value, (*prefix, key)))
return out


def changed_direct_crates() -> tuple[set[str], set[str]]:
files = subprocess.check_output(
["git", "ls-files", "*Cargo.toml"], text=True
).splitlines()
names: set[str] = set()
unknown_source: set[str] = set()
for path in files:
with open(path, "rb") as f:
after = collect_dependencies(tomllib.load(f))
before = collect_dependencies(tomllib.loads(git_show(path).decode()))
for key in set(before) | set(after):
old = before.get(key)
new = after.get(key)
if old == new:
continue
alias = key[1]
spec = new if new is not None else old
name = declaration_name(alias, spec)
names.add(name)
if isinstance(spec, dict) and any(k in spec for k in ("git", "path")):
unknown_source.add(name)
return names, unknown_source


def changed_lock_packages() -> list[dict]:
before = {package_key(p) for p in load_lock(git_show("Cargo.lock"))}
with open("Cargo.lock", "rb") as f:
after = load_lock(f.read())
return [p for p in after if package_key(p) not in before and p.get("source")]
value = info.get("created_at") if info else None
return dt.datetime.fromisoformat(value.replace("Z", "+00:00")) if value else None


def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("mode", choices=("manifest", "lockfile"))
parser.add_argument("--base-ref", required=True)
parser.add_argument("--hours", type=int, default=72)
args = parser.parse_args()

try:
before = {key(pkg) for pkg in load_lock(git_show(args.base_ref, "Cargo.lock"))}
except subprocess.CalledProcessError:
before = set()

after = load_lock(Path("Cargo.lock").read_bytes())
changed = [pkg for pkg in after if key(pkg) not in before and is_crates_io(pkg)]
cutoff = dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=args.hours)
changed = changed_lock_packages()

if args.mode == "lockfile":
bad = []
for pkg in changed:
reason = ineligible(pkg, cutoff)
if reason:
bad.append(reason)
for reason in bad:
print(f"quarantined: {reason}", file=sys.stderr)
return 1 if bad else 0

direct, unknown_source = changed_direct_crates()
blocked = set(unknown_source)
reasons = []

blocked: list[str] = []
for pkg in changed:
if pkg["name"] not in direct:
continue
reason = ineligible(pkg, cutoff)
if reason:
blocked.add(pkg["name"])
reasons.append(reason)

for name in sorted(unknown_source):
reasons.append(f"{name}: changed git/path dependency has no release timestamp")
for reason in reasons:
print(f"quarantined: {reason}", file=sys.stderr)
for name in sorted(blocked):
print(name)
when = publication_time(pkg)
if when is None:
blocked.append(f"{pkg['name']} {pkg['version']}: missing authoritative crates.io publication time")
elif when > cutoff:
blocked.append(
f"{pkg['name']} {pkg['version']}: published {when.isoformat()}, "
f"younger than {args.hours}h"
)

if blocked:
print("Cargo release-age quarantine is still active:", file=sys.stderr)
for item in blocked:
print(f"- {item}", file=sys.stderr)
return 1

print(f"Cargo release-age quarantine passed for {len(changed)} newly introduced crates.io package versions.")
return 0


Expand Down
22 changes: 22 additions & 0 deletions .github/workflows/dependency-quarantine.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
name: Cargo release-age quarantine

on:
pull_request:
paths:
- "**/Cargo.lock"
- "Cargo.lock"

permissions:
contents: read

jobs:
quarantine:
runs-on: ubuntu-24.04
timeout-minutes: 10
steps:
- uses: actions/checkout@v7.0.1
with:
fetch-depth: 0

- name: Enforce 72-hour crates.io quarantine
run: python3 .github/scripts/cargo-release-quarantine.py --base-ref "${{ github.event.pull_request.base.sha }}" --hours 72
Loading
Loading