plugin XEP-0373 and XEP-0374: Implementation of OX and OXIM:

GPGME is used as the GPG provider.

rel 374

sat/plugins/plugin_xep_0060.py

@@ -893,14 +893,20 @@ class XEP_0060(object):
             client, jid.JID(service_s) if service_s else None, nodeIdentifier, options
         )
 
-    def createNode(self, client, service, nodeIdentifier=None, options=None):
+    def createNode(
+        self,
+        client: SatXMPPClient,
+        service: jid.JID,
+        nodeIdentifier: Optional[str] = None,
+        options: Optional[Dict[str, str]] = None
+    ) -> str:
         """Create a new node
 
-        @param service(jid.JID): PubSub service,
-        @param NodeIdentifier(unicode, None): node name
-           use None to create instant node (identifier will be returned by this method)
-        @param option(dict[unicode, unicode], None): node configuration options
-        @return (unicode): identifier of the created node (may be different from requested name)
+        @param service: PubSub service,
+        @param NodeIdentifier: node name use None to create instant node (identifier will
+            be returned by this method)
+        @param option: node configuration options
+        @return: identifier of the created node (may be different from requested name)
         """
         # TODO: if pubsub service doesn't hande publish-options, configure it in a second time
         return client.pubsub_client.createNode(service, nodeIdentifier, options)

sat/plugins/plugin_xep_0374.py

