# HG changeset patch # User Goffi # Date 1715428363 -7200 # Node ID e47e29511d574a25abb552d46c557a355b700d89 # Parent 8acf46ed7f360aa8b1b2084752e6939856766dfa cli (remote-control): new `remote-control` command: 2 subcommands are available: `send` to control remotely a device, and `receive` to be controlled. Documentation will follow. rel 436 diff -r 8acf46ed7f36 -r e47e29511d57 libervia/cli/cmd_remote_control.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/libervia/cli/cmd_remote_control.py Sat May 11 13:52:43 2024 +0200 @@ -0,0 +1,383 @@ +#!/usr/bin/env python3 + + +# Libervia CLI +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + + +import asyncio +from functools import partial +import logging +import xml.etree.ElementTree as ET + +from prompt_toolkit.input import Input, create_input +from prompt_toolkit.key_binding import KeyPress +from prompt_toolkit.keys import Keys +from prompt_toolkit.patch_stdout import patch_stdout +from prompt_toolkit import PromptSession + +from libervia.backend.core import exceptions +from libervia.backend.core.i18n import _ +from libervia.backend.tools.common import data_format +from libervia.cli.constants import Const as C +from libervia.frontends.tools import aio, jid +from libervia.frontends.tools.webrtc_remote_control import WebRTCRemoteController + +from . import base +from . import xmlui_manager + +__commands__ = ["RemoteControl"] + + +class Send(base.CommandBase): + def __init__(self, host): + super(Send, self).__init__( + host, + "send", + use_progress=True, + use_verbose=True, + help=_("remote control another device"), + ) + self.remote_controller: WebRTCRemoteController | None = None + + def add_parser_options(self): + self.parser.add_argument("jid", help=_("the destination jid")) + + def send_key( + self, + key: str, + code: str, + ctrl_key: bool = False, + shift_key: bool = False, + alt_key: bool = False, + meta_key: bool = False, + ) -> None: + """ + Send the key press input. + + @param key: The key pressed. + @param code: The code of the key pressed. + @param ctrl_key: Whether the Ctrl key was pressed. + @param shift_key: Whether the Shift key was pressed. + @param alt_key: Whether the Alt key was pressed. + @param meta_key: Whether the Meta key was pressed. + """ + assert self.remote_controller is not None + event_data = { + "key": key, + "code": code, + "ctrlKey": ctrl_key, + "shiftKey": shift_key, + "altKey": alt_key, + "metaKey": meta_key, + } + # we send both events as we don't distinguish them. + for evt_type in ["keydown", "keyup"]: + key_data = {"type": evt_type, **event_data} + self.remote_controller.send_input(key_data) + print(f"Sending {key_data}") + + def handle_key_press(self, key_press: KeyPress) -> None: + """Handle key press event.""" + key = key_press.key + if key.startswith("c-"): + key = key[2:] # remove "c-" prefix + elif key.startswith("s-"): + key = key[2:] # remove "s-" prefix + self.send_key( + key=key.lower(), + # FIXME: handle properly code. + code=key.lower(), + ctrl_key=key_press.key.startswith("c-"), + shift_key=key_press.key.startswith("s-") or key.isupper(), + # FIXME: alt-key is translated to escape + key, see + # https://python-prompt-toolkit.readthedocs.io/en/master/pages/advanced_topics/key_bindings.html + alt_key=False, + # TODO + meta_key=False, + ) + + def on_keys_ready(self, input_: Input, handle_key_fut: asyncio.Future) -> None: + for key_press in input_.read_keys(): + if key_press.key == Keys.ControlC: + handle_key_fut.set_exception(KeyboardInterrupt()) + else: + self.handle_key_press(key_press) + + async def confirm_ctrl_c(self) -> bool: + """Ask user if they want to send Ctrl-C event or quit.""" + session = PromptSession() + with patch_stdout(): + while True: + response = await session.prompt_async( + "Ctrl-C pressed. Send event (e) or quit (q)? (e/q): " + ) + if response.lower() == "e": + return True + elif response.lower() == "q": + return False + + async def handle_ctrl_c(self) -> None: + """Handle Ctrl-C key press.""" + if await self.confirm_ctrl_c(): + self.send_key(key="c", code="c", ctrl_key=True) + else: + await self.host.a_quit() + + async def _on_open(self, remote_controller: WebRTCRemoteController) -> None: + input_ = create_input() + self.disp( + "Connection with peer established. Your keyboard input will be sent to " + "controlled device." + ) + + while True: + handle_key_fut = asyncio.Future() + try: + with input_.raw_mode(): + with input_.attach( + partial(self.on_keys_ready, input_, handle_key_fut) + ): + await handle_key_fut + except KeyboardInterrupt: + await self.handle_ctrl_c() + + async def start(self): + root_logger = logging.getLogger() + # we don't want any formatting for messages from webrtc + for handler in root_logger.handlers: + handler.setFormatter(None) + if self.verbosity == 0: + root_logger.setLevel(logging.ERROR) + if self.verbosity >= 1: + root_logger.setLevel(logging.WARNING) + if self.verbosity >= 2: + root_logger.setLevel(logging.DEBUG) + + aio.install_glib_asyncio_iteration() + self.remote_controller = WebRTCRemoteController( + self.host.bridge, self.profile, end_call_cb=self.host.a_quit + ) + await self.remote_controller.start( + self.args.jid, {"devices": {"keyboard": {}}}, on_open_cb=self._on_open + ) + + +class Receive(base.CommandAnswering): + def __init__(self, host): + super().__init__( + host, + "receive", + use_verbose=True, + help=_("be remote controlled by another device"), + ) + self.action_callbacks = { + C.META_TYPE_CONFIRM: self.on_confirm_action, + C.META_TYPE_NOT_IN_ROSTER_LEAK: self.on_confirm_action, + C.META_TYPE_REMOTE_CONTROL: self.on_remote_control_action, + } + self.receiver = None + + def add_parser_options(self): + self.parser.add_argument( + "-S", + "--share-screen", + choices=["yes", "no", "auto"], + default="auto", + help=_("share the screen (default: auto)"), + ) + self.parser.add_argument( + "jids", + nargs="+", + help=_("jids accepted automatically"), + ) + + async def start_webrtc( + self, from_jid: jid.JID, session_id: str, screenshare: dict, devices: dict + ) -> None: + """Start the WebRTC workflown""" + assert self.receiver is not None + root_logger = logging.getLogger() + # we don't want any formatting for messages from webrtc + for handler in root_logger.handlers: + handler.setFormatter(None) + if self.verbosity == 0: + root_logger.setLevel(logging.ERROR) + if self.verbosity >= 1: + root_logger.setLevel(logging.WARNING) + if self.verbosity >= 2: + root_logger.setLevel(logging.DEBUG) + await self.receiver.start_receiving(from_jid, session_id, screenshare) + + def get_xmlui_id(self, action_data): + # FIXME: we temporarily use ElementTree, but a real XMLUI managing module + # should be available in the futur + # TODO: XMLUI module + try: + xml_ui = action_data["xmlui"] + except KeyError: + self.disp(_("Action has no XMLUI"), 1) + else: + ui = ET.fromstring(xml_ui.encode("utf-8")) + xmlui_id = ui.get("submit") + if not xmlui_id: + self.disp(_("Invalid XMLUI received"), error=True) + return xmlui_id + + async def get_confirmation( + self, action_data: dict, pre_accepted: bool = False + ) -> tuple[str, jid.JID, bool]: + """Check if action is confirmed, and ask user otherwise + + @param action_data: Data as used in on_action_new. + @return: a tuple with: + - XMLUI ID + - sender JID + - confirmation boolean + """ + xmlui_id = self.get_xmlui_id(action_data) + if xmlui_id is None: + self.disp("Internal ERROR: xmlui_id missing", error=True) + raise exceptions.InternalError() + if ( + action_data["type"] != C.META_TYPE_REMOTE_CONTROL + and action_data.get("subtype") != C.META_TYPE_REMOTE_CONTROL + ): + self.disp(_("Ignoring confirm dialog unrelated to remote control."), 1) + raise exceptions.CancelError + try: + from_jid = jid.JID(action_data["from_jid"]) + except ValueError: + self.disp( + _('invalid "from_jid" value received, ignoring: {value}').format( + value=action_data["from_jid"] + ), + error=True, + ) + raise exceptions.DataError + except KeyError: + self.disp(_('ignoring action without "from_jid" value'), error=True) + raise exceptions.DataError + + self.disp(_("Confirmation needed for request from an entity not in roster"), 1) + + if pre_accepted: + # Session has already been accepted during pre-flight. + confirmed = True + elif action_data["type"] == C.META_TYPE_CONFIRM and not self.bare_jids: + # Sender is in roster, and we have an "accept all" policy. + confirmed = True + self.disp( + _( + f"{from_jid} automatically confirmed due to entity being in roster" + " and accept all policy." + ) + ) + elif from_jid.bare in self.bare_jids: + # If the sender is expected, we can confirm the session. + confirmed = True + self.disp(_("Sender confirmed because they are explicitly expected"), 1) + else: + # Not automatically accepted, we ask authorisation to the user. + xmlui = xmlui_manager.create(self.host, action_data["xmlui"]) + confirmed = await self.host.confirm(xmlui.dlg.message) + return xmlui_id, from_jid, confirmed + + async def on_confirm_action(self, action_data, action_id, security_limit, profile): + """Handle pre-flight remote control request""" + try: + xmlui_id, from_jid, confirmed = await self.get_confirmation(action_data) + except exceptions.InternalError: + self.host.quit_from_signal(1) + return + except (exceptions.InternalError, exceptions.DataError): + return + xmlui_data = {"answer": C.bool_const(confirmed)} + await self.host.bridge.action_launch( + xmlui_id, data_format.serialise(xmlui_data), profile_key=profile + ) + if not confirmed: + self.disp(_("Session refused for {from_jid}").format(from_jid=from_jid)) + self.host.quit_from_signal(0) + + async def on_remote_control_action( + self, action_data, action_id, security_limit, profile + ): + """Handles actual remote control request""" + try: + session_id = action_data["session_id"] + except KeyError: + self.disp( + f"Internal Error: Session ID is missing in action data: {action_data=}", + error=True, + ) + return + pre_accepted = action_data.get("pre_accepted", False) + try: + xmlui_id, from_jid, confirmed = await self.get_confirmation( + action_data, pre_accepted + ) + except exceptions.InternalError: + self.host.quit_from_signal(1) + return + except (exceptions.InternalError, exceptions.DataError): + return + if confirmed: + await self.start_webrtc( + from_jid, + session_id, + action_data.get("screenshare", {}), + action_data.get("devices", {}), + ) + xmlui_data = {"answer": C.bool_const(confirmed)} + await self.host.bridge.action_launch( + xmlui_id, data_format.serialise(xmlui_data), profile_key=profile + ) + + async def start(self): + self.bare_jids = [jid.JID(jid_).bare for jid_ in self.args.jids] + from libervia.frontends.tools.webrtc_remote_control import ( + WebRTCRemoteControlReceiver, + ) + + self.receiver = WebRTCRemoteControlReceiver( + self.host.bridge, + self.profile, + on_close_cb=self.host.a_quit, + verbose=self.verbosity >= 1, + ) + aio.install_glib_asyncio_iteration() + # FIXME: for now AUTO always do the screen sharing, but is should be disabled and + # appropriate method should be used when no desktop environment is detected. + with_screen_sharing = self.args.share_screen in ("yes", "auto") + await self.receiver.request_remote_desktop( + with_screen_sharing + ) + + self.disp(_("Waiting for controlling device…")) + await self.start_answering() + + +class RemoteControl(base.CommandBase): + subcommands = (Send, Receive) + + def __init__(self, host): + super().__init__( + host, + "remote-control", + use_profile=False, + help=_("Control or be controlled by another device."), + )