Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0334.py @ 4071:4b842c1fb686

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
refactoring: renamed `sat` package to `libervia.backend`
author Goffi <goffi@goffi.org>
date Fri, 02 Jun 2023 11:49:51 +0200
parents sat/plugins/plugin_xep_0334.py@c23cad65ae99
children 3b3cd9453d9b
comparison
equal deleted inserted replaced
4070:d10748475025 4071:4b842c1fb686
1 #!/usr/bin/env python3
2
3
4 # SAT plugin for Delayed Delivery (XEP-0334)
5 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
6 # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org)
7
8 # This program is free software: you can redistribute it and/or modify
9 # it under the terms of the GNU Affero General Public License as published by
10 # the Free Software Foundation, either version 3 of the License, or
11 # (at your option) any later version.
12
13 # This program is distributed in the hope that it will be useful,
14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 # GNU Affero General Public License for more details.
17
18 # You should have received a copy of the GNU Affero General Public License
19 # along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21 from typing import Iterable
22 from libervia.backend.core.i18n import _, D_
23 from libervia.backend.core.log import getLogger
24
25 log = getLogger(__name__)
26 from libervia.backend.core.constants import Const as C
27
28 from libervia.backend.tools.common import data_format
29
30 from wokkel import disco, iwokkel
31
32 from twisted.words.protocols.jabber import xmlstream
33 from twisted.words.xish import domish
34 from zope.interface import implementer
35 from textwrap import dedent
36
37
38 PLUGIN_INFO = {
39 C.PI_NAME: "Message Processing Hints",
40 C.PI_IMPORT_NAME: "XEP-0334",
41 C.PI_TYPE: "XEP",
42 C.PI_MODES: C.PLUG_MODE_BOTH,
43 C.PI_PROTOCOLS: ["XEP-0334"],
44 C.PI_MAIN: "XEP_0334",
45 C.PI_HANDLER: "yes",
46 C.PI_DESCRIPTION: D_("""Implementation of Message Processing Hints"""),
47 C.PI_USAGE: dedent(
48 D_(
49 """\
50 Frontends can use HINT_* constants in mess_data['extra'] in a serialized 'hints' dict.
51 Internal plugins can use directly add_hint([HINT_* constant]).
52 Will set mess_data['extra']['history'] to 'skipped' when no store is requested and message is not saved in history."""
53 )
54 ),
55 }
56
57 NS_HINTS = "urn:xmpp:hints"
58
59
60 class XEP_0334(object):
61 HINT_NO_PERMANENT_STORE = "no-permanent-store"
62 HINT_NO_STORE = "no-store"
63 HINT_NO_COPY = "no-copy"
64 HINT_STORE = "store"
65 HINTS = (HINT_NO_PERMANENT_STORE, HINT_NO_STORE, HINT_NO_COPY, HINT_STORE)
66
67 def __init__(self, host):
68 log.info(_("Message Processing Hints plugin initialization"))
69 self.host = host
70 host.trigger.add("sendMessage", self.send_message_trigger)
71 host.trigger.add("message_received", self.message_received_trigger, priority=-1000)
72
73 def get_handler(self, client):
74 return XEP_0334_handler()
75
76 def add_hint(self, mess_data, hint):
77 if hint == self.HINT_NO_COPY and not mess_data["to"].resource:
78 log.error(
79 "{hint} can only be used with full jids! Ignoring it.".format(hint=hint)
80 )
81 return
82 hints = mess_data.setdefault("hints", set())
83 if hint in self.HINTS:
84 hints.add(hint)
85 else:
86 log.error("Unknown hint: {}".format(hint))
87
88 def add_hint_elements(self, message_elt: domish.Element, hints: Iterable[str]) -> None:
89 """Add hints elements to message stanza
90
91 @param message_elt: stanza where hints must be added
92 @param hints: hints to add
93 """
94 for hint in hints:
95 if not list(message_elt.elements(NS_HINTS, hint)):
96 message_elt.addElement((NS_HINTS, hint))
97 else:
98 log.debug('Not adding {hint!r} hint: it is already present in <message>')
99
100 def _send_post_xml_treatment(self, mess_data):
101 if "hints" in mess_data:
102 self.add_hint_elements(mess_data["xml"], mess_data["hints"])
103 return mess_data
104
105 def send_message_trigger(
106 self, client, mess_data, pre_xml_treatments, post_xml_treatments
107 ):
108 """Add the hints element to the message to be sent"""
109 if "hints" in mess_data["extra"]:
110 for hint in data_format.dict2iter("hints", mess_data["extra"], pop=True):
111 self.add_hint(hint)
112
113 post_xml_treatments.addCallback(self._send_post_xml_treatment)
114 return True
115
116 def _received_skip_history(self, mess_data):
117 mess_data["history"] = C.HISTORY_SKIP
118 return mess_data
119
120 def message_received_trigger(self, client, message_elt, post_treat):
121 """Check for hints in the received message"""
122 for elt in message_elt.elements():
123 if elt.uri == NS_HINTS and elt.name in (
124 self.HINT_NO_PERMANENT_STORE,
125 self.HINT_NO_STORE,
126 ):
127 log.debug("history will be skipped for this message, as requested")
128 post_treat.addCallback(self._received_skip_history)
129 break
130 return True
131
132
133 @implementer(iwokkel.IDisco)
134 class XEP_0334_handler(xmlstream.XMPPHandler):
135
136 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
137 return [disco.DiscoFeature(NS_HINTS)]
138
139 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
140 return []