File size: 4,689 Bytes
f6588d9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""Verify the exact ep15 merged model shipped with this runtime."""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
from typing import Any, Sequence

EXPECTED_MANIFEST_SHA256 = (
    "c372e38348349baa7a20eb3ea771b2ae3f742f764f107c3b9e539d272931677b"
)
EXPECTED_FILE_COUNT = 13
EXPECTED_TOTAL_BYTES = 18_839_811_919


def _sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(8 * 1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def verify_model(model_root: Path, manifest_path: Path) -> dict[str, Any]:
    """Return a machine-readable report or fail before model import."""

    if model_root.is_symlink():
        raise RuntimeError(f"model root must not be a symlink: {model_root}")
    root = model_root.resolve(strict=True)
    manifest = manifest_path.resolve(strict=True)
    if not root.is_dir() or root.is_symlink():
        raise RuntimeError(f"model root must be a real directory: {root}")
    manifest_sha256 = _sha256(manifest)
    if manifest_sha256 != EXPECTED_MANIFEST_SHA256:
        raise RuntimeError(
            f"model manifest SHA mismatch: {manifest_sha256} != "
            f"{EXPECTED_MANIFEST_SHA256}"
        )

    entries: list[tuple[str, int, str]] = []
    for line_number, raw_line in enumerate(
        manifest.read_text(encoding="utf-8").splitlines(), start=1
    ):
        fields = raw_line.split("\t")
        if len(fields) != 3:
            raise RuntimeError(f"invalid model manifest row {line_number}")
        relative_text, bytes_text, expected_sha256 = fields
        relative = Path(relative_text)
        if (
            relative.is_absolute()
            or ".." in relative.parts
            or relative.as_posix() != relative_text
            or not bytes_text.isdigit()
            or len(expected_sha256) != 64
            or any(character not in "0123456789abcdef" for character in expected_sha256)
        ):
            raise RuntimeError(f"unsafe model manifest row {line_number}")
        entries.append((relative_text, int(bytes_text), expected_sha256))

    names = [entry[0] for entry in entries]
    if names != sorted(set(names)):
        raise RuntimeError("model manifest paths must be unique and sorted")
    if len(entries) != EXPECTED_FILE_COUNT:
        raise RuntimeError(
            f"model manifest has {len(entries)} files; expected {EXPECTED_FILE_COUNT}"
        )
    if sum(entry[1] for entry in entries) != EXPECTED_TOTAL_BYTES:
        raise RuntimeError("model manifest total byte count changed")

    observed_names: list[str] = []
    for path in root.rglob("*"):
        if path.is_symlink():
            raise RuntimeError(f"model tree contains a symlink: {path}")
        if path.is_file():
            observed_names.append(path.relative_to(root).as_posix())
        elif not path.is_dir():
            raise RuntimeError(f"model tree contains a special file: {path}")
    if sorted(observed_names) != names:
        raise RuntimeError("model regular-file set differs from the frozen manifest")

    verified: list[dict[str, Any]] = []
    for relative_text, expected_bytes, expected_sha256 in entries:
        path = root / relative_text
        observed_bytes = path.stat().st_size
        if observed_bytes != expected_bytes:
            raise RuntimeError(f"model file size changed: {relative_text}")
        observed_sha256 = _sha256(path)
        if observed_sha256 != expected_sha256:
            raise RuntimeError(f"model file SHA changed: {relative_text}")
        verified.append(
            {
                "path": relative_text,
                "bytes": observed_bytes,
                "sha256": observed_sha256,
            }
        )

    return {
        "schema_version": "procedure.ep15_merged3225.model_verify.v1",
        "status": "PASS",
        "model_root": str(root),
        "manifest": str(manifest),
        "manifest_sha256": manifest_sha256,
        "file_count": len(verified),
        "total_bytes": sum(item["bytes"] for item in verified),
        "files": verified,
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--model-root", type=Path, required=True)
    parser.add_argument("--manifest", type=Path, required=True)
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    report = verify_model(**vars(build_parser().parse_args(argv)))
    print(json.dumps(report, ensure_ascii=False, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())