diff --git a/ptlinux/identity_probe.py b/ptlinux/identity_probe.py new file mode 100644 index 0000000..cdae753 --- /dev/null +++ b/ptlinux/identity_probe.py @@ -0,0 +1,393 @@ +"""Privacy-safe checks for the Windows identity inputs used by the game client.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +from pathlib import Path +import re + +from .command import CommandRunner +from .paths import AppPaths +from .runtime import RuntimeSelection, run_runtime_prefix_command + + +IDENTITY_PROBE_SHA256 = "677cb5aa19f8dac65b982462bb30974feed05399fdd759765c3b6e802d4c838b" +IDENTITY_PROBE_TIMEOUT_SECONDS = 45.0 +IDENTITY_PROBE_PRIVACY = ( + "Only success flags, row counts, value lengths, VARIANT types, adapter type, " + "and generic-value quality flags are collected. Computer names, serial numbers, " + "MAC addresses, and WMI values never leave the Windows helper." +) +WMI_PROPERTY_NAMES = ( + "processor_id", + "processor_unique_id", + "processor_manufacturer", + "processor_max_clock_speed", + "bios_release_date", + "bios_serial_number", + "bios_identification_code", + "bios_smbios_version", + "bios_manufacturer", + "baseboard_serial_number", + "baseboard_name", + "baseboard_manufacturer", + "baseboard_model", + "video_name", + "video_driver_version", +) +HRESULT_PATTERN = re.compile(r"0x[0-9A-F]{8}\Z") + + +@dataclass(frozen=True) +class IdentityProbeResult: + helper: str + helper_sha256: str | None + helper_verified: bool + ran: bool + ready: bool | None + error: str | None + report: dict[str, object] | None + observations: tuple[str, ...] + warnings: tuple[str, ...] + + +def packaged_identity_probe() -> Path: + return Path(__file__).resolve().parent / "native" / "pt-identity-probe.exe" + + +def probe_client_identity( + paths: AppPaths, + selection: RuntimeSelection, + runner: CommandRunner, + *, + timeout_seconds: float = IDENTITY_PROBE_TIMEOUT_SECONDS, +) -> IdentityProbeResult: + """Check identity API availability without returning the underlying values.""" + + helper = packaged_identity_probe() + helper_sha256 = _sha256_if_file(helper) + helper_verified = helper_sha256 == IDENTITY_PROBE_SHA256 + if not helper_verified: + error = ( + "the packaged privacy-safe identity helper is missing" + if helper_sha256 is None + else "the packaged privacy-safe identity helper failed verification" + ) + return IdentityProbeResult( + helper=str(helper), + helper_sha256=helper_sha256, + helper_verified=False, + ran=False, + ready=None, + error=error, + report=None, + observations=(), + warnings=(error,), + ) + try: + outcome = run_runtime_prefix_command( + paths, + selection, + runner, + [str(helper)], + check=False, + timeout_seconds=timeout_seconds, + ) + except (OSError, RuntimeError) as exc: + error = f"the client-identity helper could not start: {exc}" + return IdentityProbeResult( + helper=str(helper), + helper_sha256=helper_sha256, + helper_verified=True, + ran=False, + ready=None, + error=error, + report=None, + observations=(), + warnings=(error,), + ) + if outcome.timed_out: + error = "the client-identity helper timed out" + return _failed_result(helper, helper_sha256, error, ran=True) + if outcome.returncode != 0: + status = "unknown" if outcome.returncode is None else str(outcome.returncode) + error = f"the client-identity helper exited with status {status}" + return _failed_result(helper, helper_sha256, error, ran=True) + try: + report = _parse_privacy_safe_report(outcome.stdout) + except (ValueError, json.JSONDecodeError) as exc: + error = f"the client-identity helper returned invalid data: {exc}" + return _failed_result(helper, helper_sha256, error, ran=True) + warnings = _identity_warnings(report) + return IdentityProbeResult( + helper=str(helper), + helper_sha256=helper_sha256, + helper_verified=True, + ran=True, + ready=not warnings, + error=None, + report=report, + observations=_identity_observations(report), + warnings=warnings, + ) + + +def _failed_result( + helper: Path, + helper_sha256: str | None, + error: str, + *, + ran: bool, +) -> IdentityProbeResult: + return IdentityProbeResult( + helper=str(helper), + helper_sha256=helper_sha256, + helper_verified=True, + ran=ran, + ready=None, + error=error, + report=None, + observations=(), + warnings=(error,), + ) + + +def _parse_privacy_safe_report(stdout: str) -> dict[str, object]: + lines = [line.strip() for line in stdout.splitlines() if line.strip()] + if not lines: + raise ValueError("empty output") + value = json.loads(lines[-1]) + root = _object(value, "root") + if _integer(root.get("schema"), "schema") != 2: + raise ValueError("unsupported schema") + computer = _object(root.get("computer_name"), "computer_name") + volume = _object(root.get("volume"), "volume") + adapters = _object(root.get("adapters"), "adapters") + firmware = _object(root.get("firmware"), "firmware") + wmi = _object(root.get("wmi"), "wmi") + properties = _object(wmi.get("properties"), "wmi.properties") + safe_properties: dict[str, object] = {} + for name in WMI_PROPERTY_NAMES: + entry = _object(properties.get(name), f"wmi.properties.{name}") + safe_properties[name] = { + "result": _hresult(entry.get("result"), f"{name}.result"), + "variant_type": _integer(entry.get("variant_type"), f"{name}.variant_type"), + "rows": _integer(entry.get("rows"), f"{name}.rows"), + "nonempty_rows": _integer( + entry.get("nonempty_rows"), + f"{name}.nonempty_rows", + ), + "maximum_length": _integer( + entry.get("maximum_length"), + f"{name}.maximum_length", + ), + "placeholder_rows": _integer( + entry.get("placeholder_rows"), + f"{name}.placeholder_rows", + ), + "wine_rows": _integer( + entry.get("wine_rows"), + f"{name}.wine_rows", + ), + } + return { + "schema": 2, + "computer_name": { + "success": _boolean(computer.get("success"), "computer_name.success"), + "nonempty": _boolean(computer.get("nonempty"), "computer_name.nonempty"), + "length": _integer(computer.get("length"), "computer_name.length"), + }, + "volume": { + "success": _boolean(volume.get("success"), "volume.success"), + "serial_nonzero": _boolean( + volume.get("serial_nonzero"), + "volume.serial_nonzero", + ), + }, + "adapters": { + "status": _integer(adapters.get("status"), "adapters.status"), + "count": _integer(adapters.get("count"), "adapters.count"), + "first_address_length": _integer( + adapters.get("first_address_length"), + "adapters.first_address_length", + ), + "first_address_nonzero": _boolean( + adapters.get("first_address_nonzero"), + "adapters.first_address_nonzero", + ), + "first_address_locally_administered": _boolean( + adapters.get("first_address_locally_administered"), + "adapters.first_address_locally_administered", + ), + "first_address_multicast": _boolean( + adapters.get("first_address_multicast"), + "adapters.first_address_multicast", + ), + "first_type": _integer(adapters.get("first_type"), "adapters.first_type"), + }, + "firmware": { + "raw_smbios_bytes": _integer( + firmware.get("raw_smbios_bytes"), + "firmware.raw_smbios_bytes", + ), + }, + "wmi": { + "com_initialize": _hresult(wmi.get("com_initialize"), "wmi.com_initialize"), + "com_security": _hresult(wmi.get("com_security"), "wmi.com_security"), + "connect": _hresult(wmi.get("connect"), "wmi.connect"), + "properties": safe_properties, + }, + } + + +def _identity_warnings(report: dict[str, object]) -> tuple[str, ...]: + warnings: list[str] = [] + computer = _object(report["computer_name"], "computer_name") + volume = _object(report["volume"], "volume") + adapters = _object(report["adapters"], "adapters") + wmi = _object(report["wmi"], "wmi") + properties = _object(wmi["properties"], "wmi.properties") + computer_success = _boolean(computer["success"], "computer_name.success") + computer_nonempty = _boolean(computer["nonempty"], "computer_name.nonempty") + volume_success = _boolean(volume["success"], "volume.success") + volume_serial_nonzero = _boolean(volume["serial_nonzero"], "volume.serial_nonzero") + adapter_status = _integer(adapters["status"], "adapters.status") + adapter_count = _integer(adapters["count"], "adapters.count") + first_address_length = _integer( + adapters["first_address_length"], + "adapters.first_address_length", + ) + first_address_nonzero = _boolean( + adapters["first_address_nonzero"], + "adapters.first_address_nonzero", + ) + if not computer_success or not computer_nonempty: + warnings.append("Wine did not provide the computer name used by the login request.") + if not volume_success or not volume_serial_nonzero: + warnings.append("Wine did not provide a usable C: volume identity.") + if ( + adapter_status != 0 + or adapter_count == 0 + or first_address_length < 6 + or not first_address_nonzero + ): + warnings.append( + "Wine did not provide a usable first network-adapter identity; the game " + "would send its UUID fallback instead." + ) + if _hresult(wmi["connect"], "wmi.connect") != "0x00000000": + warnings.append("Wine could not connect to ROOT\\CIMV2 for the login identity query.") + else: + if not _any_nonempty(properties, ("processor_id", "processor_unique_id")): + warnings.append("Wine returned no processor identifier through WMI.") + if not _any_nonempty( + properties, + ( + "bios_release_date", + "bios_serial_number", + "bios_identification_code", + "bios_smbios_version", + "bios_manufacturer", + ), + ): + warnings.append("Wine returned no BIOS identity fields through WMI.") + if not _any_nonempty( + properties, + ( + "baseboard_serial_number", + "baseboard_name", + "baseboard_manufacturer", + "baseboard_model", + ), + ): + warnings.append("Wine returned no baseboard identity fields through WMI.") + if not _any_nonempty(properties, ("video_name",)): + warnings.append("Wine returned no video-controller name through WMI.") + return tuple(warnings) + + +def _identity_observations(report: dict[str, object]) -> tuple[str, ...]: + observations: list[str] = [] + adapters = _object(report["adapters"], "adapters") + if _boolean( + adapters["first_address_locally_administered"], + "adapters.first_address_locally_administered", + ): + observations.append( + "The first network-adapter address is locally administered." + ) + if _boolean( + adapters["first_address_multicast"], + "adapters.first_address_multicast", + ): + observations.append("The first network-adapter address has the multicast bit set.") + wmi = _object(report["wmi"], "wmi") + properties = _object(wmi["properties"], "wmi.properties") + placeholder_names: list[str] = [] + wine_names: list[str] = [] + for name in WMI_PROPERTY_NAMES: + entry = _object(properties[name], name) + if _integer(entry["placeholder_rows"], f"{name}.placeholder_rows") > 0: + placeholder_names.append(name) + if _integer(entry["wine_rows"], f"{name}.wine_rows") > 0: + wine_names.append(name) + if placeholder_names: + observations.append( + "Common placeholder WMI values were observed in: " + + ", ".join(placeholder_names) + ) + if wine_names: + observations.append( + "Wine-labelled WMI values were observed in: " + ", ".join(wine_names) + ) + return tuple(observations) + + +def _any_nonempty(properties: dict[str, object], names: tuple[str, ...]) -> bool: + for name in names: + entry = _object(properties[name], name) + if ( + _hresult(entry["result"], f"{name}.result") == "0x00000000" + and _integer(entry["nonempty_rows"], f"{name}.nonempty_rows") > 0 + ): + return True + return False + + +def _object(value: object, label: str) -> dict[str, object]: + if not isinstance(value, dict) or not all(isinstance(key, str) for key in value): + raise ValueError(f"{label} must be an object") + return value + + +def _boolean(value: object, label: str) -> bool: + if not isinstance(value, bool): + raise ValueError(f"{label} must be a boolean") + return value + + +def _integer(value: object, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 0 or value > 0xFFFFFFFF: + raise ValueError(f"{label} must be an unsigned 32-bit integer") + return value + + +def _hresult(value: object, label: str) -> str: + if not isinstance(value, str) or HRESULT_PATTERN.fullmatch(value) is None: + raise ValueError(f"{label} must be a hexadecimal HRESULT") + return value + + +def _sha256_if_file(path: Path) -> str | None: + try: + if not path.is_file(): + return None + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + except OSError: + return None diff --git a/ptlinux/native/pt-identity-probe.exe b/ptlinux/native/pt-identity-probe.exe new file mode 100644 index 0000000..46c64ea Binary files /dev/null and b/ptlinux/native/pt-identity-probe.exe differ