diff --git a/addon/lib/protocol/__init__.py b/addon/lib/protocol/__init__.py index 745294e..3f3b306 100644 --- a/addon/lib/protocol/__init__.py +++ b/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 diff --git a/addon/lib/protocol/_restrictedUnpickling.py b/addon/lib/protocol/_restrictedUnpickling.py new file mode 100644 index 0000000..c7bfb79 --- /dev/null +++ b/addon/lib/protocol/_restrictedUnpickling.py @@ -0,0 +1,104 @@ +# RDAccess: Remote Desktop Accessibility for NVDA +# Copyright 2023-2026 Leonard de Ruijter +# 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() diff --git a/tests/_stubs.py b/tests/_stubs.py index e8189ae..7b2890c 100644 --- a/tests/_stubs.py +++ b/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.." (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 "". + """ + 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 diff --git a/tests/test_misc.py b/tests/test_misc.py index ac98a06..c1f56d3 100644 --- a/tests/test_misc.py +++ b/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.""" diff --git a/tests/test_restrictedUnpickling.py b/tests/test_restrictedUnpickling.py new file mode 100644 index 0000000..3265fc7 --- /dev/null +++ b/tests/test_restrictedUnpickling.py @@ -0,0 +1,180 @@ +# RDAccess: Remote Desktop Accessibility for NVDA +# Copyright 2026 Leonard de Ruijter +# License: GNU General Public License version 2.0 + +"""Unit tests for RestrictedUnpickler, the pickle.Unpickler subclass in +addon/lib/protocol/_restrictedUnpickling.py used by RemoteProtocolHandler._unpickle to allowlist +the classes that legitimately cross the RDAccess wire, blocking generic __reduce__-based RCE gadgets. +""" + +from __future__ import annotations + +import collections +import os +import pickle +import unittest +from typing import Any + +from autoSettingsUtils.driverSetting import BooleanDriverSetting, DriverSetting, NumericDriverSetting +from autoSettingsUtils.utils import StringParameterInfo +from inputCore import GlobalGestureMap +from lib.protocol.braille import BrailleInputGesture +from logHandler import log +from speech.commands import IndexCommand, NotASpeechCommand, PitchCommand +from synthDriverHandler import VoiceInfo + +from tests._fakes import FakeHandlerBase + +# --------------------------------------------------------------------------- +# Helper: mimics a __reduce__-based RCE gadget, the classic pickle attack shape. +# --------------------------------------------------------------------------- + + +class _ReduceGadget: + """Pickles to a call of an arbitrary global with arbitrary args. + + This is the shape a malicious peer would use to smuggle a call to e.g. os.system or + builtins.eval through pickle.loads; RestrictedUnpickler must refuse to resolve the + referenced global before the call ever happens. + """ + + def __init__(self, target: Any, args: tuple = ()): + self._target = target + self._args = args + + def __reduce__(self): + return (self._target, self._args) + + +# --------------------------------------------------------------------------- +# Rejection cases. +# --------------------------------------------------------------------------- + + +class TestRestrictedUnpicklingRejections(unittest.TestCase): + def setUp(self): + self.handler = FakeHandlerBase() + self.addCleanup(self.handler.terminate) + + def test_rejects_os_system_reduce_gadget(self): + """A payload whose __reduce__ references os.system must not be unpickled.""" + payload = pickle.dumps(_ReduceGadget(os.system, ("echo pwned",)), protocol=4) + with self.assertRaises(pickle.UnpicklingError): + self.handler._unpickle(payload) + + def test_rejects_builtins_eval_reduce_gadget(self): + """A payload whose __reduce__ references builtins.eval must not be unpickled.""" + payload = pickle.dumps(_ReduceGadget(eval, ("1+1",)), protocol=4) + with self.assertRaises(pickle.UnpicklingError): + self.handler._unpickle(payload) + + def test_rejects_non_speechcommand_name_in_speech_commands_module(self): + """The dynamic speech.commands rule must refuse names that exist in the module but are not + SpeechCommand subclasses. + """ + payload = pickle.dumps(NotASpeechCommand, protocol=4) + with self.assertRaises(pickle.UnpicklingError): + self.handler._unpickle(payload) + + def test_rejects_arbitrary_unknown_class(self): + """A class that never crosses the wire legitimately (and isn't in the allowlist) is refused.""" + payload = pickle.dumps(collections.Counter, protocol=4) + with self.assertRaises(pickle.UnpicklingError): + self.handler._unpickle(payload) + + def test_rejection_is_logged_as_error(self): + """Rejections are logged via log.error, since exceptions raised inside _bgExecutor futures + are otherwise swallowed silently. + """ + log.records.clear() + payload = pickle.dumps(_ReduceGadget(os.system, ("echo pwned",)), protocol=4) + with self.assertRaises(pickle.UnpicklingError): + self.handler._unpickle(payload) + errorRecords = [msg for level, msg in log.records if level == "error"] + self.assertTrue(errorRecords, "Expected a log.error call recording the rejected global") + # os.system's actual module is platform-specific (nt on Windows, posixpath on Unix) + module_name = os.system.__module__ + self.assertTrue( + any("system" in msg and module_name in msg for msg in errorRecords), + "Expected the rejection log to name the offending (module, name)", + ) + + +# --------------------------------------------------------------------------- +# Acceptance cases: round-trip _pickle -> _unpickle for everything that legitimately +# crosses the wire (see the audit table in the implementation plan). +# --------------------------------------------------------------------------- + + +class TestRestrictedUnpicklingAcceptance(unittest.TestCase): + def setUp(self): + self.handler = FakeHandlerBase() + self.addCleanup(self.handler.terminate) + + def _roundtrip(self, obj: Any) -> Any: + return self.handler._unpickle(self.handler._pickle(obj)) + + def test_roundtrips_speech_sequence(self): + original = ["hello", IndexCommand(1), PitchCommand(offset=5)] + restored = self._roundtrip(original) + self.assertEqual(restored[0], "hello") + self.assertIsInstance(restored[1], IndexCommand) + self.assertEqual(restored[1].index, 1) + self.assertIsInstance(restored[2], PitchCommand) + self.assertEqual(restored[2].offset, 5) + + def test_roundtrips_frozenset_of_speech_command_classes(self): + """synthDrivers/remote.py._incoming_supportedCommands sends classes, not instances.""" + original = frozenset({IndexCommand, PitchCommand}) + restored = self._roundtrip(original) + self.assertEqual(restored, original) + + def test_roundtrips_list_of_driver_settings(self): + original = [ + DriverSetting("id1", "Setting 1"), + BooleanDriverSetting("id2", "Setting 2", defaultVal=True), + NumericDriverSetting("id3", "Setting 3", defaultVal=10), + ] + restored = self._roundtrip(original) + self.assertEqual( + [type(o) for o in restored], + [DriverSetting, BooleanDriverSetting, NumericDriverSetting], + ) + self.assertEqual(restored[0].id, "id1") + self.assertEqual(restored[1].defaultVal, True) + self.assertEqual(restored[2].defaultVal, 10) + + def test_roundtrips_ordered_dict_of_voice_and_string_parameter_info(self): + original = collections.OrderedDict(( + ("voice1", VoiceInfo("voice1", "Voice One", language="en")), + ("param1", StringParameterInfo("param1", "Param One")), + )) + restored = self._roundtrip(original) + self.assertIsInstance(restored, collections.OrderedDict) + self.assertIsInstance(restored["voice1"], VoiceInfo) + self.assertEqual(restored["voice1"].language, "en") + self.assertIsInstance(restored["param1"], StringParameterInfo) + self.assertEqual(restored["param1"].displayName, "Param One") + + def test_roundtrips_global_gesture_map(self): + original = GlobalGestureMap({"kb:a": "someScript"}) + restored = self._roundtrip(original) + self.assertIsInstance(restored, GlobalGestureMap) + self.assertEqual(restored._map, original._map) + + def test_roundtrips_braille_input_gesture(self): + original = BrailleInputGesture(source="test", id="routing1", routingIndex=3) + restored = self._roundtrip(original) + self.assertIsInstance(restored, BrailleInputGesture) + self.assertEqual(restored.id, "routing1") + self.assertEqual(restored.routingIndex, 3) + + def test_roundtrips_plain_builtins_dict(self): + """Beep/wave-file kwargs are plain builtins; these never touch find_class at all.""" + original = {"frequency": 440, "length": 50, "loop": False, "left": 0.0} + restored = self._roundtrip(original) + self.assertEqual(restored, original) + + +if __name__ == "__main__": + unittest.main()