Skip to content
Merged
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
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -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
Expand Down
14 changes: 10 additions & 4 deletions docs/platforms.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
60 changes: 52 additions & 8 deletions src/sshdesk/capture/native.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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
Expand Down
16 changes: 15 additions & 1 deletion src/sshdesk/input/macos.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
from __future__ import annotations

import ctypes
import ctypes.util
from typing import Any

from .base import InputBackend
Expand Down Expand Up @@ -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."""

Expand All @@ -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"
)
Expand Down
89 changes: 89 additions & 0 deletions tests/test_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down
30 changes: 30 additions & 0 deletions tests/test_input.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import unittest
from types import SimpleNamespace
from unittest.mock import patch

from sshdesk.input.events import (
ControlEvent,
Expand All @@ -14,13 +15,42 @@
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
from sshdesk.render.kitty import PixelViewport, translate_pixel_coordinates
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()
Expand Down
Loading