SmartestHome/hosts/thin-client/agent/thinclient_agent/input_control.py

"""Keyboard and pointer injection into the focused window, driven from Home Assistant.

The point of this module is the Home Assistant mobile app: type a search term into the
kiosk's Firefox, or nudge and click the pointer, without opening a VNC viewer.

ydotool rather than xdotool: xdotool talks XTEST to an X server, which does not exist
here — this is a wlroots Wayland session. ydotool goes in the other direction, writing
to /dev/uinput as a virtual input device, so the compositor sees ordinary hardware
events and no compositor-specific protocol is involved.

SECURITY: the text entity's payload is typed verbatim, and that is the entire point of
it — the user is asking for those characters to appear in the focused field. It is
passed as a single argv element to subprocess with no shell, so it can be any string
without becoming a command. Every *other* action here is a fixed enumerated constant:
the movement buttons resolve to one of four hard-coded deltas and the click buttons to
one of two hard-coded button codes. Nothing inbound ever becomes an argv element except
that one string.

move_relative() takes an arbitrary integer delta rather than an enumerated direction,
which is why it is worth being explicit that it is NOT part of that inbound surface:
mqtt_discovery.py never wires it to a topic. Its only caller is
configs/gesture-control/gesture_pointer.py, which runs locally on this machine with no
network input of any kind. The deltas are clamped and coerced to int here anyway, so
the argv elements stay numeric whatever a caller passes.
"""

from __future__ import annotations

import logging
import shutil
import subprocess

log = logging.getLogger(__name__)

MOVE_STEP = 20

DIRECTIONS = {
    "UP": (0, -MOVE_STEP),
    "DOWN": (0, MOVE_STEP),
    "LEFT": (-MOVE_STEP, 0),
    "RIGHT": (MOVE_STEP, 0),
}

# ydotool 1.x takes a hex mask where 0x40 is "left button" and 0xC0 is
# "left button, press and release". ydotool 0.1.x (what Debian bookworm ships) takes a
# plain index instead: 0 left, 1 right, 2 middle. Which dialect is installed is decided
# in _uses_daemon_dialect() below.
CLICK_CODES = {
    "LEFT": {"modern": "0xC0", "legacy": "0"},
    "RIGHT": {"modern": "0xC1", "legacy": "1"},
}

MAX_TYPE_LENGTH = 255

# Ceiling on a single move_relative() step. A camera gesture frame that lands badly
# should nudge the pointer wrongly, not fling it off the far edge of the display.
MAX_RELATIVE_STEP = 200


class InputControl:
    def __init__(self, env_provider):
        self._env_provider = env_provider
        # ydotool 1.x split the tool into a client plus a ydotoold daemon that owns
        # /dev/uinput; 0.1.x has no daemon and opens the device itself. The presence of
        # the daemon binary is therefore also a reliable marker of which command-line
        # dialect this image got. See 1000-ydotool.hook.chroot.
        self._dialect = "modern" if shutil.which("ydotoold") else "legacy"
        self.available = shutil.which("ydotool") is not None
        if not self.available:
            log.warning("ydotool is not installed; the HA input-control entities will do nothing")
        else:
            log.info("ydotool present, using the %s command dialect", self._dialect)

    def _ydotool(self, *args: str) -> bool:
        if not self.available:
            return False
        try:
            result = subprocess.run(
                ["ydotool", *args],
                env=self._env_provider(),
                capture_output=True,
                text=True,
                timeout=15,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired) as exc:
            log.warning("ydotool %s failed: %s", args[0], exc)
            return False
        if result.returncode != 0:
            log.warning("ydotool %s: %s", args[0], result.stderr.strip())
            return False
        return True

    def type_text(self, payload: str) -> None:
        text = payload.rstrip("\n")
        if not text:
            return
        if len(text) > MAX_TYPE_LENGTH:
            # Not a security limit — the payload is safe at any length. It is a guard
            # against a stuck automation holding the keyboard for minutes on a machine
            # whose display is shared with a room full of people.
            log.warning("truncating typed text from %d to %d characters", len(text), MAX_TYPE_LENGTH)
            text = text[:MAX_TYPE_LENGTH]

        log.info("typing %d characters into the focused window", len(text))
        # `--` so text beginning with a dash is typed rather than parsed as options.
        # VERIFY: accepted by ydotool 1.x; if the installed 0.1.x build rejects it, drop
        # it here — the only consequence is that leading-dash text is misread as flags.
        self._ydotool("type", "--", text)

    def move(self, direction: str) -> None:
        delta = DIRECTIONS.get(direction.strip().upper())
        if delta is None:
            log.warning("ignoring unknown pointer direction %r", direction)
            return
        self.move_relative(*delta)

    def move_relative(self, dx: int, dy: int) -> None:
        dx = max(-MAX_RELATIVE_STEP, min(MAX_RELATIVE_STEP, int(dx)))
        dy = max(-MAX_RELATIVE_STEP, min(MAX_RELATIVE_STEP, int(dy)))
        if dx == 0 and dy == 0:
            return
        if self._dialect == "modern":
            self._ydotool("mousemove", "-x", str(dx), "-y", str(dy))
        else:
            self._ydotool("mousemove", str(dx), str(dy))

    def click(self, button: str) -> None:
        code = CLICK_CODES.get(button.strip().upper())
        if code is None:
            log.warning("ignoring unknown mouse button %r", button)
            return
        self._ydotool("click", code[self._dialect])