Is it safe to download a model file?

"""Keep the copy of the demo files inside ../widget.html equal to the ones modelfile.py builds.

    python3 embed_widget.py          rewrite the data block in widget.html
    python3 embed_widget.py --check  exit 1 if widget.html differs (test_widget.py runs the same check)

The data block holds the two demo files as hexadecimal text. The page shows what is in them and runs nothing.
"""
import json
import re
import sys
from pathlib import Path

import harmless
import modelfile as mf

WIDGET = Path(__file__).resolve().parent.parent / "widget.html"
CONFIG_BLOCK = re.compile(r'(<script type="application/json" id="mf-config">)(.*?)(</script>)', re.S)


def config():
    """What the page needs: the bytes of both demo files, the name and note of the one function, and the restricted loader's list."""
    return {"pickle": mf.demo_pickle().hex(), "safetensors": mf.demo_safetensors().hex(), "func": harmless.ran_on_open.__name__, "note": mf.NOTE,
            "allowed": [f"{m}.{n}" for m, n in mf.ALLOWED]}


def embed(html):
    """html with the data block rewritten from modelfile.py."""
    settings = json.dumps(config(), ensure_ascii=False, separators=(", ", ": "))
    return CONFIG_BLOCK.sub(lambda m: m.group(1) + "\n" + settings + "\n" + m.group(3), html, count=1)


if __name__ == "__main__":
    current = WIDGET.read_text(encoding="utf-8")
    updated = embed(current)
    if "--check" in sys.argv:
        sys.exit(0 if updated == current else 1)
    WIDGET.write_text(updated, encoding="utf-8")
    print("widget.html data block now matches modelfile.py")

"""The one function the demo pickle calls. It records a note and does nothing else."""

NOTES = []


def ran_on_open(note):
    """Called by the loader while it opens the demo pickle; keeps the note in NOTES and hands it back."""
    NOTES.append(note)
    return note

"""Two files that hold the same six numbers: a pickle that calls a function while it opens, and a safetensors file that cannot.

Standard library only. The one function the demo pickle calls is harmless.ran_on_open, which records a note and does nothing else.
The pickle is built here from Python objects at run time; no pickle file is stored anywhere.
"""
import io
import json
import pickle
import pickletools
import struct

import harmless

NAME = "layer.weight"
SHAPE = (2, 3)
VALUES = (1.5, -0.25, 0.75, 2.0, -1.0, 0.125)             # exact in float32 and in float64, so both formats give back the same numbers
NOTE = "the demo file was opened"
PICKLE_PROTOCOL = 3                                        # protocol 3 writes the import as one GLOBAL opcode with the two names as plain text
MAX_HEADER = 100_000_000                                   # 100 MB, as in the safetensors README and MAX_HEADER_SIZE in safetensors/src/tensor.rs
DTYPES = {"F32": ("f", 4), "I32": ("i", 4)}
ALLOWED = (("builtins", "range"), ("builtins", "set"), ("builtins", "frozenset"), ("builtins", "complex"), ("builtins", "slice"))
IMPORTS = ("GLOBAL", "STACK_GLOBAL", "INST")               # opcodes that ask the loader to import a name
CALLS = ("REDUCE", "OBJ", "NEWOBJ", "NEWOBJ_EX")           # opcodes that make the loader call it
STRING_PUSHES = ("SHORT_BINUNICODE", "BINUNICODE", "BINUNICODE8", "UNICODE")
MEMO = ("MEMOIZE", "PUT", "BINPUT", "LONG_BINPUT")        # bookkeeping: remember the top item


class FormatError(ValueError):
    """The bytes are not a well-formed safetensors file."""


class Opens:
    """An object whose pickle says: to rebuild me, call ran_on_open with this note."""

    def __reduce__(self):
        return (harmless.ran_on_open, (NOTE,))


def demo_pickle():
    """The six numbers plus one object that calls ran_on_open when the file is opened."""
    return pickle.dumps({NAME: list(VALUES), "note": Opens()}, protocol=PICKLE_PROTOCOL)


def plain_pickle():
    """The same six numbers and nothing else: a pickle that imports and calls nothing."""
    return pickle.dumps({NAME: list(VALUES)}, protocol=PICKLE_PROTOCOL)


# ---------- safetensors: an 8-byte length, a JSON header, then raw numbers ----------

def write_safetensors(tensors, metadata=None):
    """tensors is {name: (dtype, shape, values)}. Row-major, little-endian, no padding."""
    header, body = {}, b""
    if metadata:
        header["__metadata__"] = dict(metadata)
    for name, (dtype, shape, values) in tensors.items():
        code, _ = DTYPES[dtype]
        assert len(values) == _count(shape)
        raw = struct.pack(f"<{len(values)}{code}", *values)
        header[name] = {"dtype": dtype, "shape": list(shape), "data_offsets": [len(body), len(body) + len(raw)]}
        body += raw
    text = json.dumps(header, separators=(",", ":")).encode("utf-8")
    return struct.pack("<Q", len(text)) + text + body


def demo_safetensors():
    return write_safetensors({NAME: ("F32", SHAPE, VALUES)})


def _count(shape):
    n = 1
    for side in shape:
        n *= side
    return n


def _no_duplicates(pairs):
    out = {}
    for key, value in pairs:
        if key in out:
            raise FormatError(f"duplicate key {key!r}")
        out[key] = value
    return out


def _is_int(x):
    return isinstance(x, int) and not isinstance(x, bool)