@@ -0,0 +1,421 @@
+#!/usr/bin/env python3
+
+# Libervia plugin for OpenPGP for XMPP Instant Messaging
+# Copyright (C) 2022-2022 Tim Henkes (me@syndace.dev)
+
+# 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 typing import Dict, Optional, Set, cast
+
+from typing_extensions import Final
+from wokkel import muc  # type: ignore[import]
+
+from sat.core import exceptions
+from sat.core.constants import Const as C
+from sat.core.core_types import SatXMPPEntity
+from sat.core.i18n import _, D_
+from sat.core.log import getLogger, Logger
+from sat.core.sat_main import SAT
+from sat.core.xmpp import SatXMPPClient
+from sat.plugins.plugin_xep_0045 import XEP_0045
+from sat.plugins.plugin_xep_0334 import XEP_0334
+from sat.plugins.plugin_xep_0373 import NS_OX, XEP_0373, TrustLevel
+from sat.tools import xml_tools
+from twisted.internet import defer
+from twisted.words.protocols.jabber import jid
+from twisted.words.xish import domish
+
+
+__all__ = [  # pylint: disable=unused-variable
+    "PLUGIN_INFO",
+    "XEP_0374",
+    "NS_OXIM"
+]
+
+
+log = cast(Logger, getLogger(__name__))  # type: ignore[no-untyped-call]
+
+
+PLUGIN_INFO = {
+    C.PI_NAME: "OXIM",
+    C.PI_IMPORT_NAME: "XEP-0374",
+    C.PI_TYPE: "SEC",
+    C.PI_PROTOCOLS: [ "XEP-0374" ],
+    C.PI_DEPENDENCIES: [ "XEP-0334", "XEP-0373" ],
+    C.PI_RECOMMENDATIONS: [ "XEP-0045" ],
+    C.PI_MAIN: "XEP_0374",
+    C.PI_HANDLER: "no",
+    C.PI_DESCRIPTION: _("""Implementation of OXIM"""),
+}
+
+
+# The disco feature
+NS_OXIM: Final = "urn:xmpp:openpgp:im:0"
+
+
+class XEP_0374:
+    """
+    Plugin equipping Libervia with OXIM capabilities under the ``urn:xmpp:openpgp:im:0``
+    namespace. MUC messages are supported next to one to one messages. For trust
+    management, the two trust models "BTBV" and "manual" are supported.
+    """
+
+    def __init__(self, sat: SAT) -> None:
+        """
+        @param sat: The SAT instance.
+        """
+
+        self.__sat = sat
+
+        # Plugins
+        self.__xep_0045 = cast(Optional[XEP_0045], sat.plugins.get("XEP-0045"))
+        self.__xep_0334 = cast(XEP_0334, sat.plugins["XEP-0334"])
+        self.__xep_0373 = cast(XEP_0373, sat.plugins["XEP-0373"])
+
+        # Triggers
+        sat.trigger.add(
+            "messageReceived",
+            self.__message_received_trigger,
+            priority=100050
+        )
+        sat.trigger.add("send", self.__send_trigger, priority=0)
+
+        # Register the encryption plugin
+        sat.registerEncryptionPlugin(self, "OXIM", NS_OX, 102)
+
+    async def getTrustUI(  # pylint: disable=invalid-name
+        self,
+        client: SatXMPPClient,
+        entity: jid.JID
+    ) -> xml_tools.XMLUI:
+        """
+        @param client: The client.
+        @param entity: The entity whose device trust levels to manage.
+        @return: An XMLUI instance which opens a form to manage the trust level of all
+            devices belonging to the entity.
+        """
+
+        return await self.__xep_0373.getTrustUI(client, entity)
+
+    @staticmethod
+    def __get_joined_muc_users(
+        client: SatXMPPClient,
+        xep_0045: XEP_0045,
+        room_jid: jid.JID
+    ) -> Set[jid.JID]:
+        """
+        @param client: The client.
+        @param xep_0045: A MUC plugin instance.
+        @param room_jid: The room JID.
+        @return: A set containing the bare JIDs of the MUC participants.
+        @raise InternalError: if the MUC is not joined or the entity information of a
+            participant isn't available.
+        """
+
+        bare_jids: Set[jid.JID] = set()
+
+        try:
+            room = cast(muc.Room, xep_0045.getRoom(client, room_jid))
+        except exceptions.NotFound as e:
+            raise exceptions.InternalError(
+                "Participant list of unjoined MUC requested."
+            ) from e
+
+        for user in cast(Dict[str, muc.User], room.roster).values():
+            entity = cast(Optional[SatXMPPEntity], user.entity)
+            if entity is None:
+                raise exceptions.InternalError(
+                    f"Participant list of MUC requested, but the entity information of"
+                    f" the participant {user} is not available."
+                )
+
+            bare_jids.add(entity.jid.userhostJID())
+
+        return bare_jids
+
+    async def __message_received_trigger(
+        self,
+        client: SatXMPPClient,
+        message_elt: domish.Element,
+        post_treat: defer.Deferred
+    ) -> bool:
+        """
+        @param client: The client which received the message.
+        @param message_elt: The message element. Can be modified.
+        @param post_treat: A deferred which evaluates to a :class:`MessageData` once the
+            message has fully progressed through the message receiving flow. Can be used
+            to apply treatments to the fully processed message, like marking it as
+            encrypted.
+        @return: Whether to continue the message received flow.
+        """
+        sender_jid = jid.JID(message_elt["from"])
+        feedback_jid: jid.JID
+
+        message_type = message_elt.getAttribute("type", "unknown")
+        is_muc_message = message_type == C.MESS_TYPE_GROUPCHAT
+        if is_muc_message:
+            if self.__xep_0045 is None:
+                log.warning(
+                    "Ignoring MUC message since plugin XEP-0045 is not available."
+                )
+                # Can't handle a MUC message without XEP-0045, let the flow continue
+                # normally
+                return True
+
+            room_jid = feedback_jid = sender_jid.userhostJID()
+
+            try:
+                room = cast(muc.Room, self.__xep_0045.getRoom(client, room_jid))
+            except exceptions.NotFound:
+                log.warning(
+                    f"Ignoring MUC message from a room that has not been joined:"
+                    f" {room_jid}"
+                )
+                # Whatever, let the flow continue
+                return True
+
+            sender_user = cast(Optional[muc.User], room.getUser(sender_jid.resource))
+            if sender_user is None:
+                log.warning(
+                    f"Ignoring MUC message from room {room_jid} since the sender's user"
+                    f" wasn't found {sender_jid.resource}"
+                )
+                # Whatever, let the flow continue
+                return True
+
+            sender_user_jid = cast(Optional[jid.JID], sender_user.entity)
+            if sender_user_jid is None:
+                log.warning(
+                    f"Ignoring MUC message from room {room_jid} since the sender's bare"
+                    f" JID couldn't be found from its user information: {sender_user}"
+                )
+                # Whatever, let the flow continue
+                return True
+
+            sender_jid = sender_user_jid
+        else:
+            # I'm not sure why this check is required, this code is copied from XEP-0384
+            if sender_jid.userhostJID() == client.jid.userhostJID():
+                # TODO: I've seen this cause an exception "builtins.KeyError: 'to'", seems
+                # like "to" isn't always set.
+                feedback_jid = jid.JID(message_elt["to"])
+            else:
+                feedback_jid = sender_jid
+
+        sender_bare_jid = sender_jid.userhost()
+
+        openpgp_elt = cast(Optional[domish.Element], next(
+            message_elt.elements(NS_OX, "openpgp"),
+            None
+        ))
+
+        if openpgp_elt is None:
+            # None of our business, let the flow continue
+            return True
+
+        try:
+            payload_elt, timestamp = await self.__xep_0373.unpack_openpgp_element(
+                client,
+                openpgp_elt,
+                "signcrypt",
+                jid.JID(sender_bare_jid)
+            )
+        except Exception as e:
+            # TODO: More specific exception handling
+            log.warning(_("Can't decrypt message: {reason}\n{xml}").format(
+                reason=e,
+                xml=message_elt.toXml()
+            ))
+            client.feedback(
+                feedback_jid,
+                D_(
+                    f"An OXIM message from {sender_jid.full()} can't be decrypted:"
+                    f" {e}"
+                ),
+                { C.MESS_EXTRA_INFO: C.EXTRA_INFO_DECR_ERR }
+            )
+            # No point in further processing this message
+            return False
+
+        message_elt.children.remove(openpgp_elt)
+
+        log.debug(f"OXIM message of type {message_type} received from {sender_bare_jid}")
+
+        # Remove all body elements from the original element, since those act as
+        # fallbacks in case the encryption protocol is not supported
+        for child in message_elt.elements():
+            if child.name == "body":
+                message_elt.children.remove(child)
+
+        # Move all extension elements from the payload to the stanza root
+        # TODO: There should probably be explicitly forbidden elements here too, just as
+        # for XEP-0420
+        for child in list(payload_elt.elements()):
+            # Remove the child from the content element
+            payload_elt.children.remove(child)
+
+            # Add the child to the stanza
+            message_elt.addChild(child)
+
+        # Mark the message as trusted or untrusted. Undecided counts as untrusted here.
+        trust_level = TrustLevel.UNDECIDED  # TODO: Load the actual trust level
+        if trust_level is TrustLevel.TRUSTED:
+            post_treat.addCallback(client.encryption.markAsTrusted)
+        else:
+            post_treat.addCallback(client.encryption.markAsUntrusted)
+
+        # Mark the message as originally encrypted
+        post_treat.addCallback(
+            client.encryption.markAsEncrypted,
+            namespace=NS_OX
+        )
+
+        # Message processed successfully, continue with the flow
+        return True
+
+    async def __send_trigger(self, client: SatXMPPClient, stanza: domish.Element) -> bool:
+        """
+        @param client: The client sending this message.
+        @param stanza: The stanza that is about to be sent. Can be modified.
+        @return: Whether the send message flow should continue or not.
+        """
+        # OXIM only handles message stanzas
+        if stanza.name != "message":
+            return True
+
+        # Get the intended recipient
+        recipient = stanza.getAttribute("to", None)
+        if recipient is None:
+            raise exceptions.InternalError(
+                f"Message without recipient encountered. Blocking further processing to"
+                f" avoid leaking plaintext data: {stanza.toXml()}"
+            )
+
+        # Parse the JID
+        recipient_bare_jid = jid.JID(recipient).userhostJID()
+
+        # Check whether encryption with OXIM is requested
+        encryption = client.encryption.getSession(recipient_bare_jid)
+
+        if encryption is None:
+            # Encryption is not requested for this recipient
+            return True
+
+        if encryption["plugin"].namespace != NS_OX:
+            # Encryption is requested for this recipient, but not with OXIM
+            return True
+
+        # All pre-checks done, we can start encrypting!
+        await self.__encrypt(
+            client,
+            stanza,
+            recipient_bare_jid,
+            stanza.getAttribute("type", "unkown") == C.MESS_TYPE_GROUPCHAT
+        )
+
+        # Add a store hint if this is a message stanza
+        self.__xep_0334.addHintElements(stanza, [ "store" ])
+
+        # Let the flow continue.
+        return True
+
+    async def __encrypt(
+        self,
+        client: SatXMPPClient,
+        stanza: domish.Element,
+        recipient_jid: jid.JID,
+        is_muc_message: bool
+    ) -> None:
+        """
+        @param client: The client.
+        @param stanza: The stanza, which is modified by this call.
+        @param recipient_jid: The JID of the recipient. Can be a bare (aka "userhost") JID
+            but doesn't have to.
+        @param is_muc_message: Whether the stanza is a message stanza to a MUC room.
+
+        @warning: The calling code MUST take care of adding the store message processing
+            hint to the stanza if applicable! This can be done before or after this call,
+            the order doesn't matter.
+        """
+
+        recipient_bare_jids: Set[jid.JID]
+        feedback_jid: jid.JID
+
+        if is_muc_message:
+            if self.__xep_0045 is None:
+                raise exceptions.InternalError(
+                    "Encryption of MUC message requested, but plugin XEP-0045 is not"
+                    " available."
+                )
+
+            room_jid = feedback_jid = recipient_jid.userhostJID()
+
+            recipient_bare_jids = self.__get_joined_muc_users(
+                client,
+                self.__xep_0045,
+                room_jid
+            )
+        else:
+            recipient_bare_jids = { recipient_jid.userhostJID() }
+            feedback_jid = recipient_jid.userhostJID()
+
+        log.debug(
+            f"Intercepting message that is to be encrypted by {NS_OX} for"
+            f" {recipient_bare_jids}"
+        )
+
+        signcrypt_elt, payload_elt = \
+            self.__xep_0373.build_signcrypt_element(recipient_bare_jids)
+
+        # Move elements from the stanza to the content element.
+        # TODO: There should probably be explicitly forbidden elements here too, just as
+        # for XEP-0420
+        for child in list(stanza.elements()):
+            # Remove the child from the stanza
+            stanza.children.remove(child)
+
+            # A namespace of ``None`` can be used on domish elements to inherit the
+            # namespace from the parent. When moving elements from the stanza root to
+            # the content element, however, we don't want elements to inherit the
+            # namespace of the content element. Thus, check for elements with ``None``
+            # for their namespace and set the namespace to jabber:client, which is the
+            # namespace of the parent element.
+            if child.uri is None:
+                child.uri = C.NS_CLIENT
+                child.defaultUri = C.NS_CLIENT
+
+            # Add the child with corrected namespaces to the content element
+            payload_elt.addChild(child)
+
+        try:
+            openpgp_elt = await self.__xep_0373.build_openpgp_element(
+                client,
+                signcrypt_elt,
+                recipient_bare_jids
+            )
+        except Exception as e:
+            msg = _(
+                # pylint: disable=consider-using-f-string
+                "Can't encrypt message for {entities}: {reason}".format(
+                    entities=', '.join(jid.userhost() for jid in recipient_bare_jids),
+                    reason=e
+                )
+            )
+            log.warning(msg)
+            client.feedback(feedback_jid, msg, {
+                C.MESS_EXTRA_INFO: C.EXTRA_INFO_ENCR_ERR
+            })
+            raise e
+
+        stanza.addChild(openpgp_elt)

