diff --git a/CHANGELOG.md b/CHANGELOG.md index 235031f..f0ffa36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## Unreleased + +- capture macOS desktops with Quartz `CGDisplayCreateImage` instead of Pillow's + `screencapture` helper, which fails over SSH +- keep macOS mouse coordinates in `CGDisplayPixelsWide`/`High` space +- restore Accessibility checks on PyObjC 12, where `Quartz.AXIsProcessTrusted` + is no longer exported + ## 0.4.5 - adapt display FPS under terminal RTT/write backpressure so slow clients stop accumulating visual lag diff --git a/docs/platforms.md b/docs/platforms.md index 0b01f11..37656c5 100644 --- a/docs/platforms.md +++ b/docs/platforms.md @@ -51,10 +51,16 @@ provide none of the listed capture interfaces need a backend adapter. ## macOS host -`NativeCapture` uses Pillow's CoreGraphics path and `MacOSInput` uses Quartz. -Grant Screen Recording and Accessibility permission to the installed Python -binary. A manually launched server in the logged-in Aqua session is supported; -OpenSSH daemon access depends on macOS TCC/session policy. +`NativeCapture` uses Quartz `CGDisplayCreateImage` and `MacOSInput` uses Quartz +event injection. Pillow's macOS `ImageGrab` path shells out to `screencapture`, +which fails from OpenSSH (`could not create image from display`) even when the +Python process already has Screen Recording permission. Quartz capture stays in +that process, so SSH sessions work after TCC is granted. Newer PyObjC builds no +longer export `AXIsProcessTrusted` on the `Quartz` module; input falls back to +the ApplicationServices C API. Grant Screen Recording and Accessibility +permission to the installed Python binary. A manually launched server in the +logged-in Aqua session is supported; OpenSSH daemon access still depends on +macOS TCC/session policy. ## Windows host diff --git a/src/sshdesk/capture/native.py b/src/sshdesk/capture/native.py index 7ae97fc..da5fe4d 100644 --- a/src/sshdesk/capture/native.py +++ b/src/sshdesk/capture/native.py @@ -3,14 +3,50 @@ import hashlib import platform import time +from typing import Any from PIL import Image, ImageGrab from .base import Frame, ScreenCapture +def _quartz_bitmap_to_rgb(quartz: Any, cgimage: object) -> Image.Image: + width = int(quartz.CGImageGetWidth(cgimage)) + height = int(quartz.CGImageGetHeight(cgimage)) + bytes_per_row = int(quartz.CGImageGetBytesPerRow(cgimage)) + provider = quartz.CGImageGetDataProvider(cgimage) + copied = quartz.CGDataProviderCopyData(provider) if provider is not None else None + if width < 1 or height < 1 or copied is None: + raise RuntimeError("desktop capture failed; empty display image") + raw = bytes(copied) + if len(raw) < bytes_per_row * height: + raise RuntimeError("desktop capture failed; incomplete pixel buffer") + return ( + Image.frombuffer("RGBA", (width, height), raw, "raw", "BGRA", bytes_per_row, 1) + .convert("RGB") + .copy() + ) + + +def grab_macos_display(quartz: Any) -> Image.Image: + display = quartz.CGMainDisplayID() + cgimage = quartz.CGDisplayCreateImage(display) + if cgimage is None: + raise RuntimeError( + "desktop capture failed; grant Screen Recording permission to the SSH/Python process" + ) + image = _quartz_bitmap_to_rgb(quartz, cgimage) + logical = ( + int(quartz.CGDisplayPixelsWide(display)), + int(quartz.CGDisplayPixelsHigh(display)), + ) + if logical[0] >= 1 and logical[1] >= 1 and image.size != logical: + return image.resize(logical, Image.Resampling.BICUBIC) + return image + + class NativeCapture(ScreenCapture): - """Pillow-backed desktop capture for Windows and macOS.""" + """Native desktop capture for Windows (Pillow) and macOS (Quartz).""" def __init__(self) -> None: self.system = platform.system() @@ -19,16 +55,24 @@ def __init__(self) -> None: self._target_size: tuple[int, int] | None = None self._desktop_size = self._grab().size + def _grab_macos(self) -> Image.Image: + try: + import Quartz + except ImportError as exc: + raise RuntimeError( + "macOS capture needs the macOS extra: pip install 'sshdesk[macos]'" + ) from exc + return grab_macos_display(Quartz) + def _grab(self) -> Image.Image: + if self.system == "Darwin": + return self._grab_macos() try: - image = ImageGrab.grab(all_screens=self.system == "Windows") + image = ImageGrab.grab(all_screens=True) except OSError as exc: - permission = ( - "grant Screen Recording permission to the SSH/Python process" - if self.system == "Darwin" - else "run SSHDESK in the logged-in interactive Windows session" - ) - raise RuntimeError(f"desktop capture failed; {permission}") from exc + raise RuntimeError( + "desktop capture failed; run SSHDESK in the logged-in interactive Windows session" + ) from exc if image.mode != "RGB": image = image.convert("RGB") return image diff --git a/src/sshdesk/input/macos.py b/src/sshdesk/input/macos.py index 3eccf00..18c6fac 100644 --- a/src/sshdesk/input/macos.py +++ b/src/sshdesk/input/macos.py @@ -1,5 +1,7 @@ from __future__ import annotations +import ctypes +import ctypes.util from typing import Any from .base import InputBackend @@ -34,6 +36,18 @@ } +def process_is_trusted(quartz: Any) -> bool: + checker = getattr(quartz, "AXIsProcessTrusted", None) + if callable(checker): + return bool(checker()) + path = ctypes.util.find_library("ApplicationServices") + if not path: + return False + library = ctypes.cdll.LoadLibrary(path) + library.AXIsProcessTrusted.restype = ctypes.c_bool + return bool(library.AXIsProcessTrusted()) + + class MacOSInput(InputBackend): """macOS Quartz input injection with Accessibility permission checks.""" @@ -45,7 +59,7 @@ def __init__(self) -> None: "macOS input needs the macOS extra: pip install 'sshdesk[macos]'" ) from exc self.q: Any = Quartz - if not Quartz.AXIsProcessTrusted(): + if not process_is_trusted(Quartz): raise RuntimeError( "grant Accessibility permission to the SSH/Python process for input control" ) diff --git a/tests/test_capture.py b/tests/test_capture.py index e0da7bb..8a412c3 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -8,9 +8,98 @@ from PIL import Image from sshdesk.capture.gnome import GnomeScreenCastCapture +from sshdesk.capture.native import NativeCapture, grab_macos_display from sshdesk.capture.wayland import WaylandCapture +class _FakeQuartz: + def __init__( + self, + *, + raw: bytes, + width: int, + height: int, + bytes_per_row: int, + logical: tuple[int, int], + cgimage: object = "cgimage", + ) -> None: + self.raw = raw + self.width = width + self.height = height + self.bytes_per_row = bytes_per_row + self.logical = logical + self.cgimage = cgimage + + def CGMainDisplayID(self) -> int: + return 1 + + def CGDisplayCreateImage(self, _display: int) -> object | None: + return self.cgimage + + def CGImageGetWidth(self, _image: object) -> int: + return self.width + + def CGImageGetHeight(self, _image: object) -> int: + return self.height + + def CGImageGetBytesPerRow(self, _image: object) -> int: + return self.bytes_per_row + + def CGImageGetDataProvider(self, _image: object) -> object: + return "provider" + + def CGDataProviderCopyData(self, _provider: object) -> bytes: + return self.raw + + def CGDisplayPixelsWide(self, _display: int) -> int: + return self.logical[0] + + def CGDisplayPixelsHigh(self, _display: int) -> int: + return self.logical[1] + + +class NativeCaptureTests(unittest.TestCase): + def test_macos_quartz_capture_resizes_to_logical_pixels(self) -> None: + # 2x2 BGRA: red, green / blue, white. Logical size is 1x1. + raw = bytes( + [ + 0, 0, 255, 255, + 0, 255, 0, 255, + 255, 0, 0, 255, + 255, 255, 255, 255, + ] + ) + image = grab_macos_display( + _FakeQuartz(raw=raw, width=2, height=2, bytes_per_row=8, logical=(1, 1)) + ) + self.assertEqual(image.size, (1, 1)) + self.assertEqual(image.mode, "RGB") + + def test_macos_quartz_capture_keeps_matching_logical_size(self) -> None: + raw = bytes([0, 0, 255, 255, 255, 255, 255, 255]) + image = grab_macos_display( + _FakeQuartz(raw=raw, width=2, height=1, bytes_per_row=8, logical=(2, 1)) + ) + self.assertEqual(image.size, (2, 1)) + self.assertEqual(image.getpixel((0, 0)), (255, 0, 0)) + + def test_macos_quartz_missing_frame_asks_for_screen_recording(self) -> None: + quartz = _FakeQuartz( + raw=b"", width=0, height=0, bytes_per_row=0, logical=(1, 1), cgimage=None + ) + with self.assertRaisesRegex(RuntimeError, "Screen Recording"): + grab_macos_display(quartz) + + def test_macos_grab_does_not_call_pillow_screencapture(self) -> None: + capture = object.__new__(NativeCapture) + capture.system = "Darwin" + capture._grab_macos = lambda: Image.new("RGB", (10, 5), (1, 2, 3)) + with patch("sshdesk.capture.native.ImageGrab.grab") as grab: + image = capture._grab() + grab.assert_not_called() + self.assertEqual(image.size, (10, 5)) + + class WaylandCaptureTests(unittest.TestCase): def test_capture_helper_timeout_becomes_clean_runtime_error(self) -> None: command = ["gnome-screenshot", "-f", "/tmp/frame.png"] diff --git a/tests/test_input.py b/tests/test_input.py index b9779d2..377f090 100644 --- a/tests/test_input.py +++ b/tests/test_input.py @@ -2,6 +2,7 @@ import unittest from types import SimpleNamespace +from unittest.mock import patch from sshdesk.input.events import ( ControlEvent, @@ -14,6 +15,7 @@ MouseScrollEvent, TerminalReportEvent, ) +from sshdesk.input.macos import process_is_trusted from sshdesk.input.mutter import MutterInput from sshdesk.input.terminal import TerminalEventParser, translate_coordinates from sshdesk.render.base import Viewport @@ -21,6 +23,34 @@ from sshdesk.session.direct import DirectSession +class MacOSTrustTests(unittest.TestCase): + def test_process_is_trusted_uses_quartz_when_exported(self) -> None: + self.assertTrue(process_is_trusted(SimpleNamespace(AXIsProcessTrusted=lambda: True))) + self.assertFalse(process_is_trusted(SimpleNamespace(AXIsProcessTrusted=lambda: False))) + + def test_process_is_trusted_falls_back_to_application_services(self) -> None: + class FakeLibrary: + def __init__(self) -> None: + self.AXIsProcessTrusted = lambda: True + self.AXIsProcessTrusted.restype = None + + loaded: list[str] = [] + + def load_library(path: str) -> FakeLibrary: + loaded.append(path) + return FakeLibrary() + + with patch("sshdesk.input.macos.ctypes.util.find_library", return_value="/AS"), patch( + "sshdesk.input.macos.ctypes.cdll.LoadLibrary", side_effect=load_library + ): + self.assertTrue(process_is_trusted(SimpleNamespace())) + self.assertEqual(loaded, ["/AS"]) + + def test_process_is_trusted_is_false_without_application_services(self) -> None: + with patch("sshdesk.input.macos.ctypes.util.find_library", return_value=None): + self.assertFalse(process_is_trusted(SimpleNamespace())) + + class InputTests(unittest.TestCase): def test_key_mapping(self) -> None: parser = TerminalEventParser()