From bc4f04bd04bf9b66d59cdf42a0f13657c5954743 Mon Sep 17 00:00:00 2001 From: Valen Torassa Date: Mon, 26 Oct 2026 19:44:58 -0300 Subject: [PATCH 1/2] Opt-in experimental product ID for untested vendor-0200 decks ragnarosd --experimental-pid 0x2000 (or RAGNAROS_EXPERIMENTAL_PID) also opens that product ID after the SS552 and drives it with the SS552 protocol, so an owner of a sibling deck can see whether image writes and input work. Off by default; the daemon says EXPERIMENTAL at startup and on attach, ragnarosctl status shows it, and every raw input report is logged while it is on. A single-interface deck is still claimed. --- ragnaros/daemon.py | 69 +++++++++++++++++------ ragnaros/protocol.py | 70 ++++++++++++++++++++--- tests/test_bundled_config.py | 2 +- tests/test_daemon.py | 76 +++++++++++++++++++++++-- tests/test_protocol.py | 105 +++++++++++++++++++++++++++++++++++ 5 files changed, 294 insertions(+), 28 deletions(-) diff --git a/ragnaros/daemon.py b/ragnaros/daemon.py index 7cc03bf..c7f33e4 100755 --- a/ragnaros/daemon.py +++ b/ragnaros/daemon.py @@ -141,8 +141,12 @@ def run_action(action): class Deck: - def __init__(self): + pids = (protocol.PID,) # product IDs to open, in order + experimental_id = None # "0200:2000" while driving an untested product ID + + def __init__(self, pids=(protocol.PID,)): self.dev = None + self.pids = tuple(pids) self._wait_for_device() self.profile = load_profile(saved_profile_name()) self.last_activity = time.monotonic() @@ -199,9 +203,13 @@ def _wait_for_device(self): announced = False while True: try: - self.dev = protocol.Ragnaros() + self.dev = protocol.Ragnaros(self.pids) if announced: print("ragnaros: device attached, resuming", file=sys.stderr) + if getattr(self.dev, "experimental", False): + self.experimental_id = protocol.device_id(self.dev.pid) + print(f"ragnaros: EXPERIMENTAL: opened untested device " + f"{self.experimental_id} with the SS552 protocol", file=sys.stderr) return except RuntimeError as error: if "libusb-1.0 is required" in str(error): @@ -1017,17 +1025,20 @@ def handle_control(self, argv): return {"ok": False, "error": "empty command"} command, args = argv[0], argv[1:] if command in ("status", "state"): - return {"ok": True, "profile": self.profile.get("_name"), - "profiles": available_profiles(), - "layer": self.layer or "base", "layers": sorted(self.layers()), - "strip": self.strip_state(), "brightness": self.brightness, - "locked": self.locked, "dashboard": self.pinned == "dashboard", - "player": self.player and {k: self.player[k] - for k in ("title", "artist", "length")}, - "alert": self.alert and self.alert["summary"], - "watchers": len(self.control.subscribers) if self.control else 0, - "pomodoro": self.pomo and self.pomo["phase"], - "obs": self.obs_state} + reply = {"ok": True, "profile": self.profile.get("_name"), + "profiles": available_profiles(), + "layer": self.layer or "base", "layers": sorted(self.layers()), + "strip": self.strip_state(), "brightness": self.brightness, + "locked": self.locked, "dashboard": self.pinned == "dashboard", + "player": self.player and {k: self.player[k] + for k in ("title", "artist", "length")}, + "alert": self.alert and self.alert["summary"], + "watchers": len(self.control.subscribers) if self.control else 0, + "pomodoro": self.pomo and self.pomo["phase"], + "obs": self.obs_state} + if self.experimental_id: + reply["experimental"] = f"{self.experimental_id} (untested, SS552 protocol)" + return reply if command in ("profile", "profiles"): if not args or args[0] == "list": return {"ok": True, "profiles": available_profiles(), @@ -1164,23 +1175,49 @@ def run(self): self.dev.close() +def experimental_arg(argv): + """The value of --experimental-pid X / --experimental-pid=X, or None.""" + for i, arg in enumerate(argv): + if arg == "--experimental-pid": + if i + 1 >= len(argv): + raise ValueError("--experimental-pid needs a product ID, e.g. 0x2000") + return argv[i + 1] + if arg.startswith("--experimental-pid="): + return arg.partition("=")[2] + return None + + def main(argv=None): argv = sys.argv[1:] if argv is None else argv if argv and argv[0] in ("-h", "--help"): - print("usage: ragnarosd\n\n" + print("usage: ragnarosd [--experimental-pid PID]\n\n" "Run the Ragnaros deck daemon in the foreground (normally started by\n" "the ragnarosd.service systemd user unit).\n\n" + " --experimental-pid PID also open the untested vendor-0200 deck PID\n" + " (hex, e.g. 0x2000) with the SS552 protocol;\n" + " for testing only, see `ragnaros diagnose`\n\n" f" profiles {', '.join(paths.profile_dirs())}\n" f" assets {', '.join(paths.asset_roots())}\n" f" control {control.socket_path()} (drive it with ragnarosctl)\n\n" "Environment: RAGNAROS_CONFIG, RAGNAROS_DATA, RAGNAROS_IDLE_SECONDS,\n" - "RAGNAROS_IDLE_FACE_SECONDS, RAGNAROS_USAGE_LOG, RAGNAROS_NO_KEEPALIVE") + "RAGNAROS_IDLE_FACE_SECONDS, RAGNAROS_USAGE_LOG, RAGNAROS_NO_KEEPALIVE,\n" + f"{protocol.EXPERIMENTAL_ENV} (same as --experimental-pid)") return 0 if argv and argv[0] == "--version": from . import __version__ print(f"ragnarosd {__version__}") return 0 - Deck().run() + try: + pids = protocol.product_ids(experimental_arg(argv)) + except ValueError as error: + print(f"ragnarosd: {error}", file=sys.stderr) + return 2 + for pid in pids[1:]: + print(f"ragnaros: EXPERIMENTAL: also accepting {protocol.device_id(pid)}, which is " + "untested, and driving it with the SS552 protocol. Keys, knobs and the strip " + "may not line up. Please report what works in " + "https://github.com/ValentinTorassa/VT-Ragnaros/issues", file=sys.stderr) + Deck(pids).run() return 0 diff --git a/ragnaros/protocol.py b/ragnaros/protocol.py index 5449c3d..db33f27 100644 --- a/ragnaros/protocol.py +++ b/ragnaros/protocol.py @@ -1,11 +1,17 @@ import ctypes import ctypes.util +import os +import sys import time from .inputmap import parse_input VID = 0x0200 PID = 0x3001 +# Opt-in only: another product ID under the same vendor, driven with the +# SS552 protocol so an owner of an untested deck can see what works. +EXPERIMENTAL_ENV = "RAGNAROS_EXPERIMENTAL_PID" +LIBUSB_ERROR_NOT_FOUND = -5 PACKET_SIZE = 1024 READ_SIZE = 512 @@ -23,7 +29,37 @@ CRT = (0x43, 0x52, 0x54) -def activate_input(): +def parse_pid(text): + """A product ID under vendor 0200, in hex: '0x2000', '2000' or '0200:2000'.""" + vendor, sep, product = str(text).strip().lower().rpartition(":") + try: + vid = int(vendor, 16) if sep else VID + pid = int(product, 16) + except ValueError: + raise ValueError(f"not a USB product ID: {text!r} (hex, e.g. 0x2000)") from None + if vid != VID: + raise ValueError(f"only vendor {VID:04x} devices can be tried, not {vid:04x}") + if not 0 < pid <= 0xFFFF: + raise ValueError(f"not a USB product ID: {text!r} (hex, e.g. 0x2000)") + return pid + + +def product_ids(experimental=None): + """Product IDs to open, in order: the SS552, then the untested one a + tester opted into (--experimental-pid, else $RAGNAROS_EXPERIMENTAL_PID).""" + if experimental is None: + experimental = os.environ.get(EXPERIMENTAL_ENV, "") + if not str(experimental).strip(): + return (PID,) + pid = parse_pid(experimental) + return (PID,) if pid == PID else (PID, pid) + + +def device_id(pid): + return f"{VID:04x}:{pid:04x}" + + +def activate_input(pids=(PID,)): library = ctypes.util.find_library("usb-1.0") if not library: raise RuntimeError("libusb-1.0 is required to activate Ragnaros input") @@ -67,12 +103,21 @@ def activate_input(): raise RuntimeError(f"libusb initialization failed ({result})") handle = None + pid = None detached = [] claimed = [] try: - handle = usb.libusb_open_device_with_vid_pid(context, VID, PID) + for pid in pids: + handle = usb.libusb_open_device_with_vid_pid(context, VID, pid) + if handle: + break if not handle: - raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") + if tuple(pids) == (PID,): + raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") + raise RuntimeError( + f"cannot open {' or '.join(device_id(p) for p in pids)}: not plugged in, " + "or no permission (`ragnaros diagnose` shows the device node)") + experimental = pid != PID for interface in (0, 1): if usb.libusb_kernel_driver_active(handle, interface) == 1: result = usb.libusb_detach_kernel_driver(handle, interface) @@ -81,6 +126,8 @@ def activate_input(): detached.append(interface) for interface in (0, 1): result = usb.libusb_claim_interface(handle, interface) + if result == LIBUSB_ERROR_NOT_FOUND and experimental and interface: + continue # an untested deck may have a single interface if result < 0: raise RuntimeError(f"cannot claim HID interface {interface} ({result})") claimed.append(interface) @@ -93,13 +140,18 @@ def activate_input(): usb.libusb_close(handle) usb.libusb_exit(context) raise - return usb, context, handle, detached, claimed + return usb, context, handle, detached, claimed, pid class Ragnaros: - def __init__(self): - self.usb, self.context, self.handle, self.detached, self.claimed = activate_input() - self.path = f"libusb:{VID:04x}:{PID:04x}" + pid = PID + experimental = False # an untested product ID, opened on request + + def __init__(self, pids=(PID,)): + (self.usb, self.context, self.handle, self.detached, self.claimed, + self.pid) = activate_input(pids) + self.experimental = self.pid != PID + self.path = f"libusb:{device_id(self.pid)}" self._initialized = False def close(self): @@ -254,4 +306,8 @@ def poll(self, timeout=10): data = self.transfer(0x82, bytes(READ_SIZE), timeout) if data is None: return None + if self.experimental: + # the raw report, so a tester can show what an unknown deck sends + print(f"ragnaros: experimental input ({len(data)} bytes): " + f"{data.rstrip(bytes(1))[:64].hex(' ')}", file=sys.stderr) return parse_input(data) diff --git a/tests/test_bundled_config.py b/tests/test_bundled_config.py index d7eeffd..cd0a248 100644 --- a/tests/test_bundled_config.py +++ b/tests/test_bundled_config.py @@ -373,7 +373,7 @@ def test_gif_tools_write_what_the_panels_show(): class RecordingDev: """Stands in for protocol.Ragnaros and keeps the last image per slot.""" - def __init__(self): + def __init__(self, pids=None): self.keys, self.strip = {}, {} def send_image(self, slot, data, strip=False): diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 00856e9..c9a9bd5 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -15,10 +15,11 @@ def close(self): def test_wait_for_device_retries_until_present(monkeypatch): - calls = {"n": 0} + calls = {"n": 0, "pids": []} - def flaky(): + def flaky(pids): calls["n"] += 1 + calls["pids"].append(pids) if calls["n"] < 3: raise RuntimeError("cannot open Ragnaros USB device; check udev permissions") return FakeDev() @@ -27,11 +28,13 @@ def flaky(): monkeypatch.setattr(ragnarosd.time, "sleep", lambda s: None) deck = ragnarosd.Deck() assert calls["n"] == 3 + assert calls["pids"] == [(0x3001,)] * 3 # only the SS552 unless asked assert isinstance(deck.dev, FakeDev) + assert deck.experimental_id is None deck.dev.close() def test_wait_for_device_exits_without_libusb(monkeypatch): - def broken(): + def broken(pids): raise RuntimeError("libusb-1.0 is required to activate Ragnaros input") monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", broken) @@ -43,8 +46,69 @@ def broken(): raise AssertionError("expected SystemExit when libusb is missing") +def test_an_untested_device_is_marked_experimental(monkeypatch, capsys): + class UntestedDev(FakeDev): + pid = 0x2000 + experimental = True + + monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda pids: UntestedDev()) + deck = ragnarosd.Deck((0x3001, 0x2000)) + assert deck.experimental_id == "0200:2000" + assert "EXPERIMENTAL: opened untested device 0200:2000" in capsys.readouterr().err + + +@pytest.fixture +def decks_built(monkeypatch): + """main() up to the point it builds the Deck: which product IDs it got.""" + built = [] + + class StubDeck: + def __init__(self, pids): + built.append(pids) + + def run(self): + pass + + monkeypatch.setattr(ragnarosd, "Deck", StubDeck) + monkeypatch.delenv(ragnarosd.protocol.EXPERIMENTAL_ENV, raising=False) + return built + + +def test_main_opens_only_the_ss552_by_default(decks_built, capsys): + assert ragnarosd.main([]) == 0 + assert decks_built == [(0x3001,)] + assert "EXPERIMENTAL" not in capsys.readouterr().err + + +@pytest.mark.parametrize("argv", [["--experimental-pid", "0x2000"], + ["--experimental-pid=2000"]]) +def test_main_takes_an_experimental_pid(decks_built, capsys, argv): + assert ragnarosd.main(argv) == 0 + assert decks_built == [(0x3001, 0x2000)] + assert "EXPERIMENTAL: also accepting 0200:2000" in capsys.readouterr().err + + +def test_main_reads_the_experimental_pid_from_the_environment(decks_built, monkeypatch): + monkeypatch.setenv(ragnarosd.protocol.EXPERIMENTAL_ENV, "0x2000") + assert ragnarosd.main([]) == 0 + assert decks_built == [(0x3001, 0x2000)] + + +@pytest.mark.parametrize("argv", [["--experimental-pid", "nope"], ["--experimental-pid"]]) +def test_main_refuses_a_bad_experimental_pid(decks_built, capsys, argv): + assert ragnarosd.main(argv) == 2 + assert decks_built == [] + assert "ragnarosd:" in capsys.readouterr().err + + +def test_help_documents_the_experimental_pid(capsys): + assert ragnarosd.main(["--help"]) == 0 + out = capsys.readouterr().out + assert "--experimental-pid" in out and "RAGNAROS_EXPERIMENTAL_PID" in out + + def test_reconnect_uses_wait_loop(monkeypatch): - monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda: FakeDev()) + monkeypatch.setattr(ragnarosd.protocol, "Ragnaros", lambda pids: FakeDev()) monkeypatch.setattr(ragnarosd.time, "sleep", lambda s: None) class BareDeck(ragnarosd.Deck): @@ -276,6 +340,10 @@ def test_control_status_and_errors(monkeypatch): deck = bare_deck(PROFILE) status = deck.handle_control(["status"]) assert status["ok"] and status["layer"] == "base" and status["strip"] == "keys" + assert "experimental" not in status + deck.experimental_id = "0200:2000" + assert deck.handle_control(["status"])["experimental"].startswith("0200:2000 (untested") + deck.experimental_id = None assert deck.handle_control([])["ok"] is False assert deck.handle_control(["nonsense"])["ok"] is False assert deck.handle_control(["profile", "ghost"])["ok"] is False diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 0978620..8852a00 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,3 +1,7 @@ +import types + +import pytest + from ragnaros import protocol @@ -81,3 +85,104 @@ def test_send_image_rejects_unknown_key(): pass else: raise AssertionError("expected ValueError for unknown key") + + +# -- opt-in untested product IDs ------------------------------------------------ +def test_parse_pid_accepts_the_usual_spellings(): + assert protocol.parse_pid("0x2000") == 0x2000 + assert protocol.parse_pid("2000") == 0x2000 + assert protocol.parse_pid(" 0200:2000 ") == 0x2000 + for bad in ("", "zz", "0x10000", "0", "046d:c08c"): + with pytest.raises(ValueError): + protocol.parse_pid(bad) + + +def test_only_the_ss552_unless_a_tester_opts_in(monkeypatch): + monkeypatch.delenv(protocol.EXPERIMENTAL_ENV, raising=False) + assert protocol.product_ids() == (0x3001,) + monkeypatch.setenv(protocol.EXPERIMENTAL_ENV, "0x2000") + assert protocol.product_ids() == (0x3001, 0x2000) + assert protocol.product_ids("0x1000") == (0x3001, 0x1000) # the flag wins + assert protocol.product_ids("3001") == (0x3001,) + monkeypatch.setenv(protocol.EXPERIMENTAL_ENV, " ") + assert protocol.product_ids() == (0x3001,) + + +def fake_libusb(monkeypatch, present, interfaces=(0, 1)): + """Enough of libusb-1.0 for activate_input(): which PIDs are plugged in and + which interfaces they have. Records the calls that touch a device.""" + calls = [] + + def open_device(context, vid, pid): + calls.append(("open", vid, pid)) + return 1 if pid in present else None + + def claim(handle, interface): + if interface not in interfaces: + return protocol.LIBUSB_ERROR_NOT_FOUND + calls.append(("claim", interface)) + return 0 + + usb = types.SimpleNamespace( + libusb_init=lambda context: 0, + libusb_open_device_with_vid_pid=open_device, + libusb_kernel_driver_active=lambda handle, interface: 0, + libusb_detach_kernel_driver=lambda handle, interface: 0, + libusb_claim_interface=claim, + libusb_release_interface=lambda handle, interface: calls.append(("release", interface)), + libusb_attach_kernel_driver=lambda handle, interface: 0, + libusb_control_transfer=lambda *args: 0, + libusb_interrupt_transfer=lambda *args: 0, + libusb_close=lambda handle: calls.append(("close",)), + libusb_exit=lambda context: None, + ) + monkeypatch.setattr(protocol.ctypes.util, "find_library", lambda name: "libusb-1.0.so.0") + monkeypatch.setattr(protocol.ctypes, "CDLL", lambda name: usb) + return calls + + +def test_the_default_opens_only_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x3001, 0x2000}) + dev = protocol.Ragnaros() + assert [c for c in calls if c[0] == "open"] == [("open", 0x0200, 0x3001)] + assert (dev.pid, dev.experimental, dev.path) == (0x3001, False, "libusb:0200:3001") + + +def test_the_default_error_is_unchanged(monkeypatch): + fake_libusb(monkeypatch, present={0x2000}) + with pytest.raises(RuntimeError, match="cannot open Ragnaros USB device"): + protocol.Ragnaros() + + +def test_an_opted_in_pid_is_tried_after_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x2000}, interfaces=(0,)) + dev = protocol.Ragnaros((0x3001, 0x2000)) + assert [c for c in calls if c[0] == "open"] == [("open", 0x0200, 0x3001), + ("open", 0x0200, 0x2000)] + assert (dev.pid, dev.experimental, dev.path) == (0x2000, True, "libusb:0200:2000") + assert dev.claimed == [0] # a single-interface deck is still worth a try + + +def test_a_missing_interface_still_fails_the_ss552(monkeypatch): + calls = fake_libusb(monkeypatch, present={0x3001}, interfaces=(0,)) + with pytest.raises(RuntimeError, match="cannot claim HID interface 1"): + protocol.Ragnaros((0x3001, 0x2000)) + assert ("release", 0) in calls and ("close",) in calls + + +def test_nothing_to_open_names_both_ids(monkeypatch): + fake_libusb(monkeypatch, present=set()) + with pytest.raises(RuntimeError, match="0200:3001 or 0200:2000.*ragnaros diagnose"): + protocol.Ragnaros((0x3001, 0x2000)) + + +def test_experimental_input_is_logged_raw(capsys): + report = b"ACK\x00\x00OK\x00\x00\x0b\x01" + bytes(501) + dev = make_dev() + dev.transfer = lambda endpoint, payload, timeout: report + assert dev.poll() == ("unknown", 0x0B, 1) + assert capsys.readouterr().err == "" + dev.experimental = True + dev.poll() + err = capsys.readouterr().err + assert "experimental input (512 bytes): 41 43 4b 00 00 4f 4b 00 00 0b 01\n" in err From 68c38baede34590d5e43c9cf0de4b9f3aad16184 Mon Sep 17 00:00:00 2001 From: Valen Torassa Date: Mon, 26 Oct 2026 19:45:08 -0300 Subject: [PATCH 2/2] Add ragnaros diagnose: a read-only device report to paste into an issue For every vendor-0200 device (or --device VID:PID) it prints Markdown with the IDs and strings, interfaces and endpoints, HID report descriptors, hidraw and USB nodes with their permissions, the udev rules and usbhid quirks that cover it, and the ragnaros, kernel and distro versions. It only reads sysfs and /dev permissions, never opens the device, and leaves out the serial number, hostname and user names. Next steps for an untested deck go to stderr. README: Reporting a new device. --- README.md | 43 +++- ragnaros/cli.py | 15 +- ragnaros/diagnose.py | 458 +++++++++++++++++++++++++++++++++++++++++ tests/test_diagnose.py | 238 +++++++++++++++++++++ 4 files changed, 752 insertions(+), 2 deletions(-) create mode 100644 ragnaros/diagnose.py create mode 100644 tests/test_diagnose.py diff --git a/README.md b/README.md index 268bc42..7848440 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,7 @@ All three install the same commands: |---|---| | `ragnarosd` | the daemon (run by the `ragnarosd` user unit) | | `ragnarosctl` | drive the running daemon (see [Control CLI](#control-cli)) | -| `ragnaros` | `install-system`, `install-service`, `paths` | +| `ragnaros` | `install-system`, `install-service`, `paths`, `diagnose` (see [Reporting a new device](#reporting-a-new-device)) | | `ragnaros-fetch-gifs` | opt-in anime GIF themes for the strip (see [GIF themes](#gif-themes)) | | `ragnaros-claude-hook` | Claude Code hook (see [Notices that wait for you](#notices-that-wait-for-you)) | | `ragnaros-strip-gif` | convert any GIF to the 704x124 full-strip format | @@ -312,6 +312,47 @@ connected, so the deck becomes a blank input device; the grab is released the moment the client disconnects. Profile switches, layer changes and alerts are published on the same stream. +## Reporting a new device + +Ragnaros drives one deck, the SS552 (`0200:3001`). Other Redragon decks share its USB +vendor ID `0200` (which `lsusb` labels "TP-Link") and may or may not speak the same +protocol. If you have one, two things show whether it can be added. + +**1. A device report.** `ragnaros diagnose` prints a Markdown report on every vendor-`0200` +device plugged in: IDs and strings, interfaces and endpoints, the HID report descriptors, +the hidraw and USB device nodes with their permissions, which udev rules and usbhid quirks +cover it, and the ragnaros, kernel and distribution versions. It only reads sysfs and the +permissions in `/dev`: the deck is never opened, so it needs no root and is safe while the +daemon runs. It leaves out the serial number, the hostname and user names. + +```bash +ragnaros diagnose > report.md # --device VID:PID reports any other device +``` + +Paste `report.md` into an issue, with how many keys, knobs and screens the deck has. What +to try next goes to the terminal, not to the file. + +**2. An experimental test.** `ragnarosd --experimental-pid 0x2000` (or +`RAGNAROS_EXPERIMENTAL_PID=0x2000`) also opens that product ID and drives it with the SS552 +protocol. It is off unless you ask for it, it is marked experimental in the daemon's log and +in `ragnarosctl status`, and it logs every raw input report the deck sends. Nothing about +it is known to work. + +The udev rules only give your user the SS552, so hand yourself the deck's USB node (the +`device node` line of the report) until you unplug it, instead of widening the rules: + +```bash +sudo chown "$USER" /dev/bus/usb/005/002 # the device node from the report +systemctl --user stop ragnarosd # if it runs +ragnarosd --experimental-pid 0x2000 # stays in the foreground; Ctrl+C stops it +ragnarosctl watch --grab # in a second terminal: press every key, turn every knob +``` + +`--grab` keeps the profile's actions from running while you press things. Report whether +icons show up and on which keys (a photo helps), and paste the output of both terminals. +Unplugging the deck undoes the `chown`. From a git checkout the commands are +`python3 -m ragnaros.cli diagnose`, `python3 daemon/ragnarosd.py` and `tools/ragnarosctl`. + ## Tools The user-facing tools are installed as commands (table above). The checkout also keeps: diff --git a/ragnaros/cli.py b/ragnaros/cli.py index fe2d17f..1e8ce92 100644 --- a/ragnaros/cli.py +++ b/ragnaros/cli.py @@ -3,6 +3,7 @@ ragnaros install-system [--print] udev rules + usbhid quirk (sudo, one reboot) ragnaros install-service [--print] systemd user unit for ragnarosd ragnaros paths where profiles, assets and GIF themes live + ragnaros diagnose [--device VID:PID] read-only device report to paste in an issue The .deb and AUR packages already install the udev rules, the quirk and the user units; there you only run `systemctl --user enable --now ragnarosd`. @@ -15,7 +16,7 @@ import subprocess import sys -from . import __version__, control, paths, theme +from . import __version__, control, diagnose, paths, theme UDEV_RULES = "70-ragnaros.rules" QUIRK_CONF = "ragnaros-usbhid.conf" @@ -188,6 +189,13 @@ def show_paths(args): return 0 +def device_arg(text): + try: + return diagnose.parse_device(text) + except ValueError as error: + raise argparse.ArgumentTypeError(str(error)) from None + + def main(argv=None): parser = argparse.ArgumentParser( prog="ragnaros", description="Setup helpers for the Ragnaros deck daemon.", @@ -213,6 +221,11 @@ def main(argv=None): p = sub.add_parser("paths", help="show where profiles, assets and GIF themes live") p.set_defaults(func=show_paths) + p = sub.add_parser("diagnose", help="read-only device report to paste in an issue") + p.add_argument("--device", type=device_arg, metavar="VID:PID", + help="report this USB device instead of every vendor-0200 one") + p.set_defaults(func=diagnose.run) + args = parser.parse_args(argv) if not args.command: parser.print_help() diff --git a/ragnaros/diagnose.py b/ragnaros/diagnose.py new file mode 100644 index 0000000..b42e6c1 --- /dev/null +++ b/ragnaros/diagnose.py @@ -0,0 +1,458 @@ +"""`ragnaros diagnose`: a device report to paste into an issue. + +Read-only: everything comes from sysfs, the /dev nodes' permissions and the +udev rules directories. The device is never opened, so it needs no root and +is safe while ragnarosd owns the deck. The serial number, the hostname and +user names are left out. + +By default it reports every USB device under vendor 0200 (the SS552 and its +siblings); `--device VID:PID` picks any other one. +""" +import ctypes.util +import fnmatch +import grp +import os +import platform +import pwd +import re +import stat +import sys + +from . import __version__, protocol + +ROOT = "/" +SUPPORTED = {(protocol.VID, protocol.PID): "Redragon SS552"} +UDEV_DIRS = ("etc/udev/rules.d", "run/udev/rules.d", "usr/local/lib/udev/rules.d", + "usr/lib/udev/rules.d", "lib/udev/rules.d") +HID_DEVICE = re.compile(r"^[0-9A-Fa-f]{4}:[0-9A-Fa-f]{4}:[0-9A-Fa-f]{4}\.[0-9A-Fa-f]+$") +QUIRK = re.compile(r"^0x([0-9a-f]+):0x([0-9a-f]+):0x([0-9a-f]+)$", re.I) +QUIRK_NAMES = {0x4: "IGNORE"} +CLASSES = {"03": "HID", "ff": "vendor specific"} +BOOT_PROTOCOLS = {("01", "01"): "boot keyboard", ("01", "02"): "boot mouse"} +RULE_MATCH = re.compile( + r'\b(ATTRS?\{idVendor\}|ATTRS?\{idProduct\}|ENV\{ID_VENDOR_ID\}|ENV\{ID_MODEL_ID\}' + r'|SUBSYSTEMS?)\s*==\s*"([^"]*)"') + + +def parse_device(text): + """'0200:2000' -> (0x0200, 0x2000).""" + vendor, sep, product = text.strip().lower().partition(":") + try: + ids = int(vendor, 16), int(product, 16) + except ValueError: + ids = None + if not sep or ids is None or not all(0 <= i <= 0xFFFF for i in ids): + raise ValueError(f"expected VID:PID in hex, e.g. 0200:2000, not {text!r}") + return ids + + +def _path(root, *parts): + return os.path.join(root, *parts) + + +def read_attr(directory, name): + try: + with open(os.path.join(directory, name), errors="replace") as f: + return f.read().strip() + except OSError: + return None + + +def read_bytes(path): + try: + with open(path, "rb") as f: + return f.read() + except OSError: + return None + + +def hexdump(data, width=16): + return "\n".join(data[i:i + width].hex(" ") for i in range(0, len(data), width)) + + +# -- finding devices -------------------------------------------------------------- +def find_devices(root, vid, pid=None): + """sysfs directories of the USB devices with this vendor (and product).""" + base = _path(root, "sys", "bus", "usb", "devices") + try: + names = sorted(os.listdir(base)) + except OSError: + return [] + found = [] + for name in names: + if ":" in name: # an interface, not a device + continue + directory = os.path.realpath(os.path.join(base, name)) + try: + ids = int(read_attr(directory, "idVendor") or "", 16), \ + int(read_attr(directory, "idProduct") or "", 16) + except ValueError: + continue + if ids[0] == vid and (pid is None or ids[1] == pid): + found.append(directory) + return found + + +def interfaces(directory): + """Each interface of the device: class, driver, endpoints and HID children.""" + name = os.path.basename(directory) + result = [] + for entry in sorted(os.listdir(directory)): + if not entry.startswith(name + ":"): + continue + path = os.path.join(directory, entry) + driver = os.path.realpath(os.path.join(path, "driver")) + endpoints = [] + for ep in sorted(os.listdir(path)): + if ep.startswith("ep_"): + endpoints.append({k: read_attr(os.path.join(path, ep), k) for k in ( + "bEndpointAddress", "direction", "type", "wMaxPacketSize", + "bInterval", "interval")}) + hid = [] + for child in sorted(os.listdir(path)): + if HID_DEVICE.match(child): + hid_dir = os.path.join(path, child) + try: + nodes = sorted(os.listdir(os.path.join(hid_dir, "hidraw"))) + except OSError: + nodes = [] + hid.append({"report_descriptor": + read_bytes(os.path.join(hid_dir, "report_descriptor")), + "hidraw": nodes}) + result.append({ + "number": read_attr(path, "bInterfaceNumber"), + "class": read_attr(path, "bInterfaceClass"), + "subclass": read_attr(path, "bInterfaceSubClass"), + "protocol": read_attr(path, "bInterfaceProtocol"), + "driver": os.path.basename(driver) if os.path.lexists( + os.path.join(path, "driver")) else None, + "endpoints": endpoints, + "hid": hid, + }) + return result + + +def hid_descriptors(raw): + """{interface number: (bcdHID, report descriptor length)} from the raw + configuration descriptors, which sysfs keeps even with no driver bound.""" + found = {} + interface = None + i = 0 + while raw and i + 2 <= len(raw): + length, kind = raw[i], raw[i + 1] + if length < 2: + break + if kind == 0x04 and length >= 3: + interface = raw[i + 2] + elif kind == 0x21 and length >= 9 and interface is not None: + bcd = raw[i + 3] << 8 | raw[i + 2] + for j in range(i + 6, i + length - 2, 3): + if raw[j] == 0x22: + found[interface] = (f"{bcd >> 8:x}.{bcd & 0xFF:02x}", + raw[j + 1] | raw[j + 2] << 8) + i += length + return found + + +# -- permissions, udev, quirks ---------------------------------------------------------- +def _account(ident, lookup, mine, users): + """A system account by name; a person's account only as "you" or anonymous.""" + if ident == mine and ident != 0: + return "you" if users else "your group" + if ident < 1000: + try: + return lookup(ident)[0] + except KeyError: + return str(ident) + return "another user" if users else "a user group" + + +def node_access(path): + """'crw-rw---- root:plugdev, ACL, you: read+write', or None if absent.""" + try: + st = os.stat(path) + except OSError: + return None + owner = _account(st.st_uid, pwd.getpwuid, os.getuid(), True) + group = _account(st.st_gid, grp.getgrgid, os.getgid(), False) + try: + acl = ", ACL" if "system.posix_acl_access" in os.listxattr(path) else "" + except OSError: + acl = "" + if os.access(path, os.R_OK | os.W_OK): + you = "read+write" + elif os.access(path, os.R_OK): + you = "read only" + else: + you = "no access" + return f"{stat.filemode(st.st_mode)} {owner}:{group}{acl}, you: {you}" + + +def usbhid_quirks(root): + """[(vid, pid, flags)] the running usbhid was loaded with, or None if unknown.""" + text = read_attr(_path(root, "sys", "module", "usbhid", "parameters"), "quirks") + if text is None: + return None + quirks = [] + for item in text.split(","): + match = QUIRK.match(item.strip()) + if match: + quirks.append(tuple(int(g, 16) for g in match.groups())) + return quirks + + +def udev_rule_files(root): + """The rules files udev uses: the first of each name, /etc before /usr/lib.""" + chosen = {} + for directory in UDEV_DIRS: + try: + names = os.listdir(_path(root, directory)) + except OSError: + continue + for name in names: + if name.endswith(".rules") and name not in chosen: + chosen[name] = _path(root, directory, name) + return [chosen[name] for name in sorted(chosen)] + + +def _matches(pattern, value): + return any(fnmatch.fnmatchcase(value, alt) for alt in pattern.split("|")) + + +def rule_matches(text, vid, pid): + """Subsystems of the rules in this file that name this vendor and product + (a rule naming only the vendor covers every product).""" + found = [] + for line in text.replace("\\\n", " ").splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + keys = {"vendor": [], "product": [], "subsystem": []} + for key, value in RULE_MATCH.findall(line): + if "Vendor" in key or "VENDOR" in key: + keys["vendor"].append(value) + elif "SUBSYSTEM" in key: + keys["subsystem"].append(value) + else: + keys["product"].append(value) + if not keys["vendor"]: + continue + if all(_matches(p, f"{vid:04x}") for p in keys["vendor"]) and \ + all(_matches(p, f"{pid:04x}") for p in keys["product"]): + subsystem = "/".join(keys["subsystem"]) or "any" + if subsystem not in found: + found.append(subsystem) + return found + + +def udev_coverage(root, vid, pid): + covered = [] + for path in udev_rule_files(root): + try: + with open(path, errors="replace") as f: + subsystems = rule_matches(f.read(), vid, pid) + except OSError: + continue + if subsystems: + shown = "/" + os.path.relpath(path, root) + covered.append(f"`{shown}` ({', '.join(subsystems)})") + return covered + + +# -- the report --------------------------------------------------------------------------- +def distribution(root): + for rel in (("etc", "os-release"), ("usr", "lib", "os-release")): + try: + with open(_path(root, *rel)) as f: + for line in f: + if line.startswith("PRETTY_NAME="): + return line.partition("=")[2].strip().strip('"') + except OSError: + continue + return "unknown" + + +def table(rows): + lines = ["| | |", "|---|---|"] + lines += [f"| {key} | {value} |" for key, value in rows] + return "\n".join(lines) + + +def environment_rows(root, quirks): + experimental = os.environ.get(protocol.EXPERIMENTAL_ENV, "").strip() + return [ + ("ragnaros", __version__), + ("Python", platform.python_version()), + ("kernel", platform.release()), + ("distribution", distribution(root)), + ("libusb-1.0", ctypes.util.find_library("usb-1.0") or "not found"), + ("usbhid quirks", "unknown (usbhid not loaded?)" if quirks is None else + ", ".join(f"`0x{v:04x}:0x{p:04x}:0x{f:08x}`" for v, p, f in quirks) or "none"), + (f"`{protocol.EXPERIMENTAL_ENV}`", f"`{experimental}`" if experimental else "not set"), + ] + + +def _endpoint_row(ep): + address = ep["bEndpointAddress"] or "?" + size = ep["wMaxPacketSize"] + try: + size = str(int(size, 16) & 0x7FF) + except (TypeError, ValueError): + size = "?" + return (f"0x{address} {(ep['direction'] or '?').upper()}", ep["type"] or "?", size, + f"{ep['interval'] or '?'} (bInterval {ep['bInterval'] or '?'})") + + +def device_section(root, directory, quirks): + """(markdown, facts) for one device; facts feed the hints.""" + vid = int(read_attr(directory, "idVendor"), 16) + pid = int(read_attr(directory, "idProduct"), 16) + ident = f"{vid:04x}:{pid:04x}" + product = read_attr(directory, "product") + busnum, devnum = read_attr(directory, "busnum"), read_attr(directory, "devnum") + try: + node = f"/dev/bus/usb/{int(busnum):03d}/{int(devnum):03d}" + except (TypeError, ValueError): + node = None + access = node_access(_path(root, node.lstrip("/"))) if node else None + raw = read_bytes(os.path.join(directory, "descriptors")) or b"" + lengths = hid_descriptors(raw) + intfs = interfaces(directory) + experimental = os.environ.get(protocol.EXPERIMENTAL_ENV, "").strip() + try: + tried = vid == protocol.VID and experimental and protocol.parse_pid(experimental) == pid + except ValueError: + tried = False + + if (vid, pid) in SUPPORTED: + status = f"Supported ({SUPPORTED[(vid, pid)]})." + elif tried: + status = (f"Not supported yet; `{protocol.EXPERIMENTAL_ENV}` opts it into " + "the experimental test.") + else: + status = "Not supported yet." + quirk = [f for v, p, f in (quirks or []) if (v, p) == (vid, pid)] + rules = udev_coverage(root, vid, pid) + speed = read_attr(directory, "speed") + rows = [ + ("manufacturer", read_attr(directory, "manufacturer") or "none"), + ("product", product or "none"), + # never read: only whether the device has one + ("serial number", "present, not shown" + if os.path.exists(os.path.join(directory, "serial")) else "none"), + ("USB", f"{read_attr(directory, 'version') or '?'}, {speed or '?'} Mbit/s, " + f"bcdDevice {read_attr(directory, 'bcdDevice') or '?'}, " + f"class {read_attr(directory, 'bDeviceClass') or '?'}, " + f"{read_attr(directory, 'bNumConfigurations') or '?'} configuration(s), " + f"{(read_attr(directory, 'bNumInterfaces') or '?').strip()} interface(s), " + f"{read_attr(directory, 'bMaxPower') or '?'}"), + ("sysfs", f"`{os.path.basename(directory)}`"), + ("device node", f"`{node}` {access}" if access else f"`{node}` (missing)"), + ("usbhid quirk", ", ".join( + f"`0x{f:08x}` ({QUIRK_NAMES.get(f, 'flags')})" for f in quirk) or "none"), + ("udev rules", ", ".join(rules) or "none match"), + ] + out = [f"#### {ident} {product or ''}".rstrip(), "", status, "", table(rows), ""] + + out += ["| interface | class | driver | endpoint | type | max packet | interval |", + "|---|---|---|---|---|---|---|"] + for intf in intfs: + klass = intf["class"] or "?" + label = CLASSES.get(klass, "") + boot = BOOT_PROTOCOLS.get((intf["subclass"], intf["protocol"])) + label = f"{label}, {boot}" if boot and klass == "03" else label + head = [str(int(intf["number"], 16)) if intf["number"] else "?", + f"{klass}/{intf['subclass'] or '?'}/{intf['protocol'] or '?'}" + + (f" {label}" if label else ""), + intf["driver"] or "none"] + eps = intf["endpoints"] or [None] + for n, ep in enumerate(eps): + cells = (head if n == 0 else ["", "", ""]) + \ + (list(_endpoint_row(ep)) if ep else ["none", "", "", ""]) + out.append("| " + " | ".join(cells) + " |") + out.append("") + + for intf in intfs: + number = int(intf["number"], 16) if intf["number"] else None + declared = lengths.get(number) + size = f" ({declared[1]} bytes declared, HID {declared[0]})" if declared else "" + descriptors = [h for h in intf["hid"] if h["report_descriptor"]] + if not descriptors: + if intf["class"] == "03": + why = (f"no HID driver bound (driver: {intf['driver'] or 'none'})" + if not intf["hid"] else "not readable") + out += [f"HID report descriptor, interface {number}{size}: {why}.", ""] + continue + for hid in descriptors: + nodes = [] + for name in hid["hidraw"]: + hid_access = node_access(_path(root, "dev", name)) + nodes.append(f"`/dev/{name}` {hid_access}" if hid_access + else f"`/dev/{name}` (missing)") + out += [f"HID report descriptor, interface {number}{size}" + + (f", {', '.join(nodes)}" if nodes else ""), "", + "```", hexdump(hid["report_descriptor"]), "```", ""] + + if raw: + out += ["
Raw USB descriptors " + f"({len(raw)} bytes, as lsusb -v decodes them)", "", + "```", hexdump(raw), "```", "", "
", ""] + facts = {"id": ident, "vid": vid, "pid": pid, "node": node, + "writable": bool(access and "read+write" in access), + "supported": (vid, pid) in SUPPORTED} + return "\n".join(out), facts + + +def hints(facts): + """What to do next; printed on stderr so `> report.md` keeps only the report.""" + lines = [] + for fact in facts: + if fact["supported"]: + if not fact["writable"]: + lines.append(f"{fact['id']}: you cannot open {fact['node']}; install the udev " + "rules (`ragnaros install-system` or the .deb/AUR package) and " + "reboot once.") + continue + if fact["vid"] != protocol.VID: + continue + lines.append(f"{fact['id']} is not supported yet. To try it with the SS552 protocol " + "(experimental; README, \"Reporting a new device\"):") + steps = [] + if not fact["writable"] and fact["node"]: + steps.append((f"sudo chown \"$USER\" {fact['node']}", "until you unplug it")) + steps += [("systemctl --user stop ragnarosd", "if it runs"), + (f"ragnarosd --experimental-pid 0x{fact['pid']:04x}", "stays in the foreground"), + ("ragnarosctl watch --grab", "in a second terminal; press every key")] + lines += [f" {command:<40} # {note}" for command, note in steps] + return lines + + +def report(root=None, device=None): + """(markdown, hints, found): the report for vendor 0200, or one VID:PID.""" + root = ROOT if root is None else root + vid, pid = device if device else (protocol.VID, None) + quirks = usbhid_quirks(root) + wanted = f"{vid:04x}:{pid:04x}" if pid is not None else f"vendor {vid:04x}" + out = ["### Ragnaros device report", "", table(environment_rows(root, quirks)), ""] + facts = [] + devices = find_devices(root, vid, pid) + for directory in devices: + text, fact = device_section(root, directory, quirks) + out.append(text) + facts.append(fact) + if not devices: + out += [f"No USB device found for {wanted}.", ""] + notes = [f"No USB device found for {wanted}: is it plugged in? `lsusb` lists them" + + ("." if device else "; --device VID:PID reports another vendor's.")] + else: + notes = hints(facts) + return "\n".join(out).rstrip() + "\n", notes, bool(devices) + + +def run(args): + text, notes, found = report(device=args.device) + sys.stdout.write(text) + sys.stdout.flush() + for line in notes: + print(line, file=sys.stderr) + return 0 if found else 1 diff --git a/tests/test_diagnose.py b/tests/test_diagnose.py new file mode 100644 index 0000000..e5756ed --- /dev/null +++ b/tests/test_diagnose.py @@ -0,0 +1,238 @@ +"""`ragnaros diagnose` against a fake sysfs: read-only, and nothing private.""" +import builtins +import getpass +import os +import platform +import socket + +import pytest + +from ragnaros import cli, diagnose, paths + +# The SS552's configuration descriptors as sysfs keeps them: device + config, +# a vendor HID interface (EP 0x82 in, 0x03 out) and a boot keyboard (EP 0x81). +SS552_DESCRIPTORS = bytes.fromhex(""" + 12 01 00 02 00 00 00 40 00 02 01 30 02 00 01 02 + 03 01 09 02 42 00 02 01 00 80 e1 09 04 00 00 02 + 03 00 00 00 09 21 00 02 00 01 22 36 00 07 05 82 + 03 00 02 01 07 05 03 03 00 04 01 09 04 01 00 01 + 03 01 01 00 09 21 00 02 00 01 22 3f 00 07 05 81 + 03 08 00 0a""") +REPORT_DESCRIPTOR = bytes.fromhex("06 00 ff 09 01 a1 01 09 02 15 00 26 ff 00 75 08 95 40 81 02 c0") +SERIAL = "SECRET-SERIAL-0042" + + +def write(path, content): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb" if isinstance(content, bytes) else "w") as f: + f.write(content) + + +def endpoint(interface, address, direction, size, b_interval, interval): + for key, value in {"bEndpointAddress": address, "direction": direction, + "type": "Interrupt", "wMaxPacketSize": size, + "bInterval": b_interval, "interval": interval}.items(): + write(interface / f"ep_{address}" / key, value + "\n") + + +def plug(root, name="5-1.4", vid="0200", pid="2000", busnum=5, devnum=2, driver="usbhid"): + """A deck in the fake sysfs; with usbhid bound it gets hidraw nodes.""" + dev = root / "sys" / "bus" / "usb" / "devices" / name + for key, value in {"idVendor": vid, "idProduct": pid, "manufacturer": "HOTSPOTEKUSB", + "product": "HOTSPOTEKUSB HID DEMO", "serial": SERIAL, + "busnum": str(busnum), "devnum": str(devnum), "speed": "480", + "version": " 2.00", "bcdDevice": "0002", "bDeviceClass": "00", + "bNumConfigurations": "1", "bNumInterfaces": " 2", + "bMaxPower": "450mA"}.items(): + write(dev / key, value + "\n") + raw = bytearray(SS552_DESCRIPTORS) + raw[10:12] = int(pid, 16).to_bytes(2, "little") + write(dev / "descriptors", bytes(raw)) + drivers = root / "sys" / "bus" / "usb" / "drivers" / driver + drivers.mkdir(parents=True, exist_ok=True) + for number, (subclass, protocol) in enumerate([("00", "00"), ("01", "01")]): + intf = dev / f"{name}:1.{number}" + for key, value in {"bInterfaceNumber": f"{number:02x}", "bInterfaceClass": "03", + "bInterfaceSubClass": subclass, + "bInterfaceProtocol": protocol}.items(): + write(intf / key, value + "\n") + os.symlink(drivers, intf / "driver") + if driver == "usbhid": + hid = intf / f"0003:{vid.upper()}:{pid.upper()}.000{number + 5}" + write(hid / "report_descriptor", REPORT_DESCRIPTOR) + (hid / "hidraw" / f"hidraw{number + 3}").mkdir(parents=True) + write(root / "dev" / f"hidraw{number + 3}", "") + endpoint(dev / f"{name}:1.0", "82", "in", "0200", "01", "125us") + endpoint(dev / f"{name}:1.0", "03", "out", "0400", "01", "125us") + endpoint(dev / f"{name}:1.1", "81", "in", "0008", "0a", "64ms") + write(root / "dev" / "bus" / "usb" / f"{busnum:03d}" / f"{devnum:03d}", "") + return dev + + +@pytest.fixture +def root(tmp_path, monkeypatch): + """A machine with the SS552's udev rule and quirk installed and no deck yet.""" + monkeypatch.delenv("RAGNAROS_EXPERIMENTAL_PID", raising=False) + with open(os.path.join(paths.bundled("udev"), "70-ragnaros.rules")) as f: + write(tmp_path / "etc" / "udev" / "rules.d" / "70-ragnaros.rules", f.read()) + write(tmp_path / "usr" / "lib" / "udev" / "rules.d" / "50-udev-default.rules", + 'SUBSYSTEM=="usb", ENV{DEVTYPE}=="usb_device", MODE="0664"\n') + write(tmp_path / "sys" / "module" / "usbhid" / "parameters" / "quirks", + "0x0200:0x3001:0x00000004,(null),(null),(null)\n") + write(tmp_path / "etc" / "os-release", 'NAME="Debian"\nPRETTY_NAME="Debian GNU/Linux 13"\n') + hub = tmp_path / "sys" / "bus" / "usb" / "devices" / "usb5" + write(hub / "idVendor", "1d6b\n") + write(hub / "idProduct", "0002\n") + return tmp_path + + +def test_an_untested_sibling_gets_everything_needed_to_add_it(root): + plug(root) + text, notes, found = diagnose.report(str(root)) + assert found + assert "#### 0200:2000 HOTSPOTEKUSB HID DEMO\n\nNot supported yet." in text + assert ("| 0 | 03/00/00 HID | usbhid | 0x03 OUT | Interrupt | 1024 | 125us (bInterval 01) |\n" + "| | | | 0x82 IN | Interrupt | 512 | 125us (bInterval 01) |") in text + assert "| 1 | 03/01/01 HID, boot keyboard | usbhid | 0x81 IN | Interrupt | 8 |" in text + assert ("HID report descriptor, interface 0 (54 bytes declared, HID 2.00), " + "`/dev/hidraw3` ") in text + assert "```\n06 00 ff 09 01 a1 01 09 02 15 00 26 ff 00 75 08\n95 40 81 02 c0\n```" in text + assert "| device node | `/dev/bus/usb/005/002` " in text + assert "| udev rules | none match |" in text # 70-ragnaros.rules is for 3001 only + assert "| usbhid quirk | none |" in text + assert "| distribution | Debian GNU/Linux 13 |" in text + assert "| usbhid quirks | `0x0200:0x3001:0x00000004` |" in text + assert "12 01 00 02 00 00 00 40 00 02 00 20 02 00 01 02" in text # raw descriptors + assert "1d6b" not in text # only vendor 0200 by default + assert any("ragnarosd --experimental-pid 0x2000" in line for line in notes) + + +def test_the_report_leaves_out_serial_host_and_user(root, monkeypatch): + plug(root) + monkeypatch.setattr(platform, "node", lambda: "secret-hostname") + monkeypatch.setattr(socket, "gethostname", lambda: "secret-hostname") + text, notes, _ = diagnose.report(str(root)) + assert "| serial number | present, not shown |" in text + for private in (SERIAL, "secret-hostname", str(root)): + assert private not in text and not any(private in line for line in notes) + if os.getuid() != 0: + assert getpass.getuser() not in text + + +def test_accounts_are_named_only_when_they_are_system_ones(): + names = {0: "root", 46: "plugdev", 1000: "alice", 1001: "bob"} + + def lookup(ident): + return (names[ident],) + + assert diagnose._account(0, lookup, 1000, True) == "root" + assert diagnose._account(46, lookup, 1000, False) == "plugdev" + assert diagnose._account(1000, lookup, 1000, True) == "you" + assert diagnose._account(1001, lookup, 1000, True) == "another user" + assert diagnose._account(1001, lookup, 1000, False) == "a user group" + + +def test_the_ss552_without_a_hid_driver(root): + plug(root, pid="3001", devnum=3, driver="usbfs") + text, notes, found = diagnose.report(str(root)) + assert found and notes == [] # writable, supported: nothing to do + assert "Supported (Redragon SS552)." in text + assert ("HID report descriptor, interface 0 (54 bytes declared, HID 2.00): " + "no HID driver bound (driver: usbfs).") in text + assert "| usbhid quirk | `0x00000004` (IGNORE) |" in text + assert "| udev rules | `/etc/udev/rules.d/70-ragnaros.rules` (hidraw, usb) |" in text + + +def test_device_picks_one_vid_pid(root): + plug(root) + plug(root, name="3-2", vid="046d", pid="c08c", busnum=3, devnum=2) + text, notes, found = diagnose.report(str(root), (0x046D, 0xC08C)) + assert found and "#### 046d:c08c" in text and "0200:2000" not in text + assert notes == [] # the experimental test is for vendor 0200 only + text, notes, found = diagnose.report(str(root), (0x0200, 0x1234)) + assert not found and "No USB device found for 0200:1234." in text + + +def test_the_hint_grants_access_when_the_node_is_locked(root): + if os.getuid() == 0: + pytest.skip("root can open anything") + plug(root) + os.chmod(root / "dev" / "bus" / "usb" / "005" / "002", 0o444) + text, notes, _ = diagnose.report(str(root)) + assert "you: read only" in text + assert any(line.startswith(' sudo chown "$USER" /dev/bus/usb/005/002 ') + and line.endswith("# until you unplug it") for line in notes) + + +def test_an_opted_in_device_says_so(root, monkeypatch): + plug(root) + monkeypatch.setenv("RAGNAROS_EXPERIMENTAL_PID", "0x2000") + text, _, _ = diagnose.report(str(root)) + assert "| `RAGNAROS_EXPERIMENTAL_PID` | `0x2000` |" in text + assert "opts it into the experimental test" in text + + +def test_it_never_writes_or_opens_a_device_node(root, monkeypatch): + plug(root) + opened = [] + real_open = builtins.open + + def guarded(path, mode="r", *args, **kwargs): + opened.append((str(path), mode)) + return real_open(path, mode, *args, **kwargs) + + monkeypatch.setattr(diagnose, "open", guarded, raising=False) + diagnose.report(str(root)) + assert opened and all(mode in ("r", "rb") for _, mode in opened) + assert not [p for p, _ in opened if os.path.relpath(p, root).startswith("dev")] + assert not [p for p, _ in opened if p.endswith("serial")] + + +def test_hid_descriptor_lengths_come_from_the_raw_descriptors(): + assert diagnose.hid_descriptors(SS552_DESCRIPTORS) == {0: ("2.00", 54), 1: ("2.00", 63)} + assert diagnose.hid_descriptors(b"") == {} + assert diagnose.hid_descriptors(b"\x00\x00") == {} + + +@pytest.mark.parametrize("rule,expected", [ + ('SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="3001", MODE="0660"', []), + ('SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="3001|2000"', ["usb"]), + ('SUBSYSTEM=="hidraw", ATTRS{idVendor}=="0200", ATTRS{idProduct}=="2*"', ["hidraw"]), + ('ATTRS{idVendor}=="0200", TAG+="uaccess"', ["any"]), + ('ENV{ID_VENDOR_ID}=="0200", ENV{ID_MODEL_ID}=="2000"', ["any"]), + ('# SUBSYSTEM=="usb", ATTR{idVendor}=="0200"', []), + ('SUBSYSTEM=="usb", ATTR{idVendor}=="046d"', []), + ('SUBSYSTEM=="usb", \\\n ATTR{idVendor}=="0200", ATTR{idProduct}=="2000"', ["usb"]), +]) +def test_udev_rule_matching(rule, expected): + assert diagnose.rule_matches(rule, 0x0200, 0x2000) == expected + + +def test_etc_rules_override_packaged_ones(root): + write(root / "usr" / "lib" / "udev" / "rules.d" / "70-ragnaros.rules", + 'SUBSYSTEM=="usb", ATTR{idVendor}=="0200", ATTR{idProduct}=="2000"\n') + assert diagnose.udev_coverage(str(root), 0x0200, 0x2000) == [] + os.remove(root / "etc" / "udev" / "rules.d" / "70-ragnaros.rules") + assert diagnose.udev_coverage(str(root), 0x0200, 0x2000) == [ + "`/usr/lib/udev/rules.d/70-ragnaros.rules` (usb)"] + + +def test_cli_diagnose_report_on_stdout_hints_on_stderr(root, monkeypatch, capsys): + plug(root) + monkeypatch.setattr(diagnose, "ROOT", str(root)) + assert cli.main(["diagnose"]) == 0 + out, err = capsys.readouterr() + assert out.startswith("### Ragnaros device report\n") + assert "ragnarosd --experimental-pid" not in out + assert "ragnarosd --experimental-pid 0x2000" in err + assert cli.main(["diagnose", "--device", "0200:1234"]) == 1 + assert "No USB device found for 0200:1234" in capsys.readouterr().err + with pytest.raises(SystemExit): + cli.main(["diagnose", "--device", "nope"]) + + +def test_parse_device(): + assert diagnose.parse_device("0200:2000") == (0x0200, 0x2000) + for bad in ("0200", "0200:", "zz:2000", "0200:12345"): + with pytest.raises(ValueError): + diagnose.parse_device(bad)