scripts/storage_policy.py

#!/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 "<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())