diff options
Diffstat (limited to 'scripts')
| -rw-r--r-- | scripts/package_windows_conpty.py | 142 | ||||
| -rw-r--r-- | scripts/test_package_windows_conpty.py | 186 |
2 files changed, 327 insertions, 1 deletions
diff --git a/scripts/package_windows_conpty.py b/scripts/package_windows_conpty.py index 3c63312e..1877d6b0 100644 --- a/scripts/package_windows_conpty.py +++ b/scripts/package_windows_conpty.py @@ -49,6 +49,142 @@ def pe_machine(data: bytes) -> int: 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: @@ -124,6 +260,8 @@ def stage_bundle( 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 @@ -196,6 +334,10 @@ def validate_stage(metadata_path: Path, architecture: str, stage_dir: Path) -> N ) 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) diff --git a/scripts/test_package_windows_conpty.py b/scripts/test_package_windows_conpty.py index b92edaf3..61aec9d0 100644 --- a/scripts/test_package_windows_conpty.py +++ b/scripts/test_package_windows_conpty.py @@ -129,11 +129,20 @@ class WindowsConptyPackageTests(unittest.TestCase): encoding="utf-8", ) herdr = root / "input-herdr.exe" - herdr.write_bytes(b"herdr") + 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: @@ -148,6 +157,107 @@ class WindowsConptyPackageTests(unittest.TestCase): 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) @@ -158,6 +268,80 @@ class WindowsConptyPackageTests(unittest.TestCase): 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, |