def read_safetensors(data):
    """(tensors, metadata). Reading only unpacks numbers: nothing in the file is imported, run or evaluated."""
    if len(data) < 8:
        raise FormatError("shorter than 8 bytes")
    (n,) = struct.unpack_from("<Q", data, 0)
    if n > MAX_HEADER or 8 + n > len(data):
        raise FormatError("header length is larger than the file or the 100 MB cap")
    raw = data[8:8 + n]
    if not raw.startswith(b"{"):
        raise FormatError("header does not begin with {")
    try:
        header = json.loads(raw.decode("utf-8"), object_pairs_hook=_no_duplicates)
    except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as e:
        raise FormatError(f"header is not JSON: {e}") from e
    metadata = header.pop("__metadata__", {})
    if not isinstance(metadata, dict) or not all(isinstance(k, str) and isinstance(v, str) for k, v in metadata.items()):
        raise FormatError("__metadata__ must map text to text")
    body = data[8 + n:]
    spans, tensors = [], {}
    for name, info in header.items():
        if not isinstance(info, dict) or set(info) != {"dtype", "shape", "data_offsets"} or not isinstance(info["dtype"], str) or info["dtype"] not in DTYPES:
            raise FormatError(f"{name!r}: needs dtype, shape and data_offsets, with a known dtype")
        shape, offsets = info["shape"], info["data_offsets"]
        if not isinstance(shape, list) or not all(_is_int(s) and s >= 0 for s in shape):
            raise FormatError(f"{name!r}: bad shape")
        if not isinstance(offsets, list) or len(offsets) != 2 or not all(_is_int(o) for o in offsets):
            raise FormatError(f"{name!r}: bad data_offsets")
        begin, end = offsets
        code, size = DTYPES[info["dtype"]]
        if not 0 <= begin <= end <= len(body) or end - begin != _count(shape) * size:
            raise FormatError(f"{name!r}: offsets run past the data or do not fit the shape")
        spans.append((begin, end))
        tensors[name] = (info["dtype"], tuple(shape), struct.unpack(f"<{_count(shape)}{code}", body[begin:end]))
    expected = 0
    for begin, end in sorted(spans):
        if begin != expected:
            raise FormatError("tensors overlap or leave a hole")
        expected = end
    if expected != len(body):
        raise FormatError("bytes after the last tensor are not indexed")
    return tensors, metadata


# ---------- pickle: a list of instructions ----------

def opcodes(data):
    """[(position, name, argument)] for every opcode, read with pickletools.genops. Nothing is executed."""
    return [(pos, op.name, arg) for op, arg, pos in pickletools.genops(data)]


def kind(name):
    return "import" if name in IMPORTS else "call" if name in CALLS else "build"


def imports(data):
    """[(step, module, name)] for each import opcode; module is None when the two names are not plain text just before it."""
    found, recent = [], []
    for step, (_, name, arg) in enumerate(opcodes(data)):
        if name in MEMO:
            continue
        if name in ("GLOBAL", "INST"):
            module, _, attr = arg.partition(" ")
            found.append((step, module, attr))
        elif name == "STACK_GLOBAL":
            module, attr = ([None, None] + recent)[-2:]
            found.append((step, module, attr) if module is not None and attr is not None else (step, None, None))
        recent = (recent + [arg if name in STRING_PUSHES else None])[-2:]
    return found


def scan(data):
    """What a name-based scanner flags: every import not on the allowed list, as 'module.name', or 'unreadable'."""
    return [f"{m}.{n}" if m is not None else "unreadable" for _, m, n in imports(data) if m is None or (m, n) not in ALLOWED]


class Restricted(pickle.Unpickler):
    """The 'Restricting Globals' pattern from the Python pickle documentation: only the allowed names may be imported."""

    def find_class(self, module, name):
        if (module, name) in ALLOWED:
            return super().find_class(module, name)
        raise pickle.UnpicklingError(f"global '{module}.{name}' is forbidden")


def load_restricted(data):
    return Restricted(io.BytesIO(data)).load()


def first_refused(data):
    """The step (0-based) at which load_restricted stops, or None if it never does."""
    for step, module, name in imports(data):
        if module is None or (module, name) not in ALLOWED:
            return step
    return None


def numbers():
    """Every number the lesson prints about the two demo files."""
    p, s = demo_pickle(), demo_safetensors()
    ops = opcodes(p)
    return {"pickle_bytes": len(p), "safetensors_bytes": len(s), "header_bytes": struct.unpack_from("<Q", s, 0)[0],
            "opcodes": len(ops), "imports": sum(kind(o[1]) == "import" for o in ops), "calls": sum(kind(o[1]) == "call" for o in ops),
            "import_step": first_refused(p) + 1, "call_step": next(i for i, o in enumerate(ops, 1) if kind(o[1]) == "call"),
            "safetensors_steps": 3}

"""Prints every number the lesson quotes. Loading the demo pickle calls harmless.ran_on_open once; the safetensors file calls nothing."""
import pickle

import harmless
import modelfile as mf


def main():
    n = mf.numbers()
    print(f"pickle file: {n['pickle_bytes']} bytes, {n['opcodes']} opcodes, {n['imports']} import and {n['calls']} call")
    print(f"  import is step {n['import_step']}, call is step {n['call_step']}")
    print(f"safetensors file: {n['safetensors_bytes']} bytes, header {n['header_bytes']} bytes, {n['safetensors_steps']} reading steps, 0 imports, 0 calls")
    print("header text:", mf.demo_safetensors()[8:8 + n["header_bytes"]].decode())

    tensors, _ = mf.read_safetensors(mf.demo_safetensors())
    print("safetensors opened; notes recorded:", len(harmless.NOTES), "; numbers:", tensors[mf.NAME][2])
    loaded = pickle.loads(mf.demo_pickle())
    print("pickle opened; notes recorded:", len(harmless.NOTES), "; numbers:", tuple(loaded[mf.NAME]))
    harmless.NOTES.clear()
    try:
        mf.load_restricted(mf.demo_pickle())
    except pickle.UnpicklingError as e:
        print("restricted loader:", e, "; notes recorded:", len(harmless.NOTES))
    print("scanner flags:", mf.scan(mf.demo_pickle()), "; flags the plain pickle:", mf.scan(mf.plain_pickle()))


if __name__ == "__main__":
    main()

"""Guards for modelfile.py: what opening each demo file does, what the restricted loader and the scanner say, and that the folder holds no payload."""
import ast
import pickle
import re
import struct
from pathlib import Path

