Mercurial > libervia-backend

comparison libervia/cli/cmd_remote_control.py @ 4243:e47e29511d57

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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
author Goffi <goffi@goffi.org>
date Sat, 11 May 2024 13:52:43 +0200
parents
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4242:8acf46ed7f36 4243:e47e29511d57
1 #!/usr/bin/env python3
2
3
4 # Libervia CLI
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6
7 # This program is free software: you can redistribute it and/or modify
8 # it under the terms of the GNU Affero General Public License as published by
9 # the Free Software Foundation, either version 3 of the License, or
10 # (at your option) any later version.
11
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU Affero General Public License for more details.
16
17 # You should have received a copy of the GNU Affero General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20
21 import asyncio
22 from functools import partial
23 import logging
24 import xml.etree.ElementTree as ET
25
26 from prompt_toolkit.input import Input, create_input
27 from prompt_toolkit.key_binding import KeyPress
28 from prompt_toolkit.keys import Keys
29 from prompt_toolkit.patch_stdout import patch_stdout
30 from prompt_toolkit import PromptSession
31
32 from libervia.backend.core import exceptions
33 from libervia.backend.core.i18n import _
34 from libervia.backend.tools.common import data_format
35 from libervia.cli.constants import Const as C
36 from libervia.frontends.tools import aio, jid
37 from libervia.frontends.tools.webrtc_remote_control import WebRTCRemoteController
38
39 from . import base
40 from . import xmlui_manager
41
42 __commands__ = ["RemoteControl"]
43
44
45 class Send(base.CommandBase):
46 def __init__(self, host):
47 super(Send, self).__init__(
48 host,
49 "send",
50 use_progress=True,
51 use_verbose=True,
52 help=_("remote control another device"),
53 )
54 self.remote_controller: WebRTCRemoteController | None = None
55
56 def add_parser_options(self):
57 self.parser.add_argument("jid", help=_("the destination jid"))
58
59 def send_key(
60 self,
61 key: str,
62 code: str,
63 ctrl_key: bool = False,
64 shift_key: bool = False,
65 alt_key: bool = False,
66 meta_key: bool = False,
67 ) -> None:
68 """
69 Send the key press input.
70
71 @param key: The key pressed.
72 @param code: The code of the key pressed.
73 @param ctrl_key: Whether the Ctrl key was pressed.
74 @param shift_key: Whether the Shift key was pressed.
75 @param alt_key: Whether the Alt key was pressed.
76 @param meta_key: Whether the Meta key was pressed.
77 """
78 assert self.remote_controller is not None
79 event_data = {
80 "key": key,
81 "code": code,
82 "ctrlKey": ctrl_key,
83 "shiftKey": shift_key,
84 "altKey": alt_key,
85 "metaKey": meta_key,
86 }
87 # we send both events as we don't distinguish them.
88 for evt_type in ["keydown", "keyup"]:
89 key_data = {"type": evt_type, **event_data}
90 self.remote_controller.send_input(key_data)
91 print(f"Sending {key_data}")
92
93 def handle_key_press(self, key_press: KeyPress) -> None:
94 """Handle key press event."""
95 key = key_press.key
96 if key.startswith("c-"):
97 key = key[2:] # remove "c-" prefix
98 elif key.startswith("s-"):
99 key = key[2:] # remove "s-" prefix
100 self.send_key(
101 key=key.lower(),
102 # FIXME: handle properly code.
103 code=key.lower(),
104 ctrl_key=key_press.key.startswith("c-"),
105 shift_key=key_press.key.startswith("s-") or key.isupper(),
106 # FIXME: alt-key is translated to escape + key, see
107 # https://python-prompt-toolkit.readthedocs.io/en/master/pages/advanced_topics/key_bindings.html
108 alt_key=False,
109 # TODO
110 meta_key=False,
111 )
112
113 def on_keys_ready(self, input_: Input, handle_key_fut: asyncio.Future) -> None:
114 for key_press in input_.read_keys():
115 if key_press.key == Keys.ControlC:
116 handle_key_fut.set_exception(KeyboardInterrupt())
117 else:
118 self.handle_key_press(key_press)
119
120 async def confirm_ctrl_c(self) -> bool:
121 """Ask user if they want to send Ctrl-C event or quit."""
122 session = PromptSession()
123 with patch_stdout():
124 while True:
125 response = await session.prompt_async(
126 "Ctrl-C pressed. Send event (e) or quit (q)? (e/q): "
127 )
128 if response.lower() == "e":
129 return True
130 elif response.lower() == "q":
131 return False
132
133 async def handle_ctrl_c(self) -> None:
134 """Handle Ctrl-C key press."""
135 if await self.confirm_ctrl_c():
136 self.send_key(key="c", code="c", ctrl_key=True)
137 else:
138 await self.host.a_quit()
139
140 async def _on_open(self, remote_controller: WebRTCRemoteController) -> None:
141 input_ = create_input()
142 self.disp(
143 "Connection with peer established. Your keyboard input will be sent to "
144 "controlled device."
145 )
146
147 while True:
148 handle_key_fut = asyncio.Future()
149 try:
150 with input_.raw_mode():
151 with input_.attach(
152 partial(self.on_keys_ready, input_, handle_key_fut)
153 ):
154 await handle_key_fut
155 except KeyboardInterrupt:
156 await self.handle_ctrl_c()
157
158 async def start(self):
159 root_logger = logging.getLogger()
160 # we don't want any formatting for messages from webrtc
161 for handler in root_logger.handlers:
162 handler.setFormatter(None)
163 if self.verbosity == 0:
164 root_logger.setLevel(logging.ERROR)
165 if self.verbosity >= 1:
166 root_logger.setLevel(logging.WARNING)
167 if self.verbosity >= 2:
168 root_logger.setLevel(logging.DEBUG)
169
170 aio.install_glib_asyncio_iteration()
171 self.remote_controller = WebRTCRemoteController(
172 self.host.bridge, self.profile, end_call_cb=self.host.a_quit
173 )
174 await self.remote_controller.start(
175 self.args.jid, {"devices": {"keyboard": {}}}, on_open_cb=self._on_open
176 )
177
178
179 class Receive(base.CommandAnswering):
180 def __init__(self, host):
181 super().__init__(
182 host,
183 "receive",
184 use_verbose=True,
185 help=_("be remote controlled by another device"),
186 )
187 self.action_callbacks = {
188 C.META_TYPE_CONFIRM: self.on_confirm_action,
189 C.META_TYPE_NOT_IN_ROSTER_LEAK: self.on_confirm_action,
190 C.META_TYPE_REMOTE_CONTROL: self.on_remote_control_action,
191 }
192 self.receiver = None
193
194 def add_parser_options(self):
195 self.parser.add_argument(
196 "-S",
197 "--share-screen",
198 choices=["yes", "no", "auto"],
199 default="auto",
200 help=_("share the screen (default: auto)"),
201 )
202 self.parser.add_argument(
203 "jids",
204 nargs="+",
205 help=_("jids accepted automatically"),
206 )
207
208 async def start_webrtc(
209 self, from_jid: jid.JID, session_id: str, screenshare: dict, devices: dict
210 ) -> None:
211 """Start the WebRTC workflown"""
212 assert self.receiver is not None
213 root_logger = logging.getLogger()
214 # we don't want any formatting for messages from webrtc
215 for handler in root_logger.handlers:
216 handler.setFormatter(None)
217 if self.verbosity == 0:
218 root_logger.setLevel(logging.ERROR)
219 if self.verbosity >= 1:
220 root_logger.setLevel(logging.WARNING)
221 if self.verbosity >= 2:
222 root_logger.setLevel(logging.DEBUG)
223 await self.receiver.start_receiving(from_jid, session_id, screenshare)
224
225 def get_xmlui_id(self, action_data):
226 # FIXME: we temporarily use ElementTree, but a real XMLUI managing module
227 # should be available in the futur
228 # TODO: XMLUI module
229 try:
230 xml_ui = action_data["xmlui"]
231 except KeyError:
232 self.disp(_("Action has no XMLUI"), 1)
233 else:
234 ui = ET.fromstring(xml_ui.encode("utf-8"))
235 xmlui_id = ui.get("submit")
236 if not xmlui_id:
237 self.disp(_("Invalid XMLUI received"), error=True)
238 return xmlui_id
239
240 async def get_confirmation(
241 self, action_data: dict, pre_accepted: bool = False
242 ) -> tuple[str, jid.JID, bool]:
243 """Check if action is confirmed, and ask user otherwise
244
245 @param action_data: Data as used in on_action_new.
246 @return: a tuple with:
247 - XMLUI ID
248 - sender JID
249 - confirmation boolean
250 """
251 xmlui_id = self.get_xmlui_id(action_data)
252 if xmlui_id is None:
253 self.disp("Internal ERROR: xmlui_id missing", error=True)
254 raise exceptions.InternalError()
255 if (
256 action_data["type"] != C.META_TYPE_REMOTE_CONTROL
257 and action_data.get("subtype") != C.META_TYPE_REMOTE_CONTROL
258 ):
259 self.disp(_("Ignoring confirm dialog unrelated to remote control."), 1)
260 raise exceptions.CancelError
261 try:
262 from_jid = jid.JID(action_data["from_jid"])
263 except ValueError:
264 self.disp(
265 _('invalid "from_jid" value received, ignoring: {value}').format(
266 value=action_data["from_jid"]
267 ),
268 error=True,
269 )
270 raise exceptions.DataError
271 except KeyError:
272 self.disp(_('ignoring action without "from_jid" value'), error=True)
273 raise exceptions.DataError
274
275 self.disp(_("Confirmation needed for request from an entity not in roster"), 1)
276
277 if pre_accepted:
278 # Session has already been accepted during pre-flight.
279 confirmed = True
280 elif action_data["type"] == C.META_TYPE_CONFIRM and not self.bare_jids:
281 # Sender is in roster, and we have an "accept all" policy.
282 confirmed = True
283 self.disp(
284 _(
285 f"{from_jid} automatically confirmed due to entity being in roster"
286 " and accept all policy."
287 )
288 )
289 elif from_jid.bare in self.bare_jids:
290 # If the sender is expected, we can confirm the session.
291 confirmed = True
292 self.disp(_("Sender confirmed because they are explicitly expected"), 1)
293 else:
294 # Not automatically accepted, we ask authorisation to the user.
295 xmlui = xmlui_manager.create(self.host, action_data["xmlui"])
296 confirmed = await self.host.confirm(xmlui.dlg.message)
297 return xmlui_id, from_jid, confirmed
298
299 async def on_confirm_action(self, action_data, action_id, security_limit, profile):
300 """Handle pre-flight remote control request"""
301 try:
302 xmlui_id, from_jid, confirmed = await self.get_confirmation(action_data)
303 except exceptions.InternalError:
304 self.host.quit_from_signal(1)
305 return
306 except (exceptions.InternalError, exceptions.DataError):
307 return
308 xmlui_data = {"answer": C.bool_const(confirmed)}
309 await self.host.bridge.action_launch(
310 xmlui_id, data_format.serialise(xmlui_data), profile_key=profile
311 )
312 if not confirmed:
313 self.disp(_("Session refused for {from_jid}").format(from_jid=from_jid))
314 self.host.quit_from_signal(0)
315
316 async def on_remote_control_action(
317 self, action_data, action_id, security_limit, profile
318 ):
319 """Handles actual remote control request"""
320 try:
321 session_id = action_data["session_id"]
322 except KeyError:
323 self.disp(
324 f"Internal Error: Session ID is missing in action data: {action_data=}",
325 error=True,
326 )
327 return
328 pre_accepted = action_data.get("pre_accepted", False)
329 try:
330 xmlui_id, from_jid, confirmed = await self.get_confirmation(
331 action_data, pre_accepted
332 )
333 except exceptions.InternalError:
334 self.host.quit_from_signal(1)
335 return
336 except (exceptions.InternalError, exceptions.DataError):
337 return
338 if confirmed:
339 await self.start_webrtc(
340 from_jid,
341 session_id,
342 action_data.get("screenshare", {}),
343 action_data.get("devices", {}),
344 )
345 xmlui_data = {"answer": C.bool_const(confirmed)}
346 await self.host.bridge.action_launch(
347 xmlui_id, data_format.serialise(xmlui_data), profile_key=profile
348 )
349
350 async def start(self):
351 self.bare_jids = [jid.JID(jid_).bare for jid_ in self.args.jids]
352 from libervia.frontends.tools.webrtc_remote_control import (
353 WebRTCRemoteControlReceiver,
354 )
355
356 self.receiver = WebRTCRemoteControlReceiver(
357 self.host.bridge,
358 self.profile,
359 on_close_cb=self.host.a_quit,
360 verbose=self.verbosity >= 1,
361 )
362 aio.install_glib_asyncio_iteration()
363 # FIXME: for now AUTO always do the screen sharing, but is should be disabled and
364 # appropriate method should be used when no desktop environment is detected.
365 with_screen_sharing = self.args.share_screen in ("yes", "auto")
366 await self.receiver.request_remote_desktop(
367 with_screen_sharing
368 )
369
370 self.disp(_("Waiting for controlling device…"))
371 await self.start_answering()
372
373
374 class RemoteControl(base.CommandBase):
375 subcommands = (Send, Receive)
376
377 def __init__(self, host):
378 super().__init__(
379 host,
380 "remote-control",
381 use_profile=False,
382 help=_("Control or be controlled by another device."),
383 )