erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin XEP-0384: Fully reworked to adjust to the reworked python-omemo:

Browse source 
- support for both (modern) OMEMO under the `urn:xmpp:omemo:2` namespace and (legacy) OMEMO under the `eu.siacs.conversations.axolotl` namespace
- maintains one identity across both versions of OMEMO
- migrates data from the old plugin
- includes more features for protocol stability
- uses SCE for modern OMEMO
- fully type-checked, linted and format-checked
- added type hints to various pieces of backend code used by the plugin
- added stubs for some Twisted APIs used by the plugin under stubs/ (use `export MYPYPATH=stubs/` before running mypy)
- core (xmpp): enabled `send` trigger and made it an asyncPoint

fix 375
This commit is contained in:
Syndace 2022-08-23 21:06:24 +02:00
commit 75ae8160ae
7 changed files with 2202 additions and 1414 deletions
Download patch file Download diff file Expand all files Collapse all files

9
sat/plugins/plugin_xep_0045.py
View file

@ -26,6 +26,7 @@ from twisted.words.protocols.jabber import error as xmpp_error
from twisted.python import failure
from sat.core import exceptions
from sat.core.xmpp import SatXMPPClient
from sat.memory import memory
import time
@ -204,10 +205,10 @@ class XEP_0045(object):
return False
return True
def getRoom(self, client, room_jid):
def getRoom(self, client: SatXMPPClient, room_jid: jid.JID) -> muc.Room:
"""Retrieve Room instance from its jid
@param room_jid(jid.JID): jid of the room
@param room_jid: jid of the room
@raise exceptions.NotFound: the room has not been joined
"""
try:
@ -223,10 +224,10 @@ class XEP_0045(object):
if room_jid not in client._muc_client.joined_rooms:
raise exceptions.NotFound(_("This room has not been joined"))
def isJoinedRoom(self, client, room_jid):
def isJoinedRoom(self, client: SatXMPPClient, room_jid: jid.JID) -> bool:
"""Tell if a jid is a known and joined room
@room_jid(jid.JID): jid of the room
@room_jid: jid of the room
"""
try:
self.checkRoomJoined(client, room_jid)

45
sat/plugins/plugin_xep_0060.py
View file

@ -19,7 +19,7 @@
from collections import namedtuple
from functools import reduce
from typing import Any, Dict, List, Optional, Tuple, Union, Callable
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union, Callable
import urllib.error
import urllib.parse
import urllib.request
@ -39,6 +39,7 @@ from sat.core.constants import Const as C
from sat.core.core_types import SatXMPPEntity
from sat.core.i18n import _
from sat.core.log import getLogger
from sat.core.xmpp import SatXMPPClient
from sat.tools import utils
from sat.tools import sat_defer
from sat.tools import xml_tools
@ -538,17 +539,22 @@ class XEP_0060(object):
))
async def sendItem(
self, client, service, nodeIdentifier, payload, item_id=None, extra=None
):
self,
client: SatXMPPClient,
service: Union[jid.JID, None],
nodeIdentifier: str,
payload: domish.Element,
item_id: Optional[str] = None,
extra: Optional[Dict[str, Any]] = None
) -> Optional[str]:
"""High level method to send one item
@param service(jid.JID, None): service to send the item to
None to use PEP
@param NodeIdentifier(unicode): PubSub node to use
@param payload(domish.Element): payload of the item to send
@param item_id(unicode, None): id to use or None to create one
@param extra(dict, None): extra option, not used yet
@return (unicode, None): id of the created item
@param service: service to send the item to None to use PEP
@param NodeIdentifier: PubSub node to use
@param payload: payload of the item to send
@param item_id: id to use or None to create one
@param extra: extra options
@return: id of the created item
"""
assert isinstance(payload, domish.Element)
item_elt = domish.Element((pubsub.NS_PUBSUB, 'item'))
@ -1090,7 +1096,12 @@ class XEP_0060(object):
client, jid.JID(service_s) if service_s else None, nodeIdentifier
)
def deleteNode(self, client, service, nodeIdentifier):
def deleteNode(
self,
client: SatXMPPClient,
service: jid.JID,
nodeIdentifier: str
) -> defer.Deferred:
return client.pubsub_client.deleteNode(service, nodeIdentifier)
def _addWatch(self, service_s, node, profile_key):
@ -1132,12 +1143,12 @@ class XEP_0060(object):
def retractItems(
self,
client,
service,
nodeIdentifier,
itemIdentifiers,
notify=True,
):
client: SatXMPPClient,
service: jid.JID,
nodeIdentifier: str,
itemIdentifiers: Iterable[str],
notify: bool = True,
) -> defer.Deferred:
return client.pubsub_client.retractItems(
service, nodeIdentifier, itemIdentifiers, notify=notify
)

