#!/usr/bin/env python3 """Match every Steam runtime ELF binary to the exact upstream package it came from. For each runtime staged under client/runtime/, reads its package manifests (manifest.dpkg / manifest.txt / pressure-vessel metadata/packages.txt), downloads each listed .deb (Valve's repo.steampowered.com first, then Debian, Debian security, archive.debian.org, Ubuntu, snapshot.debian.org) plus the matching pressure-vessel release tarballs, unpacks them in a scratch dir, hashes every file, and looks up each runtime binary's SHA-256. Downloaded archives are untrusted data: they are only unpacked (bsdtar, which refuses absolute and ../ paths) and hashed, never executed. Writes decomp/runtime/PROVENANCE.tsv (one row per runtime binary, including the duplicates listed in decomp/runtime/DUPLICATES.tsv) and decomp/runtime/PROVENANCE-missing.tsv (packages that couldn't be downloaded). Usage: python3 -I tools/verify_runtime.py (re-runs reuse client/debs/ cache) """ import concurrent.futures as cf import hashlib import json import os import re import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parent.parent CACHE = ROOT / "client" / "debs" OUT = ROOT / "decomp" / "runtime" H = Path.home() / ".local/share/Steam" L = Path("/1tb/SteamLibrary/steamapps/common") # runtime name (as staged under client/runtime/) -> (install dir, Valve repo, fallback distro) RUNTIMES = { "steam-runtime": (H / "ubuntu12_32/steam-runtime", "steamrt1", "ubuntu"), "steam-runtime.old": (H / "ubuntu12_32/steam-runtime.old", "steamrt1", "ubuntu"), "SteamLinuxRuntime": (L / "SteamLinuxRuntime", "steamrt1", "ubuntu"), "steamrt64-pv-runtime": (H / "steamrt64/pv-runtime", "steamrt3c", "debian"), "SteamLinuxRuntime_soldier": (L / "SteamLinuxRuntime_soldier", "steamrt2", "debian"), "SteamLinuxRuntime_sniper": (L / "SteamLinuxRuntime_sniper", "steamrt3", "debian"), "SteamLinuxRuntime_4": (L / "SteamLinuxRuntime_4", "steamrt4", "debian"), } VALVE = "https://repo.steampowered.com/{repo}/apt/pool/{comp}/{prefix}/{src}/{file}" FALLBACK = { "debian": [ "https://deb.debian.org/debian/pool/{comp}/{prefix}/{src}/{file}", "https://security.debian.org/debian-security/pool/updates/{comp}/{prefix}/{src}/{file}", "https://archive.debian.org/debian/pool/{comp}/{prefix}/{src}/{file}", "https://archive.debian.org/debian-security/pool/updates/{comp}/{prefix}/{src}/{file}", ], "ubuntu": [ "https://old-releases.ubuntu.com/ubuntu/pool/{comp}/{prefix}/{src}/{file}", "https://archive.ubuntu.com/ubuntu/pool/{comp}/{prefix}/{src}/{file}", ], } PV = "https://repo.steampowered.com/pressure-vessel/snapshots/{ver}/{file}" PV_FILES = ["pressure-vessel-bin.tar.gz", "pressure-vessel-arm64.tar.gz", "pressure-vessel-arm64+amd64+i386.tar.gz"] # Whole-runtime image for a scout version, for packages since pruned from the apt pool. SCOUT_IMAGE = "https://repo.steampowered.com/steamrt-images-scout/snapshots/{ver}/SteamLinuxRuntime.tar.xz" UA = {"User-Agent": "steam-runtime-provenance/1 (personal research)"} def fetch(url, dest): """Download url to dest; True on success, False on 404/410, raises on other errors.""" tmp = dest.with_suffix(dest.suffix + ".part") req = urllib.request.Request(url, headers=UA) try: with urllib.request.urlopen(req, timeout=120) as r, open(tmp, "wb") as f: shutil.copyfileobj(r, f) except urllib.error.HTTPError as e: tmp.unlink(missing_ok=True) if e.code in (403, 404, 410): return False raise tmp.rename(dest) return True def parse_manifest(path): """Yield (package, arch or None, version, source) from a tab-separated dpkg manifest.""" for line in path.read_text(errors="replace").splitlines(): if not line or line.startswith("#"): continue cols = line.split("\t") if len(cols) < 2: continue pkg, _, arch = cols[0].partition(":") src = cols[2].strip() if len(cols) > 2 and cols[2].strip() else pkg src = src.split(" ")[0] # "src (ver)" form yield pkg, arch or None, cols[1], src def pool_prefix(src): return src[:4] if src.startswith("lib") and len(src) > 3 else src[0] def deb_candidates(repo, distro, pkg, arch, ver, src): noepoch = ver.split(":", 1)[1] if ":" in ver else ver archs = [arch] if arch else ["amd64", "all", "i386"] for a in archs: file = f"{pkg}_{noepoch}_{a}.deb" q = urllib.parse.quote(file) for comp in ("main", "contrib", "non-free", "non-free-firmware", "universe", "multiverse", "restricted"): fmt = dict(comp=comp, prefix=pool_prefix(src), src=src, file=q) if comp in ("main", "contrib", "non-free", "non-free-firmware"): yield file, VALVE.format(repo=repo, **fmt) for tmpl in FALLBACK[distro]: if distro == "ubuntu" and comp in ("contrib", "non-free", "non-free-firmware"): continue if distro == "debian" and comp in ("universe", "multiverse", "restricted"): continue yield file, tmpl.format(**fmt) def snapshot_lookup(pkg, ver, arch): """snapshot.debian.org machine-readable API: binary package version -> file URL.""" url = f"https://snapshot.debian.org/mr/binary/{urllib.parse.quote(pkg)}/{urllib.parse.quote(ver)}/binfiles?fileinfo=1" try: with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=60) as r: data = json.load(r) except (urllib.error.URLError, ValueError): return None for res in data.get("result", []): if arch and res.get("architecture") not in (arch, "all"): continue return "https://snapshot.debian.org/file/" + res["hash"] return None def download_pkg(job): repo, distro, pkg, arch, ver, src = job noepoch = ver.split(":", 1)[1] if ":" in ver else ver for a in ([arch] if arch else ["amd64", "all", "i386"]): cached = CACHE / f"{pkg}_{noepoch}_{a}.deb" if cached.exists(): return job, cached, json.loads(cached.with_suffix(".url").read_text()) if cached.with_suffix(".url").exists() else "" tried = set() for file, url in deb_candidates(repo, distro, pkg, arch, ver, src): if url in tried: continue tried.add(url) dest = CACHE / file try: if fetch(url, dest): dest.with_suffix(".url").write_text(json.dumps(url)) return job, dest, url except Exception: time.sleep(2) if distro == "debian": for a in ([arch] if arch else ["amd64", "all", "i386"]): url = snapshot_lookup(pkg, ver, a) time.sleep(1) # be gentle with snapshot.debian.org if url: dest = CACHE / f"{pkg}_{noepoch}_{a}.deb" try: if fetch(url, dest): dest.with_suffix(".url").write_text(json.dumps(url)) return job, dest, url except Exception: pass return job, None, "" def hash_archive(path, inner_deb_data=True): """Unpack a .deb (or tarball) in a scratch dir and return {sha256: member path} for its files.""" out = {} with tempfile.TemporaryDirectory(dir=CACHE) as tmp: tmp = Path(tmp) tar = path if inner_deb_data: subprocess.run(["bsdtar", "-xf", str(path), "-C", str(tmp), "data.tar*"], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) tars = list(tmp.glob("data.tar*")) if not tars: return out tar = tars[0] data = tmp / "data" data.mkdir() subprocess.run(["bsdtar", "-xf", str(tar), "-C", str(data)], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for dp, _, fn in os.walk(data): for f in fn: p = Path(dp) / f if p.is_symlink() or not p.is_file(): continue with open(p, "rb") as fh: out[hashlib.sha256(fh.read()).hexdigest()] = "/" + str(p.relative_to(data)) return out def main(): CACHE.mkdir(parents=True, exist_ok=True) # 1. targets: every staged unique runtime ELF, plus duplicates targets = {} # rel path under client/runtime -> sha256 stage = ROOT / "client" / "runtime" for dp, _, fn in os.walk(stage): for f in fn: p = Path(dp) / f with open(p, "rb") as fh: targets[str(p.relative_to(stage))] = hashlib.sha256(fh.read()).hexdigest() dups = {} dup_file = OUT / "DUPLICATES.tsv" if dup_file.exists(): for line in dup_file.read_text().splitlines()[1:]: d, canon = line.split("\t") dups[d] = canon print(f"{len(targets)} unique runtime binaries, {len(dups)} duplicates", flush=True) # 2. packages from every manifest jobs, pv_versions, scout_versions = set(), set(), set() for name, (inst, repo, distro) in RUNTIMES.items(): mans = [p for p in inst.rglob("manifest.dpkg")] + [p for p in inst.rglob("manifest.txt")] \ + [p for p in inst.rglob("metadata/packages.txt")] for m in mans: for pkg, arch, ver, src in parse_manifest(m): jobs.add((repo, distro, pkg, arch, ver, src)) for v in inst.rglob("pressure-vessel*/metadata/VERSION.txt"): pv_versions.add(v.read_text().strip()) if repo == "steamrt1": for v in inst.rglob("version.txt"): m = re.match(r"steam-runtime_(1\.0\.[0-9.]+)$", v.read_text().strip()) if m: scout_versions.add(m.group(1)) print(f"{len(jobs)} package versions to fetch, pressure-vessel versions: {sorted(pv_versions)}", flush=True) # 3. download (parallel, cached) and hash index = {} # sha256 -> (package, version, arch, path, url, source, repo) missing = [] done = 0 with cf.ThreadPoolExecutor(8) as ex, cf.ThreadPoolExecutor(8) as hx: hashing = [] for job, deb, url in ex.map(download_pkg, sorted(jobs)): done += 1 if done % 200 == 0: print(f" fetched {done}/{len(jobs)}", flush=True) if deb is None: missing.append(job) continue hashing.append((job, url, hx.submit(hash_archive, deb))) for job, url, fut in hashing: repo, distro, pkg, arch, ver, src = job try: for h, member in fut.result().items(): index.setdefault(h, (pkg, ver, arch or "", member, url, src, repo)) except Exception as e: print(f" could not unpack {pkg} {ver}: {e}", flush=True) for ver in sorted(pv_versions): for f in PV_FILES: dest = CACHE / f"pv-{ver}-{f}" url = PV.format(ver=ver, file=urllib.parse.quote(f)) if dest.exists() or fetch(url, dest): for h, member in hash_archive(dest, inner_deb_data=False).items(): index.setdefault(h, ("pressure-vessel", ver, "", member, url, "steam-runtime-tools", "pressure-vessel")) for ver in sorted(scout_versions): dest = CACHE / f"scout-{ver}-SteamLinuxRuntime.tar.xz" url = SCOUT_IMAGE.format(ver=ver) if dest.exists() or fetch(url, dest): for h, member in hash_archive(dest, inner_deb_data=False).items(): index.setdefault(h, ("steam-runtime image", ver, "", member, url, "steam-runtime", "steamrt-images-scout")) # 4. report def row(rel, h): m = index.get(h) if not m: return [rel, h, "unmatched"] + [""] * 7 pkg, ver, arch, member, url, src, repo = m return [rel, h, "match", pkg, ver, arch, member, src, repo, url] hdr = ["runtime_path", "sha256", "status", "package", "version", "arch", "path_in_package", "source_package", "repo", "url"] rows = [row(r, h) for r, h in sorted(targets.items())] by_rel = {r: h for r, h in targets.items()} rows += [row(d, by_rel[c]) + [] for d, c in sorted(dups.items()) if c in by_rel] with open(OUT / "PROVENANCE.tsv", "w") as f: f.write("\t".join(hdr) + "\n") for r in rows: f.write("\t".join(r) + "\n") with open(OUT / "PROVENANCE-missing.tsv", "w") as f: f.write("repo\tpackage\tarch\tversion\tsource\n") for repo, distro, pkg, arch, ver, src in sorted(missing): f.write(f"{repo}\t{pkg}\t{arch or ''}\t{ver}\t{src}\n") matched = sum(1 for r in rows if r[2] == "match") uniq_matched = sum(1 for h in targets.values() if h in index) print(f"unique binaries matched: {uniq_matched}/{len(targets)}; all rows matched: {matched}/{len(rows)}; " f"packages not downloadable: {len(missing)}", flush=True) if __name__ == "__main__": sys.exit(main())