sat/plugins/plugin_xep_0384.py

@@ -479,10 +479,10 @@ def make_session_manager(sat: SAT, profile: str) -> Type[omemo.SessionManager]:
                         xml_tools.et_elt_2_domish_elt(element),
                         item_id=str(bundle.device_id),
                         extra={
-                            xep_0060.EXTRA_PUBLISH_OPTIONS: {
-                                xep_0060.OPT_MAX_ITEMS: "max"
+                            XEP_0060.EXTRA_PUBLISH_OPTIONS: {
+                                XEP_0060.OPT_MAX_ITEMS: "max"
                             },
-                            xep_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
+                            XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
                         }
                     )
                 except (error.StanzaError, Exception) as e:
@@ -519,8 +519,8 @@ def make_session_manager(sat: SAT, profile: str) -> Type[omemo.SessionManager]:
                         xml_tools.et_elt_2_domish_elt(element),
                         item_id=xep_0060.ID_SINGLETON,
                         extra={
-                            xep_0060.EXTRA_PUBLISH_OPTIONS: { xep_0060.OPT_MAX_ITEMS: 1 },
-                            xep_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
+                            XEP_0060.EXTRA_PUBLISH_OPTIONS: { XEP_0060.OPT_MAX_ITEMS: 1 },
+                            XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "publish_without_options"
                         }
                     )
                 except Exception as e:
