view libervia/backend/plugins/plugin_xep_0166/models.py @ 4219:1b5cf2ee1d86

plugin XEP-0384, XEP-0391: download missing devices list: when a peer jid was not in our roster, devices list was not retrieved, resulting in failed en/decryption. This patch does check it and download missing devices list in necessary. There is no subscription managed yet, so the list won't be updated in case of new devices, this should be addressed at some point.
author Goffi <goffi@goffi.org>
date Tue, 05 Mar 2024 17:31:36 +0100
parents 0da563780ffc
children e11b13418ba6
line wrap: on
line source

#!/usr/bin/env python3

# Libervia plugin for Jingle (XEP-0166)
# 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 <http://www.gnu.org/licenses/>.


import abc
from dataclasses import dataclass
from typing import Awaitable, Callable, Union

from twisted.internet import defer
from twisted.words.xish import domish

from libervia.backend.core import exceptions
from libervia.backend.core.core_types import SatXMPPEntity
from libervia.backend.core.i18n import _

class BaseApplicationHandler(abc.ABC):

    @abc.abstractmethod
    async def jingle_preflight(
        self,
        client: SatXMPPEntity,
        session: dict,
        description_elt: domish.Element
    ) -> None:
        """Called when preparation steps are needed by a plugin

        Notably used by XEP-0353 when a initiation message is received
        """
        pass

    @abc.abstractmethod
    async def jingle_preflight_info(
        self,
        client: SatXMPPEntity,
        session: dict,
        info_type: str,
        info_data: dict|None = None
    ) -> None:
        """Called when we have new information during preflight

        Notably used by XEP-0353 to signal ringing
        """
        pass

    @abc.abstractmethod
    async def jingle_preflight_cancel(
        self,
        client: SatXMPPEntity,
        session: dict,
        cancel_error: exceptions.CancelError
    ) -> None:
        """Called when preflight initiation is cancelled

        Notably used by XEP-0353 when an initiation message is cancelled
        """
        pass

    @abc.abstractmethod
    def jingle_request_confirmation(
        self,
        client: SatXMPPEntity,
        action: str,
        session: dict,
        content_name: str,
        desc_elt: domish.Element,
    ) -> Union[
        Callable[..., Union[bool, defer.Deferred]],
        Callable[..., Awaitable[bool]]
    ]:
        """If present, it is called on when session must be accepted.
        If not present, a generic accept dialog will be used.

        @param session: Jingle Session
        @param desc_elt: <description> element
        @return: True if the session is accepted.
            A Deferred can be returned.
        """
        pass

    @abc.abstractmethod
    def jingle_session_init(
        self,
        client: SatXMPPEntity,
        session: dict,
        content_name: str,
        *args, **kwargs
    ) -> Union[
        Callable[..., domish.Element],
        Callable[..., Awaitable[domish.Element]]
    ]:
        """Must return the domish.Element used for initial content.

        @param client: SatXMPPEntity instance
        @param session: Jingle Session
        @param content_name: Name of the content
        @return: The domish.Element used for initial content
        """
        pass

    @abc.abstractmethod
    def jingle_handler(
        self,
        client: SatXMPPEntity,
        action: str,
        session: dict,
        content_name: str,
        transport_elt: domish.Element
    ) -> Union[
        Callable[..., None],
        Callable[..., Awaitable[None]]
    ]:
        """Called on several actions to negotiate the application or transport.

        @param client: SatXMPPEntity instance
        @param action: Jingle action
        @param session: Jingle Session
        @param content_name: Name of the content
        @param transport_elt: Transport element
        """
        pass

    @abc.abstractmethod
    def jingle_terminate(
        self,
        client: SatXMPPEntity,
        action: str,
        session: dict,
        content_name: str,
        reason_elt: domish.Element
    ) -> Union[
        Callable[..., None],
        Callable[..., Awaitable[None]]
    ]:
        """Called on session terminate, with reason_elt.
        May be used to clean session.

        @param reason_elt: Reason element
        """
        pass


class BaseTransportHandler(abc.ABC):

    @abc.abstractmethod
    def jingle_session_init(
        self,
        client: SatXMPPEntity,
        session: dict,
        content_name: str,
        *args, **kwargs
    ) -> Union[
        Callable[..., domish.Element],
        Callable[..., Awaitable[domish.Element]]
    ]:
        """Must return the domish.Element used for initial content.

        @param client: SatXMPPEntity instance
        @param session: Jingle Session
        @param content_name: Name of the content
        @return: The domish.Element used for initial content
        """
        pass

    @abc.abstractmethod
    def jingle_handler(
        self,
        client: SatXMPPEntity,
        action: str,
        session: dict,
        content_name: str,
        reason_elt: domish.Element
    ) -> Union[
        Callable[..., None],
        Callable[..., Awaitable[None]]
    ]:
        """Called on several actions to negotiate the application or transport.

        @param client: SatXMPPEntity instance
        @param action: Jingle action
        @param session: Jingle Session
        @param content_name: Name of the content
        @param reason_elt: <reason> element
        """
        pass


@dataclass(frozen=True)
class ApplicationData:
    namespace: str
    handler: BaseApplicationHandler


@dataclass(frozen=True)
class TransportData:
    namespace: str
    handler: BaseTransportHandler
    priority: int


@dataclass(frozen=True)
class ContentData:
    application: ApplicationData
    app_args: list
    app_kwargs: dict
    transport_data: dict
    content_name: str