286 lines
8.0 KiB
Python
Executable File
286 lines
8.0 KiB
Python
Executable File
#!/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<name>.+):\s+(?P<id>[A-Fa-f0-9]{32,64})"
|
|
r"(?:\s+on\s+.+?)?\s+\((?P<state>[^)]+)\)\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:]))
|