Read busctl as JSON so device names keep their characters

A phone named "Gib's iPhone" with a typographic apostrophe was shown as
"Gib\342\200\231s iPhone".

busctl's default text output escapes every non-ASCII byte in octal, and
escapes it into the output rather than into a quoted string a
shell-style parser can undo, so shlex handed back the escape sequences
as literal characters and they went straight to the page. Apostrophes
were only the visible case: accents, emoji, quotes and backslashes were
all affected, and a name containing a quote could have split a field.

Property and method reads now use --json=short, which returns real
UTF-8, and the parsers read a document rather than splitting words.
That removes the class rather than unescaping octal by hand.

The fixtures were the reason this stayed invisible: every test fed the
text form and passed against output the helper is no longer asking for.
They now carry what busctl actually emits in the mode used, plus a case
for a non-ASCII name and one asserting the old text form is refused
rather than parsed wrongly.

Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
This commit is contained in:
Gabriel Brown
2026-08-19 11:28:36 -04:00
parent eeb49c5aff
commit dbd1472e6b
2 changed files with 93 additions and 53 deletions
+34 -20
View File
@@ -7,7 +7,6 @@ from __future__ import annotations
import json import json
import pathlib import pathlib
import re import re
import shlex
import subprocess import subprocess
import sys import sys
from collections.abc import Callable from collections.abc import Callable
@@ -88,33 +87,44 @@ def normalize_device_line(
} }
# busctl properties are read with --json=short, not in its default text form.
#
# The text form escapes every non-ASCII byte in octal, and it escapes them into
# the OUTPUT rather than into a quoted string a shell-style parser can undo -- so
# a phone named "Gib's iPhone" with a typographic apostrophe arrived as the
# literal characters "Gib\342\200\231s iPhone" and was displayed that way.
# That is not specific to apostrophes: any name with an accent, an emoji, a
# quote, or a backslash was affected the same way.
#
# The JSON form returns real UTF-8 and needs no unescaping, which is why these
# parse a document rather than splitting words.
def parse_property(output: str, expected: str) -> object | None:
"""The value of a busctl --json=short property read, or None if it is not
the type asked for."""
try:
payload = json.loads(output)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(payload, dict) or payload.get("type") != expected:
return None
return payload.get("data")
def parse_loaded_plugins(output: str) -> list[str]: def parse_loaded_plugins(output: str) -> list[str]:
try: data = parse_property(output, "as")
parts = shlex.split(output) if not isinstance(data, list):
except ValueError:
return [] return []
if len(parts) < 2 or parts[0] != "as": return [str(item) for item in data]
return []
try:
count = int(parts[1])
except ValueError:
return []
return parts[2 : 2 + max(0, count)]
def parse_string_property(output: str) -> str: def parse_string_property(output: str) -> str:
try: data = parse_property(output, "s")
parts = shlex.split(output) return data if isinstance(data, str) else ""
except ValueError:
return ""
return parts[1] if len(parts) == 2 and parts[0] == "s" else ""
def parse_bool_property(output: str) -> bool | None: def parse_bool_property(output: str) -> bool | None:
parts = output.split() data = parse_property(output, "b")
if len(parts) != 2 or parts[0] != "b" or parts[1] not in {"true", "false"}: return data if isinstance(data, bool) else None
return None
return parts[1] == "true"
def run_command( def run_command(
@@ -141,6 +151,7 @@ def loaded_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]
[ [
"busctl", "busctl",
"--user", "--user",
"--json=short",
"call", "call",
"org.kde.kdeconnect", "org.kde.kdeconnect",
device_object(device_id), device_object(device_id),
@@ -157,6 +168,7 @@ def supported_plugins(device_id: str, runner: Runner = subprocess.run) -> list[s
[ [
"busctl", "busctl",
"--user", "--user",
"--json=short",
"get-property", "get-property",
"org.kde.kdeconnect", "org.kde.kdeconnect",
device_object(device_id), device_object(device_id),
@@ -178,6 +190,7 @@ def reported_type(device_id: str, runner: Runner = subprocess.run) -> str:
[ [
"busctl", "busctl",
"--user", "--user",
"--json=short",
"get-property", "get-property",
"org.kde.kdeconnect", "org.kde.kdeconnect",
device_object(device_id), device_object(device_id),
@@ -198,6 +211,7 @@ def device_property(
[ [
"busctl", "busctl",
"--user", "--user",
"--json=short",
"get-property", "get-property",
"org.kde.kdeconnect", "org.kde.kdeconnect",
device_object(device_id), device_object(device_id),
+59 -33
View File
@@ -89,8 +89,8 @@ class KdeConnectBridgeTest(unittest.TestCase):
def test_busctl_plugin_output_is_normalized(self) -> None: def test_busctl_plugin_output_is_normalized(self) -> None:
output = ( output = (
'as 5 "kdeconnect_ping" "kdeconnect_share" ' '{"type":"as","data":["kdeconnect_ping","kdeconnect_share",'
'"kdeconnect_clipboard" "kdeconnect_findmyphone" "unrelated"\n' '"kdeconnect_clipboard","kdeconnect_findmyphone","unrelated"]}\n'
) )
self.assertEqual( self.assertEqual(
@@ -104,16 +104,42 @@ class KdeConnectBridgeTest(unittest.TestCase):
], ],
) )
def test_device_name_keeps_its_typographic_characters(self) -> None:
"""A phone named "Gib's iPhone" is displayed that way.
busctl's default TEXT output escapes every non-ASCII byte in octal, and
escapes it into the output rather than into a quoted string a
shell-style parser can undo -- so the name arrived as the literal
characters "Gib\\342\\200\\231s iPhone" and was shown on the Home &
Phone page exactly like that. Apostrophes were only the visible case;
accents, emoji, quotes and backslashes were all affected.
"""
output = '{"type":"s","data":"Gib\u2019s iPhone"}\n'
self.assertEqual(bridge.parse_string_property(output), "Gib\u2019s iPhone")
def test_octal_escaped_name_is_not_accepted_as_a_value(self) -> None:
"""The old text form must not parse at all, rather than parse wrongly.
Reading it as a value is what produced the mangled name; refusing it
means a future change back to text output fails loudly instead of
displaying escape sequences to someone.
"""
self.assertEqual(bridge.parse_string_property('s "Gib\\342\\200\\231s iPhone"\n'), "")
self.assertEqual(bridge.parse_bool_property("b true\n"), None)
self.assertEqual(bridge.parse_loaded_plugins('as 1 "kdeconnect_ping"\n'), [])
def test_offline_device_falls_back_to_supported_plugins(self) -> None: def test_offline_device_falls_back_to_supported_plugins(self) -> None:
def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: def runner(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]:
member = command[-1] member = command[-1]
if member == "loadedPlugins": if member == "loadedPlugins":
return subprocess.CompletedProcess(command, 0, "as 0\n", "") return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "")
if member == "supportedPlugins": if member == "supportedPlugins":
return subprocess.CompletedProcess( return subprocess.CompletedProcess(
command, command,
0, 0,
'as 3 "kdeconnect_share" "kdeconnect_clipboard" "kdeconnect_findmyphone"\n', '{"type":"as","data":["kdeconnect_share",'
'"kdeconnect_clipboard","kdeconnect_findmyphone"]}\n',
"", "",
) )
raise AssertionError(command) raise AssertionError(command)
@@ -139,17 +165,17 @@ class KdeConnectBridgeTest(unittest.TestCase):
return subprocess.CompletedProcess(command, 0, "0 devices found\n", "") return subprocess.CompletedProcess(command, 0, "0 devices found\n", "")
if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
return subprocess.CompletedProcess(command, 0, f"└─ {device_path}\n", "") return subprocess.CompletedProcess(command, 0, f"└─ {device_path}\n", "")
if command[:3] == ["busctl", "--user", "call"]: if command[:4] == ["busctl", "--user", "--json=short", "call"]:
return subprocess.CompletedProcess(command, 0, "as 0\n", "") return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "")
if command[:3] == ["busctl", "--user", "get-property"]: if command[:4] == ["busctl", "--user", "--json=short", "get-property"]:
values = { values = {
"name": 's "Fixture iPhone"\n', "name": '{"type":"s","data":"Fixture iPhone"}\n',
"type": 's "phone"\n', "type": '{"type":"s","data":"phone"}\n',
"isPaired": "b true\n", "isPaired": '{"type":"b","data":true}\n',
"isReachable": "b false\n", "isReachable": '{"type":"b","data":false}\n',
"supportedPlugins": ( "supportedPlugins": (
'as 3 "kdeconnect_share" "kdeconnect_clipboard" ' '{"type":"as","data":["kdeconnect_share",'
'"kdeconnect_findmyphone"\n' '"kdeconnect_clipboard","kdeconnect_findmyphone"]}\n'
), ),
} }
return subprocess.CompletedProcess(command, 0, values[command[-1]], "") return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
@@ -180,14 +206,14 @@ class KdeConnectBridgeTest(unittest.TestCase):
if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
path = f"/modules/kdeconnect/devices/{device_id}" path = f"/modules/kdeconnect/devices/{device_id}"
return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
if command[:3] == ["busctl", "--user", "call"]: if command[:4] == ["busctl", "--user", "--json=short", "call"]:
return subprocess.CompletedProcess(command, 0, "as 0\n", "") return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "")
if command[:3] == ["busctl", "--user", "get-property"]: if command[:4] == ["busctl", "--user", "--json=short", "get-property"]:
values = { values = {
"name": 's "Fixture iPhone"\n', "name": '{"type":"s","data":"Fixture iPhone"}\n',
"type": 's "phone"\n', "type": '{"type":"s","data":"phone"}\n',
"isPaired": "b true\n", "isPaired": '{"type":"b","data":true}\n',
"isReachable": "b false\n", "isReachable": '{"type":"b","data":false}\n',
} }
if command[-1] == "supportedPlugins": if command[-1] == "supportedPlugins":
raise subprocess.TimeoutExpired(command, 8) raise subprocess.TimeoutExpired(command, 8)
@@ -217,12 +243,12 @@ class KdeConnectBridgeTest(unittest.TestCase):
if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
path = f"/modules/kdeconnect/devices/{device_id}" path = f"/modules/kdeconnect/devices/{device_id}"
return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
if command[:3] == ["busctl", "--user", "get-property"]: if command[:4] == ["busctl", "--user", "--json=short", "get-property"]:
values = { values = {
"name": 's "Nearby Stranger"\n', "name": '{"type":"s","data":"Nearby Stranger"}\n',
"type": 's "phone"\n', "type": '{"type":"s","data":"phone"}\n',
"isPaired": "b false\n", "isPaired": '{"type":"b","data":false}\n',
"isReachable": "b true\n", "isReachable": '{"type":"b","data":true}\n',
} }
return subprocess.CompletedProcess(command, 0, values[command[-1]], "") return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
raise AssertionError(command) raise AssertionError(command)
@@ -240,16 +266,16 @@ class KdeConnectBridgeTest(unittest.TestCase):
if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]: if command == ["busctl", "--user", "tree", "org.kde.kdeconnect"]:
path = f"/modules/kdeconnect/devices/{dbus_id}" path = f"/modules/kdeconnect/devices/{dbus_id}"
return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "") return subprocess.CompletedProcess(command, 0, f"└─ {path}\n", "")
if command[:3] == ["busctl", "--user", "call"]: if command[:4] == ["busctl", "--user", "--json=short", "call"]:
return subprocess.CompletedProcess(command, 0, "as 0\n", "") return subprocess.CompletedProcess(command, 0, '{"type":"as","data":[]}\n', "")
if command[:3] == ["busctl", "--user", "get-property"]: if command[:4] == ["busctl", "--user", "--json=short", "get-property"]:
is_dbus_device = dbus_id in command[4] is_dbus_device = dbus_id in command[4]
values = { values = {
"name": 's "Fixture iPhone"\n', "name": '{"type":"s","data":"Fixture iPhone"}\n',
"type": 's "phone"\n' if is_dbus_device else 's "desktop"\n', "type": '{"type":"s","data":"phone"}\n' if is_dbus_device else '{"type":"s","data":"desktop"}\n',
"isPaired": "b true\n", "isPaired": '{"type":"b","data":true}\n',
"isReachable": "b false\n", "isReachable": '{"type":"b","data":false}\n',
"supportedPlugins": "as 0\n", "supportedPlugins": '{"type":"as","data":[]}\n',
} }
return subprocess.CompletedProcess(command, 0, values[command[-1]], "") return subprocess.CompletedProcess(command, 0, values[command[-1]], "")
raise AssertionError(command) raise AssertionError(command)