XCB: Verify visual, and improve error messages

The MSS object, when it's created, will now ensure that the X11
visual's format is exactly what we think it is: 32bpp, as a 24-bit
BGRx or 32-bit BGRA image, in that order, with no extra scanline
padding, color indexing, etc.  This is true of all modern X servers in
real-world situations, but it's not guaranteed.  For instance,
VNC-based servers can be run in depth 16 with RGB565 formats.

The error messages for protocol-level errors (such as bad coordinates)
are only available if the xcb-util-errors library is installed.
(That's libxcb-errors.so.0, in the libxcb-errors0 package on
Debian/Ubuntu.)  This isn't widely installed, and is mostly useful to
developers, so it's optional.  The connection-level errors (like the
hint to check $DISPLAY) are available everywhere.

Author: jholveck

src/mss/linux/xcb.py

@@ -171,36 +171,65 @@ class XError(ScreenShotError):
 class XProtoError(XError):
     """Exception indicating server-reported errors."""
 
-    def __init__(self, _xcb_conn: XcbConnection, xcb_err: XcbGenericErrorStructure) -> None:
+    def __init__(self, xcb_conn: XcbConnection, xcb_err: XcbGenericErrorStructure) -> None:
         if isinstance(xcb_err, _Pointer):
             xcb_err = xcb_err.contents
-        # I would verify isinstance(xcb_err,
-        # XcbGenericErrorStructure), but ruff doesn't really give me a
-        # great way to throw an appropriate exception here.
-        # FIXME Get the text descriptions, if possible.  That's why we
-        # got the connection handle: that can be necessary to get error
-        # descriptions for extensions.
-        super().__init__(
-            "X11 Protocol Error",
-            details={
-                "error_code": xcb_err.error_code,
-                "sequence": xcb_err.sequence,
-                "resource_id": xcb_err.resource_id,
-                "minor_code": xcb_err.minor_code,
-                "major_code": xcb_err.major_code,
-                "full_sequence": xcb_err.full_sequence,
-            },
-        )
+        assert isinstance(xcb_err, XcbGenericErrorStructure)  # noqa: S101
+
+        details = {
+            "error_code": xcb_err.error_code,
+            "sequence": xcb_err.sequence,
+            "resource_id": xcb_err.resource_id,
+            "minor_code": xcb_err.minor_code,
+            "major_code": xcb_err.major_code,
+            "full_sequence": xcb_err.full_sequence,
+        }
+
+        # xcb-errors is a library to get descriptive error strings, instead of reporting the raw codes.  This is not
+        # installed by default on most systems, but is quite helpful for developers.  We use it if it exists, but don't
+        # force the matter.  We can't do this when we format the error message, since the XCB connection may be gone.
+        if xcb_errors:
+            # We don't try to reuse the error context, since it's per-connection, and probably will only be used once.
+            ctx = POINTER(XcbErrorsContext)()
+            ctx_new_setup = xcb_errors.xcb_errors_context_new(xcb_conn, ctx)
+            if ctx_new_setup == 0:
+                try:
+                    # Some of these may return NULL, but some are guaranteed.
+                    ext_name = POINTER(c_char_p)()
+                    error_name = xcb_errors.xcb_errors_get_name_for_error(ctx, xcb_err.error_code, ext_name)
+                    details["error"] = error_name.decode("ascii", errors="replace")
+                    if ext_name:
+                        details["extension"] = ext_name.decode("ascii", errors="replace")
+                    major_name = xcb_errors.xcb_errors_get_name_for_major_code(ctx, xcb_err.major_code)
+                    details["major_name"] = major_name.decode("ascii", errors="replace")
+                    minor_name = xcb_errors.xcb_errors_get_name_for_minor_code(
+                        ctx, xcb_err.major_code, xcb_err.minor_code
+                    )
+                    if minor_name:
+                        details["minor_name"] = minor_name.decode("ascii", errors="replace")
+                finally:
+                    xcb_errors.xcb_errors_context_free(ctx)
+
+        super().__init__("X11 Protocol Error", details=details)
 
     def __str__(self) -> str:
         msg = super().__str__()
