"""Installed, fail-closed reproduction receipt for Mixle's release claims."""
from __future__ import annotations
import argparse
import base64
import binascii
import csv
import hashlib
import io
import json
import platform
import subprocess
import sys
import types
import zipfile
from importlib.metadata import PackageNotFoundError, distribution, version
from pathlib import Path
from typing import Any
_EXPECTATIONS = {
"gaussian_fit_mu": {"expected": 3.0, "absolute_tolerance": 0.2},
"gaussian_fit_sigma": {"expected": 2.0, "absolute_tolerance": 0.2},
"scalar_vectorized_agree": {"expected": True},
"serialization_score_equal": {"expected": True},
"auto_selects": {"expected": "GaussianEstimator"},
"deterministic_sample": {"expected": 1.701912, "absolute_tolerance": 1e-6},
}
def _pkg_version(name: str) -> str:
try:
return version(name)
except PackageNotFoundError:
return "not installed"
def _sha256_file(path: Path) -> tuple[str, int]:
digest = hashlib.sha256()
size = 0
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1 << 20), b""):
digest.update(chunk)
size += len(chunk)
return digest.hexdigest(), size
def _full_git_commit(repository: Path) -> str:
"""Resolve the source checkout containing this module, never the caller's working directory."""
try:
result = subprocess.run(
["git", "-C", str(repository), "rev-parse", "HEAD"],
check=False,
capture_output=True,
text=True,
timeout=10,
)
except (OSError, subprocess.SubprocessError):
return "unknown"
commit = result.stdout.strip().lower()
if result.returncode != 0 or len(commit) != 40 or any(character not in "0123456789abcdef" for character in commit):
return "unknown"
return commit
def _is_installer_bytecode(name: str) -> bool:
"""True for a byte-compiled module the INSTALLER produced, not a distributed file.
Deliberately narrow: the name must end in ``.pyc`` AND sit directly inside a ``__pycache__``
directory, which is the only place ``pip``'s compileall step writes. A ``.pyc`` shipped
anywhere else in a distribution is not installer output and stays subject to the hash rule.
"""
if not name.endswith(".pyc"):
return False
parts = name.split("/")
return len(parts) >= 2 and parts[-2] == "__pycache__"
def _bytecode_matches_source(pyc: Path, source: Path) -> bool | None:
"""True when ``pyc`` is what compiling ``source`` produces; None when it cannot be decided.
Exempting installer bytecode from the hash rule left the bytecode itself unauthenticated: a
``.pyc`` may differ from its verified ``.py`` and still be what actually executes. CPython's
default timestamp invalidation does not close this -- an attacker who copies the source's mtime
and size into the header gets the tampered bytecode loaded without a recompile.
So the bytecode is compared against a recompilation of the source it claims to come from, using
the optimization level the ``.pyc`` records and the TRUSTED source path -- ``source`` as located
through the installed distribution -- as the compile filename. An earlier version took the
filename from the ``.pyc`` itself, so an altered ``co_filename`` was copied into the expected
object and accepted although tracebacks and code inspection observe it (systems review pass 6);
the file being checked must not supply the value it is checked against. Measured against real
pip-generated bytecode (810/810 modules) the trusted path IS what pip compiled with, so this
reproduces byte-for-byte; a relocated environment, whose bytecode names the old prefix, is
refused rather than excused (recompile it: ``python -m compileall -f``). ``None`` is returned
when the comparison cannot be performed at all (unreadable, truncated, or a marshal format this
interpreter cannot load), so the caller can treat "undecidable" differently from "wrong".
"""
import marshal
try:
raw = pyc.read_bytes()
payload = marshal.loads(raw[16:]) # nosec B302 # the payload is the installed distribution's own __pycache__ bytecode, which CPython will unmarshal AND execute on the very next import of that module; this verifier unmarshals WITHOUT executing, precisely to compare it against a recompilation of the hash-verified source. Refusing to examine it would leave the bytes that actually execute unverified -- the second-pass SYS-02 finding this function closes. Undecidable payloads (the except below) fail closed.
optimize = 2 if ".opt-2." in pyc.name else (1 if ".opt-1." in pyc.name else -1)
recompiled = compile(source.read_bytes(), str(source), "exec", dont_inherit=True, optimize=optimize)
return _code_equal(recompiled, payload)
except (OSError, ValueError, EOFError, TypeError, SyntaxError, AttributeError):
return None
# The comparison used to split code-object fields into "behavioural" and "metadata" and skip the
# latter. That classification was wrong twice: co_exceptiontable was left out (SYS3-02), and then
# co_filename was filed as metadata even though ``sys._getframe().f_code.co_filename`` reads it
# (SYS4-02) -- and measured directly, co_firstlineno and the line table are readable the same way
# via ``f_lineno``. Any field the program can observe can change its output, so NO field is
# metadata for this purpose. There is also no need for a carve-out: the source is recompiled with
# the TRUSTED installed source path as its filename (never the ``.pyc``'s own -- systems review pass
# 6), which is exactly what pip compiled with, so filename, first line and line tables are
# deterministic functions of the same source text and match by construction -- verified on all 810
# modules of a real installation, where every data field matched. What remains is:
#
# * every ``co_*`` DATA attribute is compared for equality;
# * the two ``co_*`` METHODS (``co_lines``, ``co_positions``) return fresh iterators, which compare
# unequal as objects, so they are compared by materialised output;
# * ``co_consts`` is walked recursively so nested code objects get the same treatment;
# * the field set is taken from the LIVE interpreter, so a future CPython field is compared rather
# than silently skipped -- an unknown field can only make the check stricter, never looser.
_CODE_METHOD_FIELDS = frozenset({"co_lines", "co_positions"})
def _code_fields() -> tuple[str, ...]:
"""Every ``co_*`` name on this interpreter's code objects, data and method alike."""
probe = compile("pass", "<probe>", "exec")
return tuple(sorted(name for name in dir(probe) if name.startswith("co_")))
def _code_equal(left: Any, right: Any) -> bool:
"""Equality of two code objects on every observable field, recursing through nested code.
Deliberately NOT ``marshal.dumps(a) == marshal.dumps(b)``: marshal encodes back-references for
shared and interned objects, and that sharing differs between a freshly compiled object and one
loaded from a file. Comparing the serialized bytes therefore reports differences that do not
exist -- measured at 9 false mismatches over 810 files on a CLEAN install, which would have
reinstated the very defect this check exists to fix (refusing ordinary installations).
"""
if type(left) is not type(right):
return False
if not isinstance(left, types.CodeType):
# ``nan != nan``, so two identical code objects each holding a NaN constant compared UNEQUAL
# and a legitimate module was refused -- the original SYS-02 defect returning through a
# numeric edge. Compare floats (and complex parts) by identity-of-representation instead;
# for everything else plain equality is what we want.
if isinstance(left, float) or isinstance(left, complex):
return repr(left) == repr(right)
if isinstance(left, tuple) or isinstance(left, frozenset):
if len(left) != len(right):
return False
if isinstance(left, tuple):
return all(_code_equal(a, b) for a, b in zip(left, right))
return sorted(map(repr, left)) == sorted(map(repr, right))
return bool(left == right)
for attribute in _code_fields():
if attribute == "co_consts":
continue
if attribute in _CODE_METHOD_FIELDS:
try:
if list(getattr(left, attribute)()) != list(getattr(right, attribute)()):
return False
except Exception: # noqa: BLE001 - an unreadable table is a difference, not a pass
return False
continue
if attribute == "co_lnotab":
continue # deprecated alias of co_linetable/co_lines; reading it only emits a warning
if getattr(left, attribute, None) != getattr(right, attribute, None):
return False
if len(left.co_consts) != len(right.co_consts):
return False
return all(_code_equal(a, b) for a, b in zip(left.co_consts, right.co_consts))
def _bytecode_source_path(name: str) -> str | None:
"""Map ``pkg/__pycache__/mod.cpython-312.pyc`` back to the ``pkg/mod.py`` it was compiled from.
Returns ``None`` when the name does not decompose, so an undecodable bytecode path is treated
as unaccounted-for rather than silently exempted.
"""
parts = name.split("/")
if len(parts) < 2 or parts[-2] != "__pycache__":
return None
stem = parts[-1].split(".")[0]
if not stem:
return None
return "/".join([*parts[:-2], f"{stem}.py"])
def _orphaned_package_files() -> int:
"""Count importable mixle files present with no distribution metadata behind them.
A wheel install that fails partway -- a corrupt or truncated archive -- leaves the files it had
already unpacked in place, and pip writes the ``.dist-info`` LAST. The result imports and
answers ``mixle.__version__`` as ``0+unknown`` while being a fragment of a package (SYS-06).
"""
try:
import mixle
root = Path(next(iter(mixle.__path__)))
except Exception: # noqa: BLE001 - nothing importable is a real answer, not an error
return 0
try:
return sum(1 for path in root.rglob("*") if path.is_file())
except OSError:
return 0
[docs]
def installed_content_provenance() -> dict[str, Any]:
"""Digest the installed distribution's RECORD identities and verify hashed files."""
try:
dist = distribution("mixle")
except PackageNotFoundError:
# Distinguish "no mixle here" from "a partially installed mixle here". Both fail closed,
# but they are different situations with different remedies, and reporting them
# identically understated the second: an importable package with no metadata behind it is
# a broken install that must be REMOVED, not an absent one that can simply be installed
# over (SYS-06). Installing on top leaves whatever orphaned files the new wheel does not
# happen to overwrite.
orphaned = _orphaned_package_files()
if orphaned:
return {
"artifact": "mixle.installed_content/v1",
"verified": False,
"reason": (
f"partial install: {orphaned} importable mixle files are present with no "
f"distribution metadata, so this environment is a fragment of a package rather "
f"than an installation of one. Remove the orphaned package directory and "
f"install a hash-verified artifact; installing over it can leave orphans behind."
),
"orphaned_file_count": orphaned,
}
return {
"artifact": "mixle.installed_content/v1",
"verified": False,
"reason": "mixle distribution metadata is not installed",
}
entries: list[dict[str, Any]] = []
failures: list[str] = []
hashed_names = {str(item) for item in (dist.files or ()) if item.hash is not None}
# A partial install does not always erase the metadata: an interrupted or replaced install can
# leave a .dist-info that resolves while its RECORD describes files that are not there. That
# case took the ordinary path below and reported plain "missing" failures, missing the
# partial-install classification and its remedy, which only fired when metadata was absent
# entirely. Recognise it here from the RECORD's own accounting.
recorded = [item for item in (dist.files or ()) if item.hash is not None]
absent = [item for item in recorded if not Path(dist.locate_file(item)).is_file()]
if recorded and len(absent) == len(recorded):
return {
"artifact": "mixle.installed_content/v1",
"verified": False,
"reason": (
f"partial install: the distribution metadata lists {len(recorded)} files and none "
f"of them are present. Remove the distribution and install a hash-verified "
f"artifact; installing over it can leave orphans behind."
),
"missing_recorded_file_count": len(absent),
}
bytecode_exempt: list[str] = []
for item in sorted(dist.files or (), key=str):
if item.hash is None:
# Skipping unhashed installed entries is the same tamper vector as the wheel side
# (SYS-RR8-2): only RECORD may legitimately lack a hash. Installer-written metadata
# (INSTALLER, REQUESTED, direct_url.json) is not distributed code and is exempt.
name = str(item)
if _is_installer_bytecode(name):
# Ordinary `pip install` byte-compiles every module and writes those .pyc rows to
# RECORD with no hash, because it generates them AFTER hashing the distributed
# files. Treating them as tampering made `mixle-reproduce` report passed:false on
# every normal installation (SYS-02) -- the check refused the standard install path
# rather than any actual modification.
#
# They are exempted from the HASH rule, not from verification. Each must correspond
# to a distributed source file that is itself hashed and verified below, and the
# bytecode must be what compiling that source produces. The first version of this
# exemption stopped at the source check and recorded, in this comment, that it did
# not authenticate the bytecode -- which left the bytes that actually execute
# unverified, since CPython's timestamp invalidation is trivially satisfied by
# copying the source's mtime and size into the .pyc header.
source = _bytecode_source_path(name)
if source is None or source not in hashed_names:
failures.append(f"{item}: installer bytecode has no verified source file")
else:
source_path = Path(dist.locate_file(source))
matched = _bytecode_matches_source(Path(dist.locate_file(item)), source_path)
if matched is True:
bytecode_exempt.append(name)
elif matched is False:
failures.append(f"{item}: bytecode does not match its verified source")
else:
failures.append(f"{item}: bytecode could not be checked against its source")
elif not name.endswith(".dist-info/RECORD") and not name.startswith(("mixle-", "mixle.")):
failures.append(f"{item}: installed entry carries no RECORD hash")
elif name.endswith((".py", ".so", ".pyd", ".json")) and ".dist-info/" not in name:
failures.append(f"{item}: installed code entry carries no RECORD hash")
continue
path = Path(dist.locate_file(item))
if not path.is_file():
failures.append(f"{item}: missing")
continue
algorithm = item.hash.mode
if algorithm != "sha256":
failures.append(f"{item}: unsupported RECORD hash {algorithm}")
continue
digest, size = _sha256_file(path)
expected = base64.urlsafe_b64encode(bytes.fromhex(digest)).decode("ascii").rstrip("=")
if expected != item.hash.value:
failures.append(f"{item}: RECORD hash mismatch")
entries.append({"path": str(item), "sha256": digest, "size_bytes": size})
canonical = json.dumps(entries, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
return {
"artifact": "mixle.installed_content/v1",
"digest": hashlib.sha256(canonical).hexdigest(),
"file_count": len(entries),
"installer_bytecode_exempt_count": len(bytecode_exempt),
"verified": bool(entries) and not failures,
"failures": failures,
# The per-file identities the digest above is derived from, so a caller can bind the
# installation to a specific wheel's RECORD rather than only to a summary digest.
"entries": entries,
}
def _expected_hex(value: str, *, length: int, parameter: str) -> str:
"""Normalize a caller-supplied expected digest; refuse only what could never match.
Case and surrounding whitespace are presentation, not identity -- a digest pasted in
uppercase names the same bytes -- so both are normalized rather than refused (legitimate
input must not be rejected). Anything that is not hex of the right length after that can
never equal a shape-valid provenance field, and reporting it as a provenance "mismatch"
would present the caller's typo as evidence of tampering; it is a ValueError instead.
"""
text = value.strip().lower()
if len(text) != length or any(character not in "0123456789abcdef" for character in text):
raise ValueError(f"{parameter} must be a {length}-character hexadecimal digest")
return text
[docs]
def wheel_provenance(
path: Path,
*,
expected_commit: str | None = None,
expected_content_sha256: str | None = None,
) -> dict[str, Any]:
"""Verify a Mixle wheel's identity, RECORD self-consistency, and provenance record shape.
VERIFIED, fail-closed. A wheel that cannot be identified at all raises ``ValueError``
(unreadable archive, wrong METADATA/provenance member count, name/version differing from
the installed distribution, ill-shaped or dirty provenance record). Every check past
identification lands in the returned ``problems`` list instead, and any entry makes
``verified`` False -- a tampered-but-identifiable wheel gets a receipt naming the
tampering rather than no receipt:
* RECORD self-consistency: every hashed RECORD claim is recomputed from the archived
member it names, every member must be claimed, duplicate and root-escaping names are
refused, and exactly one unhashed self-referencing RECORD row is allowed. This is
``_wheel_record_hashes``, the same check ``subject_binding`` builds on.
* ``expected_commit=`` / ``expected_content_sha256=``, when given, are compared against
the record's ``source_commit`` / ``source_content_sha256`` (case-insensitively).
NOT verified, stated so this record cannot be presented as more than it is:
* Provenance field VALUES are shape-validated only: a well-formed but false
``source_commit``/``source_tree``/``source_content_sha256`` -- 40 zeros is a valid
shape -- passes unless the caller supplies the matching ``expected_*`` argument.
* RECORD self-consistency proves the archive matches its own manifest, not that the
manifest is the released one: a wheel rebuilt with an internally consistent RECORD,
including an extra module listed with a correct hash, still self-verifies here.
Binding to the bytes that actually execute is ``subject_binding``'s job, and
``build_receipt`` ANDs the two verdicts.
* The reported wheel ``sha256`` is an identity for external comparison, not a check.
``source_content_sha256`` is one key name over two populations (T3-05). The digest
covers whatever tree the build ran from, and the sibling fields
``source_content_file_count`` / ``source_content_universe`` declare that population: a
wheel built from a git checkout -- and an sdist's embedded copy of this record -- digest
the full checkout (~2,000 files), while a wheel built from an unpacked sdist digests
only the packaged subset (~800 files). Two records disagreeing on this key while their
file counts differ is therefore a POPULATION difference, not tampering. The
cross-artifact check is the sdist record's ``sdist_content_sha256``: the digest of the
packaged subset, present in every sdist-embedded record (and carried into wheels whose
record was adopted through the sdist, beside ``source_content_carried_through_sdist``),
which equals the ``source_content_sha256`` a wheel built from that sdist derives for
itself. ``expected_content_sha256`` is compared against THIS record's declared
population, so draw the expectation from the same universe the record names.
"""
if not path.is_file() or path.suffix != ".whl":
raise ValueError("reproduction artifact must be an existing .whl file")
digest, size = _sha256_file(path)
try:
with zipfile.ZipFile(path) as archive:
metadata_names = [name for name in archive.namelist() if name.endswith(".dist-info/METADATA")]
provenance_names = [name for name in archive.namelist() if name == "mixle/_build_provenance.json"]
if len(metadata_names) != 1 or len(provenance_names) != 1:
raise ValueError("wheel must contain exactly one METADATA and one Mixle build-provenance record")
metadata = archive.read(metadata_names[0]).decode("utf-8")
provenance = json.loads(archive.read(provenance_names[0]))
except Exception as exc: # noqa: BLE001 -- untrusted archive: decompressor/feature errors are not ValueErrors
# zipfile surfaces a corrupt member as zlib.error or lzma.LZMAError, an unsupported method,
# version or flag as NotImplementedError, and an encrypted member as RuntimeError; none is a
# subclass of the classes first enumerated here, so `mixle-reproduce --wheel <such wheel>`
# exited with a traceback instead of a receipt (SYS5-03's sibling, found by attacking the
# repair). Every failure to read the archive is one refusal with the cause named.
raise ValueError(f"reproduction artifact is not a valid Mixle wheel: {type(exc).__name__}: {exc}") from exc
fields = {}
for line in metadata.splitlines():
if ": " in line:
key, value = line.split(": ", 1)
fields.setdefault(key, value)
installed_version = _pkg_version("mixle")
if fields.get("Name", "").lower() != "mixle" or fields.get("Version") != installed_version:
raise ValueError("wheel name/version does not match the installed Mixle distribution")
if not isinstance(provenance, dict) or provenance.get("artifact") != "mixle.build_provenance/v1":
raise ValueError("wheel build provenance has an unsupported schema")
for field in ("source_commit", "source_tree"):
value = provenance.get(field)
if not isinstance(value, str) or len(value) != 40 or any(c not in "0123456789abcdef" for c in value):
raise ValueError(f"wheel build provenance has invalid {field}")
content_digest = provenance.get("source_content_sha256")
if (
not isinstance(content_digest, str)
or len(content_digest) != 64
or any(c not in "0123456789abcdef" for c in content_digest)
):
raise ValueError("wheel build provenance has invalid source_content_sha256")
# SYS-08 added the population these fields describe -- count and selection rule -- because the
# bare digest name implied a single canonical "source" digest that does not exist. Emitting the
# fields is only half the repair: a verifier that does not REQUIRE them lets a record omit or
# mistype them and still pass, so the population stays undeclared in exactly the receipts that
# matter (SYS3-07). Both must be present and well-typed; a stated universe must be non-empty.
file_count = provenance.get("source_content_file_count")
if isinstance(file_count, bool) or not isinstance(file_count, int) or file_count <= 0:
raise ValueError("wheel build provenance has invalid or missing source_content_file_count")
universe = provenance.get("source_content_universe")
if not isinstance(universe, str) or not universe.strip():
raise ValueError("wheel build provenance has invalid or missing source_content_universe")
if provenance.get("source_dirty") is not False:
raise ValueError("release reproduction requires a wheel built from a clean source candidate")
# T3-06: ``verified`` was a hardcoded literal True -- the function hashed the wheel only to
# REPORT a digest and could not fail for altered member bytes, a member deleted from the
# RECORD, or a smuggled unlisted member, while the module docstring said "fail-closed". The
# RECORD self-consistency check that subject_binding already builds on runs here too, so this
# record's verdict reflects the archive it describes. Failures are collected, not raised: past
# identification, a receipt naming the tampering is strictly more useful than a traceback.
problems: list[str] = []
record_hashed_entry_count: int | None = None
try:
record_hashed_entry_count = len(_wheel_record_hashes(path))
except Exception as exc: # noqa: BLE001 -- untrusted archive, same boundary as subject_binding
problems.append(f"wheel RECORD self-consistency failed: {type(exc).__name__}: {exc}")
if expected_commit is not None:
expected = _expected_hex(expected_commit, length=40, parameter="expected_commit")
if provenance["source_commit"] != expected:
problems.append(f"source_commit is {provenance['source_commit']}, expected {expected}")
if expected_content_sha256 is not None:
expected = _expected_hex(expected_content_sha256, length=64, parameter="expected_content_sha256")
if provenance["source_content_sha256"] != expected:
problems.append(f"source_content_sha256 is {provenance['source_content_sha256']}, expected {expected}")
return {
"artifact": "mixle.wheel_provenance/v1",
"filename": path.name,
"sha256": digest,
"size_bytes": size,
"version": installed_version,
"build": provenance,
# None when the RECORD itself failed, so a reader can tell "no entries verified" from
# "the count happened to be zero" (the latter is itself refused by _wheel_record_hashes).
"record_hashed_entry_count": record_hashed_entry_count,
"problems": problems,
"verified": not problems,
}
def _record_rows(text: str) -> list[tuple[str, str, str]]:
"""Split a RECORD into ``(path, hash, size)`` rows with the CSV semantics the wheel spec gives it.
RECORD is CSV -- PEP 376 specifies the ``csv`` module with ``,`` as delimiter and ``"`` as the
quote character, which is what installers read it with -- so a path containing a comma MUST
be quoted and any path MAY be. The previous parser used
``rsplit(",", 2)``: a wheel whose unchanged first path was legally quoted installed fine with
pip, but the resolver kept the quote characters as part of the filename and refused the wheel
as claiming a member that was not in the archive (SYS5-04). ``csv.reader`` over the raw text
(``newline=""`` so the reader, not the file layer, owns line ends) accepts LF and CRLF alike and
unquotes exactly as an installer does; ``strict=True`` turns malformed quoting into a refusal
with a precise message instead of a silently tolerated field. Blank lines are skipped, as
before; every other row must carry exactly three fields, or the RECORD is refused -- a
non-quoted comma is a four-field row, never a path.
"""
rows: list[tuple[str, str, str]] = []
try:
for index, row in enumerate(csv.reader(io.StringIO(text, newline=""), strict=True), start=1):
if not row or (len(row) == 1 and not row[0].strip()):
continue
if len(row) != 3:
raise ValueError(f"wheel RECORD row {index} has {len(row)} fields; expected exactly 3 (path,hash,size)")
rows.append((row[0], row[1], row[2]))
except csv.Error as exc:
raise ValueError(f"wheel RECORD is not well-formed CSV: {exc}") from exc
return rows
def _wheel_record_hashes(path: Path) -> dict[str, str]:
"""Parse the wheel's RECORD into ``{path: sha256-hex}`` for every hashed entry."""
with zipfile.ZipFile(path) as archive:
record_names = [name for name in archive.namelist() if name.endswith(".dist-info/RECORD")]
if len(record_names) != 1:
raise ValueError("wheel must contain exactly one RECORD")
text = archive.read(record_names[0]).decode("utf-8")
hashes: dict[str, str] = {}
unhashed: list[str] = []
for name, hash_field, _size in _record_rows(text):
if not hash_field:
# RECORD cannot hash itself; ANY other unhashed row is the tamper vector -- strip a
# file's hash and the comparison silently skipped it, so altered code installed and
# received a passing receipt (SYS-RR8-2). Collected, then rejected below.
unhashed.append(name)
continue
algorithm, _, value = hash_field.partition("=")
if algorithm != "sha256" or not value:
raise ValueError(f"wheel RECORD entry {name!r} uses unsupported hash {algorithm!r}")
if name in hashes:
raise ValueError(f"wheel RECORD lists {name!r} more than once")
try:
hashes[name] = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).hex()
except (ValueError, binascii.Error) as exc:
raise ValueError(f"wheel RECORD entry {name!r} has a malformed hash") from exc
if len(hashes[name]) != 64:
raise ValueError(f"wheel RECORD entry {name!r} has a malformed sha256 digest")
unexpected = [name for name in unhashed if not name.endswith(".dist-info/RECORD")]
if unexpected:
raise ValueError(f"wheel RECORD omits hashes for {len(unexpected)} entrie(s): {sorted(unexpected)[:5]}")
if len(unhashed) != 1:
raise ValueError("wheel RECORD must contain exactly one self-referencing unhashed row")
if not hashes:
raise ValueError("wheel RECORD carries no hashed entries")
# The RECORD is a set of CLAIMS the wheel makes about its own members. Returning them without
# checking them turned every caller into a consumer of unverified assertions: a wheel whose
# member bytes had been altered while its RECORD stayed stale still "declared" the original
# digests, and a binding that compared an installation against those declarations authorized
# bytes nobody had reviewed (SYS4-01). Every RECORD claim is now recomputed from the archived
# member it names, and -- the neighbouring case -- every archived member must be claimed by the
# RECORD, so an extra file smuggled into the archive is refused too. Only after both hold do
# the declared hashes mean anything, and only then are they returned.
with zipfile.ZipFile(path) as archive:
infos = [info for info in archive.infolist() if not info.is_dir()]
member_names = [info.filename for info in infos]
# A zip may legally hold two entries with one name, and ``ZipFile.read(name)`` returns one
# of them while an installer may extract the other -- so a duplicate lets a member exist
# in two versions, only one of which the RECORD check sees. Refuse the ambiguity outright.
seen: set[str] = set()
for name in member_names:
if name in seen:
raise ValueError(f"wheel archive holds member {name!r} more than once")
seen.add(name)
# A member or RECORD row that escapes the archive root (``..`` segments, an absolute path,
# a drive letter) is a path-traversal payload, not a package file. Refuse before hashing.
for name in list(member_names) + list(hashes):
parts = Path(name).parts
if name.startswith(("/", "\\")) or ".." in parts or (len(name) > 1 and name[1] == ":"):
raise ValueError(f"wheel member path escapes the archive root: {name!r}")
members = set(member_names)
for name, declared in hashes.items():
if name not in members:
raise ValueError(f"wheel RECORD claims {name!r}, which is not in the archive")
actual = hashlib.sha256(archive.read(name)).hexdigest()
if actual != declared:
raise ValueError(
f"wheel member {name!r} does not match its RECORD hash: archive has {actual}, RECORD claims {declared}"
)
unclaimed = members - set(hashes) - set(unhashed)
if unclaimed:
raise ValueError(
f"wheel archive holds {len(unclaimed)} member(s) the RECORD does not claim: {sorted(unclaimed)[:5]}"
)
return hashes
[docs]
def subject_binding(path: Path, build: dict[str, Any]) -> dict[str, Any]:
"""Bind the receipt's SUBJECT wheel to the distribution that is actually executing.
Name/version equality is not identity: any clean ``mixle==X`` wheel matches the installed
version string, so the adversarial review presented an older 0.8.0 wheel as the subject while
a newer 0.8.0 build executed, and the receipt reported ``passed`` for bytes that never ran
(SYS-RR7-3). Two comparisons close that: the installed package's embedded build provenance
must equal the subject wheel's (source commit, tree, and content digest), and every hashed
entry in the subject wheel's RECORD must appear in the installed RECORD with the same digest.
The installed RECORD may carry extra installer-written files (INSTALLER, direct_url.json);
the wheel's own entries are the code, and those are what must match.
"""
mismatches: list[str] = []
# Establish WHICH CODE RAN before trusting anything found beside it. A PYTHONPATH shadow
# package supplied its own __init__/reproduction modules while extending __path__ to the real
# installation, and produced a receipt byte-identical to the legitimate one (SYS-RR8-3):
# provenance was read next to the shadow while distribution() independently located the real
# metadata. Reading provenance first would let a copied-in provenance file decide the outcome,
# so the import-path identity is settled up front.
import mixle as _installed
from mixle import reproduction as _running
package_roots = [Path(p).resolve() for p in getattr(_installed, "__path__", [])]
try:
distribution_root = Path(distribution("mixle").locate_file("mixle")).resolve()
except PackageNotFoundError:
return {
"artifact": "mixle.subject_binding/v1",
"verified": False,
"mismatches": ["mixle distribution metadata is not installed"],
}
if len(package_roots) != 1:
mismatches.append(
f"imported mixle has {len(package_roots)} package roots; a release replay requires exactly one"
)
for root in package_roots:
if root != distribution_root:
mismatches.append(f"imported package root {root} is not the installed distribution at {distribution_root}")
running_file = Path(_running.__file__).resolve()
if not running_file.is_relative_to(distribution_root):
mismatches.append(f"executing module {running_file} is outside the installed distribution")
provenance_path = Path(_installed.__file__).resolve().parent / "_build_provenance.json"
try:
installed_provenance = json.loads(provenance_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
return {
"artifact": "mixle.subject_binding/v1",
"verified": False,
"mismatches": [*mismatches, f"installed build provenance unreadable: {exc}"],
}
for field in ("source_commit", "source_tree", "source_content_sha256"):
if installed_provenance.get(field) != build.get(field):
mismatches.append(
f"build provenance {field}: wheel={build.get(field)!r} installed={installed_provenance.get(field)!r}"
)
try:
wheel_hashes = _wheel_record_hashes(path)
except Exception as exc: # noqa: BLE001 -- untrusted archive, same boundary as wheel_provenance
# append rather than replace: the import-path mismatches found above are the more
# diagnostic half of a shadowed-install failure and must not be discarded here
return {
"artifact": "mixle.subject_binding/v1",
"verified": False,
"mismatches": [*mismatches, f"{type(exc).__name__}: {exc}"],
}
installed_hashes: dict[str, str] = {}
try:
dist = distribution("mixle")
except PackageNotFoundError:
mismatches.append("mixle distribution metadata is not installed")
else:
for item in dist.files or ():
if item.hash is not None and item.hash.mode == "sha256":
value = item.hash.value
installed_hashes[str(item)] = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).hex()
for name, digest in sorted(wheel_hashes.items()):
if installed_hashes.get(name) != digest:
mismatches.append(f"RECORD content differs at {name}")
if len(mismatches) >= 10:
mismatches.append("further RECORD mismatches suppressed")
break
return {
"artifact": "mixle.subject_binding/v1",
"verified": not mismatches,
"compared_entries": len(wheel_hashes),
# recorded so two receipts can be compared on WHERE the code ran, not only on its digests
"executing_module": str(running_file),
"package_roots": [str(root) for root in package_roots],
"distribution_root": str(distribution_root),
"mismatches": mismatches,
}
[docs]
def source_tree_provenance() -> dict[str, Any]:
"""Identify an explicitly requested development checkout by its own repository root."""
repository = Path(__file__).resolve().parents[1]
commit = _full_git_commit(repository)
return {
"artifact": "mixle.source_tree_provenance/v1",
"commit": commit,
"repository": str(repository),
"verified": commit != "unknown",
}
def environment() -> dict[str, Any]:
return {
"python": platform.python_version(),
"platform": platform.platform(),
"machine": platform.machine(),
"mixle": _pkg_version("mixle"),
"numpy": _pkg_version("numpy"),
"scipy": _pkg_version("scipy"),
"installed_content": installed_content_provenance(),
}
[docs]
def claim_checks() -> dict[str, Any]:
"""Run deterministic, seeded measurements backing the release claims."""
import numpy as np
import mixle.stats as st
from mixle.inference.estimation import optimize
from mixle.utils.automatic import get_estimator
from mixle.utils.serialization import ensure_pysp_serialization_registry, from_serializable, to_serializable
ensure_pysp_serialization_registry()
data = np.random.RandomState(0).normal(3.0, 2.0, 2000).tolist()
fitted = optimize(data, st.GaussianEstimator(), max_its=20, out=None)
g = st.GaussianDistribution(1.0, 2.0)
xs = list(g.sampler(seed=0).sample(32))
seq = np.asarray(g.seq_log_density(g.dist_to_encoder().seq_encode(xs)), dtype=float)
scalar = np.array([float(g.log_density(x)) for x in xs])
back = from_serializable(to_serializable(g))
return {
"gaussian_fit_mu": round(float(fitted.mu), 4),
"gaussian_fit_sigma": round(float(np.sqrt(fitted.sigma2)), 4),
"scalar_vectorized_agree": bool(np.allclose(seq, scalar, atol=1e-9)),
"serialization_score_equal": (round(float(back.log_density(0.5)), 6) == round(float(g.log_density(0.5)), 6)),
"auto_selects": type(get_estimator(np.random.RandomState(1).normal(0, 1, 1500).tolist())).__name__,
"deterministic_sample": round(float(st.GammaDistribution(2.0, 1.5).sampler(seed=7).sample(1)[0]), 6),
}
[docs]
def evaluate_claims(observed: object) -> dict[str, dict[str, Any]]:
"""Bind every observed measurement to a required value/tolerance and pass state."""
if not isinstance(observed, dict) or set(observed) != set(_EXPECTATIONS):
missing = sorted(set(_EXPECTATIONS) - set(observed) if isinstance(observed, dict) else _EXPECTATIONS)
extra = sorted(set(observed) - set(_EXPECTATIONS)) if isinstance(observed, dict) else []
raise ValueError(f"claim checks do not match the required protocol; missing={missing}, extra={extra}")
results = {}
for name, expectation in _EXPECTATIONS.items():
actual = observed[name]
expected = expectation["expected"]
tolerance = expectation.get("absolute_tolerance")
if tolerance is None:
passed = type(actual) is type(expected) and actual == expected
else:
passed = (
isinstance(actual, (int, float))
and not isinstance(actual, bool)
and abs(float(actual) - float(expected)) <= tolerance
)
results[name] = {
"observed": actual,
"expected": expected,
"absolute_tolerance": tolerance,
"passed": passed,
}
return results
[docs]
def build_receipt(*, wheel: Path | None, allow_source_tree: bool) -> tuple[dict[str, Any], bool]:
"""Build a versioned receipt and its overall fail-closed verdict."""
if wheel is not None and allow_source_tree:
raise ValueError("--wheel and --source-tree are mutually exclusive")
if wheel is not None:
artifact = wheel_provenance(wheel)
# the receipt's subject must be the wheel whose code EXECUTES, not merely one sharing its
# version string (SYS-RR7-3)
artifact["installed_binding"] = subject_binding(wheel, artifact["build"])
artifact["verified"] = bool(artifact["verified"]) and bool(artifact["installed_binding"]["verified"])
elif allow_source_tree:
artifact = source_tree_provenance()
else:
artifact = {
"artifact": "mixle.unverified_artifact/v1",
"verified": False,
"reason": "provide --wheel for release reproduction or --source-tree for development",
}
checks = evaluate_claims(claim_checks())
installed = environment()
passed = (
bool(artifact.get("verified"))
and bool(installed["installed_content"].get("verified"))
and all(check["passed"] for check in checks.values())
)
return (
{
"artifact": "mixle.reproduction_receipt/v2",
"passed": passed,
# The receipt names its own coverage so it cannot be presented as more than it is
# (SYS-RR7-4): the installed command runs the self-contained claim checks against the
# bound artifact; the reproduction-bundle entries are SOURCE-CHECKOUT receipts,
# produced separately, because the wheel ships neither the bundle nor its inputs.
"scope": {
"claim_checks": sorted(_EXPECTATIONS),
"reproduction_bundle_entries": (
"not included; replay each entry from the source checkout with "
"`python scripts/run_repro_entry.py --entry <id>` and attach those receipts"
),
},
"subject": artifact,
"environment": installed,
"checks": checks,
},
passed,
)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Reproduce Mixle's release claims and emit a fail-closed receipt.")
parser.add_argument("--wheel", type=Path, help="exact installed wheel to hash and verify")
parser.add_argument(
"--source-tree",
action="store_true",
help="development-only: bind the receipt to the checkout containing mixle.reproduction",
)
parser.add_argument("--out", type=Path, default=None, help="write the JSON receipt here (default: stdout)")
args = parser.parse_args(argv)
try:
receipt, passed = build_receipt(wheel=args.wheel, allow_source_tree=args.source_tree)
text = json.dumps(receipt, indent=2, sort_keys=True, allow_nan=False)
if args.out is not None:
args.out.write_text(text + "\n", encoding="utf-8")
print(f"wrote {args.out}")
else:
print(text)
except (OSError, TypeError, ValueError) as exc:
print(str(exc), file=sys.stderr)
return 2
return 0 if passed else 1
if __name__ == "__main__":
raise SystemExit(main())