import pytest

import harmless
import modelfile as mf

FOLDER = Path(__file__).resolve().parent.parent


@pytest.fixture(autouse=True)
def clean_notes():
    harmless.NOTES.clear()
    yield
    harmless.NOTES.clear()


def make(header_text, body=b""):
    """A safetensors file with a header written by hand, for the cases the writer would never produce."""
    raw = header_text.encode("utf-8")
    return struct.pack("<Q", len(raw)) + raw + body


def test_opening_the_pickle_calls_the_function_and_still_gives_the_numbers():
    """Guards the lesson's headline: one call happens while the file opens, and nothing about the result shows it."""
    loaded = pickle.loads(mf.demo_pickle())
    assert harmless.NOTES == [mf.NOTE]
    assert loaded[mf.NAME] == list(mf.VALUES) and loaded["note"] == mf.NOTE


def test_opening_the_safetensors_file_calls_nothing_even_with_hostile_looking_text():
    """Guards the claim that reading a weight-only file runs nothing: text in names and metadata is only text, and the numbers match the pickle's."""
    tensors, _ = mf.read_safetensors(mf.demo_safetensors())
    assert tensors == {mf.NAME: ("F32", mf.SHAPE, mf.VALUES)} and harmless.NOTES == []
    odd = {"harmless.ran_on_open": ("F32", (1,), (1.5,)), "c harmless\nran_on_open\n": ("I32", (2,), (1, -2))}
    meta = {"note": "call harmless.ran_on_open when you read this", "format": "pt"}
    tensors, metadata = mf.read_safetensors(mf.write_safetensors(odd, meta))
    assert tensors == {k: (d, s, v) for k, (d, s, v) in odd.items()} and metadata == meta and harmless.NOTES == []


def test_pickle_bytes_are_not_a_safetensors_file():
    """Guards the two formats being told apart by shape: a pickle fed to the safetensors reader is refused and nothing is called."""
    for data in (mf.demo_pickle(), mf.plain_pickle(), b"", b"\x80\x03"):
        with pytest.raises(mf.FormatError):
            mf.read_safetensors(data)
    assert harmless.NOTES == []


def test_malformed_safetensors_files_are_each_refused():
    """Guards the checks the README lists (offsets in range, no overlap, no hole, no unindexed bytes, size matches shape, header length, header start,
    duplicate keys, known dtype, text-only metadata) and the type checks behind them (entry, keys, shape, offsets, dtype, text, nesting): each broken
    file raises FormatError, never a TypeError or a struct error, and calls nothing. The lesson quotes the number of rows."""
    a = '"a":{"dtype":"F32","shape":[3],"data_offsets":[%d,%d]}'
    not_utf8 = b'{"\xff":1}'
    good = make("{" + a % (0, 12) + "}", bytes(12))
    bad = {
        "past the end": make('{"a":{"dtype":"F32","shape":[6],"data_offsets":[0,24]}}', bytes(12)),
        "size does not fit shape": make('{"a":{"dtype":"F32","shape":[2],"data_offsets":[0,12]}}', bytes(12)),
        "overlap": make("{" + a % (0, 12) + "," + a.replace('"a"', '"b"') % (8, 20) + "}", bytes(20)),
        "hole": make("{" + a % (0, 12) + "," + a.replace('"a"', '"b"') % (16, 28) + "}", bytes(28)),
        "bytes after the last tensor": make("{" + a % (0, 12) + "}", bytes(16)),
        "header longer than the file": struct.pack("<Q", 10 ** 6) + good[8:],
        "header over the cap": struct.pack("<Q", mf.MAX_HEADER + 1) + good[8:],
        "header does not start with {": make(" {" + a % (0, 12) + "}", bytes(12)),
        "unknown dtype": make('{"a":{"dtype":"XX","shape":[3],"data_offsets":[0,12]}}', bytes(12)),
        "duplicate key": make("{" + a % (0, 12) + "," + a % (0, 12) + "}", bytes(12)),
        "metadata value is a number": make('{"__metadata__":{"n":1},' + a % (0, 12) + "}", bytes(12)),
        "boolean in the shape": make('{"a":{"dtype":"F32","shape":[true],"data_offsets":[0,4]}}', bytes(4)),
        "not JSON": make("{nope", bytes(4)),
        "header is not UTF-8": struct.pack("<Q", len(not_utf8)) + not_utf8,
        "header nested too deep": make('{"a":' + "[" * 1000000 + "]" * 1000000 + "}"),
        "entry is not an object": make('{"a":["dtype","shape","data_offsets"]}'),
        "entry misses a key": make('{"a":{"dtype":"F32","shape":[3]}}'),
        "entry has an extra key": make('{"a":{"dtype":"F32","shape":[3],"data_offsets":[0,12],"x":1}}', bytes(12)),
        "dtype is a list": make('{"a":{"dtype":["F32"],"shape":[3],"data_offsets":[0,12]}}', bytes(12)),
        "shape is a number": make('{"a":{"dtype":"F32","shape":3,"data_offsets":[0,12]}}', bytes(12)),
        "negative sides that multiply to the size": make('{"a":{"dtype":"F32","shape":[-1,-3],"data_offsets":[0,12]}}', bytes(12)),
        "offsets are a number": make('{"a":{"dtype":"F32","shape":[3],"data_offsets":5}}', bytes(12)),
        "three offsets": make('{"a":{"dtype":"F32","shape":[3],"data_offsets":[0,12,12]}}', bytes(12)),
        "boolean offset": make('{"a":{"dtype":"F32","shape":[1],"data_offsets":[false,4]}}', bytes(4)),
        "float offset": make('{"a":{"dtype":"F32","shape":[1],"data_offsets":[0,4.0]}}', bytes(4)),
        "negative begin offset": make('{"a":{"dtype":"F32","shape":[2],"data_offsets":[-4,4]}}', bytes(4)),
    }
    lesson = (FOLDER / "lesson.md").read_text(encoding="utf-8")
    assert lesson.count(f"{len(bad)} broken files") == 2, "the lesson quotes how many broken files the reader refuses"
    assert mf.read_safetensors(good)[0]["a"] == ("F32", (3,), (0.0, 0.0, 0.0))
    for what, data in bad.items():
        with pytest.raises(mf.FormatError):
            mf.read_safetensors(data)
        assert harmless.NOTES == [], what