-        err = self.details
+        details = self.details
+        error_desc = f"{details['error_code']} ({details['error']})" if "error" in details else details["error_code"]
+        major_desc = (
+            f"{details['major_code']} ({details['major_name']})" if "major_name" in details else details["major_code"]
+        )
+        minor_desc = (
+            f"{details['minor_code']} ({details['minor_name']})" if "minor_name" in details else details["minor_code"]
+        )
+        ext_desc = f"\n  Extension:  {details['ext_name']}" if "ext_desc" in details else ""
         msg += (
-            f"\nX Error of failed request:  {err['error_code']}"
-            f"\n  Major opcode of failed request:  {err['major_code']}"
-            f"\n  Minor opcode of failed request:  {err['minor_code']}"
-            f"\n  Resource id in failed request:  {err['resource_id']}"
-            f"\n  Serial number of failed request:  {err['full_sequence']}"
+            f"\nX Error of failed request:  {error_desc}"
+            f"\n  Major opcode of failed request:  {major_desc}"
+            f"{ext_desc}"
+            f"\n  Minor opcode of failed request:  {minor_desc}"
+            f"\n  Resource id in failed request:  {details['resource_id']}"
+            f"\n  Serial number of failed request:  {details['full_sequence']}"
         )
         return msg
 
@@ -234,7 +263,6 @@ def check(self, xcb_conn: XcbConnection) -> None:
 
         This will raise an exception if there is an error.
         """
-        self._xcb_handled_ = True
         err_p = libxcb.xcb_request_check(xcb_conn, self)
         if not err_p:
             return
@@ -555,11 +583,30 @@ class XcbQueryExtensionReply(Structure):
 # for many core X requests.
 XCB_NONE = XID(0)
 XCB_ATOM_WINDOW = XcbAtom(33)
+XCB_CONN_ERROR = 1
+XCB_CONN_CLOSED_EXT_NOTSUPPORTED = 2
+XCB_CONN_CLOSED_MEM_INSUFFICIENT = 3
+XCB_CONN_CLOSED_REQ_LEN_EXCEED = 4
+XCB_CONN_CLOSED_PARSE_ERR = 5
+XCB_CONN_CLOSED_INVALID_SCREEN = 6
+XCB_CONN_CLOSED_FDPASSING_FAILED = 7
 XCB_IMAGE_FORMAT_Z_PIXMAP = 2
 XCB_IMAGE_ORDER_LSB_FIRST = 0
 XCB_VISUAL_CLASS_TRUE_COLOR = 4
 XCB_VISUAL_CLASS_DIRECT_COLOR = 5
 
+# I don't know of error descriptions for the XCB connection errors being accessible through a library (a la strerror),
+# and the ones in xcb.h's comments aren't too great, so I wrote these.
+XCB_CONN_ERRMSG = {
+    XCB_CONN_ERROR: "connection lost or could not be established",
+    XCB_CONN_CLOSED_EXT_NOTSUPPORTED: "extension not supported",
+    XCB_CONN_CLOSED_MEM_INSUFFICIENT: "memory exhausted",
+    XCB_CONN_CLOSED_REQ_LEN_EXCEED: "request length longer than server accepts",
+    XCB_CONN_CLOSED_PARSE_ERR: "display is unset or invalid (check $DISPLAY)",
+    XCB_CONN_CLOSED_INVALID_SCREEN: "server does not have a screen matching the requested display",
+    XCB_CONN_CLOSED_FDPASSING_FAILED: "could not pass file descriptor",
+}
+
 # randr
 
 
@@ -769,6 +816,19 @@ class XcbXfixesGetCursorImageReply(Structure):
 XCB_XFIXES_MAJOR_VERSION = 6
 XCB_XFIXES_MINOR_VERSION = 0
 
+# xcb-errors
+
+
+class XcbErrorsContext(Structure):
+    """A context for using this library.
+
+    Create a context with xcb_errors_context_new() and destroy it with xcb_errors_context_free(). Except for
+    xcb_errors_context_free(), all functions in this library are thread-safe and can be called from multiple threads at
+    the same time, even on the same context.
+    """
+
+    # Opaque
+
 
 #### ctypes initialization
 
@@ -784,7 +844,7 @@ def initialize() -> None:
         if _INITIALIZE_DONE:
             return
 
-        global libc, libxcb, randr, xcb_randr_id, render, xcb_render_id, xfixes, xcb_xfixes_id
+        global libc, libxcb, randr, xcb_randr_id, render, xcb_render_id, xfixes, xcb_xfixes_id, xcb_errors
 
         # We don't use the cached versions that ctypes.cdll exposes as
         # attributes, since other libraries may be doing their own
@@ -958,6 +1018,24 @@ def initialize() -> None:
         xfixes.xcb_xfixes_get_cursor_image_cursor_image_length.argtypes = [POINTER(XcbXfixesGetCursorImageReply)]
         xfixes.xcb_xfixes_get_cursor_image_cursor_image_length.restype = c_int
 
+        # xcb_errors is an optional library, mostly only useful to developers.
+        # We use the qualified .so name, since it's subject to change.
+        try:
+            xcb_errors = cdll.LoadLibrary("libxcb-errors.so.0")
+        except Exception:  # noqa: BLE001
+            xcb_errors = None
+        else:
+            xcb_errors.xcb_errors_context_new.argtypes = [POINTER(XcbConnection), POINTER(POINTER(XcbErrorsContext))]
+            xcb_errors.xcb_errors_context_new.restype = c_int
+            xcb_errors.xcb_errors_context_free.argtypes = [POINTER(XcbErrorsContext)]
+            xcb_errors.xcb_errors_context_free.restype = None
+            xcb_errors.xcb_errors_get_name_for_major_code.argtypes = [POINTER(XcbErrorsContext), c_uint8]
+            xcb_errors.xcb_errors_get_name_for_major_code.restype = c_char_p
+            xcb_errors.xcb_errors_get_name_for_minor_code.argtypes = [POINTER(XcbErrorsContext), c_uint8, c_uint16]
+            xcb_errors.xcb_errors_get_name_for_minor_code.restype = c_char_p
+            xcb_errors.xcb_errors_get_name_for_error.argtypes = [POINTER(XcbErrorsContext), c_uint8, POINTER(c_char_p)]
+            xcb_errors.xcb_errors_get_name_for_error.restype = c_char_p
+
         _INITIALIZE_DONE = True
 
 
@@ -1290,22 +1368,32 @@ def connect(display: str | bytes | None = None) -> tuple[XcbConnection, int]:
 
     pref_screen_num = c_int()
     conn_p = libxcb.xcb_connect(display, pref_screen_num)
-    if libxcb.xcb_connection_has_error(conn_p) != 0:
-        # FIXME Get the error information
-        msg = "Cannot connect to display"
+
+    # We still get a connection object even if the connection fails.
+    conn_err = libxcb.xcb_connection_has_error(conn_p)
+    if conn_err != 0:
+        msg = "Cannot connect to display: "
+        conn_errmsg = XCB_CONN_ERRMSG.get(conn_err)
+        if conn_errmsg:
+            msg += conn_errmsg
+        else:
+            msg += f"error code {conn_err}"
         raise XError(msg)
 
     return conn_p.contents, pref_screen_num.value
 
 
 def disconnect(conn: XcbConnection) -> None:
-    error_status = libxcb.xcb_connection_has_error(conn)
-    # XCB won't free its connection structures until we disconnect,
-    # even in the event of an error.
+    conn_err = libxcb.xcb_connection_has_error(conn)
+    # XCB won't free its connection structures until we disconnect, even in the event of an error.
     libxcb.xcb_disconnect(conn)
-    if error_status != 0:
-        # FIXME Get the error information
-        msg = "Connection closed due to errors"
+    if conn_err != 0:
+        msg = "Connection to X server closed: "
+        conn_errmsg = XCB_CONN_ERRMSG.get(conn_err)
+        if conn_errmsg:
+            msg += conn_errmsg
+        else:
+            msg += f"error code {conn_err}"
         raise XError(msg)
 
 

src/mss/linux/xgetimage.py

@@ -11,13 +11,17 @@
 from . import xcb
 from .xcb import (
     XCB_IMAGE_FORMAT_Z_PIXMAP,
+    XCB_IMAGE_ORDER_LSB_FIRST,
     XCB_RANDR_MAJOR_VERSION,
     XCB_RANDR_MINOR_VERSION,
+    XCB_VISUAL_CLASS_DIRECT_COLOR,
+    XCB_VISUAL_CLASS_TRUE_COLOR,
     XCB_XFIXES_MAJOR_VERSION,
     XCB_XFIXES_MINOR_VERSION,
     connect,
     disconnect,
     initialize,
+    xcb_depth_visuals,
     xcb_get_geometry,
     xcb_get_image,
     xcb_get_image_data,
@@ -27,14 +31,19 @@
     xcb_randr_get_screen_resources_current,
     xcb_randr_get_screen_resources_current_crtcs,
     xcb_randr_query_version,
+    xcb_screen_allowed_depths,
+    xcb_setup_pixmap_formats,
     xcb_setup_roots,
     xcb_xfixes_get_cursor_image,
     xcb_xfixes_get_cursor_image_cursor_image,
     xcb_xfixes_query_version,
 )
 
 SUPPORTED_DEPTHS = {24, 32}
-SUPPORTED_BYTES_PER_PIXEL = 32
+SUPPORTED_BITS_PER_PIXEL = 32
+SUPPORTED_RED_MASK = 0xFF0000
+SUPPORTED_GREEN_MASK = 0x00FF00
+SUPPORTED_BLUE_MASK = 0x0000FF
 ALL_PLANES = 0xFFFFFFFF  # XCB doesn't define AllPlanes
 
 
@@ -47,7 +56,7 @@ class MSS(MSSBase):
     * XFixes: Including the cursor.
     """
 