@@ -546,7 +546,6 @@ def make_session_manager(sat: SAT, profile: str) -> Type[omemo.SessionManager]:
                         client,
                         jid.JID(bare_jid),
                         node,
-                        max_items=None,
                         item_ids=[ str(device_id) ]
                     )
                 except Exception as e:
@@ -653,11 +652,11 @@ def make_session_manager(sat: SAT, profile: str) -> Type[omemo.SessionManager]:
                     xml_tools.et_elt_2_domish_elt(element),
                     item_id=xep_0060.ID_SINGLETON,
                     extra={
-                        xep_0060.EXTRA_PUBLISH_OPTIONS: {
-                            xep_0060.OPT_MAX_ITEMS: 1,
-                            xep_0060.OPT_ACCESS_MODEL: "open"
+                        XEP_0060.EXTRA_PUBLISH_OPTIONS: {
+                            XEP_0060.OPT_MAX_ITEMS: 1,
+                            XEP_0060.OPT_ACCESS_MODEL: "open"
                         },
-                        xep_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
+                        XEP_0060.EXTRA_ON_PRECOND_NOT_MET: "raise"
                     }
                 )
             except (error.StanzaError, Exception) as e:

sat/plugins/plugin_xep_0420.py

@@ -22,8 +22,10 @@ import enum
 import secrets
 import string
 from typing import Dict, NamedTuple, Optional, Set, Tuple, cast
