| |
| """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): |
| |
| |
| 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(): |
| |
| 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): |
| |
| |
| 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 "<stdin>", 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()) |
|
|