-    def __init__(self, /, **kwargs: Any) -> None:
+    def __init__(self, /, **kwargs: Any) -> None:  # noqa: PLR0912
         super().__init__(**kwargs)
 
         display = kwargs.get("display", b"")
@@ -72,6 +81,71 @@ def __init__(self, /, **kwargs: Any) -> None:
         # We don't probe the XFixes presence or version until we need it.
         self._xfixes_ready = None
 
+        # Probe the visuals (and related information), and make sure that our drawable is in an acceptable format.
+        # These iterations and tests don't involve any traffic with the server; it's all stuff that was included in the
+        # connection setup.  Effectively all modern setups will be acceptable, but we verify to be sure.
+
+        # Currently, we assume that the drawable we're capturing is the root; when we add single-window capture, we'll
+        # have to ask the server for its depth and visual.
+        assert self.root == self.drawable  # noqa: S101
+        self.drawable_depth = pref_screen.root_depth
+        self.drawable_visual_id = pref_screen.root_visual.value
+        # Server image byte order
+        if xcb_setup.image_byte_order != XCB_IMAGE_ORDER_LSB_FIRST:
+            msg = "Only X11 servers using LSB-First images are supported."
+            raise ScreenShotError(msg)
+        # Depth
+        if self.drawable_depth not in SUPPORTED_DEPTHS:
+            msg = f"Only screens of color depth 24 or 32 are supported, not {self.drawable_depth}"
+            raise ScreenShotError(msg)
+        # Format (i.e., bpp, padding)
+        for format_ in xcb_setup_pixmap_formats(xcb_setup):
+            if format_.depth == self.drawable_depth:
+                break
+        else:
+            msg = "Internal error: drawable's depth not found in screen's supported formats"
+            raise ScreenShotError(msg)
+        drawable_format = format_
+        if drawable_format.bits_per_pixel != SUPPORTED_BITS_PER_PIXEL:
+            msg = (
+                f"Only screens at 32 bpp (regardless of color depth) are supported; "
+                f"got {drawable_format.bits_per_pixel} bpp"
+            )
+            raise ScreenShotError(msg)
+        if drawable_format.scanline_pad != SUPPORTED_BITS_PER_PIXEL:
+            # To clarify the padding: the scanline_pad is the multiple that the scanline gets padded to.  If there is
+            # no padding, then it will be the same as one pixel's size.
+            msg = "Screens with scanline padding are not supported"
+            raise ScreenShotError(msg)
+        # Visual, the interpretation of pixels (like indexed, grayscale, etc).  (Visuals are arranged by depth, so we
+        # iterate over the depths first.)
+        for xcb_depth in xcb_screen_allowed_depths(pref_screen):
+            if xcb_depth.depth == self.drawable_depth:
+                break
+        else:
+            msg = "Internal error: drawable's depth not found in screen's supported depths"
+            raise ScreenShotError(msg)
+        for visual_info in xcb_depth_visuals(xcb_depth):
+            if visual_info.visual_id.value == self.drawable_visual_id:
+                break
+        else:
+            msg = "Internal error: drawable's visual not found in screen's supported visuals"
+            raise ScreenShotError(msg)
+        if visual_info.class_ not in {XCB_VISUAL_CLASS_TRUE_COLOR, XCB_VISUAL_CLASS_DIRECT_COLOR}:
+            msg = "Only TrueColor and DirectColor visuals are supported"
+            raise ScreenShotError(msg)
+        if (
+            visual_info.red_mask != SUPPORTED_RED_MASK
+            or visual_info.green_mask != SUPPORTED_GREEN_MASK
+            or visual_info.blue_mask != SUPPORTED_BLUE_MASK
+        ):
+            # There are two ways to phrase this layout: BGRx accounts for the byte order, while xRGB implies the native
+            # word order.  Since we return the data as a byte array, we use the former.  By the time we get to this
+            # point, we've already checked the endianness and depth, so this is pretty much never going to happen
+            # anyway.
+            msg = "Only visuals with BGRx ordering are supported"
+            raise ScreenShotError(msg)
+
     def close(self) -> None:
         if self.conn is not None:
             disconnect(self.conn)