def test_restricted_loader_stops_at_the_import_before_anything_is_called():
    """Guards the fix: the demo file is refused at its import with no note recorded, a plain pickle still loads, and an allowed name is still imported and called."""
    with pytest.raises(pickle.UnpicklingError, match=r"global 'harmless\.ran_on_open' is forbidden"):
        mf.load_restricted(mf.demo_pickle())
    assert harmless.NOTES == []
    for wrong in (b"charmless\nrange\n.", b"cbuiltins\nran_on_open\n."):             # an allowed name from the wrong module, an allowed module with a name it lacks
        with pytest.raises(pickle.UnpicklingError, match="is forbidden"):
            mf.load_restricted(wrong)
    assert mf.load_restricted(mf.plain_pickle()) == {mf.NAME: list(mf.VALUES)}
    with_range = pickle.dumps({"r": range(3)}, protocol=mf.PICKLE_PROTOCOL)
    assert mf.load_restricted(with_range) == {"r": range(3)}
    assert [mf.kind(o[1]) for o in mf.opcodes(with_range)].count("import") == 1                # it did import and call: the allowed list is a list of names
    four = pickle.dumps({mf.NAME: list(mf.VALUES), "note": mf.Opens()}, protocol=4)            # protocol 4 spells the import as STACK_GLOBAL
    assert "STACK_GLOBAL" in [o[1] for o in mf.opcodes(four)]
    with pytest.raises(pickle.UnpicklingError, match=r"global 'harmless\.ran_on_open' is forbidden"):
        mf.load_restricted(four)
    assert harmless.NOTES == []


def test_opcode_listing_finds_one_import_then_one_call_without_running_anything():
    """Guards the numbers the lesson quotes (28 opcodes, import at step 19, call at step 25) and that listing is read-only."""
    ops = mf.opcodes(mf.demo_pickle())
    kinds = [mf.kind(o[1]) for o in ops]
    assert len(ops) == 28 and kinds.count("import") == 1 and kinds.count("call") == 1
    assert kinds.index("import") + 1 == 19 and kinds.index("call") + 1 == 25 and ops[18][2] == "harmless ran_on_open"
    assert [o[1] for o in mf.opcodes(mf.plain_pickle())].count("REDUCE") == 0 and not mf.imports(mf.plain_pickle())
    assert [mf.kind(name) for name in ("REDUCE", "OBJ", "NEWOBJ", "NEWOBJ_EX")] == ["call"] * 4              # the four steps the lesson says call
    assert harmless.NOTES == []


def test_scanner_flags_names_not_intent_and_flags_what_it_cannot_read():
    """Guards the lesson's limit on scanners: the harmless demo is flagged because its name is not on the list, allowed names pass,
    and an import whose names are not plain text just before it is flagged as unreadable instead of passed."""
    assert mf.scan(mf.demo_pickle()) == ["harmless.ran_on_open"] and mf.scan(mf.plain_pickle()) == []
    assert mf.scan(pickle.dumps(range(3), protocol=mf.PICKLE_PROTOCOL)) == []
    assert mf.scan(pickle.dumps(mf.Opens(), protocol=4)) == ["harmless.ran_on_open"]
    from_memo = b"\x80\x04\x8c\x08builtins\x94h\x00\x93."                    # STACK_GLOBAL whose second name comes from the memo, not from a push
    assert mf.scan(from_memo) == ["unreadable"] and mf.first_refused(from_memo) == 4
    assert mf.first_refused(mf.demo_pickle()) == 18 and mf.first_refused(mf.plain_pickle()) is None


def test_the_numbers_the_lesson_quotes():
    """Guards the lesson text against a changed demo: a different value, name or protocol moves every size."""
    assert mf.numbers() == {"pickle_bytes": 158, "safetensors_bytes": 100, "header_bytes": 68, "opcodes": 28, "imports": 1, "calls": 1,
                            "import_step": 19, "call_step": 25, "safetensors_steps": 3}
    header = mf.demo_safetensors()[8:76].decode()
    assert header == '{"layer.weight":{"dtype":"F32","shape":[2,3],"data_offsets":[0,24]}}'


def test_the_only_function_the_demo_calls_records_a_note_and_nothing_else():
    """Guards the safety rule: harmless.py imports nothing, holds one list and one function, and the function appends to the list and returns its argument."""
    tree = ast.parse((FOLDER / "code" / "harmless.py").read_text(encoding="utf-8"))
    body = [n for n in tree.body if not (isinstance(n, ast.Expr) and isinstance(n.value, ast.Constant))]
    assert [type(n).__name__ for n in body] == ["Assign", "FunctionDef"] and body[0].targets[0].id == "NOTES" and body[1].name == "ran_on_open"
    fn = [n for n in body[1].body if not (isinstance(n, ast.Expr) and isinstance(n.value, ast.Constant))]
    assert ast.unparse(fn[0]) == "NOTES.append(note)" and ast.unparse(fn[1]) == "return note" and len(fn) == 2
    assert not any(isinstance(n, (ast.Import, ast.ImportFrom)) for n in ast.walk(tree))


