Files
herdr/scripts/test_package_windows_conpty.py
T
JJ Liebig f70a1f835e fix(windows): reject dynamic msvc runtime in packaging (#4200)
The v0.8.2 Windows archive shipped an herdr.exe that imported
VCRUNTIME140.dll and the api-ms-win-crt-* Universal CRT, none of which the
archive provides. Static CRT linkage was restored for x86_64-pc-windows-msvc
in .cargo/config.toml, but nothing verified the packaged executable, so the
dependency could silently regress.

Parse the PE import and delay-import tables and reject herdr.exe when it
depends on the dynamic Microsoft C/C++ runtime, including release and debug
variants. Check the input while staging and re-check the staged executable
before archiving so a direct archive cannot bypass the guard.

refs #3129
2026-09-15 18:14:07 +02:00

374 lines
15 KiB
Python

from __future__ import annotations
import hashlib
import io
import json
import struct
import tempfile
import unittest
import urllib.error
import zipfile
from pathlib import Path
from unittest import mock
from scripts import package_windows_conpty as package
class WindowsConptyPackageTests(unittest.TestCase):
def test_pinned_metadata_and_notices_are_consistent(self) -> None:
metadata = package.load_metadata(package.DEFAULT_METADATA)
self.assertEqual(metadata["package"]["id"], "Microsoft.Windows.Console.ConPTY")
self.assertEqual(metadata["package"]["version"], "1.24.260710001")
self.assertEqual(
{item["destination"] for item in metadata["bundles"]["x86_64"]["files"]},
{
"conpty/conpty.dll",
"conpty/x64/OpenConsole.exe",
"conpty/arm64/OpenConsole.exe",
},
)
loader = (
package.PROJECT_ROOT / "vendor/portable-pty/src/win/psuedocon.rs"
).read_text(encoding="utf-8")
installer = (package.PROJECT_ROOT / "distribution/install.ps1").read_text(
encoding="utf-8"
)
for item in metadata["bundles"]["x86_64"]["files"]:
self.assertIn(item["sha256"], loader)
self.assertNotIn(item["sha256"], installer)
self.assertIn('Get-Content -LiteralPath $markerPath -Raw', installer)
self.assertIn('$filesProperty.Value.PSObject.Properties[$relative]', installer)
for notice in metadata["notices"]:
source = package.PROJECT_ROOT / notice["source"]
self.assertEqual(package.sha256_file(source), notice["sha256"])
def test_powershell_wrapper_verifies_package_and_signatures(self) -> None:
wrapper = (package.PROJECT_ROOT / "scripts/package_windows_conpty.ps1").read_text(
encoding="utf-8"
)
self.assertIn('"nuget", "verify", "--all"', wrapper)
self.assertIn("Get-AuthenticodeSignature", wrapper)
self.assertIn('conpty\\arm64\\OpenConsole.exe', wrapper)
self.assertIn('conpty\\x64\\OpenConsole.exe', wrapper)
self.assertIn('conpty\\conpty.dll', wrapper)
self.assertIn('"*Microsoft Corporation*"', wrapper)
def test_package_download_retries_server_errors_with_a_finite_timeout(self) -> None:
payload = b"package"
with tempfile.TemporaryDirectory() as temporary:
destination = Path(temporary) / "conpty.nupkg"
metadata = {
"url": "https://example.invalid/conpty.nupkg",
"sha256": hashlib.sha256(payload).hexdigest(),
}
server_error = urllib.error.HTTPError(
metadata["url"], 504, "Gateway Time-out", {}, None
)
with (
mock.patch.object(
package.urllib.request,
"urlopen",
side_effect=[server_error, io.BytesIO(payload)],
) as urlopen,
mock.patch.object(package.time, "sleep") as sleep,
):
package.acquire_package(metadata, destination)
self.assertEqual(
urlopen.call_args_list,
[
mock.call(
metadata["url"], timeout=package.DOWNLOAD_TIMEOUT_SECONDS
),
mock.call(
metadata["url"], timeout=package.DOWNLOAD_TIMEOUT_SECONDS
),
],
)
sleep.assert_called_once_with(1)
def test_stage_and_archive_validate_exact_package(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
dll = self._pe(0x8664, b"dll")
x64_host = self._pe(0x8664, b"x64")
arm64_host = self._pe(0xAA64, b"arm64")
files = [
self._file("runtimes/win-x64/native/conpty.dll", "conpty/conpty.dll", dll, "0x8664"),
self._file(
"build/native/runtimes/x64/OpenConsole.exe",
"conpty/x64/OpenConsole.exe",
x64_host,
"0x8664",
),
self._file(
"build/native/runtimes/arm64/OpenConsole.exe",
"conpty/arm64/OpenConsole.exe",
arm64_host,
"0xaa64",
),
]
nupkg = root / "conpty.nupkg"
self._write_package(nupkg, files, {item["source"]: data for item, data in zip(files, (dll, x64_host, arm64_host))})
metadata_path = root / "conpty.json"
metadata_path.write_text(
json.dumps(
{
"schema_version": 1,
"package": {
"id": "Microsoft.Windows.Console.ConPTY",
"version": "1.24.260710001",
"url": nupkg.as_uri(),
"sha256": package.sha256_file(nupkg),
"license": "MIT",
},
"bundles": {"x86_64": {"files": files}},
"notices": [],
}
),
encoding="utf-8",
)
herdr = root / "input-herdr.exe"
herdr.write_bytes(self._pe_with_imports(0x8664, ["KERNEL32.dll"]))
stage = root / "stage"
package.stage_bundle(metadata_path, "x86_64", nupkg, herdr, stage)
package.validate_stage(metadata_path, "x86_64", stage)
(stage / "herdr.exe").write_bytes(
self._pe_with_imports(0x8664, ["MSVCP140D.dll"])
)
with self.assertRaisesRegex(ValueError, "dynamic Microsoft"):
package.validate_stage(metadata_path, "x86_64", stage)
(stage / "herdr.exe").write_bytes(
self._pe_with_imports(0x8664, ["KERNEL32.dll"])
)
output = root / "herdr.zip"
package.archive_bundle(metadata_path, "x86_64", stage, output)
with zipfile.ZipFile(output) as archive:
self.assertEqual(
set(archive.namelist()),
package.expected_stage_files(
package.load_metadata(metadata_path), "x86_64"
),
)
(stage / "conpty" / "conpty.dll").write_bytes(b"tampered")
with self.assertRaisesRegex(ValueError, "staged file hash mismatch"):
package.validate_stage(metadata_path, "x86_64", stage)
def test_pe_imported_dlls_reads_import_and_delay_import_tables(self) -> None:
data = self._pe_with_imports(
0x8664,
["KERNEL32.dll", "api-ms-win-core-synch-l1-2-0.dll"],
["USER32.dll"],
)
self.assertEqual(
set(package.pe_imported_dlls(data)),
{"KERNEL32.dll", "api-ms-win-core-synch-l1-2-0.dll", "USER32.dll"},
)
pe32 = self._pe_with_imports(0x8664, ["VCRUNTIME140.dll"], magic=0x10B)
self.assertEqual(package.pe_imported_dlls(pe32), ["VCRUNTIME140.dll"])
def test_pe_imported_dlls_handles_va_based_delay_imports(self) -> None:
# VA-based delay-import descriptors predate the RVA flag and are only
# representable with a 32-bit image base.
data = self._pe_with_imports(
0x8664,
["KERNEL32.dll"],
["MSVCP140.dll"],
magic=0x10B,
delay_imports_use_va=True,
)
self.assertEqual(
set(package.pe_imported_dlls(data)),
{"KERNEL32.dll", "MSVCP140.dll"},
)
with self.assertRaisesRegex(ValueError, "dynamic Microsoft"):
package.validate_static_msvc_runtime(data, "herdr.exe")
def test_pe_imported_dlls_respects_declared_directory_size(self) -> None:
# The second descriptor is outside the declared import-directory size,
# so it must not be treated as an import even though it names a
# dynamic runtime.
outside = self._pe_with_imports(
0x8664,
["KERNEL32.dll", "VCRUNTIME140.dll"],
import_directory_size=20,
)
self.assertEqual(package.pe_imported_dlls(outside), ["KERNEL32.dll"])
package.validate_static_msvc_runtime(outside, "herdr.exe")
inside = self._pe_with_imports(0x8664, ["KERNEL32.dll", "VCRUNTIME140.dll"])
with self.assertRaisesRegex(ValueError, "dynamic Microsoft"):
package.validate_static_msvc_runtime(inside, "herdr.exe")
def test_dynamic_msvc_runtime_imports_are_rejected(self) -> None:
for dll in (
"VCRUNTIME140.dll",
"vcruntime140_1.dll",
"vcruntime140d.dll",
"MSVCP140.dll",
"MSVCP140D.dll",
"MSVCR120.dll",
"MSVCP120.dll",
"concrt140d.dll",
"ucrtbase.dll",
"ucrtbased.dll",
"api-ms-win-crt-runtime-l1-1-0.dll",
):
with self.subTest(dll=dll):
with self.assertRaisesRegex(ValueError, "dynamic Microsoft"):
package.validate_static_msvc_runtime(
self._pe_with_imports(0x8664, [dll]), "herdr.exe"
)
def test_static_crt_imports_are_accepted(self) -> None:
data = self._pe_with_imports(
0x8664,
[
"KERNEL32.dll",
"ntdll.dll",
"msvcrt.dll",
"api-ms-win-core-synch-l1-2-0.dll",
],
["USER32.dll"],
)
package.validate_static_msvc_runtime(data, "herdr.exe")
self.assertFalse(package.is_dynamic_msvc_runtime("msvcrt.dll"))
def test_stage_rejects_dynamic_crt_executable(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
metadata_path = root / "conpty.json"
metadata_path.write_text(
json.dumps(
{"schema_version": 1, "bundles": {"x86_64": {}}, "notices": []}
),
encoding="utf-8",
)
herdr = root / "herdr.exe"
herdr.write_bytes(self._pe_with_imports(0x8664, ["VCRUNTIME140.dll"]))
with self.assertRaisesRegex(ValueError, "dynamic Microsoft"):
package.stage_bundle(
metadata_path,
"x86_64",
root / "missing.nupkg",
herdr,
root / "stage",
)
@staticmethod
def _pe(machine: int, payload: bytes) -> bytes:
data = bytearray(0x80)
data[:2] = b"MZ"
struct.pack_into("<I", data, 0x3C, 0x40)
data[0x40:0x44] = b"PE\0\0"
struct.pack_into("<H", data, 0x44, machine)
return bytes(data) + payload
@staticmethod
def _pe_with_imports(
machine: int,
imports: list[str],
delay_imports: list[str] | None = None,
magic: int = 0x20B,
delay_imports_use_va: bool = False,
import_directory_size: int | None = None,
) -> bytes:
delay_imports = delay_imports or []
pe_offset = 0x40
optional_size = 0xF0 if magic == 0x20B else 0xE0
data_directory_offset = 112 if magic == 0x20B else 96
image_base = 0x140000000 if magic == 0x20B else 0x400000
section_rva = 0x1000
section_raw = 0x400
import_size = 20 * (len(imports) + 1)
delay_size = 32 * (len(delay_imports) + 1) if delay_imports else 0
section = bytearray(import_size + delay_size)
name_rvas: dict[str, int] = {}
for name in [*imports, *delay_imports]:
if name not in name_rvas:
name_rvas[name] = section_rva + len(section)
section += name.encode("ascii") + b"\0"
for index, name in enumerate(imports):
struct.pack_into("<I", section, index * 20 + 12, name_rvas[name])
for index, name in enumerate(delay_imports):
offset = import_size + index * 32
if delay_imports_use_va:
# Attributes leave the RVA flag unset, so the name is a VA.
struct.pack_into("<I", section, offset + 4, image_base + name_rvas[name])
else:
struct.pack_into("<I", section, offset, 1)
struct.pack_into("<I", section, offset + 4, name_rvas[name])
data = bytearray(section_raw + len(section))
data[:2] = b"MZ"
struct.pack_into("<I", data, 0x3C, pe_offset)
data[pe_offset : pe_offset + 4] = b"PE\0\0"
struct.pack_into("<H", data, pe_offset + 4, machine)
struct.pack_into("<H", data, pe_offset + 6, 1)
struct.pack_into("<H", data, pe_offset + 20, optional_size)
optional_offset = pe_offset + 24
struct.pack_into("<H", data, optional_offset, magic)
if magic == 0x20B:
struct.pack_into("<Q", data, optional_offset + 24, image_base)
else:
struct.pack_into("<I", data, optional_offset + 28, image_base)
struct.pack_into("<I", data, optional_offset + data_directory_offset - 4, 16)
struct.pack_into(
"<II",
data,
optional_offset + data_directory_offset + 8,
section_rva,
import_size if import_directory_size is None else import_directory_size,
)
if delay_imports:
struct.pack_into(
"<II",
data,
optional_offset + data_directory_offset + 13 * 8,
section_rva + import_size,
delay_size,
)
section_header_offset = pe_offset + 24 + optional_size
data[section_header_offset : section_header_offset + 8] = b".rdata\0\0"
struct.pack_into("<I", data, section_header_offset + 8, len(section))
struct.pack_into("<I", data, section_header_offset + 12, section_rva)
struct.pack_into("<I", data, section_header_offset + 16, len(section))
struct.pack_into("<I", data, section_header_offset + 20, section_raw)
data[section_raw : section_raw + len(section)] = section
return bytes(data)
@staticmethod
def _file(source: str, destination: str, data: bytes, machine: str) -> dict[str, str]:
return {
"source": source,
"destination": destination,
"sha256": hashlib.sha256(data).hexdigest(),
"pe_machine": machine,
}
@staticmethod
def _write_package(
path: Path, files: list[dict[str, str]], payloads: dict[str, bytes]
) -> None:
nuspec = """<?xml version="1.0" encoding="utf-8"?>
<package xmlns="http://schemas.microsoft.com/packaging/2011/08/nuspec.xsd">
<metadata>
<id>Microsoft.Windows.Console.ConPTY</id>
<version>1.24.260710001</version>
<license type="expression">MIT</license>
</metadata>
</package>
"""
with zipfile.ZipFile(path, "w") as archive:
archive.writestr("Microsoft.Windows.Console.ConPTY.nuspec", nuspec)
for item in files:
archive.writestr(item["source"], payloads[item["source"]])
if __name__ == "__main__":
unittest.main()