Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions core/mouse_hook_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,27 @@ def hid_runtime_state(self):
connected_device=self._connected_device,
)

def _should_intercept_events(self) -> bool:
"""True only when the platform hook should block, remap, or dispatch
OS-level mouse events to the engine.

Mouser exists to remap a Logitech mouse's buttons. The global event
taps on macOS (CGEventTap) and Windows (WH_MOUSE_LL) see events
from every input device the OS knows about -- when no Logitech is
currently bound to this host (KVM switched to another machine,
the device is mid-reconnect after sleep, or the user simply has
not plugged one in) those hooks must stay completely out of the
way, otherwise xbutton clicks and scroll events from a trackpad
or generic USB mouse get swallowed and routed through Mouser's
remap pipeline.

Linux's evdev hook only attaches once a Logitech source device
has been resolved, so it is naturally gated -- but consult this
property defensively before dispatching there as well so the
contract stays platform-uniform.
"""
return self._connected_device is not None

def dump_device_info(self):
hg = getattr(self, "_hid_gesture", None)
if hg and hasattr(hg, "dump_device_info"):
Expand Down
12 changes: 12 additions & 0 deletions core/mouse_hook_macos.py
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,18 @@ def _event_tap_callback(self, proxy, event_type, cg_event, refcon):
return cg_event
except Exception:
pass

# KVM / cold-start guard: when no Logitech is currently bound to
# this host, the CGEventTap must be a complete pass-through. The
# tap sees events from every mouse the OS knows about, so without
# this guard a trackpad swipe or a generic USB mouse's xbutton
# click would get routed through Mouser's remap pipeline -- the
# exact failure mode users hit when their KVM switches the
# Logitech to another machine while Mouser keeps running on
# this one.
if not self._should_intercept_events():
return cg_event

mouse_event = None
should_block = False

Expand Down
10 changes: 10 additions & 0 deletions core/mouse_hook_windows.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,16 @@ def _low_level_handler_inner(self, nCode, wParam, lParam):
if flags & INJECTED_FLAG:
return CallNextHookEx(self._hook, nCode, wParam, lParam)

# KVM / cold-start guard: when no Logitech is currently bound to
# this host, the WH_MOUSE_LL hook must be a complete pass-through.
# The hook sees events from every input device, so without this
# guard a trackpad scroll or generic USB mouse's xbutton click
# would still run through Mouser's remap pipeline -- the exact
# failure mode users hit when their KVM switches the Logitech
# to another machine while Mouser keeps running here.
if not self._should_intercept_events():
return CallNextHookEx(self._hook, nCode, wParam, lParam)

if wParam == WM_XBUTTONDOWN:
xbutton = hiword(mouse_data)
if xbutton == XBUTTON1:
Expand Down
124 changes: 124 additions & 0 deletions tests/test_mouse_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,35 @@ def test_status_callback_is_optional(self):

self.assertEqual(messages, ["Linux evdev remapping restored."])

def test_should_intercept_events_defaults_to_false(self):
"""Fresh hook, no Logitech bound -- platform taps must stand
down so non-Logitech mice are not silently remapped."""
hook = BaseMouseHook()

self.assertFalse(hook._should_intercept_events())

def test_should_intercept_events_flips_on_hid_connect(self):
hook = BaseMouseHook()
device = SimpleNamespace(name="MX Master 3S")
hook._hid_gesture = SimpleNamespace(connected_device=device)

hook._on_hid_connect()

self.assertTrue(hook._should_intercept_events())

def test_should_intercept_events_flips_off_on_hid_disconnect(self):
"""KVM switch / sleep wake: the moment HID++ drops the device,
the next OS event must pass through untouched."""
hook = BaseMouseHook()
device = SimpleNamespace(name="MX Master 3S")
hook._hid_gesture = SimpleNamespace(connected_device=device)
hook._on_hid_connect()
self.assertTrue(hook._should_intercept_events())

hook._on_hid_disconnect()

self.assertFalse(hook._should_intercept_events())



