File size: 4,550 Bytes
335be33 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | #!/usr/bin/env python3
"""Verify the release checksum and the required GGUF schema without dependencies."""
from __future__ import annotations
import argparse
import hashlib
import json
import struct
from collections import Counter
from pathlib import Path
from typing import BinaryIO
SCALAR_FORMATS = {
0: "<B",
1: "<b",
2: "<H",
3: "<h",
4: "<I",
5: "<i",
6: "<f",
7: "<?",
10: "<Q",
11: "<q",
12: "<d",
}
def read_exact(stream: BinaryIO, size: int) -> bytes:
value = stream.read(size)
if len(value) != size:
raise ValueError("unexpected end of GGUF file")
return value
def read_scalar(stream: BinaryIO, value_type: int):
fmt = SCALAR_FORMATS[value_type]
return struct.unpack(fmt, read_exact(stream, struct.calcsize(fmt)))[0]
def read_string(stream: BinaryIO, keep: bool = True):
size = read_scalar(stream, 10)
value = read_exact(stream, size)
return value.decode("utf-8") if keep else None
def read_value(stream: BinaryIO, value_type: int, keep: bool = True):
if value_type in SCALAR_FORMATS:
value = read_scalar(stream, value_type)
return value if keep else None
if value_type == 8:
return read_string(stream, keep)
if value_type == 9:
element_type = read_scalar(stream, 4)
count = read_scalar(stream, 10)
values = [read_value(stream, element_type, keep) for _ in range(count)]
return values if keep else None
raise ValueError(f"unsupported GGUF metadata type {value_type}")
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def inspect(path: Path, expected_keys: set[str]) -> dict:
with path.open("rb") as stream:
if read_exact(stream, 4) != b"GGUF":
raise ValueError("invalid GGUF magic")
version = read_scalar(stream, 4)
tensor_count = read_scalar(stream, 10)
metadata_count = read_scalar(stream, 10)
metadata = {}
for _ in range(metadata_count):
key = read_string(stream)
value_type = read_scalar(stream, 4)
value = read_value(stream, value_type, key in expected_keys)
if key in expected_keys:
metadata[key] = value
tensor_types = Counter()
offsets = []
for _ in range(tensor_count):
read_string(stream, keep=False)
dimensions = read_scalar(stream, 4)
for _ in range(dimensions):
read_scalar(stream, 10)
tensor_types[str(read_scalar(stream, 4))] += 1
offsets.append(read_scalar(stream, 10))
return {
"version": version,
"tensor_count": tensor_count,
"metadata": metadata,
"tensor_type_counts": dict(tensor_types),
"aligned": all(offset % 32 == 0 for offset in offsets),
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("artifact", nargs="?", type=Path)
parser.add_argument("--manifest", type=Path, default=Path("manifest/release.json"))
args = parser.parse_args()
expected = json.loads(args.manifest.read_text())
artifact = args.artifact or Path(expected["artifact"]["file"])
failures = []
actual_size = artifact.stat().st_size
actual_hash = sha256(artifact)
if actual_size != expected["artifact"]["size"]:
failures.append(f"size: expected {expected['artifact']['size']}, got {actual_size}")
if actual_hash != expected["artifact"]["sha256"]:
failures.append(f"sha256: expected {expected['artifact']['sha256']}, got {actual_hash}")
gguf_expected = expected["gguf"]
actual = inspect(artifact, set(gguf_expected["metadata"]))
for key in ("version", "tensor_count", "tensor_type_counts", "aligned"):
if actual[key] != gguf_expected[key]:
failures.append(f"{key}: expected {gguf_expected[key]!r}, got {actual[key]!r}")
for key, value in gguf_expected["metadata"].items():
if actual["metadata"].get(key) != value:
failures.append(
f"metadata {key}: expected {value!r}, got {actual['metadata'].get(key)!r}"
)
if failures:
raise SystemExit("verification failed:\n- " + "\n- ".join(failures))
print(f"verified {artifact}: {actual_size} bytes {actual_hash}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|