#!/usr/bin/env python3 """Fail-closed checks for code Git objects and asset LFS pointers.""" import argparse from contextlib import contextmanager import os from pathlib import Path, PurePosixPath import re import subprocess import sys import shutil MAX_CODE_BYTES = 2 * 1024 * 1024 PAYLOAD_SUFFIXES = frozenset( ".png .jpg .jpeg .webp .gif .bmp .tga .tif .tiff .exr .hdr .svg .ico " ".dds .ktx .ktx2 .texture .glb .gltf .obj .fbx .blend .mesh " ".wav .ogg .mp3 .flac .bank .mp4 .webm .mov .ttf .otf .woff .woff2 " ".exe .dll .pdb .so .dylib .a .lib .bin .spv .snapshot .cube " ".zip .7z .rar .tar .gz .xz .zst .pdf .doc .docx .pptx .xlsx " ".pk3 .bsp .aas .md3 .md4 .mdc .mds .md5mesh .md5anim .iqm .ase " ".ixpl .plugin .ixm .mb3dat .mb3txt .qvm .roq .icns .dat .map .prt .srf .rad .skin .shader .arena .bot .voice .crosshair".split() ) PAYLOAD_DIRS = frozenset({"assets", "assets-src", "compiled-assets", "compiled-shaders", ".cache", ".work", "target", "captures", "references", "packages"}) POINTER = re.compile(rb"version https://git-lfs.github.com/spec/v1\noid sha256:[0-9a-f]{64}\nsize [0-9]+\n") def git(*args, input=None): return subprocess.check_output(["git", *args], input=input) def check_code(path, data): name = PurePosixPath(path) if any(part.lower() in PAYLOAD_DIRS for part in name.parts): raise ValueError(f"{path}: asset/build/cache directory is not allowed in code Git") if name.suffix.lower() in PAYLOAD_SUFFIXES or ".so." in name.name.lower(): raise ValueError(f"{path}: asset or binary extension belongs under assets/") if len(data) > MAX_CODE_BYTES: raise ValueError(f"{path}: exceeds the {MAX_CODE_BYTES}-byte code limit") if b"\0" in data: raise ValueError(f"{path}: binary data belongs under assets/") try: data.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError(f"{path}: code/documentation must be UTF-8 text") from exc if data.startswith(b"version https://git-lfs.github.com/spec/v1\n"): raise ValueError(f"{path}: LFS pointers belong under assets/") def check_asset(path, data): # Git LFS specifies that an empty file is its own empty pointer. # This contains no payload bytes and must still have filter=lfs attributes. if data and not POINTER.fullmatch(data): raise ValueError(f"{path}: asset Git object is not a canonical LFS pointer; run setup and git add --renormalize") def is_asset(path): parts = PurePosixPath(path).parts return parts[0] == "assets" or "references" in parts[:-1] @contextmanager def blob_reader(): # One bounded streaming process, rather than two Git launches per asset. process = subprocess.Popen(["git", "cat-file", "--batch"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) try: yield process finally: process.stdin.close() process.stdout.close() if process.poll() is None: process.terminate() process.wait() def validate_blob(path, oid, reader): reader.stdin.write(oid.encode() + b"\n") reader.stdin.flush() header = reader.stdout.readline().split() if len(header) != 3 or header[1] != b"blob": raise ValueError(f"{path}: invalid or missing Git blob") size = int(header[2]) limit = 256 if is_asset(path) else MAX_CODE_BYTES if size > limit: raise ValueError(f"{path}: Git blob is {size} bytes; limit is {limit}") data = reader.stdout.read(size) if len(data) != size or reader.stdout.read(1) != b"\n": raise ValueError(f"{path}: truncated Git batch output") (check_asset if is_asset(path) else check_code)(path, data) def check_index(): paths = [] with blob_reader() as reader: for entry in git("ls-files", "--stage", "-z").split(b"\0"): if not entry: continue header, path_bytes = entry.split(b"\t", 1) permissions, oid, stage = header.decode().split() path = os.fsdecode(path_bytes) if stage != "0" or permissions not in {"100644", "100755"}: raise ValueError(f"{path}: unresolved, symlink or nested repository entry is not allowed") validate_blob(path, oid, reader) paths.append(path) if paths: raw = git("check-attr", "--cached", "-z", "--stdin", "filter", input=b"\0".join(os.fsencode(p) for p in paths) + b"\0").split(b"\0") for i in range(0, len(raw) - 1, 3): path, _, value = raw[i:i + 3] path = os.fsdecode(path) expected = b"lfs" if is_asset(path) else b"nanobrush-code" if value != expected: raise ValueError(f"{path}: expected filter={expected.decode()}, found {value.decode()}") asset_paths = [p for p in paths if is_asset(p)] if asset_paths: raw = git("check-attr", "--cached", "-z", "--stdin", "text", "diff", "merge", input=b"\0".join(os.fsencode(p) for p in asset_paths) + b"\0").split(b"\0") for i in range(0, len(raw) - 1, 3): path, attr, value = raw[i:i + 3] expected = b"unset" if attr == b"text" else b"lfs" if value != expected: raise ValueError(f"{os.fsdecode(path)}: incorrect asset {attr.decode()} attribute") def check_push_with_reader(reader): # Validate all newly reachable commits, including files deleted later. # A new remote branch checks its entire reachable history. seen = set() for line in sys.stdin: _, local_oid, _, remote_oid = line.split() if set(local_oid) == {"0"}: continue revisions = [local_oid] if set(remote_oid) != {"0"}: exists = subprocess.run(["git", "cat-file", "-e", remote_oid], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if exists.returncode == 0: revisions.append("^" + remote_oid) for commit in git("rev-list", *revisions).decode().splitlines(): for entry in git("ls-tree", "-rz", commit).split(b"\0"): if not entry: continue header, path = entry.split(b"\t", 1) permissions, kind, oid = header.decode().split() if permissions not in {"100644", "100755"} or kind != "blob": raise ValueError(f"{os.fsdecode(path)}: unsupported Git tree entry") key = (os.fsdecode(path), oid) if key not in seen: validate_blob(*key, reader) seen.add(key) def check_push(): with blob_reader() as reader: check_push_with_reader(reader) def check_setup(): if not shutil.which("git-xet"): raise ValueError("git-xet is missing; enter nix develop") required = { "filter.nanobrush-code.required": "true", "filter.lfs.required": "true", "lfs.customtransfer.xet.path": "git-xet", "lfs.customtransfer.xet.args": "transfer", "lfs.basictransfersonly": "false", "core.hooksPath": ".githooks", } for key, value in required.items(): if git("config", "--local", "--get", key).decode().strip() != value: raise ValueError(f"{key}: rerun scripts/setup-repository.py") for key in ("filter.nanobrush-code.clean", "filter.lfs.clean", "filter.lfs.process"): if not git("config", "--local", "--get", key).strip(): raise ValueError(f"{key}: missing filter driver") def check_tree(directory): """Check the asset-free source used by Nix validation/build derivations.""" if not directory: raise ValueError("check-tree requires a source directory") root = Path(directory) for path in root.rglob("*"): if path.is_symlink(): raise ValueError(f"{path}: source symlink is not allowed") if path.is_file(): with path.open("rb") as stream: check_code(path.relative_to(root).as_posix(), stream.read(MAX_CODE_BYTES + 1)) if "* text=auto eol=lf filter=nanobrush-code" not in (root / ".gitattributes").read_text().splitlines(): raise ValueError("Code export lacks the required default clean filter") def main(): parser = argparse.ArgumentParser() parser.add_argument("command", choices=["clean", "check", "push", "check-tree"]) parser.add_argument("--require-setup", action="store_true") parser.add_argument("path", nargs="?") args = parser.parse_args() try: if args.require_setup: check_setup() if args.command == "clean": data = sys.stdin.buffer.read(MAX_CODE_BYTES + 1) check_code(args.path or "", data) sys.stdout.buffer.write(data) elif args.command == "check-tree": check_tree(args.path) elif args.command == "push": check_push() else: check_index() except (ValueError, subprocess.CalledProcessError) as exc: print(f"Storage policy: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())