libervia-backend/libervia/backend/plugins/plugin_misc_remote_control.py

#!/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 tx_xmpp import disco, itx_xmpp
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(itx_xmpp.IDisco)
class RemoteControl_handler(XMPPHandler):
    def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
        return [disco.DiscoFeature(NS_REMOTE_CONTROL)]

    def getDiscoItems(self, requestor, target, nodeIdentifier=""):
        return []