class BaseMouseHookDispatchQueueTests(unittest.TestCase):
Expand Down Expand Up @@ -899,6 +928,15 @@ def _make_hook(self):
hook.invert_vscroll = True
hook.block(mouse_hook.MouseEvent.HSCROLL_LEFT)
hook.block(mouse_hook.MouseEvent.HSCROLL_RIGHT)
# The event-tap callback now early-returns when no Logitech is bound
# (so KVM / cold-start scenarios never silently remap a trackpad or
# generic mouse). Pin a stub device here so the trackpad-filter
# tests exercise the path they actually mean to.
hook._connected_device = SimpleNamespace(
key="mx_master_3s",
thumb_button_via_hid=False,
gesture_via_sense_panel=False,
)
return hook

def _mock_get_field(self, is_continuous, source_user_data=0):
Expand Down Expand Up @@ -995,5 +1033,91 @@ def _get(event, field):
self.assertEqual(event.event_type, mouse_hook.MouseEvent.HSCROLL_RIGHT)


@unittest.skipUnless(sys.platform == "darwin", "macOS-only tests")
class MacOSPassthroughWhenNoDeviceTests(unittest.TestCase):
"""End-to-end pin of the KVM pass-through contract on the macOS event
tap. The CGEventTap is global -- it sees events from every input
device the OS knows about -- so any failure here is a user-visible
regression: trackpad swipes get inverted, generic-mouse xbuttons
get swallowed, and so on."""

_kCGEventScrollWheel = 22
_kCGEventOtherMouseDown = 25

def setUp(self):
self.mock_quartz = MagicMock(name="Quartz")
self.mock_quartz.kCGEventScrollWheel = self._kCGEventScrollWheel
self.mock_quartz.kCGEventOtherMouseDown = self._kCGEventOtherMouseDown
mouse_hook.Quartz = self.mock_quartz

def tearDown(self):
if hasattr(mouse_hook, "Quartz") and isinstance(
mouse_hook.Quartz, MagicMock):
del mouse_hook.Quartz

def _bare_hook(self):
hook = mouse_hook.MouseHook()
hook._running = True
hook._tap = MagicMock(name="tap")
hook.invert_vscroll = True
hook.invert_hscroll = True
hook.block(mouse_hook.MouseEvent.XBUTTON1_DOWN)
hook.block(mouse_hook.MouseEvent.HSCROLL_RIGHT)
# No _connected_device assignment -- this is exactly the KVM state
# we are pinning behavior for.
return hook

def test_scroll_passes_through_when_no_logitech_connected(self):
hook = self._bare_hook()
cg_event = MagicMock(name="cg_event")
self.mock_quartz.CGEventGetIntegerValueField.return_value = 0

result = hook._event_tap_callback(
None, self._kCGEventScrollWheel, cg_event, None)

self.assertIs(result, cg_event)
self.assertTrue(hook._dispatch_queue.empty())

def test_xbutton_passes_through_when_no_logitech_connected(self):
hook = self._bare_hook()
cg_event = MagicMock(name="cg_event")
self.mock_quartz.CGEventGetIntegerValueField.return_value = 0

result = hook._event_tap_callback(
None, self._kCGEventOtherMouseDown, cg_event, None)

self.assertIs(result, cg_event)
self.assertTrue(hook._dispatch_queue.empty())

def test_intercept_resumes_after_logitech_reconnects(self):
"""KVM switches back: the very next event after _on_hid_connect
must be intercepted, not waited-on for some other trigger."""
hook = self._bare_hook()
cg_event = MagicMock(name="cg_event")
self.mock_quartz.CGEventGetIntegerValueField.return_value = 0

first = hook._event_tap_callback(
None, self._kCGEventScrollWheel, cg_event, None)
self.assertIs(first, cg_event)

hook._connected_device = SimpleNamespace(
key="mx_master_3s",
thumb_button_via_hid=False,
gesture_via_sense_panel=False,
)

def _get(event, field):
if field == self.mock_quartz.kCGScrollWheelEventFixedPtDeltaAxis2:
return 5 * 65536
return 0
self.mock_quartz.CGEventGetIntegerValueField.side_effect = _get

second = hook._event_tap_callback(
None, self._kCGEventScrollWheel, cg_event, None)

self.assertIsNone(second)
self.assertFalse(hook._dispatch_queue.empty())


if __name__ == "__main__":
unittest.main()
Loading