3
sat/plugins/plugin_xep_0082.py
View file

@ -17,9 +17,6 @@
# 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/>.
# Type-check with `mypy --strict`
# Lint with `pylint`
from sat.core.constants import Const as C
from sat.core.i18n import D_
from sat.core.sat_main import SAT

8
sat/plugins/plugin_xep_0334.py
View file

@ -18,6 +18,7 @@
# 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 Iterable
from sat.core.i18n import _, D_
from sat.core.log import getLogger
@ -29,6 +30,7 @@ from sat.tools.common import data_format
from wokkel import disco, iwokkel
from twisted.words.protocols.jabber import xmlstream
from twisted.words.xish import domish
from zope.interface import implementer
from textwrap import dedent
@ -83,11 +85,11 @@ class XEP_0334(object):
else:
log.error("Unknown hint: {}".format(hint))
def addHintElements(self, message_elt, hints):
def addHintElements(self, message_elt: domish.Element, hints: Iterable[str]) -> None:
"""Add hints elements to message stanza
@param message_elt(domish.Element): stanza where hints must be added
@param hints(iterable(unicode)): hints to add
@param message_elt: stanza where hints must be added
@param hints: hints to add
"""
for hint in hints:
message_elt.addElement((NS_HINTS, hint))

8
sat/plugins/plugin_xep_0359.py
View file

@ -18,6 +18,7 @@
# 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 Optional
import uuid
from zope.interface import implementer
from twisted.words.protocols.jabber import xmlstream
@ -26,6 +27,7 @@ from sat.core.constants import Const as C
from sat.core import exceptions
from sat.core.i18n import _
from sat.core.log import getLogger
from twisted.words.xish import domish
log = getLogger(__name__)
@ -101,11 +103,11 @@ class XEP_0359(object):
sid_elt["by"] = client.jid.userhost() if by is None else by.userhost()
sid_elt["id"] = stanza_id
def getOriginId(self, element):
def getOriginId(self, element: domish.Element) -> Optional[str]:
"""Return origin-id if found in element
@param element(domish.Element): element to parse
@return (unicode, None): origin-id if found
@param element: element to parse
@return: origin-id if found
"""
try:
origin_elt = next(element.elements(NS_SID, "origin-id"))

3204
sat/plugins/plugin_xep_0384.py
View file

File diff suppressed because it is too large Load diff

39
sat/plugins/plugin_xep_0420.py
View file