def test_no_payload_words_and_no_stored_model_files_in_the_lesson_folder():
    """Guards the first rule of the lesson: no word that starts a program or runs text, no old palette colours, and no pickle or checkpoint file on disk.
    The words are spelt in pieces so this file does not hold them. test_widget.py starts Node on the widget's own script and so names the process module."""
    banned = ["os" + ".system", "ev" + "al(", "ex" + "ec(", "sub" + "process", "0072" + "B2", "56B4" + "E9", "D55E" + "00", "__imp" + "ort__", "pop" + "en("]
    files = [p for p in FOLDER.rglob("*") if p.is_file()]
    assert files and not [p.name for p in files if p.suffix in (".pkl", ".pt", ".pth", ".bin", ".ckpt", ".pyc")]
    assert not [p for p in FOLDER.rglob("__pycache__")]
    for path in files:
        text = path.read_text(encoding="utf-8")
        for word in banned:
            if word == "sub" + "process" and path.name == "test_widget.py":
                continue
            assert word.lower() not in text.lower(), (path.name, word)
    lesson = (FOLDER / "lesson.md").read_text(encoding="utf-8")
    assert not re.search(r"\]\([^)]*\.(pkl|pt|pth|bin|ckpt)\)", lesson), "no model file is offered for download"


def test_lesson_code_blocks_are_verbatim_excerpts_and_its_numbers_are_the_codes():
    """The lesson quotes these files. If the code changes and the lesson does not, the reader sees code that was never tested or numbers that moved."""
    lesson = (FOLDER / "lesson.md").read_text(encoding="utf-8")
    source = (FOLDER / "code" / "modelfile.py").read_text(encoding="utf-8") + "\n" + (FOLDER / "code" / "harmless.py").read_text(encoding="utf-8")
    blocks = re.findall(r"^```python\n(.*?)^```", lesson, flags=re.S | re.M)
    assert len(blocks) >= 6, "the lesson has no python blocks to check"
    for block in blocks:
        assert block.rstrip("\n") in source, f"not in the code folder:\n{block}"
    n = mf.numbers()
    header = mf.demo_safetensors()[8:8 + n["header_bytes"]].decode()
    for fact in (f"{n['pickle_bytes']}-byte pickle", f"{n['safetensors_bytes']}-byte safetensors", f"{n['opcodes']} steps", f"step {n['import_step']}",
                 f"step {n['call_step']}", f"{n['header_bytes']} bytes", f"```json\n{header}\n```", "8 + 68 + 24 = 100"):
        assert fact in lesson, fact

"""Guards for ../widget.html: its data block, its size, that its readers see the same opcodes and header as modelfile.py, and what a reader sees."""
import json
import pickle
import re
import struct
import subprocess

import pytest

import harmless
import modelfile as mf
from embed_widget import CONFIG_BLOCK, WIDGET, config, embed

HTML = WIDGET.read_text(encoding="utf-8")
ROOT = HTML[HTML.index('<div class="mf-root">'):HTML.index("<script")]
OUTSIDE = re.sub(r"<details>.*?</details>", "", ROOT, flags=re.S)

# Reads the byte readers out of the widget between its marker comments, runs them with Node on files that Python built, and compares every
# opcode (position, end, name, argument), the step where the restricted loader stops, and every safetensors field, with what Python found.
PARITY = r"""
import fs from 'node:fs';
const [fixture, widget] = process.argv.slice(2);
const fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const html = fs.readFileSync(widget, 'utf8');
const js = html.slice(html.indexOf('/* ---------- readers ('), html.indexOf('/* ---------- the words'));
const api = new Function("'use strict';" + js + '; return {toBytes, num, readOps, firstRefused, readSafe};')();
const bad = [];
let checks = 0;
const same = (a, b) => JSON.stringify(a) === JSON.stringify(b);
fx.pickles.forEach((p) => {
  const ops = api.readOps(api.toBytes(p.hex));
  checks += 2;
  if (!same(ops.map((o) => [o.pos, o.end, o.name, o.arg]), p.ops)) bad.push(['ops', p.name, ops.slice(0, 3)]);
  if (api.firstRefused(ops, fx.allowed) !== p.first) bad.push(['first refused', p.name, api.firstRefused(ops, fx.allowed), p.first]);
});
fx.safe.forEach((s) => {
  const got = api.readSafe(api.toBytes(s.hex));
  checks++;
  if (!same(got, s.read)) bad.push(['safetensors', s.name, got, s.read]);
});
[[2, '2.0'], [1.5, '1.5'], [-0.25, '-0.25'], [0.125, '0.125'], [-1, '-1.0']].forEach(([x, t]) => { checks++; if (api.num(x) !== t) bad.push(['num', x, api.num(x)]); });
console.log(JSON.stringify({ checks, bad: bad.slice(0, 4) }));
process.exit(bad.length ? 1 : 0);
"""

