1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
# This file is part of Gajim.
#
# Gajim is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# Gajim 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Gajim. If not, see <http://www.gnu.org/licenses/>.
# Util module
from __future__ import annotations
from typing import Any
import logging
from functools import partial
from functools import wraps
from logging import LoggerAdapter
import nbxmpp
from nbxmpp.const import MessageType
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import JID
from nbxmpp.protocol import Message
from nbxmpp.structs import EMEData
from nbxmpp.structs import MessageProperties
from nbxmpp.task import Task
from gajim.common import app
from gajim.common import types
from gajim.common.const import EME_MESSAGES
from gajim.common.const import KindConstant
from gajim.common.events import MessageUpdated
from gajim.common.modules.misc import parse_correction
def from_xs_boolean(value: str | bool) -> bool:
if isinstance(value, bool):
return value
if value in ('1', 'true', 'True'):
return True
if value in ('0', 'false', 'False', ''):
return False
raise ValueError(f'Cant convert {value} to python boolean')
def to_xs_boolean(value: bool | None) -> str:
# Convert to xs:boolean ('true', 'false')
# from a python boolean (True, False) or None
if value is True:
return 'true'
if value is False:
return 'false'
if value is None:
return 'false'
raise ValueError(f'Cant convert {value} to xs:boolean')
def event_node(node: nbxmpp.Node) -> Any:
def event_node_decorator(func: Any):
@wraps(func)
def func_wrapper(self: Any,
_con: types.xmppClient,
_stanza: Message,
properties: MessageProperties
) -> Any:
if not properties.is_pubsub_event:
return
if properties.pubsub_event.node != node:
return
func(self, _con, _stanza, properties)
return func_wrapper
return event_node_decorator
def store_publish(func: Any):
@wraps(func)
def func_wrapper(self: Any, *args: Any, **kwargs: Any):
# pylint: disable=protected-access
if not app.account_is_connected(self._account):
self._stored_publish = partial(func, self, *args, **kwargs)
return None
return func(self, *args, **kwargs)
return func_wrapper
def get_eme_message(eme_data: EMEData) -> str:
try:
return EME_MESSAGES[eme_data.namespace]
except KeyError:
return EME_MESSAGES['fallback'] % eme_data.name
class LogAdapter(LoggerAdapter):
def process(self, msg: str, kwargs: Any) -> tuple[str, Any]:
return f'({self.extra["account"]}) {msg}', kwargs
def as_task(func):
@wraps(func)
def func_wrapper(self,
*args,
timeout=None,
callback=None,
user_data=None,
**kwargs):
task_ = Task(func(self, *args, **kwargs))
task_.set_timeout(timeout)
app.register_task(self, task_)
task_.set_finalize_func(app.remove_task, id(self))
task_.set_user_data(user_data)
if callback is not None:
task_.add_done_callback(callback)
task_.start()
return task_
return func_wrapper
def check_if_message_correction(properties: MessageProperties,
account: str,
jid: JID,
msgtxt: str,
kind: KindConstant,
timestamp: float,
logger: LoggerAdapter[logging.Logger]) -> bool:
correct_id = parse_correction(properties)
if correct_id is None:
return False
if properties.type not in (MessageType.GROUPCHAT, MessageType.CHAT):
logger.warning('Ignore correction with message type: %s',
properties.type)
return False
nickname = None
if properties.type.is_groupchat:
if jid.is_bare:
logger.warning(
'Ignore correction from bare groupchat jid: %s', jid)
return False
nickname = jid.resource
jid = jid.new_as_bare()
elif not properties.is_muc_pm:
jid = jid.new_as_bare()
successful = app.storage.archive.try_message_correction(
account,
jid,
nickname,
msgtxt,
correct_id,
kind,
timestamp)
if not successful:
logger.info('Message correction not successful')
return False
nickname = properties.muc_nickname or properties.nickname
event = MessageUpdated(account=account,
jid=jid,
msgtxt=msgtxt,
nickname=nickname,
properties=properties,
correct_id=correct_id)
app.ged.raise_event(event)
return True
def prepare_stanza(stanza: Message, plaintext: str) -> None:
delete_nodes(stanza, 'encrypted', Namespace.OMEMO_TEMP)
delete_nodes(stanza, 'body')
stanza.setBody(plaintext)
def delete_nodes(stanza: Message,
name: str,
namespace: str | None = None
) -> None:
nodes = stanza.getTags(name, namespace=namespace)
for node in nodes:
stanza.delChild(node)
|
