398 lines
16 KiB
Python
398 lines
16 KiB
Python
"""Independent identities for immutable Python wheel release artifacts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
from email.parser import BytesParser
|
|
from email.policy import compat32
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import stat
|
|
import zipfile
|
|
|
|
|
|
WHEEL_PAYLOAD_ALGORITHM = "govoplan-wheel-declared-payload-v1"
|
|
INSTALLED_PAYLOAD_ALGORITHM = "govoplan-installed-record-payload-v1"
|
|
MAX_WHEEL_BYTES = 512 * 1024 * 1024
|
|
MAX_WHEEL_ENTRIES = 10_000
|
|
MAX_WHEEL_UNCOMPRESSED_BYTES = 1024 * 1024 * 1024
|
|
MAX_WHEEL_ENTRY_BYTES = 64 * 1024 * 1024
|
|
MAX_METADATA_BYTES = 1024 * 1024
|
|
_PACKAGE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
|
|
_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]{0,127}$")
|
|
_PYTHON_REF = re.compile(
|
|
r"/(?P<repo>govoplan-[a-z0-9-]+)[.]git@v(?P<version>[^\s;]+)$"
|
|
)
|
|
|
|
|
|
class ArtifactIdentityError(ValueError):
|
|
"""A built artifact cannot provide a bounded immutable identity."""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PythonWheelIdentity:
|
|
package_name: str
|
|
package_version: str
|
|
archive_sha256: str
|
|
archive_size: int
|
|
payload_sha256: str
|
|
payload_file_count: int
|
|
requires_installer_receipt: bool
|
|
|
|
def catalog_payload(self) -> dict[str, object]:
|
|
return {
|
|
"artifact_kind": "python-wheel",
|
|
"package_name": self.package_name,
|
|
"package_version": self.package_version,
|
|
"archive_sha256": self.archive_sha256,
|
|
"archive_size": self.archive_size,
|
|
"installed_payload": {
|
|
"algorithm": WHEEL_PAYLOAD_ALGORITHM,
|
|
"sha256": self.payload_sha256,
|
|
"file_count": self.payload_file_count,
|
|
},
|
|
"requires_installer_receipt": self.requires_installer_receipt,
|
|
}
|
|
|
|
|
|
def inspect_python_wheel(path: Path | str) -> PythonWheelIdentity:
|
|
"""Hash a regular built wheel and derive its install-stable payload identity."""
|
|
|
|
wheel = Path(path).expanduser()
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(wheel, flags)
|
|
except OSError as exc:
|
|
raise ArtifactIdentityError("built artifact cannot be opened safely") from exc
|
|
try:
|
|
archive_sha256, archive_size, initial = _hash_open_artifact(descriptor)
|
|
os.lseek(descriptor, 0, os.SEEK_SET)
|
|
try:
|
|
with os.fdopen(os.dup(descriptor), "rb") as artifact_handle:
|
|
with zipfile.ZipFile(artifact_handle) as archive:
|
|
(
|
|
package_name,
|
|
package_version,
|
|
payload_rows,
|
|
requires_receipt,
|
|
) = _inspect_wheel_archive(archive)
|
|
except (OSError, zipfile.BadZipFile, RuntimeError) as exc:
|
|
raise ArtifactIdentityError(
|
|
"built Python artifact is not a readable wheel"
|
|
) from exc
|
|
final = os.fstat(descriptor)
|
|
if _stat_fingerprint(final) != _stat_fingerprint(initial):
|
|
raise ArtifactIdentityError("built artifact changed while it was inspected")
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
return PythonWheelIdentity(
|
|
package_name=package_name,
|
|
package_version=package_version,
|
|
archive_sha256=archive_sha256,
|
|
archive_size=archive_size,
|
|
payload_sha256=payload_identity_sha256(
|
|
algorithm=WHEEL_PAYLOAD_ALGORITHM, rows=payload_rows
|
|
),
|
|
payload_file_count=len(payload_rows),
|
|
requires_installer_receipt=requires_receipt,
|
|
)
|
|
|
|
|
|
def _inspect_wheel_archive(
|
|
archive: zipfile.ZipFile,
|
|
) -> tuple[str, str, list[dict[str, object]], bool]:
|
|
infos = archive.infolist()
|
|
if not infos or len(infos) > MAX_WHEEL_ENTRIES:
|
|
raise ArtifactIdentityError("wheel entry count is empty or exceeds its limit")
|
|
_validate_members(infos)
|
|
metadata_members = [
|
|
item
|
|
for item in infos
|
|
if not item.is_dir()
|
|
and PurePosixPath(item.filename).name == "METADATA"
|
|
and PurePosixPath(item.filename).parent.name.endswith(".dist-info")
|
|
]
|
|
if len(metadata_members) != 1:
|
|
raise ArtifactIdentityError(
|
|
"wheel must contain exactly one dist-info METADATA file"
|
|
)
|
|
metadata_member = metadata_members[0]
|
|
dist_info = PurePosixPath(metadata_member.filename).parent
|
|
metadata_bytes = _read_member(
|
|
archive, metadata_member, max_bytes=MAX_METADATA_BYTES
|
|
)
|
|
parsed = BytesParser(policy=compat32).parsebytes(metadata_bytes)
|
|
package_name = normalize_package_name(str(parsed.get("Name") or ""))
|
|
package_version = str(parsed.get("Version") or "").strip()
|
|
if _PACKAGE.fullmatch(package_name) is None:
|
|
raise ArtifactIdentityError("wheel METADATA has an invalid package name")
|
|
if _VERSION.fullmatch(package_version) is None:
|
|
raise ArtifactIdentityError("wheel METADATA has an invalid package version")
|
|
|
|
payload_rows: list[dict[str, object]] = []
|
|
requires_receipt = False
|
|
for info in infos:
|
|
if info.is_dir():
|
|
continue
|
|
normalized, transformed = _wheel_payload_path(
|
|
PurePosixPath(info.filename), dist_info=dist_info
|
|
)
|
|
requires_receipt = requires_receipt or transformed
|
|
if normalized is None:
|
|
continue
|
|
encoded = _read_member(archive, info, max_bytes=MAX_WHEEL_ENTRY_BYTES)
|
|
payload_rows.append(
|
|
{
|
|
"path": normalized,
|
|
"sha256": hashlib.sha256(encoded).hexdigest(),
|
|
"size": len(encoded),
|
|
}
|
|
)
|
|
payload_rows.sort(key=lambda item: str(item["path"]))
|
|
if not payload_rows:
|
|
raise ArtifactIdentityError("wheel has no immutable payload entries")
|
|
if len({str(item["path"]) for item in payload_rows}) != len(payload_rows):
|
|
raise ArtifactIdentityError(
|
|
"wheel payload paths are ambiguous after installation mapping"
|
|
)
|
|
entry_points = f"{dist_info.as_posix()}/entry_points.txt"
|
|
if any(item.filename == entry_points for item in infos):
|
|
entry_point_info = next(item for item in infos if item.filename == entry_points)
|
|
entry_point_text = _read_member(
|
|
archive, entry_point_info, max_bytes=MAX_METADATA_BYTES
|
|
).decode("utf-8", errors="replace")
|
|
if re.search(r"(?m)^\s*\[(?:console|gui)_scripts\]\s*$", entry_point_text):
|
|
requires_receipt = True
|
|
return package_name, package_version, payload_rows, requires_receipt
|
|
|
|
|
|
def selected_artifact_identity_issues(payload: object) -> tuple[str, ...]:
|
|
"""Report selected Python releases without a matching built-wheel identity."""
|
|
|
|
if not isinstance(payload, dict):
|
|
return ("candidate catalog must be a JSON object",)
|
|
release = payload.get("release")
|
|
selected_units = release.get("selected_units") if isinstance(release, dict) else None
|
|
artifacts = release.get("artifacts") if isinstance(release, dict) else None
|
|
if not isinstance(selected_units, list) or not selected_units:
|
|
return ("release.selected_units is missing or empty",)
|
|
entries: list[object] = [payload.get("core_release")]
|
|
modules = payload.get("modules")
|
|
if isinstance(modules, list):
|
|
entries.extend(modules)
|
|
package_by_repo: dict[str, tuple[str, str]] = {}
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
python_ref = entry.get("python_ref")
|
|
match = _PYTHON_REF.search(python_ref) if isinstance(python_ref, str) else None
|
|
package_name = entry.get("python_package")
|
|
version = entry.get("version")
|
|
if (
|
|
match is not None
|
|
and isinstance(package_name, str)
|
|
and _PACKAGE.fullmatch(package_name) is not None
|
|
and isinstance(version, str)
|
|
):
|
|
package_by_repo[match.group("repo")] = (package_name, version)
|
|
artifacts_by_package: dict[str, dict[str, object]] = {}
|
|
malformed_artifacts = False
|
|
if not isinstance(artifacts, list):
|
|
artifacts = []
|
|
for artifact in artifacts:
|
|
if not isinstance(artifact, dict):
|
|
malformed_artifacts = True
|
|
continue
|
|
package_name = artifact.get("package_name")
|
|
if not isinstance(package_name, str) or package_name in artifacts_by_package:
|
|
malformed_artifacts = True
|
|
continue
|
|
if not _catalog_artifact_shape_valid(artifact):
|
|
malformed_artifacts = True
|
|
continue
|
|
artifacts_by_package[package_name] = artifact
|
|
issues: list[str] = []
|
|
if malformed_artifacts:
|
|
issues.append("release.artifacts contains malformed or duplicate identities")
|
|
seen_repos: set[str] = set()
|
|
for unit in selected_units:
|
|
if not isinstance(unit, dict):
|
|
issues.append("release.selected_units contains a malformed entry")
|
|
continue
|
|
repo = unit.get("repo")
|
|
if not isinstance(repo, str) or repo in seen_repos:
|
|
issues.append("release.selected_units contains a duplicate or invalid repository")
|
|
continue
|
|
seen_repos.add(repo)
|
|
expected = package_by_repo.get(repo)
|
|
if expected is None:
|
|
continue # Selected non-Python repositories have no wheel identity.
|
|
package_name, version = expected
|
|
artifact = artifacts_by_package.get(package_name)
|
|
if artifact is None:
|
|
issues.append(
|
|
f"selected Python repository {repo} has no built artifact identity for {package_name} {version}"
|
|
)
|
|
elif artifact.get("package_version") != version:
|
|
issues.append(
|
|
f"selected Python repository {repo} artifact identity has version {artifact.get('package_version')!r}, expected {version!r}"
|
|
)
|
|
return tuple(issues)
|
|
|
|
|
|
def _catalog_artifact_shape_valid(value: dict[str, object]) -> bool:
|
|
if set(value) != {
|
|
"artifact_kind",
|
|
"package_name",
|
|
"package_version",
|
|
"archive_sha256",
|
|
"archive_size",
|
|
"installed_payload",
|
|
"requires_installer_receipt",
|
|
}:
|
|
return False
|
|
package_name = value.get("package_name")
|
|
version = value.get("package_version")
|
|
archive_sha256 = value.get("archive_sha256")
|
|
archive_size = value.get("archive_size")
|
|
installed_payload = value.get("installed_payload")
|
|
return (
|
|
value.get("artifact_kind") == "python-wheel"
|
|
and isinstance(package_name, str)
|
|
and _PACKAGE.fullmatch(package_name) is not None
|
|
and isinstance(version, str)
|
|
and _VERSION.fullmatch(version) is not None
|
|
and isinstance(archive_sha256, str)
|
|
and re.fullmatch(r"[0-9a-f]{64}", archive_sha256) is not None
|
|
and isinstance(archive_size, int)
|
|
and not isinstance(archive_size, bool)
|
|
and 0 < archive_size <= MAX_WHEEL_BYTES
|
|
and isinstance(value.get("requires_installer_receipt"), bool)
|
|
and isinstance(installed_payload, dict)
|
|
and set(installed_payload) == {"algorithm", "sha256", "file_count"}
|
|
and installed_payload.get("algorithm") == WHEEL_PAYLOAD_ALGORITHM
|
|
and isinstance(installed_payload.get("sha256"), str)
|
|
and re.fullmatch(r"[0-9a-f]{64}", str(installed_payload["sha256"]))
|
|
is not None
|
|
and isinstance(installed_payload.get("file_count"), int)
|
|
and not isinstance(installed_payload.get("file_count"), bool)
|
|
and 0 < int(installed_payload["file_count"]) <= MAX_WHEEL_ENTRIES
|
|
)
|
|
|
|
|
|
def payload_identity_sha256(
|
|
*, algorithm: str, rows: list[dict[str, object]]
|
|
) -> str:
|
|
payload = {"algorithm": algorithm, "files": rows}
|
|
encoded = json.dumps(
|
|
payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def record_hash_hex(value: str) -> str | None:
|
|
try:
|
|
decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
|
|
except (ValueError, TypeError):
|
|
return None
|
|
return decoded.hex() if len(decoded) == hashlib.sha256().digest_size else None
|
|
|
|
|
|
def normalize_package_name(value: str) -> str:
|
|
return re.sub(r"[-_.]+", "-", value.strip().lower())
|
|
|
|
|
|
def _hash_open_artifact(descriptor: int) -> tuple[str, int, os.stat_result]:
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
initial = os.fstat(descriptor)
|
|
if not stat.S_ISREG(initial.st_mode) or initial.st_size > MAX_WHEEL_BYTES:
|
|
raise ArtifactIdentityError("built artifact must be a bounded regular file")
|
|
while total < initial.st_size:
|
|
chunk = os.read(descriptor, min(1024 * 1024, initial.st_size - total))
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
digest.update(chunk)
|
|
final = os.fstat(descriptor)
|
|
if total != initial.st_size or _stat_fingerprint(final) != _stat_fingerprint(initial):
|
|
raise ArtifactIdentityError("built artifact changed while it was hashed")
|
|
return digest.hexdigest(), total, initial
|
|
|
|
|
|
def _stat_fingerprint(value: os.stat_result) -> tuple[int, int, int, int, int]:
|
|
return (
|
|
value.st_dev,
|
|
value.st_ino,
|
|
value.st_size,
|
|
value.st_mtime_ns,
|
|
value.st_ctime_ns,
|
|
)
|
|
|
|
|
|
def _validate_members(infos: list[zipfile.ZipInfo]) -> None:
|
|
names: set[str] = set()
|
|
total = 0
|
|
for info in infos:
|
|
path = PurePosixPath(info.filename.replace("\\", "/"))
|
|
if (
|
|
not info.filename
|
|
or path.is_absolute()
|
|
or ".." in path.parts
|
|
or any(ord(character) < 32 for character in info.filename)
|
|
or info.flag_bits & 0x1
|
|
):
|
|
raise ArtifactIdentityError("wheel contains an unsafe member")
|
|
normalized = path.as_posix()
|
|
if normalized in names:
|
|
raise ArtifactIdentityError("wheel contains duplicate members")
|
|
names.add(normalized)
|
|
if _zip_member_is_symlink(info):
|
|
raise ArtifactIdentityError("wheel contains a symbolic-link member")
|
|
total += info.file_size
|
|
if info.file_size > MAX_WHEEL_ENTRY_BYTES or total > MAX_WHEEL_UNCOMPRESSED_BYTES:
|
|
raise ArtifactIdentityError("wheel uncompressed payload exceeds its limit")
|
|
|
|
|
|
def _zip_member_is_symlink(info: zipfile.ZipInfo) -> bool:
|
|
return stat.S_ISLNK((info.external_attr >> 16) & 0o177777)
|
|
|
|
|
|
def _read_member(
|
|
archive: zipfile.ZipFile, info: zipfile.ZipInfo, *, max_bytes: int
|
|
) -> bytes:
|
|
if info.file_size > max_bytes:
|
|
raise ArtifactIdentityError("wheel member exceeds its size limit")
|
|
with archive.open(info, "r") as handle:
|
|
encoded = handle.read(max_bytes + 1)
|
|
if len(encoded) > max_bytes or len(encoded) != info.file_size:
|
|
raise ArtifactIdentityError("wheel member size is inconsistent")
|
|
return encoded
|
|
|
|
|
|
def _wheel_payload_path(
|
|
path: PurePosixPath, *, dist_info: PurePosixPath
|
|
) -> tuple[str | None, bool]:
|
|
if path.parent == dist_info and path.name in {"RECORD", "RECORD.jws", "RECORD.p7s"}:
|
|
return None, False
|
|
parts = path.parts
|
|
data_prefix = dist_info.name.removesuffix(".dist-info") + ".data"
|
|
if parts and parts[0] == data_prefix:
|
|
if len(parts) < 3:
|
|
raise ArtifactIdentityError("wheel data member has no installation target")
|
|
scheme = parts[1]
|
|
remainder = PurePosixPath(*parts[2:]).as_posix()
|
|
if scheme in {"purelib", "platlib"}:
|
|
return remainder, False
|
|
if scheme == "data":
|
|
return f"@data/{remainder}", False
|
|
if scheme in {"scripts", "headers"}:
|
|
return None, True
|
|
raise ArtifactIdentityError("wheel uses an unsupported installation scheme")
|
|
return path.as_posix(), False
|