+from typing_extensions import Final
 
 from lxml import etree
+from sat.core import exceptions
 
 from sat.core.constants import Const as C
 from sat.core.i18n import D_
@@ -68,7 +70,7 @@ PLUGIN_INFO = {
 }
 
 
-NS_SCE = "urn:xmpp:sce:1"
+NS_SCE: Final = "urn:xmpp:sce:1"
 
 
 class ProfileRequirementsNotMet(Exception):
@@ -114,7 +116,8 @@ class SCECustomAffix(ABC):
             remain. Do not modify.
         @return: An affix element to include in the envelope. The element must have the
             name :attr:`element_name` and must validate using :attr:`element_schema`.
-        @raise ValueError: if the affix couldn't be built.
+        @raise ValueError: if the affix couldn't be built due to missing information on
+            the stanza.
         """
 
     @abstractmethod
@@ -384,7 +387,7 @@ class XEP_0420:  # pylint: disable=invalid-name
             by the decryption scheme utilizing SCE.
         @return: The parsed and processed values of all affixes that were present on the
             envelope, notably including the timestamp.
-        @raise ValueError: if the serialized envelope element is malformed.
+        @raise exceptions.ParsingError: if the serialized envelope element is malformed.
         @raise ProfileRequirementsNotMet: if one or more affixes required by the profile
             are missing from the envelope.
         @raise AffixVerificationFailed: if an affix included in the envelope fails to
@@ -399,7 +402,9 @@ class XEP_0420:  # pylint: disable=invalid-name
         try:
             envelope_serialized_string = envelope_serialized.decode("utf-8")
         except UnicodeError as e:
-            raise ValueError("Serialized envelope can't bare parsed as utf-8.") from e
+            raise exceptions.ParsingError(
+                "Serialized envelope can't bare parsed as utf-8."
+            ) from e
 
         custom_affixes = set(profile.custom_policies.keys())
 
@@ -420,7 +425,9 @@ class XEP_0420:  # pylint: disable=invalid-name
         try:
             etree.fromstring(envelope_serialized_string, parser)
         except etree.XMLSyntaxError as e:
-            raise ValueError("Serialized envelope doesn't pass schema validation.") from e
+            raise exceptions.ParsingError(
+                "Serialized envelope doesn't pass schema validation."
+            ) from e
 
         # Prepare the envelope and content elements
         envelope = cast(domish.Element, ElementParser()(envelope_serialized_string))
@@ -452,7 +459,7 @@ class XEP_0420:  # pylint: disable=invalid-name
             timestamp_value = None if time_element is None else \
                 XEP_0082.parse_datetime(time_element["stamp"])
         except ValueError as e:
-            raise AffixVerificationFailed("Malformed time affix") from e
+            raise AffixVerificationFailed("Malformed time affix.") from e
 
         # The to affix is verified by comparing the to attribute of the stanza with the
         # JID referenced by the affix. Note that only bare JIDs are compared as per the

sat/tools/xmpp_datetime.py

@@ -80,12 +80,15 @@ def parse_date(value: str) -> date:
     @param value: A string containing date information formatted according to the Date
         profile specified in XEP-0082.
     @return: The date parsed from the input string.
-    @raise ValueError: if the input string is not correctly formatted.
+    @raise exceptions.ParsingError: if the input string is not correctly formatted.
     """
     # CCYY-MM-DD
 
     # The Date profile of XEP-0082 is equal to the ISO 8601 format.
