Harden protocol against pickle RCE with a restricted unpickler

addon/lib/protocol/__init__.py

@@ -27,6 +27,7 @@
 from hwIo.base import IoBase
 from logHandler import log
 
+from ._restrictedUnpickling import restrictedLoads
 from .braille import BrailleAttribute, BrailleCommand
 from .speech import SpeechAttribute, SpeechCommand
 
@@ -517,7 +518,7 @@ def _pickle(self, obj: Any):
 		return pickle.dumps(obj, protocol=4)
 
 	def _unpickle(self, payload: bytes) -> Any:
-		res = pickle.loads(payload)
+		res = restrictedLoads(payload)
 		if isinstance(res, AutoPropertyObject):
 			res.invalidateCache()
 		return res

addon/lib/protocol/_restrictedUnpickling.py

@@ -0,0 +1,104 @@
+# RDAccess: Remote Desktop Accessibility for NVDA
+# Copyright 2023-2026 Leonard de Ruijter <alderuijter@gmail.com>
+# License: GNU General Public License version 2.0
+"""Allowlist-restricted unpickling for payloads received over the RDAccess wire.
+
+``pickle.loads`` on untrusted data is an arbitrary-code-execution primitive, and RDAccess
+deserializes payloads that arrive over the DVC named pipe from a peer across the RDP trust
+boundary. This module confines unpickling to an allowlist of the classes that legitimately
+cross the wire, blocking the generic ``__reduce__``-based gadget class (``os.system``,
+``builtins.eval``, ``subprocess.*``, etc.) while staying wire-compatible with existing peers.
+"""
+
+from __future__ import annotations
+
+import io
+import pickle
+from typing import Any
+
+from logHandler import log
+
+_allowedUnpickleGlobals: dict[tuple[str, str], type] | None = None
+
+
+def _getAllowedGlobals() -> dict[tuple[str, str], type]:
+	"""Return the allowlist of classes RestrictedUnpickler may resolve, keyed by
+	``(cls.__module__, cls.__qualname__)``.
+
+	The classes are imported lazily (only on first call, then cached) rather than at module import
+	time, so that import order of NVDA's internal modules is unaffected and unit tests can stub the
+	modules before this is ever invoked. Keys are built from the imported class objects themselves
+	(not hardcoded strings) so they always match whatever string the pickling side actually wrote,
+	which matters in particular for RDAccess's own BrailleInputGesture: its __module__ is whatever
+	addonHandler.Addon.loadModule names the add-on module as, identical on both sides of the wire.
+	"""
+	global _allowedUnpickleGlobals
+	if _allowedUnpickleGlobals is None:
+		from collections import OrderedDict
+
+		from autoSettingsUtils.driverSetting import (
+			BooleanDriverSetting,
+			DriverSetting,
+			NumericDriverSetting,
+		)
+		from autoSettingsUtils.utils import StringParameterInfo
+		from inputCore import GlobalGestureMap
+		from synthDriverHandler import VoiceInfo
+
+		from .braille import BrailleInputGesture
+
+		classes: tuple[type, ...] = (
+			DriverSetting,
+			NumericDriverSetting,
+			BooleanDriverSetting,
+			StringParameterInfo,
+			VoiceInfo,
+			GlobalGestureMap,
+			OrderedDict,
+			BrailleInputGesture,
+		)
+		allowedGlobals: dict[tuple[str, str], type] = {
+			(str(cls.__module__), str(cls.__qualname__)): cls for cls in classes
+		}
+		# Compat alias: peers on older NVDA pickled this under its pre-move module name.
+		allowedGlobals[("driverHandler", "StringParameterInfo")] = StringParameterInfo
+		_allowedUnpickleGlobals = allowedGlobals
+	return _allowedUnpickleGlobals
+
+
+class RestrictedUnpickler(pickle.Unpickler):
+	"""Restricts unpickling to an allowlist of classes that legitimately cross the RDAccess wire.
+
+	pickle.loads on untrusted data is an arbitrary-code-execution primitive: a crafted payload's
+	__reduce__ can reference any importable global (os.system, builtins.eval, subprocess.*, etc.).
+	Overriding find_class so it only ever resolves allowlisted classes blocks that gadget class
+	entirely, while remaining wire-compatible with existing RDAccess peers (builtin containers and
+	scalars never go through find_class; only global/class lookups do).
+	"""
+
+	def find_class(self, module: str, name: str) -> Any:
+		if module == "speech.commands":
+			# Speech command classes are allowed dynamically so future NVDA-added commands work
+			# without an add-on update, while anything else in the module is still refused.
+			import speech.commands
+
+			obj = getattr(speech.commands, name, None)
+			if (
+				isinstance(obj, type)
+				and obj.__module__ == module
+				and issubclass(obj, speech.commands.SpeechCommand)
+			):
+				return obj
+		else:
+			cls = _getAllowedGlobals().get((module, name))
+			if cls is not None:
+				return cls
+		# Exceptions raised inside _bgExecutor futures are otherwise swallowed silently, so log here
+		# to keep rejected payloads diagnosable.
+		log.error(f"Refusing to unpickle disallowed global: module={module!r}, name={name!r}")
+		raise pickle.UnpicklingError(f"Global '{module}.{name}' is forbidden")
+
+
+def restrictedLoads(payload: bytes) -> Any:
+	"""Unpickle ``payload`` through RestrictedUnpickler's allowlist."""
+	return RestrictedUnpickler(io.BytesIO(payload)).load()

