Mercurial > libervia-backend
view libervia/backend/plugins/plugin_misc_remote_control.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia: an XMPP client # Copyright (C) 2009-2024 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 <http://www.gnu.org/licenses/>. from twisted.internet import defer from twisted.words.protocols.jabber import jid from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.xish import domish from wokkel import disco, iwokkel from zope.interface import implementer from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import D_, _ from libervia.backend.core.log import getLogger from libervia.backend.tools import xml_tools from libervia.backend.tools.common import data_format from .plugin_xep_0166 import BaseApplicationHandler log = getLogger(__name__) NS_REMOTE_CONTROL = "urn:xmpp:jingle:apps:remote-control:0" PLUGIN_INFO = { C.PI_NAME: "Jingle Remove Control", C.PI_IMPORT_NAME: "RemoteControl", C.PI_TYPE: C.PLUG_TYPE_MISC, C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0166", "XEP-0167"], C.PI_MAIN: "RemoteControl", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""Remote control devices with Jingle."""), } class RemoteControl(BaseApplicationHandler): def __init__(self, host): log.info(f'Plugin "{PLUGIN_INFO[C.PI_NAME]}" initialization') self.host = host # FIXME: to be removed once host is accessible from global var self._j = host.plugins["XEP-0166"] # We need higher priority than XEP-0167 application, as we have a RemoteControl # session and not a call one when this application is used. self._j.register_application(NS_REMOTE_CONTROL, self, priority=1000) self._rtp = host.plugins["XEP-0167"] host.register_namespace("remote-control", NS_REMOTE_CONTROL) host.bridge.add_method( "remote_control_start", ".plugin", in_sign="sss", out_sign="s", method=self._remote_control_start, async_=True, ) def get_handler(self, client): return RemoteControl_handler() # bridge methods def _remote_control_start( self, peer_jid_s: str, extra_s: str, profile: str, ) -> defer.Deferred[str]: client = self.host.get_client(profile) extra = data_format.deserialise(extra_s) d = defer.ensureDeferred( self.remote_control_start( client, jid.JID(peer_jid_s), extra, ) ) d.addCallback(data_format.serialise) return d async def remote_control_start( self, client: SatXMPPEntity, peer_jid: jid.JID, extra: dict ) -> dict: """Start a remote control session. @param peer_jid: destinee jid @return: progress id """ if not extra: raise exceptions.DataError('"extra" must be set.') # webrtc is always used for remote control extra["webrtc"] = True content = { "app_ns": NS_REMOTE_CONTROL, # XXX: for now only unidirectional device exist, but future extensions mays be # bidirectional, and which case "senders" would be set to "both" "senders": self._j.ROLE_INITIATOR, "app_kwargs": { "extra": extra, }, } try: call_data = content["app_kwargs"]["extra"]["call_data"] except KeyError: raise exceptions.DataError('"call_data" must be set in "extra".') metadata = self._rtp.parse_call_data(call_data) try: application_data = call_data["application"] except KeyError: raise exceptions.DataError('"call_data" must have an application media.') try: content["transport_data"] = { "sctp-port": metadata["sctp-port"], "max-message-size": metadata.get("max-message-size", 65536), "local_ice_data": { "ufrag": metadata["ice-ufrag"], "pwd": metadata["ice-pwd"], "candidates": application_data.pop("ice-candidates"), "fingerprint": application_data.pop("fingerprint", {}), }, } name = application_data.get("id") if name: content["name"] = name except KeyError as e: raise exceptions.DataError(f"Mandatory key is missing: {e}") contents = [content] contents.extend(self._rtp.get_contents(call_data, metadata)) session_id = await self._j.initiate( client, peer_jid, contents, call_type=C.META_SUBTYPE_CALL_REMOTE_CONTROL, metadata=metadata, peer_metadata={}, ) return {"session_id": session_id} # jingle callbacks def _get_confirm_msg( self, client: SatXMPPEntity, peer_jid: jid.JID, ) -> tuple[bool, str, str]: """Get confirmation message to display to user. This is the message to show when a remote-control request is received.""" if client.roster and peer_jid.userhostJID() not in client.roster: is_in_roster = False confirm_msg = D_( "Somebody not in your contact list ({peer_jid}) wants to control " "remotely this device. Accepting this will give full control of your " " device to this person, and leak your presence and probably your IP " "address. Do not accept if you don't trust this person!\n" "Do you accept?" ).format(peer_jid=peer_jid) confirm_title = D_("Remote Control Request From an Unknown Contact") else: is_in_roster = True confirm_msg = D_( "{peer_jid} wants to control your device. Accepting will give full " "control of your device, like if they were in front of your computer. " "Only accept if you absolute trust this person.\n" "Do you accept?" ).format(peer_jid=peer_jid) confirm_title = D_("Remote Control Request.") return (is_in_roster, confirm_msg, confirm_title) async def jingle_preflight( self, client: SatXMPPEntity, session: dict, description_elt: domish.Element ) -> None: """Perform preflight checks for an incoming call session. Check if the calls is audio only or audio/video, then, prompts the user for confirmation. @param client: The client instance. @param session: Jingle session. @param description_elt: The description element. It's parent attribute is used to determine check siblings to see if it's an audio only or audio/video call. @raises exceptions.CancelError: If the user doesn't accept the incoming call. """ session_id = session["id"] peer_jid = session["peer_jid"] is_in_roster, confirm_msg, confirm_title = self._get_confirm_msg(client, peer_jid) if is_in_roster: action_type = C.META_TYPE_CONFIRM else: action_type = C.META_TYPE_NOT_IN_ROSTER_LEAK action_extra = { "type": action_type, "session_id": session_id, "from_jid": peer_jid.full(), } action_extra["subtype"] = C.META_TYPE_REMOTE_CONTROL accepted = await xml_tools.defer_confirm( self.host, confirm_msg, confirm_title, profile=client.profile, action_extra=action_extra, ) if accepted: session["pre_accepted"] = True return accepted async def jingle_preflight_info( self, client: SatXMPPEntity, session: dict, info_type: str, info_data: dict | None = None, ) -> None: pass async def jingle_preflight_cancel( self, client: SatXMPPEntity, session: dict, cancel_error: exceptions.CancelError ) -> None: """The remote control has been rejected""" def jingle_session_init( self, client: SatXMPPEntity, session: dict, content_name: str, extra: dict ) -> domish.Element: """Initializes a jingle session. @param client: The client instance. @param session: Jingle session. @param content_name: Name of the content. @param extra: Extra data. @return: <description> element. """ desc_elt = domish.Element((NS_REMOTE_CONTROL, "description")) devices = extra.get("devices") or {} for name, data in devices.items(): device_elt = desc_elt.addElement((NS_REMOTE_CONTROL, "device")) device_elt["type"] = name return desc_elt async def jingle_request_confirmation( self, client: SatXMPPEntity, action: str, session: dict, content_name: str, desc_elt: domish.Element, ) -> bool: """Requests confirmation from the user for a Jingle session's incoming call. This method checks the content type of the Jingle session (audio or video) based on the session's contents. Confirmation is requested only for the first content; subsequent contents are automatically accepted. This means, in practice, that the call confirmation is prompted only once for both audio and video contents. @param client: The client instance. @param action: The action type associated with the Jingle session. @param session: Jingle session. @param content_name: Name of the content being checked. @param desc_elt: The description element associated with the content. @return: True if the call is accepted by the user, False otherwise. """ content_data = session["contents"][content_name] role = session["role"] if role == self._j.ROLE_INITIATOR: raise NotImplementedError return True elif role == self._j.ROLE_RESPONDER: # We are the controlled entity. return await self._remote_control_request_conf( client, session, content_data, content_name ) else: raise exceptions.InternalError(f"Invalid role {role!r}") async def _remote_control_request_conf( self, client: SatXMPPEntity, session: dict, content_data: dict, content_name: str, ) -> bool: """Handle user permission.""" peer_jid = session["peer_jid"] pre_accepted = session.get("pre_accepted", False) __, confirm_msg, confirm_title = self._get_confirm_msg(client, peer_jid) contents = session["contents"] action_extra = { "pre_accepted": pre_accepted, "type": C.META_TYPE_REMOTE_CONTROL, "devices": content_data["application_data"]["devices"], "session_id": session["id"], "from_jid": peer_jid.full(), } for name, content in contents.items(): if name == content_name: continue if content["application"].namespace == self._rtp.namespace: media = content["application_data"]["media"] action_extra.setdefault("screenshare", {})[media] = {} return await xml_tools.defer_confirm( self.host, confirm_msg, confirm_title, profile=client.profile, action_extra=action_extra, ) async def jingle_handler(self, client, action, session, content_name, desc_elt): content_data = session["contents"][content_name] application_data = content_data["application_data"] if action == self._j.A_PREPARE_CONFIRMATION: devices = application_data["devices"] = {} for device_elt in desc_elt.elements(NS_REMOTE_CONTROL, "device"): try: device_type = device_elt.attributes["type"] except KeyError: log.warning(f"Invalide device element: {device_elt.toXml()}") else: # The dict holds data for current type of devices. For now it is unused # has the spec doesn't define any device data, but it may be used by # future extensions. devices[device_type] = {} elif action == self._j.A_SESSION_INITIATE: # FIXME: for now we automatically accept keyboard, mouse and wheel and nothing # else. Must actually reflect user choices and local devices. to_remove = [] for device_elt in desc_elt.elements(NS_REMOTE_CONTROL, "device"): if device_elt.getAttribute("type") not in ("keyboard", "wheel", "mouse"): to_remove.append(device_elt) for elt in to_remove: elt.parent.children.remove(elt) return desc_elt def jingle_terminate( self, client: SatXMPPEntity, action: str, session: dict, content_name: str, reason_elt: domish.Element, ) -> None: reason, text = self._j.parse_reason_elt(reason_elt) data = {"reason": reason} if text: data["text"] = text self.host.bridge.call_ended( session["id"], data_format.serialise(data), client.profile ) @implementer(iwokkel.IDisco) class RemoteControl_handler(XMPPHandler): def getDiscoInfo(self, requestor, target, nodeIdentifier=""): return [disco.DiscoFeature(NS_REMOTE_CONTROL)] def getDiscoItems(self, requestor, target, nodeIdentifier=""): return []