# Runs the page script against a small stand-in for the DOM, moves the slider, presses the buttons and the checkbox, and reads back what a reader sees.
PAGE = r"""
import fs from 'node:fs';
const [widget, fixture] = process.argv.slice(2);
const html = fs.readFileSync(widget, 'utf8');
const fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const config = html.match(/<script type="application\/json" id="mf-config">([\s\S]*?)<\/script>/)[1];
const code = html.match(/<script>([\s\S]*?)<\/script>/)[1];

let active = null;
function el(tag) {
  const e = { tag, children: [], listeners: {}, value: '', checked: false, disabled: false, style: {}, attrs: {}, focus: () => { active = e; } };
  let text = '';                                     // as in a browser, setting the text of an element drops its children
  Object.defineProperty(e, 'textContent', { get: () => text, set: (v) => { text = v; e.children.length = 0; } });
  e.setAttribute = (k, v) => { e.attrs[k] = v; };
  e.addEventListener = (type, f) => (e.listeners[type] = e.listeners[type] || []).push(f);
  e.appendChild = (c) => (e.children.push(c), c);
  return e;
}
function open() {
  const page = { root: el('html'), messages: [], parent: { other: true } };
  fx.ids.forEach((id) => { page[id] = el('x'); });
  page['mf-config'].textContent = config;
  const doc = { getElementById: (id) => page[id], createElement: el, documentElement: page.root, get activeElement() { return active; } };
  const win = { addEventListener: (type, f) => page.messages.push(f) };
  new Function('document', 'window', 'parent', code)(doc, win, page.parent);
  return page;
}
const fire = (e, type) => (e.listeners[type || 'click'] || []).forEach((f) => f({ target: e }));
const bad = [];
const want = (what, got, expected) => { if (JSON.stringify(got) !== JSON.stringify(expected)) bad.push([what, got, expected]); };
const states = (page, id) => page[id].children.map((c) => c.attrs['data-s']);
const lit = (pre) => pre.children.filter((c) => c.attrs['class'] === 'mf-on').map((c) => c.textContent);
const step = (page, n) => { page['mf-step'].value = String(n); fire(page['mf-step'], 'input'); };

const first = open();
want('opening: headline', first['mf-head'].textContent, fx.text.opening);
want('opening: slider', [first['mf-step'].value, first['mf-step'].max, first['mf-n'].textContent, first['mf-max'].textContent], ['28', '28', '28', '28']);
want('opening: sizes', [first['mf-psize'].textContent, first['mf-ssize'].textContent], [String(fx.pickleBytes), String(fx.safeBytes)]);
want('opening: counts', [first['mf-pcount'].textContent, first['mf-scount'].textContent], [fx.text.one, fx.text.none]);
want('opening: pickle strip', states(first, 'mf-pstrip'), Array(27).fill('done').concat(['now']));
want('opening: safetensors strip', states(first, 'mf-sstrip'), ['done', 'done', 'now']);
want('opening: kinds', first['mf-pstrip'].children.map((c) => c.attrs['data-k']).map((k, i) => (k === 'build' ? '' : i + k)).filter(Boolean), ['18import', '24call']);
want('opening: labels over the strip', first['mf-marks'].children.map((c) => c.textContent), ['import', 'call']);
want('opening: safetensors line', first['mf-sline'].textContent, fx.text.safeDone);
want('opening: next is off, nothing spoken', [first['mf-next'].disabled, first['mf-live'].textContent], [true, '']);

const page = open();
step(page, 19);
want('step 19: line', page['mf-pline'].textContent, fx.text.importLine);
want('step 19: headline', page['mf-head'].textContent, fx.text.after19);
want('step 19: strip', states(page, 'mf-pstrip').indexOf('now'), 18);
want('step 19: counts', page['mf-pcount'].textContent, fx.text.none);
want('step 19: bytes outlined', lit(page['mf-phex']), [fx.hex.importStep]);
want('dragging stays silent', page['mf-live'].textContent, '');
step(page, 25);
want('step 25: line', page['mf-pline'].textContent, fx.text.callLine);
want('step 25: counts', page['mf-pcount'].textContent, fx.text.one);
want('step 25: bytes outlined', lit(page['mf-phex']), [fx.hex.callStep]);
fire(page['mf-step'], 'change');
want('release speaks', page['mf-live'].textContent, page['mf-head'].textContent + ' ' + page['mf-pline'].textContent);
step(page, 2);
want('step 2: safetensors bytes', [page['mf-sline'].textContent, lit(page['mf-shex'])], [fx.text.safeStep2, [fx.hex.header]]);
want('step 2: safetensors strip', states(page, 'mf-sstrip'), ['done', 'now', 'todo']);

// every slider stop, with the restricted loader off and on: calls count REDUCE steps reached, and the stop is at the first refused import
[false, true].forEach((restricted) => {
  const p = open();
  p['mf-gate'].checked = restricted;
  for (let n = 0; n <= 28; n++) {
    step(p, n);
    const stopped = restricted && n > fx.first;
    const calls = stopped ? 0 : fx.calls[n];
    want('stop ' + n + ' restricted ' + restricted + ': counts', p['mf-pcount'].textContent, 'Functions called: ' + calls + ' · Notes recorded: ' + calls);
    want('stop ' + n + ': stopped words', p['mf-head'].textContent.startsWith('The restricted loader stopped'), stopped);
    want('stop ' + n + ': safetensors calls', p['mf-scount'].textContent, fx.text.none);
    want('stop ' + n + ': next', p['mf-next'].disabled, n >= 28 || stopped);
    if (stopped) {
      const s = states(p, 'mf-pstrip');
      want('stopped strip ' + n, [s[fx.first], s.slice(fx.first + 1).every((x) => x === 'skip'), s.slice(0, fx.first).every((x) => x === 'done')], ['stop', true, true]);
    }
  }
});

// the restricted loader at the end of the file says where it stopped and that nothing ran; unticking restores the opening
const gate = open();
gate['mf-gate'].checked = true; fire(gate['mf-gate'], 'change');
want('restricted: headline', gate['mf-head'].textContent, fx.text.stopped);
want('restricted: line', gate['mf-pline'].textContent, fx.text.stopLine);
want('restricted: speaks', gate['mf-live'].textContent, fx.text.stopped + ' ' + fx.text.stopLine);
want('restricted: safetensors unchanged', gate['mf-sline'].textContent, fx.text.safeDone);
gate['mf-gate'].checked = false; fire(gate['mf-gate'], 'change');
want('unticked: headline', gate['mf-head'].textContent, fx.text.opening);

// the restricted loader never shows a step past its stop: the counter, the spoken value and the slider say where it is, and Next is off
const stop = open();
stop['mf-gate'].checked = true; step(stop, 25);
want('restricted at 25: counter, spoken value, slider, next', [stop['mf-n'].textContent, stop['mf-step'].attrs['aria-valuetext'], stop['mf-step'].value, stop['mf-next'].disabled],
     [String(fx.first + 1), 'step ' + (fx.first + 1) + ' of 28', String(fx.first + 1), true]);

// the bytes outlined in the safetensors dump at its three reading steps: the length, the header, the numbers
[[1, 8], [2, 68], [3, 24]].forEach(([n, bytes]) => {
  step(page, n);
  want('safetensors step ' + n + ': bytes outlined', lit(page['mf-shex']).map((t) => t.split(' ').length), [bytes]);
});

// Next step and Start over
const btn = open();
fire(btn['mf-reset']);
want('start over', [btn['mf-step'].value, btn['mf-pline'].textContent, btn['mf-sline'].textContent, btn['mf-next'].disabled], ['0', fx.text.notStarted, 'Not started.', false]);
btn['mf-live'].textContent = '';                  // Start over spoke; only the click on Next may fill it now
fire(btn['mf-next']);
want('next advances by one step', [btn['mf-step'].value, btn['mf-n'].textContent, btn['mf-live'].textContent !== ''], ['1', '1', true]);
btn['mf-next'].focus();
for (let i = 0; i < 40; i++) fire(btn['mf-next']);
want('next stops at the end', [btn['mf-step'].value, btn['mf-next'].disabled, btn['mf-head'].textContent, active === btn['mf-step']], ['28', true, fx.text.opening, true]);

// the host page theme: only the parent window may set it, and only to light or dark
const themed = open();
const send = (data, source) => themed.messages.forEach((f) => f({ data, source }));
send({ type: 'spicyshark:theme', theme: 'dark' }, { other: false });
send({ type: 'spicyshark:theme', theme: 'purple' }, themed.parent);
send({ type: 'other', theme: 'dark' }, themed.parent);
send(null, themed.parent);
want('ignored messages', themed.root.attrs, {});
send({ type: 'spicyshark:theme', theme: 'dark' }, themed.parent);
want('theme from the parent', themed.root.attrs, { 'data-theme': 'dark' });

console.log(JSON.stringify({ bad: bad.slice(0, 5) }));
process.exit(bad.length ? 1 : 0);
"""

