From 3a903934c031941177f2075d83f134c15e0609d5 Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Mon, 17 Aug 2026 13:22:31 -0400 Subject: [PATCH] Add the Panama KDE Connect bridge --- .../dot/quickshell/scripts/panama-kdeconnect | 285 ++++++++++++++++++ .../quickshell/kdeconnect-helper-contract.sh | 29 ++ tests/quickshell/kdeconnect_bridge_test.py | 108 +++++++ 3 files changed, 422 insertions(+) create mode 100755 config/dot/quickshell/scripts/panama-kdeconnect create mode 100755 tests/quickshell/kdeconnect-helper-contract.sh create mode 100644 tests/quickshell/kdeconnect_bridge_test.py diff --git a/config/dot/quickshell/scripts/panama-kdeconnect b/config/dot/quickshell/scripts/panama-kdeconnect new file mode 100755 index 0000000..79b67e0 --- /dev/null +++ b/config/dot/quickshell/scripts/panama-kdeconnect @@ -0,0 +1,285 @@ +#!/usr/bin/env python3 + +"""Capability-aware KDE Connect boundary for the Panama shell.""" + +from __future__ import annotations + +import json +import pathlib +import re +import shlex +import subprocess +import sys +from collections.abc import Callable + + +DEVICE_ID = re.compile(r"^[A-Fa-f0-9]{32,64}$") +DEVICE_LINE = re.compile( + r"^-\s+(?P.+):\s+(?P[A-Fa-f0-9]{32,64})" + r"(?:\s+on\s+.+?)?\s+\((?P[^)]+)\)\s*$" +) +PLUGIN_ACTIONS = { + "kdeconnect_clipboard": "clipboard", + "kdeconnect_findmyphone": "ring", + "kdeconnect_ping": "ping", + "kdeconnect_share": "share", +} +DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices" +Runner = Callable[..., subprocess.CompletedProcess[str]] + + +def compact_json(value: dict[str, object]) -> str: + return json.dumps(value, separators=(",", ":")) + + +def validate_device_id(value: str) -> str: + if not DEVICE_ID.fullmatch(value): + raise ValueError("invalid-device") + return value + + +def validate_file(value: pathlib.Path) -> pathlib.Path: + try: + resolved = value.expanduser().resolve(strict=True) + except OSError as error: + raise ValueError("invalid-file") from error + if not resolved.is_file(): + raise ValueError("invalid-file") + return resolved + + +def inferred_type(name: str) -> str: + normalized = name.casefold() + if "iphone" in normalized or "phone" in normalized: + return "phone" + if "tablet" in normalized or "ipad" in normalized: + return "tablet" + return "device" + + +def normalize_device_line( + line: str, + plugins: list[str], + device_type: str = "", +) -> dict[str, object]: + match = DEVICE_LINE.fullmatch(line.strip()) + if match is None: + raise ValueError("invalid-device-line") + + state = match.group("state").casefold() + actions = sorted( + { + action + for plugin, action in PLUGIN_ACTIONS.items() + if plugin in plugins + } + ) + name = match.group("name").strip() + return { + "id": match.group("id"), + "name": name, + "type": device_type or inferred_type(name), + "paired": "paired" in state, + "reachable": "reachable" in state, + "actions": actions, + } + + +def parse_loaded_plugins(output: str) -> list[str]: + try: + parts = shlex.split(output) + except ValueError: + return [] + if len(parts) < 2 or parts[0] != "as": + return [] + try: + count = int(parts[1]) + except ValueError: + return [] + return parts[2 : 2 + max(0, count)] + + +def parse_string_property(output: str) -> str: + try: + parts = shlex.split(output) + except ValueError: + return "" + return parts[1] if len(parts) == 2 and parts[0] == "s" else "" + + +def run_command( + command: list[str], + *, + runner: Runner = subprocess.run, + timeout: float = 8, +) -> subprocess.CompletedProcess[str]: + return runner( + command, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + + +def device_object(device_id: str) -> str: + return f"{DEVICE_OBJECT_PREFIX}/{validate_device_id(device_id)}" + + +def loaded_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]: + result = run_command( + [ + "busctl", + "--user", + "call", + "org.kde.kdeconnect", + device_object(device_id), + "org.kde.kdeconnect.device", + "loadedPlugins", + ], + runner=runner, + ) + return parse_loaded_plugins(result.stdout) if result.returncode == 0 else [] + + +def reported_type(device_id: str, runner: Runner = subprocess.run) -> str: + result = run_command( + [ + "busctl", + "--user", + "get-property", + "org.kde.kdeconnect", + device_object(device_id), + "org.kde.kdeconnect.device", + "type", + ], + runner=runner, + ) + return parse_string_property(result.stdout) if result.returncode == 0 else "" + + +def collect_status(runner: Runner = subprocess.run) -> dict[str, object]: + try: + listing = run_command( + ["kdeconnect-cli", "--list-devices"], + runner=runner, + ) + except (FileNotFoundError, subprocess.TimeoutExpired): + return {"available": False, "devices": [], "error": "unavailable"} + + if listing.returncode != 0: + return {"available": False, "devices": [], "error": "unavailable"} + + devices: list[dict[str, object]] = [] + for line in listing.stdout.splitlines(): + if not line.lstrip().startswith("-"): + continue + match = DEVICE_LINE.fullmatch(line.strip()) + if match is None: + continue + device_id = match.group("id") + try: + device = normalize_device_line( + line, + loaded_plugins(device_id, runner), + reported_type(device_id, runner), + ) + except ValueError: + continue + devices.append(device) + + devices.sort( + key=lambda device: ( + not bool(device["reachable"]), + not bool(device["paired"]), + str(device["name"]).casefold(), + ) + ) + return {"available": True, "devices": devices, "error": ""} + + +def action_command( + action: str, + device_id: str, + file_path: pathlib.Path | None = None, +) -> list[str]: + device_id = validate_device_id(device_id) + options = { + "ring": ["--ring"], + "clipboard": ["--send-clipboard"], + } + if action == "share" and file_path is not None: + return [ + "kdeconnect-cli", + "-d", + device_id, + "--share", + str(validate_file(file_path)), + ] + if action not in options: + raise ValueError("unsupported-action") + return ["kdeconnect-cli", "-d", device_id, *options[action]] + + +def invoke_action( + action: str, + device_id: str, + file_path: pathlib.Path | None = None, + runner: Runner = subprocess.run, +) -> dict[str, object]: + device_id = validate_device_id(device_id) + status = collect_status(runner) + device = next( + ( + item + for item in status["devices"] + if isinstance(item, dict) and item.get("id") == device_id + ), + None, + ) + if not device or not device.get("paired") or not device.get("reachable"): + return {"ok": False, "action": action, "error": "device-offline"} + if action not in device.get("actions", []): + return {"ok": False, "action": action, "error": "unsupported-action"} + + command = action_command(action, device_id, file_path) + try: + result = run_command(command, runner=runner, timeout=3600) + except (FileNotFoundError, subprocess.TimeoutExpired): + return {"ok": False, "action": action, "error": "action-failed"} + if result.returncode != 0: + return {"ok": False, "action": action, "error": "action-failed"} + + response: dict[str, object] = {"ok": True, "action": action, "error": ""} + if action == "share" and file_path is not None: + response["fileName"] = validate_file(file_path).name + return response + + +def print_result(value: dict[str, object]) -> None: + sys.stdout.write(compact_json(value) + "\n") + + +def main(argv: list[str]) -> int: + if not argv or argv[0] == "status": + print_result(collect_status()) + return 0 + + command = argv[0] + try: + if command == "ring" and len(argv) == 2: + result = invoke_action("ring", argv[1]) + elif command == "send-clipboard" and len(argv) == 2: + result = invoke_action("clipboard", argv[1]) + elif command == "send-file" and len(argv) == 3: + result = invoke_action("share", argv[1], pathlib.Path(argv[2])) + else: + result = {"ok": False, "action": command, "error": "invalid-command"} + except ValueError as error: + result = {"ok": False, "action": command, "error": str(error)} + print_result(result) + return 0 if bool(result.get("ok")) else 2 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/quickshell/kdeconnect-helper-contract.sh b/tests/quickshell/kdeconnect-helper-contract.sh new file mode 100755 index 0000000..cae51df --- /dev/null +++ b/tests/quickshell/kdeconnect-helper-contract.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash + +set -euo pipefail + +fail() { + printf 'KDE Connect helper contract: %s\n' "$1" >&2 + exit 1 +} + +project_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +helper="$project_root/config/dot/quickshell/scripts/panama-kdeconnect" + +[[ -x "$helper" ]] || fail 'helper is missing or not executable' + +status="$($helper status)" +jq -e ' + .available == true and + (.devices | type == "array" and length >= 1) and + ([.devices[] | (keys | sort) == (["actions", "id", "name", "paired", "reachable", "type"] | sort)] | all) and + ([.devices[] | (.id | test("^[A-Fa-f0-9]{32,64}$"))] | all) and + ([.devices[].actions[] | . == "clipboard" or . == "ping" or . == "ring" or . == "share"] | all) and + ([.devices[] | select(.paired)] | length >= 1) +' <<<"$status" >/dev/null || fail 'live status shape is invalid' + +paired_count="$(jq '[.devices[] | select(.paired)] | length' <<<"$status")" +reachable_count="$(jq '[.devices[] | select(.reachable)] | length' <<<"$status")" +printf 'KDE Connect helper contract: PASS (%s paired, %s reachable; identities redacted)\n' \ + "$paired_count" \ + "$reachable_count" diff --git a/tests/quickshell/kdeconnect_bridge_test.py b/tests/quickshell/kdeconnect_bridge_test.py new file mode 100644 index 0000000..3a33f45 --- /dev/null +++ b/tests/quickshell/kdeconnect_bridge_test.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +import pathlib +import tempfile +import unittest + + +ROOT = pathlib.Path(__file__).resolve().parents[2] +HELPER = ROOT / "config/dot/quickshell/scripts/panama-kdeconnect" + +loader = importlib.machinery.SourceFileLoader("panama_kdeconnect", str(HELPER)) +spec = importlib.util.spec_from_loader(loader.name, loader) +bridge = importlib.util.module_from_spec(spec) +loader.exec_module(bridge) + + +class KdeConnectBridgeTest(unittest.TestCase): + def test_status_omits_network_and_exposes_supported_ios_actions(self) -> None: + line = ( + "- Gib's iPhone: 90B54E84548E4F109268EC2D0E7D6930 " + "on 192.168.1.7 via LAN (paired and reachable)" + ) + plugins = [ + "kdeconnect_clipboard", + "kdeconnect_findmyphone", + "kdeconnect_ping", + "kdeconnect_share", + "kdeconnect_shareinputdevices", + ] + + device = bridge.normalize_device_line(line, plugins) + + self.assertEqual(device["name"], "Gib's iPhone") + self.assertEqual(device["type"], "phone") + self.assertTrue(device["paired"]) + self.assertTrue(device["reachable"]) + self.assertEqual(device["actions"], ["clipboard", "ping", "ring", "share"]) + self.assertNotIn("192.168.1.7", json.dumps(device)) + self.assertNotIn("LAN", json.dumps(device)) + + def test_status_parses_an_offline_paired_device(self) -> None: + line = "- Pocket: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA (paired)" + + device = bridge.normalize_device_line(line, ["kdeconnect_share"]) + + self.assertTrue(device["paired"]) + self.assertFalse(device["reachable"]) + self.assertEqual(device["actions"], ["share"]) + + def test_invalid_device_id_is_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "invalid-device"): + bridge.validate_device_id("phone; shutdown") + + def test_send_file_requires_a_regular_file(self) -> None: + with tempfile.TemporaryDirectory() as directory: + with self.assertRaisesRegex(ValueError, "invalid-file"): + bridge.validate_file(pathlib.Path(directory)) + + def test_send_file_resolves_a_regular_file(self) -> None: + with tempfile.TemporaryDirectory() as directory: + file_path = pathlib.Path(directory) / "hello world.txt" + file_path.write_text("fixture", encoding="utf-8") + + self.assertEqual(bridge.validate_file(file_path), file_path.resolve()) + + def test_action_commands_use_separate_arguments(self) -> None: + device_id = "90B54E84548E4F109268EC2D0E7D6930" + self.assertEqual( + bridge.action_command("ring", device_id), + ["kdeconnect-cli", "-d", device_id, "--ring"], + ) + self.assertEqual( + bridge.action_command("clipboard", device_id), + ["kdeconnect-cli", "-d", device_id, "--send-clipboard"], + ) + + def test_unknown_action_is_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "unsupported-action"): + bridge.action_command( + "unlock", + "90B54E84548E4F109268EC2D0E7D6930", + ) + + def test_busctl_plugin_output_is_normalized(self) -> None: + output = ( + 'as 5 "kdeconnect_ping" "kdeconnect_share" ' + '"kdeconnect_clipboard" "kdeconnect_findmyphone" "unrelated"\n' + ) + + self.assertEqual( + bridge.parse_loaded_plugins(output), + [ + "kdeconnect_ping", + "kdeconnect_share", + "kdeconnect_clipboard", + "kdeconnect_findmyphone", + "unrelated", + ], + ) + + +if __name__ == "__main__": + unittest.main()