diff --git a/pyproject.toml b/pyproject.toml
index 249c1af8e..b54c9f489 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -63,6 +63,7 @@ qvm-prefs = "qubesadmin.tools.qvm_prefs:main"
qvm-remove = "qubesadmin.tools.qvm_remove:main"
qvm-run = "qubesadmin.tools.qvm_run:main"
qvm-service = "qubesadmin.tools.qvm_service:main"
+qvm-set-wallpaper = "qubesadmin.tools.qvm_set_wallpaper:main"
qvm-shutdown = "qubesadmin.tools.qvm_shutdown:main"
qvm-start = "qubesadmin.tools.qvm_start:main"
qvm-start-daemon = "qubesadmin.tools.qvm_start_daemon:main"
diff --git a/qubesadmin/tests/tools/qvm_set_wallpaper.py b/qubesadmin/tests/tools/qvm_set_wallpaper.py
new file mode 100644
index 000000000..edfb8dfb7
--- /dev/null
+++ b/qubesadmin/tests/tools/qvm_set_wallpaper.py
@@ -0,0 +1,273 @@
+# -*- encoding: utf-8 -*-
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, see .
+
+# pylint: disable=missing-docstring
+
+import os
+import struct
+import tempfile
+import zlib
+from unittest.mock import patch, MagicMock
+
+import qubesadmin.tests
+import qubesadmin.tools.qvm_set_wallpaper
+
+
+class TC_00_rgba_to_png(qubesadmin.tests.QubesTestCase):
+
+ def test_000_valid_png_signature(self):
+ """Output starts with the 8-byte PNG magic signature."""
+ rgba = b"\xff\x00\x00\xff" * 4 # 2x2 red image
+ png = qubesadmin.tools.qvm_set_wallpaper.rgba_to_png(rgba, 2, 2)
+ self.assertTrue(png.startswith(b"\x89PNG\r\n\x1a\n"))
+
+ def test_001_valid_png_chunks(self):
+ """First chunk is IHDR and file ends with IEND."""
+ rgba = b"\xff\x00\x00\xff" * 4
+ png = qubesadmin.tools.qvm_set_wallpaper.rgba_to_png(rgba, 2, 2)
+ # After 8-byte signature, first chunk should be IHDR
+ self.assertEqual(png[12:16], b"IHDR")
+ # PNG must end with IEND
+ self.assertTrue(png.endswith(b"\x00\x00\x00\x00IEND\xaeB`\x82"))
+
+ def test_002_ihdr_dimensions(self):
+ """IHDR chunk encodes the correct width and height."""
+ rgba = b"\x00" * (3 * 5 * 4)
+ png = qubesadmin.tools.qvm_set_wallpaper.rgba_to_png(rgba, 3, 5)
+ # IHDR data starts at offset 16 (8 sig + 4 len + 4 type)
+ width, height = struct.unpack(">II", png[16:24])
+ self.assertEqual(width, 3)
+ self.assertEqual(height, 5)
+
+ def test_003_roundtrip_idat(self):
+ """Decompressed IDAT contains filter-byte-prefixed scanlines."""
+ rgba = b"\xff\x00\x00\xff\x00\xff\x00\xff" # 2x1
+ png = qubesadmin.tools.qvm_set_wallpaper.rgba_to_png(rgba, 2, 1)
+ # Find IDAT chunk
+ idx = png.index(b"IDAT")
+ idat_len = struct.unpack(">I", png[idx - 4 : idx])[0]
+ idat_data = png[idx + 4 : idx + 4 + idat_len]
+ raw = zlib.decompress(idat_data)
+ # Should be filter byte (0) + 2 pixels * 4 bytes = 9 bytes
+ self.assertEqual(len(raw), 1 + 2 * 4)
+ self.assertEqual(raw[0:1], b"\x00") # filter byte
+ self.assertEqual(raw[1:], rgba)
+
+
+def _make_service_response(width, height, rgba_data):
+ """Build a qubes.GetImageRGBA response."""
+ header = "{} {}\n".format(width, height).encode()
+ return header + rgba_data
+
+
+class TC_01_qvm_set_wallpaper(qubesadmin.tests.QubesTestCase):
+
+ def setUp(self):
+ super().setUp()
+ self.app.expected_calls[("dom0", "admin.vm.List", None, None)] = (
+ b"0\x00vm class=AppVM state=Running\n"
+ )
+ self.tmpdir = tempfile.mkdtemp()
+ self.output_path = os.path.join(self.tmpdir, "wallpaper.png")
+
+ def tearDown(self):
+ if os.path.exists(self.output_path):
+ os.unlink(self.output_path)
+ os.rmdir(self.tmpdir)
+ super().tearDown()
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_000_success(self, mock_set_wp):
+ """Valid RGBA response produces a PNG and sets wallpaper."""
+ rgba = b"\xff\x00\x00\xff" * 4
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ _make_service_response(2, 2, rgba)
+ )
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/to/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 0)
+ self.assertTrue(os.path.exists(self.output_path))
+ with open(self.output_path, "rb") as png_file:
+ self.assertTrue(png_file.read().startswith(b"\x89PNG"))
+ mock_set_wp.assert_called_once_with(self.output_path)
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_001_service_sends_filepath(self, _mock_set_wp):
+ """VM filepath is sent as stdin to the qrexec service."""
+ rgba = b"\xff\x00\x00\xff" * 4
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ _make_service_response(2, 2, rgba)
+ )
+ qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/home/user/photo.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ # service_calls has two entries per call: one from run_service
+ # (with kwargs) and one from the input_callback (with input bytes)
+ self.assertEqual(self.app.service_calls[0][1], "qubes.GetImageRGBA")
+ self.assertEqual(
+ self.app.service_calls[1],
+ ("vm", "qubes.GetImageRGBA", b"/home/user/photo.jpg"),
+ )
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_002_invalid_header_no_newline(self, mock_set_wp):
+ """Response without a newline-terminated header fails."""
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ b"garbage without newline"
+ )
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 1)
+ mock_set_wp.assert_not_called()
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_003_invalid_header_not_numbers(self, mock_set_wp):
+ """Non-numeric dimensions in header fail."""
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ b"abc def\n\x00"
+ )
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 1)
+ mock_set_wp.assert_not_called()
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_004_negative_dimensions(self, mock_set_wp):
+ """Negative image dimensions are rejected."""
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ b"-1 10\n\x00"
+ )
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 1)
+ mock_set_wp.assert_not_called()
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_005_data_size_mismatch(self, mock_set_wp):
+ """RGBA payload size not matching dimensions fails."""
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ _make_service_response(2, 2, b"\x00" * 8)
+ )
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 1)
+ mock_set_wp.assert_not_called()
+
+ @patch("qubesadmin.tools.qvm_set_wallpaper.set_wallpaper")
+ def test_006_wallpaper_set_failure(self, mock_set_wp):
+ """DE command failure returns 1 but still saves the PNG."""
+ rgba = b"\xff\x00\x00\xff" * 4
+ self.app.expected_service_calls[("vm", "qubes.GetImageRGBA")] = (
+ _make_service_response(2, 2, rgba)
+ )
+ mock_set_wp.side_effect = FileNotFoundError("xfconf-query not found")
+ ret = qubesadmin.tools.qvm_set_wallpaper.main(
+ ["vm", "/path/img.jpg", "--output", self.output_path],
+ app=self.app,
+ )
+ self.assertEqual(ret, 1)
+ # File should still be saved
+ self.assertTrue(os.path.exists(self.output_path))
+
+
+class TC_02_set_wallpaper(qubesadmin.tests.QubesTestCase):
+
+ @patch("subprocess.run")
+ @patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "XFCE"})
+ def test_000_xfce(self, mock_run):
+ """XFCE: sets all /last-image xfconf properties."""
+ mock_run.return_value = MagicMock(
+ stdout="/backdrop/screen0/monitor0/workspace0/last-image\n"
+ "/backdrop/screen0/monitor1/workspace0/last-image\n",
+ )
+ qubesadmin.tools.qvm_set_wallpaper.set_wallpaper("/tmp/wp.png")
+ # First call: list properties
+ mock_run.assert_any_call(
+ ["xfconf-query", "-c", "xfce4-desktop", "-l"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ # Should set both monitors
+ mock_run.assert_any_call(
+ [
+ "xfconf-query",
+ "-c",
+ "xfce4-desktop",
+ "-p",
+ "/backdrop/screen0/monitor0/workspace0/last-image",
+ "-s",
+ "/tmp/wp.png",
+ ],
+ check=True,
+ )
+ mock_run.assert_any_call(
+ [
+ "xfconf-query",
+ "-c",
+ "xfce4-desktop",
+ "-p",
+ "/backdrop/screen0/monitor1/workspace0/last-image",
+ "-s",
+ "/tmp/wp.png",
+ ],
+ check=True,
+ )
+
+ @patch("subprocess.run")
+ @patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "GNOME"})
+ def test_001_gnome(self, mock_run):
+ """GNOME: sets picture-uri via gsettings."""
+ qubesadmin.tools.qvm_set_wallpaper.set_wallpaper("/tmp/wp.png")
+ mock_run.assert_called_once_with(
+ [
+ "gsettings",
+ "set",
+ "org.gnome.desktop.background",
+ "picture-uri",
+ "file:///tmp/wp.png",
+ ],
+ check=True,
+ )
+
+ @patch("subprocess.run")
+ @patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "KDE"})
+ def test_002_kde(self, mock_run):
+ """KDE: calls plasma-apply-wallpaperimage."""
+ qubesadmin.tools.qvm_set_wallpaper.set_wallpaper("/tmp/wp.png")
+ mock_run.assert_called_once_with(
+ ["plasma-apply-wallpaperimage", "/tmp/wp.png"],
+ check=True,
+ )
+
+ @patch("subprocess.run")
+ @patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": ""})
+ def test_003_unknown_de(self, mock_run):
+ """Unknown DE: logs warning, no subprocess calls."""
+ qubesadmin.tools.qvm_set_wallpaper.set_wallpaper("/tmp/wp.png")
+ mock_run.assert_not_called()
diff --git a/qubesadmin/tools/qvm_set_wallpaper.py b/qubesadmin/tools/qvm_set_wallpaper.py
new file mode 100644
index 000000000..1ee6f1904
--- /dev/null
+++ b/qubesadmin/tools/qvm_set_wallpaper.py
@@ -0,0 +1,203 @@
+# -*- encoding: utf-8 -*-
+#
+# The Qubes OS Project, http://www.qubes-os.org
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation; either version 2.1 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License along
+# with this program; if not, see .
+
+"""Safely set desktop wallpaper from a VM image"""
+
+import logging
+import os
+import struct
+import subprocess
+import sys
+import zlib
+
+import qubesadmin
+import qubesadmin.tools
+from qubesadmin.app import QubesBase
+from qubesadmin.tools import QubesArgumentParser
+
+DEFAULT_OUTPUT_DIR = os.path.expanduser("~/.local/share/qubes-wallpaper")
+DEFAULT_OUTPUT_PATH = os.path.join(DEFAULT_OUTPUT_DIR, "wallpaper.png")
+
+
+def get_parser() -> QubesArgumentParser:
+ """Create :py:class:`argparse.ArgumentParser` suitable for
+ :program:`qvm-set-wallpaper`.
+ """
+ parser = qubesadmin.tools.QubesArgumentParser(
+ description="Safely set desktop wallpaper from a VM image",
+ vmname_nargs=1,
+ )
+ parser.add_argument(
+ "filepath",
+ metavar="FILEPATH",
+ help="path to image file within the source VM",
+ )
+ parser.add_argument(
+ "--output",
+ "-o",
+ metavar="PATH",
+ default=DEFAULT_OUTPUT_PATH,
+ help="output path for the converted PNG (default: %(default)s)",
+ )
+ return parser
+
+
+def rgba_to_png(data: bytes, width: int, height: int) -> bytes:
+ """Encode raw RGBA data as a PNG file.
+
+ :param data: raw RGBA pixel data
+ :param width: image width in pixels
+ :param height: image height in pixels
+ :returns: PNG file contents
+ """
+
+ def _chunk(chunk_type, chunk_data):
+ raw = chunk_type + chunk_data
+ return (
+ struct.pack(">I", len(chunk_data))
+ + raw
+ + struct.pack(">I", zlib.crc32(raw) & 0xFFFFFFFF)
+ )
+
+ signature = b"\x89PNG\r\n\x1a\n"
+
+ # IHDR: width, height, bit depth 8, color type 6 (RGBA)
+ ihdr_data = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0)
+ ihdr = _chunk(b"IHDR", ihdr_data)
+
+ # IDAT: filter byte 0 (None) before each scanline, then zlib compress
+ row_size = width * 4
+ raw_lines = b""
+ for row in range(height):
+ raw_lines += b"\x00" + data[row * row_size : (row + 1) * row_size]
+ idat = _chunk(b"IDAT", zlib.compress(raw_lines))
+
+ iend = _chunk(b"IEND", b"")
+
+ return signature + ihdr + idat + iend
+
+
+def set_wallpaper(path: str) -> None:
+ """Detect desktop environment and set wallpaper.
+
+ :param path: absolute path to the wallpaper image
+ """
+ path = os.path.abspath(path)
+ desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
+
+ if "xfce" in desktop:
+ # List all backdrop properties and set every last-image one
+ result = subprocess.run(
+ ["xfconf-query", "-c", "xfce4-desktop", "-l"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+ for line in result.stdout.splitlines():
+ if line.endswith("/last-image"):
+ subprocess.run(
+ ["xfconf-query", "-c", "xfce4-desktop", "-p",
+ line, "-s", path],
+ check=True,
+ )
+ elif any(de in desktop for de in ("gnome", "cinnamon", "unity")):
+ uri = "file://" + path
+ subprocess.run(
+ ["gsettings", "set", "org.gnome.desktop.background",
+ "picture-uri", uri],
+ check=True,
+ )
+ elif "kde" in desktop:
+ subprocess.run(
+ ["plasma-apply-wallpaperimage", path],
+ check=True,
+ )
+ else:
+ logging.warning(
+ "Unknown desktop environment. Wallpaper saved to: %s\n"
+ "Set it manually in your desktop environment settings."
+ "Feel free to raise a Github issue so that we can add support"
+ " for your environment.",
+ path,
+ )
+
+
+def main(args: list[str] | None = None, app: QubesBase | None = None) -> int:
+ """Main function of :program:`qvm-set-wallpaper`."""
+ app = app or qubesadmin.Qubes()
+ parser = get_parser()
+ args = parser.parse_args(args, app=app)
+ vm = args.domains.pop()
+ filepath = args.filepath
+ output_path = args.output
+
+ try:
+ stdout, _stderr = vm.run_service_for_stdio(
+ "qubes.GetImageRGBA", input=filepath.encode()
+ )
+ except subprocess.CalledProcessError as e:
+ parser.print_error(
+ "Failed to get image from '{}': {}".format(vm.name, e)
+ )
+ return 1
+
+ # Parse header: "width height\n"
+ try:
+ header_end = stdout.index(b"\n")
+ header = stdout[:header_end].decode("ascii")
+ width_s, height_s = header.split()
+ width, height = int(width_s), int(height_s)
+ except (ValueError, IndexError) as e:
+ parser.print_error(
+ "Invalid response from qubes.GetImageRGBA: {}".format(e)
+ )
+ return 1
+
+ if width <= 0 or height <= 0:
+ parser.print_error(
+ "Invalid image dimensions: {}x{}".format(width, height)
+ )
+ return 1
+
+ rgba_data = stdout[header_end + 1 :]
+ expected_size = width * height * 4
+ if len(rgba_data) != expected_size:
+ parser.print_error(
+ "Expected {} bytes of RGBA data, got {}".format(
+ expected_size, len(rgba_data)
+ )
+ )
+ return 1
+
+ png_data = rgba_to_png(rgba_data, width, height)
+
+ os.makedirs(os.path.dirname(output_path), exist_ok=True)
+ with open(output_path, "wb") as png_file:
+ png_file.write(png_data)
+
+ try:
+ set_wallpaper(output_path)
+ except (subprocess.CalledProcessError, FileNotFoundError) as e:
+ parser.print_error("Failed to set wallpaper: {}".format(e))
+ logging.error("Wallpaper saved to: %s", output_path)
+ return 1
+
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main()) # pragma: no cover