PICKLES = {
    "demo": mf.demo_pickle(),
    "plain": mf.plain_pickle(),
    "range": pickle.dumps({"r": range(3), "s": {1, 2}}, protocol=3),
    "many kinds": pickle.dumps({"a": (1, None, True, False, 70000, -5, 300), "b": [1, 2, 3], "c": (1,), "d": (1, 2), "e": (), "f": 0.5, "g": "xé"}, protocol=3),
    "two imports": pickle.dumps([mf.Opens(), range(2)], protocol=3),
}


def hexed(data, begin, end):
    return " ".join(f"{b:02x}" for b in data[begin:end])


def run_node(tmp_path, script, fixture, args):
    (tmp_path / "fixture.json").write_text(json.dumps(fixture), encoding="utf-8")
    (tmp_path / "check.mjs").write_text(script, encoding="utf-8")
    run = subprocess.run(["node", str(tmp_path / "check.mjs"), *args(tmp_path)], capture_output=True, text=True, timeout=120)
    assert run.returncode == 0, run.stdout + run.stderr
    return json.loads(run.stdout)


def arg_text(arg):
    return "" if arg is None else repr(arg) if isinstance(arg, float) else str(arg)


def pickle_fixture(name, data):
    ops = mf.opcodes(data)
    ends = [o[0] for o in ops[1:]] + [len(data)]
    first = mf.first_refused(data)
    return {"name": name, "hex": data.hex(), "ops": [[pos, end, op, arg_text(arg)] for (pos, op, arg), end in zip(ops, ends)], "first": -1 if first is None else first}


def safe_fixture(name, data):
    tensors, _ = mf.read_safetensors(data)
    (n,) = struct.unpack_from("<Q", data, 0)
    header = json.loads(data[8:8 + n])
    read = {k: {"dtype": d, "shape": list(s), "offsets": header[k]["data_offsets"], "values": list(v)} for k, (d, s, v) in tensors.items()}
    return {"name": name, "hex": data.hex(), "read": {"headerBytes": n, "tensors": read}}


def test_widget_data_block_matches_the_python_constants():
    """Guards a constant edit that was not copied into the widget (run embed_widget.py)."""
    assert embed(HTML) == HTML


def test_embed_rewrites_a_widget_whose_data_block_drifted():
    """Guards embed() skipping the block: a hand-edited note or file must be put back."""
    drifted = HTML.replace(mf.NOTE, "something else", 1).replace('"builtins.range"', '"builtins.set"', 1)
    assert drifted != HTML and embed(drifted) == HTML
    assert json.loads(CONFIG_BLOCK.search(HTML).group(2)) == json.loads(json.dumps(config()))
    assert bytes.fromhex(config()["pickle"]) == mf.demo_pickle() and bytes.fromhex(config()["safetensors"]) == mf.demo_safetensors()


def test_every_element_the_script_looks_up_exists_inside_the_root():
    """Guards a widget that never starts: the parity run covers only the readers, so a misspelt id passed there and threw in the browser."""
    script = HTML[HTML.index("<script>"):]
    lookups = set(re.findall(r"\$\('([\w-]+)'\)", script)) | {f"mf-{k}" for k in ("head", "pline", "pcount", "sline", "scount")}
    assert "$('mf-' + k)" in script                        # the five text lines are looked up by suffix; they are added above
    missing = sorted(i for i in lookups if i != "mf-config" and f'id="{i}"' not in ROOT)
    assert not missing, missing
    assert 'id="mf-config"' in HTML


def test_widget_is_small_asks_for_nothing_online_and_runs_no_text():
    assert len(HTML.splitlines()) <= 300 and len(HTML.encode("utf-8")) < 40 * 1024
    page = CONFIG_BLOCK.sub("", HTML)
    for banned in ("http://", "https://", "fetch(", "XMLHttpRequest", "WebSocket", "sendBeacon", "import(", "new Function", "innerHTML", "<form", 'type="submit"'):
        assert banned not in page, banned
    assert " src=" not in page and "@import" not in page


def test_widget_shows_at_most_five_controls_and_hides_the_rest():
    """Guards the first screen at 390 px: the slider, the two buttons and the checkbox are all there is; the byte dumps sit in a closed details element."""
    assert len(re.findall(r"<(?:select|button|input|textarea)\b", OUTSIDE)) <= 5
    assert "<details><summary>More settings</summary>" in ROOT and "<details open" not in ROOT
    assert re.search(r"<details><summary>More settings</summary>.*?id=\"mf-phex\".*?id=\"mf-shex\"", ROOT, re.S)


