Added windows identity checker.

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