-    return date.fromisoformat(value)
+    try:
+        return date.fromisoformat(value)
+    except ValueError as e:
+        raise exceptions.ParsingError() from e
 
 
 def format_datetime(
@@ -125,13 +128,16 @@ def parse_datetime(value: str) -> datetime:
     @param value: A string containing datetime information formatted according to the
         DateTime profile specified in XEP-0082.
     @return: The datetime parsed from the input string.
-    @raise ValueError: if the input string is not correctly formatted.
+    @raise exceptions.ParsingError: if the input string is not correctly formatted.
     """
     # CCYY-MM-DDThh:mm:ss[.sss]TZD
 
     value, microsecond = __parse_fraction_of_a_second(value)
 
-    result = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z")
+    try:
+        result = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S%z")
+    except ValueError as e:
+        raise exceptions.ParsingError() from e
 
     if microsecond is not None:
         result = result.replace(microsecond=microsecond)
@@ -167,7 +173,7 @@ def parse_time(value: str) -> time:
     @param value: A string containing time information formatted according to the Time
         profile specified in XEP-0082.
     @return: The time parsed from the input string.
-    @raise ValueError: if the input string is not correctly formatted.
+    @raise exceptions.ParsingError: if the input string is not correctly formatted.
     """
     # hh:mm:ss[.sss][TZD]
 
@@ -177,7 +183,10 @@ def parse_time(value: str) -> time:
     # profile, except that it doesn't handle the letter Z as time zone information for
     # UTC. This can be fixed with a simple string replacement of 'Z' with "+00:00", which
     # is another way to represent UTC.
-    result = time.fromisoformat(value.replace('Z', "+00:00"))
+    try:
+        result = time.fromisoformat(value.replace('Z', "+00:00"))
+    except ValueError as e:
+        raise exceptions.ParsingError() from e
 
     if microsecond is not None:
         result = result.replace(microsecond=microsecond)

setup.py

@@ -52,13 +52,14 @@ install_requires = [
     'urwid-satext == 0.9.*',
     'wokkel >= 18.0.0, < 19.0.0',
     'omemo >= 1.0.0, < 2',
-    'twomemo >= 1.0.0, < 2',
-    'oldmemo >= 1.0.0, < 2',
+    'twomemo[xml] >= 1.0.0, < 2',
+    'oldmemo[xml] >= 1.0.0, < 2',
     'pyyaml < 7.0.0',
     'sqlalchemy >= 1.4',
     'alembic',
     'aiosqlite',
     'txdbus',
+    'xmlschema',
 ]
 
 extras_require = {

tests/unit/test_plugin_xep_0082.py

@@ -19,6 +19,7 @@
 from datetime import date, datetime, time, timezone
 
 import pytest
+from sat.core import exceptions
 
 from sat.plugins.plugin_xep_0082 import XEP_0082
 
@@ -104,7 +105,7 @@ def test_datetime_parsing() -> None:  # pylint: disable=missing-function-docstri
     assert XEP_0082.parse_datetime("1969-07-20T21:56:15-05:00") == value
 
     # Without timezone, without a fraction of a second
-    with pytest.raises(ValueError):
+    with pytest.raises(exceptions.ParsingError):
         XEP_0082.parse_datetime("1969-07-21T02:56:15")
 
     # With timezone 'Z', with a fraction of a second consisting of two digits
@@ -123,7 +124,7 @@ def test_datetime_parsing() -> None:  # pylint: disable=missing-function-docstri
     assert XEP_0082.parse_datetime("1969-07-20T21:56:15.-05:00") == value
 
     # Without timezone, with a fraction of a second consisting of six digits
-    with pytest.raises(ValueError):
+    with pytest.raises(exceptions.ParsingError):
         XEP_0082.parse_datetime("1969-07-21T02:56:15.123456")
 
 

tests/unit/test_plugin_xep_0373.py

@@ -0,0 +1,146 @@
+from datetime import datetime, timedelta, timezone
+from sat.plugins.plugin_xep_0373 import XEP_0373, NS_OX
+from sat.tools.xmpp_datetime import parse_datetime
+
+import pytest
+from twisted.words.protocols.jabber import jid
+
+
+a = jid.JID("foo@example.com")
+b = jid.JID("bar@example.com")
+
+
+def test_signcrypt_element_args() -> None:
+    with pytest.raises(ValueError):
+        XEP_0373.build_signcrypt_element([])
+
+
+def test_signcrypt_element() -> None:
+    signcrypt_elt, payload_elt = XEP_0373.build_signcrypt_element([ a, b ])
+    payload_elt.addElement("signcrypt-test-content", content="signcrypt test content")
+
+    rpad_elt = next(signcrypt_elt.elements(NS_OX, "rpad"))
+    time_elt = next(signcrypt_elt.elements(NS_OX, "time"))
+
+    rpad = str(rpad_elt)
+    timestamp = parse_datetime(time_elt["stamp"])
+
+    signcrypt_elt.children.remove(rpad_elt)
+    signcrypt_elt.children.remove(time_elt)
+
+    assert rpad
+    assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
+    assert signcrypt_elt.toXml() == (
+        "<signcrypt xmlns='urn:xmpp:openpgp:0'>"
+            "<to jid='foo@example.com'/>"
+            "<to jid='bar@example.com'/>"
+            "<payload>"
+                "<signcrypt-test-content>signcrypt test content</signcrypt-test-content>"
+            "</payload>"
+        "</signcrypt>"
+    )
+
+
+def test_sign_element_args() -> None:
+    with pytest.raises(ValueError):
+        XEP_0373.build_sign_element([], True)
+
+
+def test_sign_element_with_rpad() -> None:
+    sign_elt, payload_elt = XEP_0373.build_sign_element([ a, b ], True)
+    payload_elt.addElement("sign-test-content", content="sign test content")
+
+    rpad_elt = next(sign_elt.elements(NS_OX, "rpad"))
+    time_elt = next(sign_elt.elements(NS_OX, "time"))
+
+    rpad = str(rpad_elt)
+    timestamp = parse_datetime(time_elt["stamp"])
+
+    sign_elt.children.remove(rpad_elt)
+    sign_elt.children.remove(time_elt)
+
+    assert rpad
+    assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
+    assert sign_elt.toXml() == (
+        "<sign xmlns='urn:xmpp:openpgp:0'>"
+            "<to jid='foo@example.com'/>"
+            "<to jid='bar@example.com'/>"
+            "<payload>"
+                "<sign-test-content>sign test content</sign-test-content>"
+            "</payload>"
+        "</sign>"
+    )
+
+
+def test_sign_element_without_rpad() -> None:
+    sign_elt, payload_elt = XEP_0373.build_sign_element([ a, b ], False)
+    payload_elt.addElement("sign-test-content", content="sign test content")
+
+    rpad_elt = next(sign_elt.elements(NS_OX, "rpad"), None)
+    time_elt = next(sign_elt.elements(NS_OX, "time"))
+
+    timestamp = parse_datetime(time_elt["stamp"])
+
+    sign_elt.children.remove(time_elt)
+
+    assert rpad_elt is None
+    assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
+    assert sign_elt.toXml() == (
+        "<sign xmlns='urn:xmpp:openpgp:0'>"
+            "<to jid='foo@example.com'/>"
+            "<to jid='bar@example.com'/>"
+            "<payload>"
+                "<sign-test-content>sign test content</sign-test-content>"
+            "</payload>"
+        "</sign>"
+    )
+
+
+def test_crypt_element_with_recipients() -> None:
+    crypt_elt, payload_elt = XEP_0373.build_crypt_element([ a, b ])
+    payload_elt.addElement("crypt-test-content", content="crypt test content")
+
+    rpad_elt = next(crypt_elt.elements(NS_OX, "rpad"))
+    time_elt = next(crypt_elt.elements(NS_OX, "time"))
+
+    rpad = str(rpad_elt)
+    timestamp = parse_datetime(time_elt["stamp"])
+
+    crypt_elt.children.remove(rpad_elt)
+    crypt_elt.children.remove(time_elt)
+
+    assert rpad
+    assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
+    assert crypt_elt.toXml() == (
+        "<crypt xmlns='urn:xmpp:openpgp:0'>"
+            "<to jid='foo@example.com'/>"
+            "<to jid='bar@example.com'/>"
+            "<payload>"
+                "<crypt-test-content>crypt test content</crypt-test-content>"
+            "</payload>"
+        "</crypt>"
+    )
+
+
+def test_crypt_element_without_recipients() -> None:
+    crypt_elt, payload_elt = XEP_0373.build_crypt_element([])
+    payload_elt.addElement("crypt-test-content", content="crypt test content")
+
+    rpad_elt = next(crypt_elt.elements(NS_OX, "rpad"))
+    time_elt = next(crypt_elt.elements(NS_OX, "time"))
+
+    rpad = str(rpad_elt)
+    timestamp = parse_datetime(time_elt["stamp"])
+
+    crypt_elt.children.remove(rpad_elt)
+    crypt_elt.children.remove(time_elt)
+
+    assert rpad
+    assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
+    assert crypt_elt.toXml() == (
+        "<crypt xmlns='urn:xmpp:openpgp:0'>"
+            "<payload>"
+                "<crypt-test-content>crypt test content</crypt-test-content>"
+            "</payload>"
+        "</crypt>"
+    )

tests/unit/test_plugin_xep_0420.py

@@ -20,6 +20,7 @@ from datetime import datetime, timezone
 from typing import Callable, cast
 
 import pytest
+from sat.core import exceptions
 
 from sat.plugins.plugin_xep_0334 import NS_HINTS
 from sat.plugins.plugin_xep_0420 import (
@@ -435,7 +436,7 @@ def test_custom_affixes() -> None:  # pylint: disable=missing-function-docstring
         </body>
     </message>""")
 
-    with pytest.raises(ValueError):
+    with pytest.raises(exceptions.ParsingError):
         XEP_0420.unpack_stanza(
             unpacking_profile,
             stanza,
@@ -558,5 +559,5 @@ def test_schema_validation() -> None:  # pylint: disable=missing-function-docstr
         <unknown-affix unknown-attr="unknown"/>
     </envelope>""".encode("utf-8")
 
-    with pytest.raises(ValueError):
+    with pytest.raises(exceptions.ParsingError):
         XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)