tests/_stubs.py

@@ -145,3 +145,132 @@ class SpeechManager:
 		MAX_INDEX = 9999
 
 	speechManager.SpeechManager = SpeechManager
+
+	_installSpeechCommandsStub(speech)
+	_installDriverSettingAndSynthVoiceStubs()
+	_installInputCoreStub()
+
+
+def _setStubIdentity(cls: type, module: str) -> None:
+	"""Set __module__ and __qualname__ on a class defined inside a function.
+
+	Classes defined here get __module__ == "tests._stubs" and __qualname__ containing
+	"install.<locals>." (or similar) by default; both must be corrected so pickled payloads in
+	tests carry the same module/qualname strings as real NVDA would produce. Pickle refuses to
+	pickle-by-reference a class whose __qualname__ contains "<locals>".
+	"""
+	cls.__module__ = module
+	cls.__qualname__ = cls.__name__
+
+
+def _installSpeechCommandsStub(speech: types.ModuleType) -> None:
+	speechCommands = _module("speech.commands")
+	speech.commands = speechCommands
+
+	class SpeechCommand:
+		pass
+
+	class IndexCommand(SpeechCommand):
+		def __init__(self, index: int):
+			self.index = index
+
+		def __eq__(self, other: Any) -> bool:
+			return type(self) is type(other) and self.index == other.index
+
+	class PitchCommand(SpeechCommand):
+		def __init__(self, offset: int = 0):
+			self.offset = offset
+
+		def __eq__(self, other: Any) -> bool:
+			return type(self) is type(other) and self.offset == other.offset
+
+	class NotASpeechCommand:
+		"""Exists in speech.commands but is not a SpeechCommand subclass; used to test that the
+		dynamic find_class rule for speech.commands rejects it.
+		"""
+
+	for cls in (SpeechCommand, IndexCommand, PitchCommand, NotASpeechCommand):
+		_setStubIdentity(cls, "speech.commands")
+
+	speechCommands.SpeechCommand = SpeechCommand
+	speechCommands.IndexCommand = IndexCommand
+	speechCommands.PitchCommand = PitchCommand
+	speechCommands.NotASpeechCommand = NotASpeechCommand
+
+
+def _installDriverSettingAndSynthVoiceStubs() -> None:
+	from baseObject import AutoPropertyObject
+
+	autoSettingsUtils = _module("autoSettingsUtils")
+	autoSettingsDriverSetting = _module("autoSettingsUtils.driverSetting")
+	autoSettingsUtils.driverSetting = autoSettingsDriverSetting
+	autoSettingsUtilsUtils = _module("autoSettingsUtils.utils")
+	autoSettingsUtils.utils = autoSettingsUtilsUtils
+
+	class StringParameterInfo:
+		def __init__(self, id: str, displayName: str):
+			self.id = id
+			self.displayName = displayName
+
+		def __eq__(self, other: Any) -> bool:
+			return type(self) is type(other) and self.id == other.id and self.displayName == other.displayName
+
+	class DriverSetting(AutoPropertyObject):
+		def __init__(
+			self,
+			id: str,
+			displayNameWithAccelerator: str,
+			availableInSettingsRing: bool = False,
+			defaultVal: Any = None,
+			displayName: str | None = None,
+			useConfig: bool = True,
+		):
+			self.id = id
+			self.displayNameWithAccelerator = displayNameWithAccelerator
+			self.displayName = displayName or displayNameWithAccelerator
+			self.availableInSettingsRing = availableInSettingsRing
+			self.defaultVal = defaultVal
+			self.useConfig = useConfig
+
+	class NumericDriverSetting(DriverSetting):
+		pass
+
+	class BooleanDriverSetting(DriverSetting):
+		pass
+
+	_setStubIdentity(StringParameterInfo, "autoSettingsUtils.utils")
+	for cls in (DriverSetting, NumericDriverSetting, BooleanDriverSetting):
+		_setStubIdentity(cls, "autoSettingsUtils.driverSetting")
+
+	autoSettingsUtilsUtils.StringParameterInfo = StringParameterInfo
+	autoSettingsDriverSetting.DriverSetting = DriverSetting
+	autoSettingsDriverSetting.NumericDriverSetting = NumericDriverSetting
+	autoSettingsDriverSetting.BooleanDriverSetting = BooleanDriverSetting
+
+	synthDriverHandler = _module("synthDriverHandler")
+
+	class VoiceInfo(StringParameterInfo):
+		def __init__(self, id: str, displayName: str, language: str | None = None):
+			self.language = language
+			super().__init__(id, displayName)
+
+	_setStubIdentity(VoiceInfo, "synthDriverHandler")
+	synthDriverHandler.VoiceInfo = VoiceInfo
+
+
+def _installInputCoreStub() -> None:
+	inputCore = _module("inputCore")
+
+	class GlobalGestureMap:
+		def __init__(self, entries: Any = None):
+			self._map: dict = {}
+			self.lastUpdateContainedError = False
+			self.fileName: str | None = None
+			if entries:
+				self.update(entries)
+
+		def update(self, entries: Any):
+			self._map.update(entries)
+
+	_setStubIdentity(GlobalGestureMap, "inputCore")
+	inputCore.GlobalGestureMap = GlobalGestureMap

