mirror of
https://github.com/herdrdev/herdr.git
synced 2026-10-07 16:00:41 +00:00
492 lines
20 KiB
Python
492 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
import zipfile
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
DEFAULT_METADATA = PROJECT_ROOT / "packaging" / "windows" / "conpty.json"
|
|
MARKER_PATH = PurePosixPath("conpty/herdr-conpty.json")
|
|
DOWNLOAD_TIMEOUT_SECONDS = 60
|
|
|
|
|
|
def sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def load_metadata(path: Path) -> dict[str, Any]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if data.get("schema_version") != 1:
|
|
raise ValueError("unsupported ConPTY metadata schema")
|
|
return data
|
|
|
|
|
|
def pe_machine(data: bytes) -> int:
|
|
if len(data) < 0x40 or data[:2] != b"MZ":
|
|
raise ValueError("file is not a PE image")
|
|
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
|
if pe_offset + 6 > len(data) or data[pe_offset : pe_offset + 4] != b"PE\0\0":
|
|
raise ValueError("file has an invalid PE header")
|
|
return struct.unpack_from("<H", data, pe_offset + 4)[0]
|
|
|
|
|
|
# Imported DLL name prefixes that identify the dynamic Microsoft C/C++
|
|
# runtime. Matching by prefix covers release and debug variants
|
|
# (msvcp140d.dll, vcruntime140_1.dll, ucrtbased.dll, ...), older versioned
|
|
# runtimes (msvcr120.dll, msvcp120.dll), and the api-ms-win-crt-* Universal
|
|
# CRT API sets. Shipping any of them would make the archive depend on a Visual
|
|
# C++ Redistributable or on the Universal CRT that is not part of the
|
|
# documented install contents.
|
|
DYNAMIC_MSVC_RUNTIME_PREFIXES = (
|
|
"vcruntime140",
|
|
"msvcp",
|
|
"msvcr",
|
|
"concrt140",
|
|
"ucrtbase",
|
|
"api-ms-win-crt-",
|
|
)
|
|
|
|
# msvcrt.dll is the legacy CRT that ships with Windows and is always present,
|
|
# so depending on it does not require the Visual C++ Redistributable.
|
|
SYSTEM_CRT_DLLS = frozenset({"msvcrt.dll"})
|
|
|
|
|
|
def is_dynamic_msvc_runtime(dll: str) -> bool:
|
|
"""Return True when a PE import names a dynamic MSVC/UCRT runtime DLL."""
|
|
lowered = dll.lower()
|
|
if lowered in SYSTEM_CRT_DLLS:
|
|
return False
|
|
return lowered.startswith(DYNAMIC_MSVC_RUNTIME_PREFIXES)
|
|
|
|
|
|
def _pe_sections(data: bytes) -> tuple[list[tuple[int, int, int]], int, int]:
|
|
"""Return PE sections, the data-directory offset, and the image base."""
|
|
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
|
number_of_sections = struct.unpack_from("<H", data, pe_offset + 6)[0]
|
|
optional_size = struct.unpack_from("<H", data, pe_offset + 20)[0]
|
|
optional_offset = pe_offset + 24
|
|
if optional_offset + optional_size > len(data):
|
|
raise ValueError("file has a truncated PE optional header")
|
|
magic = struct.unpack_from("<H", data, optional_offset)[0]
|
|
if magic == 0x20B:
|
|
data_directory_offset = optional_offset + 112
|
|
image_base = struct.unpack_from("<Q", data, optional_offset + 24)[0]
|
|
elif magic == 0x10B:
|
|
data_directory_offset = optional_offset + 96
|
|
image_base = struct.unpack_from("<I", data, optional_offset + 28)[0]
|
|
else:
|
|
raise ValueError(f"unsupported PE optional header magic: 0x{magic:04x}")
|
|
|
|
sections: list[tuple[int, int, int]] = []
|
|
section_offset = optional_offset + optional_size
|
|
for index in range(number_of_sections):
|
|
header = section_offset + index * 40
|
|
if header + 40 > len(data):
|
|
raise ValueError("file has truncated PE section headers")
|
|
virtual_size, virtual_address, raw_size, raw_pointer = struct.unpack_from(
|
|
"<IIII", data, header + 8
|
|
)
|
|
sections.append((virtual_address, max(virtual_size, raw_size), raw_pointer))
|
|
return sections, data_directory_offset, image_base
|
|
|
|
|
|
def pe_imported_dlls(data: bytes) -> list[str]:
|
|
"""Return the DLL names from the PE import and delay-import tables."""
|
|
if len(data) < 0x40 or data[:2] != b"MZ":
|
|
raise ValueError("file is not a PE image")
|
|
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
|
if pe_offset + 24 > len(data) or data[pe_offset : pe_offset + 4] != b"PE\0\0":
|
|
raise ValueError("file has an invalid PE header")
|
|
|
|
sections, data_directory_offset, image_base = _pe_sections(data)
|
|
|
|
def rva_to_offset(rva: int) -> int | None:
|
|
for virtual_address, virtual_size, raw_pointer in sections:
|
|
if virtual_address <= rva < virtual_address + virtual_size:
|
|
return raw_pointer + (rva - virtual_address)
|
|
return None
|
|
|
|
def read_c_string(offset: int | None) -> str:
|
|
if offset is None or offset < 0 or offset >= len(data):
|
|
raise ValueError("PE import name points outside the file")
|
|
end = data.find(b"\0", offset)
|
|
if end == -1:
|
|
raise ValueError("PE import name is not NUL-terminated")
|
|
return data[offset:end].decode("ascii", errors="replace")
|
|
|
|
# Directory index 1 is the import table, index 13 is the delay-import
|
|
# table. Delay-import descriptors store addresses as VAs unless their
|
|
# attributes set the RVA flag (bit 1), so normalize those to RVAs.
|
|
names: list[str] = []
|
|
for directory_index, descriptor_size, name_field, has_attributes in (
|
|
(1, 20, 12, False),
|
|
(13, 32, 4, True),
|
|
):
|
|
entry_offset = data_directory_offset + directory_index * 8
|
|
if entry_offset + 8 > len(data):
|
|
continue
|
|
directory_rva, directory_size = struct.unpack_from("<II", data, entry_offset)
|
|
if directory_rva == 0:
|
|
continue
|
|
directory_offset = rva_to_offset(directory_rva)
|
|
if directory_offset is None:
|
|
continue
|
|
# Bound each walk by the directory's declared size so bytes past the
|
|
# terminator are not parsed as descriptors. Fall back to the file end
|
|
# if a linker omits the size.
|
|
directory_end = (
|
|
min(directory_offset + directory_size, len(data))
|
|
if directory_size
|
|
else len(data)
|
|
)
|
|
position = directory_offset
|
|
while position + descriptor_size <= directory_end:
|
|
if data[position : position + descriptor_size] == b"\0" * descriptor_size:
|
|
break
|
|
name_value = struct.unpack_from("<I", data, position + name_field)[0]
|
|
if name_value == 0:
|
|
break
|
|
if has_attributes and not struct.unpack_from("<I", data, position)[0] & 1:
|
|
name_value -= image_base
|
|
names.append(read_c_string(rva_to_offset(name_value)))
|
|
position += descriptor_size
|
|
return names
|
|
|
|
|
|
def validate_static_msvc_runtime(data: bytes, label: str) -> None:
|
|
"""Reject an executable that depends on the dynamic Microsoft C runtime."""
|
|
offending = sorted(
|
|
{dll for dll in pe_imported_dlls(data) if is_dynamic_msvc_runtime(dll)}
|
|
)
|
|
if offending:
|
|
raise ValueError(
|
|
f"{label} depends on the dynamic Microsoft C/C++ runtime "
|
|
f"({', '.join(offending)}); the Windows archive must link the CRT "
|
|
"statically so it runs on a clean Windows install"
|
|
)
|
|
|
|
|
|
def validate_nuspec(archive: zipfile.ZipFile, package: dict[str, Any]) -> None:
|
|
nuspec_name = f"{package['id']}.nuspec"
|
|
try:
|
|
root = ET.fromstring(archive.read(nuspec_name).decode("utf-8-sig"))
|
|
except KeyError as error:
|
|
raise ValueError(f"package is missing {nuspec_name}") from error
|
|
namespace = {"n": "http://schemas.microsoft.com/packaging/2011/08/nuspec.xsd"}
|
|
metadata = root.find("n:metadata", namespace)
|
|
if metadata is None:
|
|
raise ValueError("package NuSpec is missing metadata")
|
|
expected = {
|
|
"id": package["id"],
|
|
"version": package["version"],
|
|
"license": package["license"],
|
|
}
|
|
for field, expected_value in expected.items():
|
|
element = metadata.find(f"n:{field}", namespace)
|
|
actual = element.text.strip() if element is not None and element.text else None
|
|
if actual != expected_value:
|
|
raise ValueError(
|
|
f"package NuSpec {field} mismatch: expected {expected_value!r}, got {actual!r}"
|
|
)
|
|
|
|
|
|
def acquire_package(package: dict[str, Any], package_path: Path) -> None:
|
|
package_path.parent.mkdir(parents=True, exist_ok=True)
|
|
if not package_path.exists():
|
|
for attempt in range(3):
|
|
try:
|
|
with urllib.request.urlopen(
|
|
package["url"], timeout=DOWNLOAD_TIMEOUT_SECONDS
|
|
) as response, package_path.open("wb") as output:
|
|
shutil.copyfileobj(response, output)
|
|
break
|
|
except urllib.error.HTTPError as error:
|
|
if error.code < 500 or attempt == 2:
|
|
raise
|
|
error.close()
|
|
time.sleep(2**attempt)
|
|
actual = sha256_file(package_path)
|
|
if actual != package["sha256"]:
|
|
raise ValueError(
|
|
f"ConPTY package hash mismatch: expected {package['sha256']}, got {actual}"
|
|
)
|
|
|
|
|
|
def marker_data(metadata: dict[str, Any], architecture: str) -> bytes:
|
|
bundle = metadata["bundles"][architecture]
|
|
marker = {
|
|
"schema_version": 1,
|
|
"package": metadata["package"]["id"],
|
|
"version": metadata["package"]["version"],
|
|
"architecture": architecture,
|
|
"files": {
|
|
item["destination"]: item["sha256"] for item in bundle["files"]
|
|
},
|
|
}
|
|
return (json.dumps(marker, indent=2, sort_keys=True) + "\n").encode()
|
|
|
|
|
|
def stage_bundle(
|
|
metadata_path: Path,
|
|
architecture: str,
|
|
package_path: Path,
|
|
herdr_exe: Path,
|
|
output_dir: Path,
|
|
) -> None:
|
|
metadata = load_metadata(metadata_path)
|
|
if architecture not in metadata["bundles"]:
|
|
raise ValueError(f"unsupported Windows architecture: {architecture}")
|
|
if output_dir.exists():
|
|
raise ValueError(f"output directory already exists: {output_dir}")
|
|
if not herdr_exe.is_file():
|
|
raise ValueError(f"Herdr executable does not exist: {herdr_exe}")
|
|
|
|
validate_static_msvc_runtime(herdr_exe.read_bytes(), herdr_exe.name)
|
|
|
|
acquire_package(metadata["package"], package_path)
|
|
bundle = metadata["bundles"][architecture]
|
|
metadata_root = metadata_path.resolve().parent
|
|
|
|
with zipfile.ZipFile(package_path) as archive, tempfile.TemporaryDirectory(
|
|
prefix="herdr-conpty-stage-", dir=output_dir.parent
|
|
) as temporary:
|
|
validate_nuspec(archive, metadata["package"])
|
|
staging = Path(temporary) / "bundle"
|
|
staging.mkdir()
|
|
shutil.copy2(herdr_exe, staging / "herdr.exe")
|
|
|
|
for item in bundle["files"]:
|
|
try:
|
|
payload = archive.read(item["source"])
|
|
except KeyError as error:
|
|
raise ValueError(f"package is missing {item['source']}") from error
|
|
actual_hash = sha256_bytes(payload)
|
|
if actual_hash != item["sha256"]:
|
|
raise ValueError(
|
|
f"hash mismatch for {item['source']}: expected {item['sha256']}, got {actual_hash}"
|
|
)
|
|
actual_machine = pe_machine(payload)
|
|
expected_machine = int(item["pe_machine"], 16)
|
|
if actual_machine != expected_machine:
|
|
raise ValueError(
|
|
f"PE machine mismatch for {item['source']}: "
|
|
f"expected {item['pe_machine']}, got 0x{actual_machine:04x}"
|
|
)
|
|
destination = staging / PurePosixPath(item["destination"])
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_bytes(payload)
|
|
|
|
marker = staging / MARKER_PATH
|
|
marker.parent.mkdir(parents=True, exist_ok=True)
|
|
marker.write_bytes(marker_data(metadata, architecture))
|
|
|
|
for notice in metadata["notices"]:
|
|
source = (metadata_root.parent.parent / notice["source"]).resolve()
|
|
actual_hash = sha256_file(source)
|
|
if actual_hash != notice["sha256"]:
|
|
raise ValueError(
|
|
f"notice hash mismatch for {source}: expected {notice['sha256']}, got {actual_hash}"
|
|
)
|
|
destination = staging / PurePosixPath(notice["destination"])
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source, destination)
|
|
|
|
staging.rename(output_dir)
|
|
|
|
|
|
def install_local_bundle(
|
|
metadata_path: Path, architecture: str, package_path: Path, herdr_exe: Path
|
|
) -> None:
|
|
release_dir = herdr_exe.resolve().parent
|
|
destination = release_dir / "conpty"
|
|
with tempfile.TemporaryDirectory(prefix="herdr-conpty-install-", dir=release_dir) as temporary:
|
|
stage = Path(temporary) / "stage"
|
|
stage_bundle(metadata_path, architecture, package_path, herdr_exe, stage)
|
|
staged_bundle = stage / "conpty"
|
|
if destination.exists() or destination.is_symlink():
|
|
staged_files = {
|
|
path.relative_to(staged_bundle)
|
|
for path in staged_bundle.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
installed_files = {
|
|
path.relative_to(destination)
|
|
for path in destination.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
matches = (
|
|
destination.is_dir()
|
|
and not destination.is_symlink()
|
|
and not any(path.is_symlink() for path in destination.rglob("*"))
|
|
and installed_files == staged_files
|
|
and all(
|
|
not (destination / relative).is_symlink()
|
|
and sha256_file(destination / relative) == sha256_file(staged_bundle / relative)
|
|
for relative in staged_files
|
|
)
|
|
)
|
|
# ponytail: pin upgrades require stopping Herdr and removing the old local bundle.
|
|
if not matches:
|
|
raise ValueError(
|
|
f"local ConPTY bundle at {destination} differs from the pinned version; "
|
|
"stop Herdr, remove that directory, then run just build again"
|
|
)
|
|
|
|
for notice in load_metadata(metadata_path)["notices"]:
|
|
relative = PurePosixPath(notice["destination"])
|
|
target = release_dir / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(stage / relative, target)
|
|
if not destination.exists():
|
|
staged_bundle.rename(destination)
|
|
|
|
|
|
def build_local_bundle(metadata_path: Path) -> None:
|
|
executable = None
|
|
with subprocess.Popen(
|
|
["cargo", "build", "--release", "--locked", "--message-format=json"],
|
|
cwd=PROJECT_ROOT,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
encoding="utf-8",
|
|
) as cargo:
|
|
assert cargo.stdout is not None
|
|
for line in cargo.stdout:
|
|
if not line.startswith("{"):
|
|
print(line, end="")
|
|
continue
|
|
message = json.loads(line)
|
|
if message.get("reason") == "compiler-message":
|
|
print(message["message"]["rendered"], end="", file=sys.stderr)
|
|
elif (
|
|
message.get("reason") == "compiler-artifact"
|
|
and message["manifest_path"] == str(PROJECT_ROOT / "Cargo.toml")
|
|
and message["target"]["name"] == "herdr"
|
|
and "bin" in message["target"]["kind"]
|
|
):
|
|
executable = Path(message["executable"])
|
|
status = cargo.wait()
|
|
if status:
|
|
raise SystemExit(status)
|
|
if executable is None:
|
|
raise ValueError("cargo did not report the Herdr executable")
|
|
|
|
target = Path(os.environ.get("CARGO_TARGET_DIR", "target"))
|
|
if not target.is_absolute():
|
|
target = PROJECT_ROOT / target
|
|
install_local_bundle(
|
|
metadata_path,
|
|
"x86_64",
|
|
target / "conpty" / "Microsoft.Windows.Console.ConPTY.nupkg",
|
|
executable,
|
|
)
|
|
|
|
|
|
def expected_stage_files(metadata: dict[str, Any], architecture: str) -> set[str]:
|
|
files = {"herdr.exe", MARKER_PATH.as_posix()}
|
|
files.update(item["destination"] for item in metadata["bundles"][architecture]["files"])
|
|
files.update(item["destination"] for item in metadata["notices"])
|
|
return files
|
|
|
|
|
|
def validate_stage(metadata_path: Path, architecture: str, stage_dir: Path) -> None:
|
|
metadata = load_metadata(metadata_path)
|
|
actual = {
|
|
path.relative_to(stage_dir).as_posix()
|
|
for path in stage_dir.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
expected = expected_stage_files(metadata, architecture)
|
|
if actual != expected:
|
|
raise ValueError(
|
|
f"bundle layout mismatch; missing={sorted(expected - actual)}, unexpected={sorted(actual - expected)}"
|
|
)
|
|
if (stage_dir / MARKER_PATH).read_bytes() != marker_data(metadata, architecture):
|
|
raise ValueError("bundle marker does not match pinned ConPTY metadata")
|
|
# The executable is not hash-pinned (it changes every build), so re-check
|
|
# it here to cover a direct archive of an existing stage or a swap after
|
|
# staging.
|
|
validate_static_msvc_runtime((stage_dir / "herdr.exe").read_bytes(), "herdr.exe")
|
|
for item in metadata["bundles"][architecture]["files"]:
|
|
path = stage_dir / PurePosixPath(item["destination"])
|
|
actual_hash = sha256_file(path)
|
|
if actual_hash != item["sha256"]:
|
|
raise ValueError(f"staged file hash mismatch for {path}")
|
|
for notice in metadata["notices"]:
|
|
path = stage_dir / PurePosixPath(notice["destination"])
|
|
actual_hash = sha256_file(path)
|
|
if actual_hash != notice["sha256"]:
|
|
raise ValueError(f"staged notice hash mismatch for {path}")
|
|
|
|
|
|
def archive_bundle(
|
|
metadata_path: Path, architecture: str, stage_dir: Path, output_path: Path
|
|
) -> None:
|
|
validate_stage(metadata_path, architecture, stage_dir)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
|
|
for path in sorted(stage_dir.rglob("*")):
|
|
if path.is_file():
|
|
archive.write(path, path.relative_to(stage_dir).as_posix())
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Package Herdr with Microsoft's pinned ConPTY runtime")
|
|
parser.add_argument("--metadata", type=Path, default=DEFAULT_METADATA)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
stage = subparsers.add_parser("stage")
|
|
stage.add_argument("--architecture", choices=("x86_64",), default="x86_64")
|
|
stage.add_argument("--package", type=Path, required=True)
|
|
stage.add_argument("--herdr-exe", type=Path, required=True)
|
|
stage.add_argument("--output-dir", type=Path, required=True)
|
|
|
|
archive = subparsers.add_parser("archive")
|
|
archive.add_argument("--architecture", choices=("x86_64",), default="x86_64")
|
|
archive.add_argument("--stage-dir", type=Path, required=True)
|
|
archive.add_argument("--output", type=Path, required=True)
|
|
subparsers.add_parser("build-local")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
if args.command == "stage":
|
|
stage_bundle(
|
|
args.metadata,
|
|
args.architecture,
|
|
args.package,
|
|
args.herdr_exe,
|
|
args.output_dir,
|
|
)
|
|
elif args.command == "archive":
|
|
archive_bundle(args.metadata, args.architecture, args.stage_dir, args.output)
|
|
else:
|
|
build_local_bundle(args.metadata)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|