@@ -159,18 +233,13 @@ def _grab_impl(self, monitor: Monitor, /) -> ScreenShot:
         # we clear the image structure.
         img_data = bytearray(img_data_arr)
 
-        # FIXME In xcb.main(), there's a more complete implementation
-        # of checking the bit depth.  This includes checking the
-        # visual type, byte order, and also determines whether the
-        # alpha channel is real or just padding.
-        #
-        # In XCB, we don't get the bits per pixel with the image; it's
-        # actually part of the visual data.  (Xlib puts the necessary
-        # visual data in the image structure automatically.)  So for
-        # now, we just check the depth, until I get around to checking
-        # the full visual.
-        if img_reply.depth not in SUPPORTED_DEPTHS:
-            msg = f"[GetImage] depth value not implemented: {img_reply.depth}."
+        if img_reply.depth != self.drawable_depth or img_reply.visual.value != self.drawable_visual_id:
+            # This should never happen; a window can't change its visual.
+            msg = (
+                "Server returned an image with a depth or visual different than it initially reported: "
+                f"expected {self.drawable_depth},{hex(self.drawable_visual_id)}, "
+                f"got {img_reply.depth},{hex(img_reply.visual.value)}"
+            )
             raise ScreenShotError(msg)
 
         return self.cls_image(img_data, monitor)
@@ -184,7 +253,7 @@ def _cursor_impl_check_xfixes(self) -> bool:
         # We can work with 2.0 and later, but not sure about the
         # actual minimum version we can use.  That's ok; everything
         # these days is much more modern.
-        return not (reply.major_version, reply.minor_version) < (2, 0)
+        return (reply.major_version, reply.minor_version) >= (2, 0)
 
     def _cursor_impl(self) -> ScreenShot:
         """Retrieve all cursor data. Pixels have to be RGBx."""

src/mss/linux/xlib.py

@@ -271,7 +271,7 @@ def __str__(self) -> str:
             if self.details["request_code"] >= X_FIRST_EXTENSION_OPCODE:
                 msg += f"\n  Minor opcode of failed request:  {self.details['minor_code']}"
             msg += (
-                f"\n  Resource id in failed request:  {self.details['resourceid']}"
+                f"\n  Resource id in failed request:  {self.details['resourceid'].value}"
                 f"\n  Serial number of failed request:  {self.details['serial']}"
             )
         return msg