scan_rust.py

import subprocess
import json
from pathlib import Path


def nix_eval(expr: str, parse: bool = False):
    result = subprocess.run(
        ["nix-instantiate", "--eval-only", "--expr", expr],
        capture_output=True,
        text=True,
    )
    output = result.stdout.strip()
    if output:
        return json.loads(output) if parse else output
    return None


def nix_build(expr: str):
    result = subprocess.run(
        ["nix-build", "--no-out-link", "--expr", expr],
        capture_output=True,
        text=True,
    )
    return result.stdout.strip()


def rg_check(path: Path, pattern: str) -> bool:
    result = subprocess.run(
        ["rg", "-q", "--max-count", "1", pattern, str(path)],
        check=False,
        capture_output=True,
    )
    return result.returncode == 0


by_name = set()
python_modules = set()
other = set()

with open("rust-packages.txt") as f:
    for line in f.readlines():
        comp = line.split("/")
        if len(comp) > 3:
            if line.startswith("pkgs/by-name/"):
                by_name.add(comp[3])
            elif line.startswith("pkgs/development/python-modules/"):
                python_modules.add("python3Packages." + comp[3])
            else:
                other.add(line.strip())

PATTERN = 'name = "async-tar|tokio-tar|astral-tokio-tar"'
nixpkgs_config = "config.allowUnsupportedSystem = true;" + "config.allowUnfree = true;"
nixpkgs = "(import ../nixpkgs {" + nixpkgs_config + "})"

packages = sorted(list(by_name | python_modules))
affected = open("affected.txt", "w")
no_lockfile = open("no-lockfile.txt", "w")
for pkg in packages:
    try:
        lockfile = nix_eval(f"{nixpkgs}.{pkg}.cargoDeps.lockFile")
        if lockfile:
            lockfile = Path(lockfile)
        else:
            sourceRoot = nix_eval(f"{nixpkgs}.{pkg}.sourceRoot", parse=True)
            cargo_root = nix_eval(f"{nixpkgs}.{pkg}.cargoRoot", parse=True)
            src = nix_build(f"{nixpkgs}.{pkg}.src")
            lock_dir = Path(src)
            if sourceRoot:
                lock_dir = lock_dir / Path(*Path(sourceRoot).parts[1:])
            if cargo_root:
                lock_dir = lock_dir / cargo_root
            lockfile = lock_dir / "Cargo.lock"
        print(pkg, lockfile)
        if lockfile.is_file():
            if rg_check(lockfile, PATTERN):
                print("-> affected")
                affected.write(pkg + " " + str(lockfile) + "\n")
                affected.flush()
        else:
            print("-> lockfile does not exist")
            no_lockfile.write(pkg + "\n")
            no_lockfile.flush()
    except Exception:
        print("Error in", pkg)
        no_lockfile.write(pkg + "\n")
        no_lockfile.flush()
affected.close()
no_lockfile.close()