@ -16,15 +16,12 @@
# 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/>.
# Type-check with `mypy --strict --disable-error-code no-untyped-call`
# Lint with `pylint`
from abc import ABC, abstractmethod
from datetime import datetime
import enum
import secrets
import string
from typing import Dict, Iterator, List, NamedTuple, Optional, Set, Tuple, Union, cast
from typing import Dict, NamedTuple, Optional, Set, Tuple, cast
from lxml import etree
@ -55,7 +52,7 @@ __all__ = [ # pylint: disable=unused-variable
]
log = cast(Logger, getLogger(__name__))
log = cast(Logger, getLogger(__name__)) # type: ignore[no-untyped-call]
PLUGIN_INFO = {
@ -291,19 +288,15 @@ class XEP_0420: # pylint: disable=invalid-name
# Note the serialized byte size of the content element before adding any children
empty_content_byte_size = len(content.toXml().encode("utf-8"))
# Just for type safety
stanza_children = cast(List[Union[domish.Element, str]], stanza.children)
content_children = cast(List[Union[domish.Element, str]], content.children)
# Move elements that are not explicitly forbidden from being encrypted from the
# stanza to the content element.
for child in list(cast(Iterator[domish.Element], stanza.elements())):
for child in list(stanza.elements()):
if (
child.uri not in XEP_0420.MUST_BE_PLAINTEXT_NAMESPACES
and (child.uri, child.name) not in XEP_0420.MUST_BE_PLAINTEXT_ELEMENTS
):
# Remove the child from the stanza
stanza_children.remove(child)
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
@ -316,7 +309,7 @@ class XEP_0420: # pylint: disable=invalid-name
child.defaultUri = C.NS_CLIENT
# Add the child with corrected namespaces to the content element
content_children.append(child)
content.addChild(child)
# Add the affixes requested by the profile
if profile.rpad_policy is not SCEAffixPolicy.NOT_NEEDED:
@ -345,7 +338,7 @@ class XEP_0420: # pylint: disable=invalid-name
time_element["stamp"] = XEP_0082.format_datetime()
if profile.to_policy is not SCEAffixPolicy.NOT_NEEDED:
recipient = cast(Optional[str], stanza.getAttribute("to", None))
recipient = stanza.getAttribute("to", None)
if recipient is None:
raise ValueError(
"<to/> affix requested, but stanza doesn't have the 'to' attribute"
@ -356,7 +349,7 @@ class XEP_0420: # pylint: disable=invalid-name
to_element["jid"] = jid.JID(recipient).userhost()
if profile.from_policy is not SCEAffixPolicy.NOT_NEEDED:
sender = cast(Optional[str], stanza.getAttribute("from", None))
sender = stanza.getAttribute("from", None)
if sender is None:
raise ValueError(
"<from/> affix requested, but stanza doesn't have the 'from'"
@ -370,7 +363,7 @@ class XEP_0420: # pylint: disable=invalid-name
if policy is not SCEAffixPolicy.NOT_NEEDED:
envelope.addChild(affix.create(stanza))
return cast(str, envelope.toXml()).encode("utf-8")
return envelope.toXml().encode("utf-8")
@staticmethod
def unpack_stanza(
@ -431,7 +424,7 @@ class XEP_0420: # pylint: disable=invalid-name
# Prepare the envelope and content elements
envelope = cast(domish.Element, ElementParser()(envelope_serialized_string))
content = cast(domish.Element, next(envelope.elements(NS_SCE, "content")))
content = next(envelope.elements(NS_SCE, "content"))
# Verify the affixes
rpad_element = cast(
@ -468,7 +461,7 @@ class XEP_0420: # pylint: disable=invalid-name
if to_element is not None:
recipient_value = jid.JID(to_element["jid"])
recipient_actual = cast(Optional[str], stanza.getAttribute("to", None))
recipient_actual = stanza.getAttribute("to", None)
if recipient_actual is None:
raise AffixVerificationFailed(
"'To' affix is included in the envelope, but the stanza is lacking a"
@ -491,7 +484,7 @@ class XEP_0420: # pylint: disable=invalid-name
if from_element is not None:
sender_value = jid.JID(from_element["jid"])
sender_actual = cast(Optional[str], stanza.getAttribute("from", None))
sender_actual = stanza.getAttribute("from", None)
if sender_actual is None:
raise AffixVerificationFailed(
"'From' affix is included in the envelope, but the stanza is lacking"
@ -551,13 +544,9 @@ class XEP_0420: # pylint: disable=invalid-name
+ custom_missing_string
)
# Just for type safety
content_children = cast(List[Union[domish.Element, str]], content.children)
stanza_children = cast(List[Union[domish.Element, str]], stanza.children)
# Move elements that are not explicitly forbidden from being encrypted from the
# content element to the stanza.
for child in list(cast(Iterator[domish.Element], content.elements())):
for child in list(content.elements()):
if (
child.uri in XEP_0420.MUST_BE_PLAINTEXT_NAMESPACES
or (child.uri, child.name) in XEP_0420.MUST_BE_PLAINTEXT_ELEMENTS
@ -568,10 +557,10 @@ class XEP_0420: # pylint: disable=invalid-name
)
else:
# Remove the child from the content element
content_children.remove(child)
content.children.remove(child)
# Add the child to the stanza
stanza_children.append(child)
stanza.addChild(child)
return SCEAffixValues(
rpad_value,