tests/test_misc.py

@@ -11,31 +11,13 @@
 import gc
 import unittest
 import weakref
+from unittest import mock
 
 import queueHandler  # noqa: E402 — must follow tests import so stubs are installed
-from baseObject import AutoPropertyObject  # noqa: E402 — same reason
+from autoSettingsUtils.driverSetting import DriverSetting  # noqa: E402 — same reason
 
 from tests._fakes import FakeHandlerBase  # bootstrap runs via tests/__init__ import
 
-# ---------------------------------------------------------------------------
-# Module-level helper required by test 5 (must be picklable by reference).
-# ---------------------------------------------------------------------------
-
-
-class _PickleProbe(AutoPropertyObject):
-	"""AutoPropertyObject whose invalidateCache sets a flag for inspection."""
-
-	cachePropertiesByDefault = True
-
-	def __init__(self):
-		super().__init__()
-		self.cacheInvalidated = False
-
-	def invalidateCache(self):
-		self.cacheInvalidated = True
-		super().invalidateCache()
-
-
 # ---------------------------------------------------------------------------
 # Tests
 # ---------------------------------------------------------------------------
@@ -93,12 +75,18 @@ def test_roundtrip_plain_structure(self):
 		self.assertEqual(restored, original)
 
 	def test_unpickle_calls_invalidate_cache_on_auto_property_object(self):
-		"""_unpickle calls invalidateCache on AutoPropertyObject results."""
-		probe = _PickleProbe()
-		raw = self.handler._pickle(probe)
-		result = self.handler._unpickle(raw)
-		self.assertIsInstance(result, _PickleProbe)
-		self.assertTrue(result.cacheInvalidated)
+		"""_unpickle calls invalidateCache on AutoPropertyObject results.
+
+		DriverSetting is used as the probe (rather than an ad hoc AutoPropertyObject subclass)
+		because RestrictedUnpickler only resolves allowlisted classes; see
+		tests/test_restrictedUnpickling.py for the allowlist itself.
+		"""
+		original = DriverSetting("id1", "Setting 1")
+		raw = self.handler._pickle(original)
+		with mock.patch.object(DriverSetting, "invalidateCache") as invalidateCache:
+			result = self.handler._unpickle(raw)
+		self.assertIsInstance(result, DriverSetting)
+		invalidateCache.assert_called_once_with()
 
 	def test_unpickle_does_not_call_invalidate_cache_on_plain_objects(self):
 		"""_unpickle does not call invalidateCache on non-AutoPropertyObject results."""