def test_widget_keeps_the_site_theme_listener_and_palette():
    assert "d.type !== 'spicyshark:theme'" in HTML and "e.source !== parent" in HTML and "parent === window" in HTML
    for pair in ("--bg: light-dark(#ffffff, #0a0f17)", "--panel: light-dark(#ebf0f6, #131b28)", "--ink: light-dark(#0a1626, #f4f7fb)",
                 "--mute: light-dark(#44566b, #a3b1c4)", "--ctl: light-dark(#66768a, #7a8a9f)", "--ink-on: light-dark(#f6f8fb, #05070c)",
                 "--green: light-dark(#00855F, #009E73)", "--orange: light-dark(#b36200, #E69F00)", ":root[data-theme=light]", ":root[data-theme=dark]"):
        assert pair in HTML, pair
    assert not set(re.findall(r"#[0-9a-fA-F]{3,8}\b", HTML)) - {"#ffffff", "#0a0f17", "#ebf0f6", "#131b28", "#0a1626", "#f4f7fb", "#44566b", "#a3b1c4",
                                                               "#66768a", "#7a8a9f", "#f6f8fb", "#05070c", "#00855F", "#009E73", "#b36200", "#E69F00"}


def test_widget_says_what_it_does_and_does_not_do():
    """Guards the sentence the lesson promises: the page shows what Python's loader would do and runs nothing itself, visible without opening anything."""
    assert "The widget shows what Python&rsquo;s loader would do with the bytes; it only reads them and writes words, and runs nothing itself." in OUTSIDE
    leaves_out = re.search(r"<details><summary>What this toy leaves out</summary>(.*?)</details>", ROOT, re.S).group(1)
    for said in ("millions or billions of numbers", "you can add types to its allowed list", "could name any function Python can import", "no holes"):
        assert said in leaves_out, said
    assert ".mf-root .mf-btns button { flex: 1 1 auto; white-space: nowrap; }" in HTML      # a button wraps to the next row, never its label
    assert 'class="mf-sr" id="mf-live" role="status" aria-live="polite" aria-atomic="true"' in ROOT


def test_widget_readers_see_what_python_sees(tmp_path):
    """Guards the widget's JavaScript drifting from modelfile.py: the lesson prints Python's numbers and the reader sees the widget's, so the opcode
    list, the step where the restricted loader stops and the safetensors header must agree on the demo files and on other pickles."""
    pickles = [pickle_fixture(name, data) for name, data in PICKLES.items()]
    safe = [safe_fixture("demo", mf.demo_safetensors()),
            safe_fixture("two tensors", mf.write_safetensors({"a": ("F32", (3,), (0.5, 1.0, -2.0)), "b": ("I32", (2,), (7, -9))}, {"format": "pt"}))]
    assert pickles[0]["first"] == 18 and pickles[1]["first"] == -1 and pickles[2]["first"] == -1 and pickles[4]["first"] > 0
    allowed = config()["allowed"]
    result = run_node(tmp_path, PARITY, {"pickles": pickles, "safe": safe, "allowed": allowed}, lambda d: [str(d / "fixture.json"), str(WIDGET)])
    assert result["bad"] == [] and result["checks"] == 2 * len(pickles) + len(safe) + 5


def test_widget_page_shows_what_the_lesson_names(tmp_path):
    """Guards the part a reader sees: the opening state (one function called by the pickle, none by safetensors), the import at step 19 and the call at
    step 25, every slider stop with the restricted loader off and on, the stop cell, the buttons, what is spoken on release, and the theme message."""
    ops = mf.opcodes(mf.demo_pickle())
    calls = [sum(1 for o in ops[:n] if o[1] == "REDUCE") for n in range(len(ops) + 1)]
    n = mf.numbers()
    fx = {"ids": sorted(set(re.findall(r'id="(mf-[\w-]+)"', HTML))), "pickleBytes": n["pickle_bytes"], "safeBytes": n["safetensors_bytes"], "calls": calls,
          "first": mf.first_refused(mf.demo_pickle()),
          "hex": {"importStep": hexed(mf.demo_pickle(), ops[18][0], ops[19][0]), "callStep": hexed(mf.demo_pickle(), ops[24][0], ops[25][0]),
                  "header": hexed(mf.demo_safetensors(), 8, 8 + n["header_bytes"])},
          "text": {"opening": "Opening the pickle called a function; opening the safetensors file called nothing.",
                   "after19": "After step 19 the pickle has called nothing; the safetensors file has called nothing.",
                   "stopped": "The restricted loader stopped the pickle at step 19 and called nothing. The safetensors file opens as before.",
                   "importLine": "Step 19 of 28 (GLOBAL): IMPORT harmless.ran_on_open. A file could name any function Python can import here.",
                   "callLine": 'Step 25 of 28 (REDUCE): CALL ran_on_open("the demo file was opened"). It records a note and does nothing else.',
                   "stopLine": "Step 19 of 28: the loader meets the import of harmless.ran_on_open, which is not on its list of 5 names. It stops with an error. "
                               "Nothing was called and the numbers did not load.",
                   "notStarted": "Not started. Move the slider or press Next step.",
                   "one": "Functions called: 1 · Notes recorded: 1", "none": "Functions called: 0 · Notes recorded: 0",
                   "safeStep2": 'Step 2 of 3: read the 68-byte header. It is plain text: layer.weight is F32, shape [2, 3], at bytes 0 to 24.',
                   "safeDone": "Step 3 of 3: read those 24 bytes as 6 numbers: 1.5, -0.25, 0.75, 2.0, -1.0, 0.125. Done: nothing was imported or called."}}
    result = run_node(tmp_path, PAGE, fx, lambda d: [str(WIDGET), str(d / "fixture.json")])
    assert result == {"bad": []}