| ... |
... |
@@ -74,7 +74,7 @@ |
|
74
|
74
|
from nbxmpp.structs import ProxyData
|
|
75
|
75
|
from nbxmpp.protocol import JID
|
|
76
|
76
|
from nbxmpp.protocol import InvalidJid
|
|
77
|
|
-from OpenSSL.crypto import load_certificate
|
|
|
77
|
+from OpenSSL.crypto import X509, load_certificate
|
|
78
|
78
|
from OpenSSL.crypto import FILETYPE_PEM
|
|
79
|
79
|
from gi.repository import Gio
|
|
80
|
80
|
from gi.repository import GLib
|
| ... |
... |
@@ -116,20 +116,20 @@ class InvalidFormat(Exception): |
|
116
|
116
|
pass
|
|
117
|
117
|
|
|
118
|
118
|
|
|
119
|
|
-def parse_jid(jidstring):
|
|
|
119
|
+def parse_jid(jidstring: str) -> str:
|
|
120
|
120
|
try:
|
|
121
|
121
|
return str(validate_jid(jidstring))
|
|
122
|
122
|
except Exception as error:
|
|
123
|
123
|
raise InvalidFormat(error)
|
|
124
|
124
|
|
|
125
|
|
-def idn_to_ascii(host):
|
|
|
125
|
+def idn_to_ascii(host: str) -> str:
|
|
126
|
126
|
"""
|
|
127
|
127
|
Convert IDN (Internationalized Domain Names) to ACE (ASCII-compatible
|
|
128
|
128
|
encoding)
|
|
129
|
129
|
"""
|
|
130
|
130
|
from encodings import idna
|
|
131
|
131
|
labels = idna.dots.split(host)
|
|
132
|
|
- converted_labels = []
|
|
|
132
|
+ converted_labels: list[str] = []
|
|
133
|
133
|
for label in labels:
|
|
134
|
134
|
if label:
|
|
135
|
135
|
converted_labels.append(idna.ToASCII(label).decode('utf-8'))
|
| ... |
... |
@@ -137,19 +137,19 @@ def idn_to_ascii(host): |
|
137
|
137
|
converted_labels.append('')
|
|
138
|
138
|
return ".".join(converted_labels)
|
|
139
|
139
|
|
|
140
|
|
-def ascii_to_idn(host):
|
|
|
140
|
+def ascii_to_idn(host: str) -> str:
|
|
141
|
141
|
"""
|
|
142
|
142
|
Convert ACE (ASCII-compatible encoding) to IDN (Internationalized Domain
|
|
143
|
143
|
Names)
|
|
144
|
144
|
"""
|
|
145
|
145
|
from encodings import idna
|
|
146
|
146
|
labels = idna.dots.split(host)
|
|
147
|
|
- converted_labels = []
|
|
|
147
|
+ converted_labels: list[str] = []
|
|
148
|
148
|
for label in labels:
|
|
149
|
149
|
converted_labels.append(idna.ToUnicode(label))
|
|
150
|
150
|
return ".".join(converted_labels)
|
|
151
|
151
|
|
|
152
|
|
-def puny_encode_url(url):
|
|
|
152
|
+def puny_encode_url(url: str) -> Optional[str]:
|
|
153
|
153
|
_url = url
|
|
154
|
154
|
if '//' not in _url:
|
|
155
|
155
|
_url = '//' + _url
|
| ... |
... |
@@ -158,10 +158,10 @@ def puny_encode_url(url): |
|
158
|
158
|
p_loc = idn_to_ascii(o.hostname)
|
|
159
|
159
|
except Exception:
|
|
160
|
160
|
log.debug('urlparse failed: %s', url)
|
|
161
|
|
- return False
|
|
|
161
|
+ return None
|
|
162
|
162
|
return url.replace(o.hostname, p_loc)
|
|
163
|
163
|
|
|
164
|
|
-def parse_resource(resource):
|
|
|
164
|
+def parse_resource(resource: str) -> Optional[str]:
|
|
165
|
165
|
"""
|
|
166
|
166
|
Perform stringprep on resource and return it
|
|
167
|
167
|
"""
|
| ... |
... |
@@ -260,48 +260,43 @@ def get_uf_role(role: Union[Role, str], plural: bool = False) -> str: |
|
260
|
260
|
role = role.value
|
|
261
|
261
|
|
|
262
|
262
|
if role == 'none':
|
|
263
|
|
- role_name = Q_('?Group Chat Contact Role:None')
|
|
264
|
|
- elif role == 'moderator':
|
|
|
263
|
+ return Q_('?Group Chat Contact Role:None')
|
|
|
264
|
+ if role == 'moderator':
|
|
265
|
265
|
if plural:
|
|
266
|
|
- role_name = _('Moderators')
|
|
267
|
|
- else:
|
|
268
|
|
- role_name = _('Moderator')
|
|
269
|
|
- elif role == 'participant':
|
|
|
266
|
+ return _('Moderators')
|
|
|
267
|
+ return _('Moderator')
|
|
|
268
|
+ if role == 'participant':
|
|
270
|
269
|
if plural:
|
|
271
|
|
- role_name = _('Participants')
|
|
272
|
|
- else:
|
|
273
|
|
- role_name = _('Participant')
|
|
274
|
|
- elif role == 'visitor':
|
|
|
270
|
+ return _('Participants')
|
|
|
271
|
+ return _('Participant')
|
|
|
272
|
+ if role == 'visitor':
|
|
275
|
273
|
if plural:
|
|
276
|
|
- role_name = _('Visitors')
|
|
277
|
|
- else:
|
|
278
|
|
- role_name = _('Visitor')
|
|
279
|
|
- return role_name
|
|
|
274
|
+ return _('Visitors')
|
|
|
275
|
+ return _('Visitor')
|
|
|
276
|
+ return ''
|
|
280
|
277
|
|
|
281
|
278
|
def get_uf_affiliation(affiliation: Union[Affiliation, str],
|
|
282
|
|
- plural: bool = False) -> str:
|
|
|
279
|
+ plural: bool = False
|
|
|
280
|
+ ) -> str:
|
|
283
|
281
|
'''Get a nice and translated affilition for muc'''
|
|
284
|
282
|
if not isinstance(affiliation, str):
|
|
285
|
283
|
affiliation = affiliation.value
|
|
286
|
284
|
|
|
287
|
285
|
if affiliation == 'none':
|
|
288
|
|
- affiliation_name = Q_('?Group Chat Contact Affiliation:None')
|
|
289
|
|
- elif affiliation == 'owner':
|
|
|
286
|
+ return Q_('?Group Chat Contact Affiliation:None')
|
|
|
287
|
+ if affiliation == 'owner':
|
|
290
|
288
|
if plural:
|
|
291
|
|
- affiliation_name = _('Owners')
|
|
292
|
|
- else:
|
|
293
|
|
- affiliation_name = _('Owner')
|
|
294
|
|
- elif affiliation == 'admin':
|
|
|
289
|
+ return _('Owners')
|
|
|
290
|
+ return _('Owner')
|
|
|
291
|
+ if affiliation == 'admin':
|
|
295
|
292
|
if plural:
|
|
296
|
|
- affiliation_name = _('Administrators')
|
|
297
|
|
- else:
|
|
298
|
|
- affiliation_name = _('Administrator')
|
|
299
|
|
- elif affiliation == 'member':
|
|
|
293
|
+ return _('Administrators')
|
|
|
294
|
+ return _('Administrator')
|
|
|
295
|
+ if affiliation == 'member':
|
|
300
|
296
|
if plural:
|
|
301
|
|
- affiliation_name = _('Members')
|
|
302
|
|
- else:
|
|
303
|
|
- affiliation_name = _('Member')
|
|
304
|
|
- return affiliation_name
|
|
|
297
|
+ return _('Members')
|
|
|
298
|
+ return _('Member')
|
|
|
299
|
+ return ''
|
|
305
|
300
|
|
|
306
|
301
|
def get_uf_relative_time(timestamp: int) -> str:
|
|
307
|
302
|
date_time = datetime.fromtimestamp(timestamp)
|
| ... |
... |
@@ -389,23 +384,23 @@ def exec_command(command: str, |
|
389
|
384
|
process = subprocess.Popen(args)
|
|
390
|
385
|
app.thread_interface(process.wait)
|
|
391
|
386
|
|
|
392
|
|
-def build_command(executable, parameter):
|
|
|
387
|
+def build_command(executable: str, parameter: str) -> str:
|
|
393
|
388
|
# we add to the parameter (can hold path with spaces)
|
|
394
|
389
|
# "" so we have good parsing from shell
|
|
395
|
|
- parameter = parameter.replace('"', '\\"') # but first escape "
|
|
|
390
|
+ parameter = parameter.replace('"', '\\"') # but first escape "
|
|
396
|
391
|
command = f'{executable} "{parameter}"'
|
|
397
|
392
|
return command
|
|
398
|
393
|
|
|
399
|
394
|
def get_file_path_from_dnd_dropped_uri(uri: str) -> str:
|
|
400
|
|
- path = urllib.parse.unquote(uri) # escape special chars
|
|
401
|
|
- path = path.strip('\r\n\x00') # remove \r\n and NULL
|
|
|
395
|
+ path = urllib.parse.unquote(uri) # escape special chars
|
|
|
396
|
+ path = path.strip('\r\n\x00') # remove \r\n and NULL
|
|
402
|
397
|
# get the path to file
|
|
403
|
|
- if re.match('^file:///[a-zA-Z]:/', path): # windows
|
|
404
|
|
- path = path[8:] # 8 is len('file:///')
|
|
405
|
|
- elif path.startswith('file://'): # nautilus, rox
|
|
406
|
|
- path = path[7:] # 7 is len('file://')
|
|
407
|
|
- elif path.startswith('file:'): # xffm
|
|
408
|
|
- path = path[5:] # 5 is len('file:')
|
|
|
398
|
+ if re.match('^file:///[a-zA-Z]:/', path): # windows
|
|
|
399
|
+ path = path[8:] # 8 is len('file:///')
|
|
|
400
|
+ elif path.startswith('file://'): # nautilus, rox
|
|
|
401
|
+ path = path[7:] # 7 is len('file://')
|
|
|
402
|
+ elif path.startswith('file:'): # xffm
|
|
|
403
|
+ path = path[5:] # 5 is len('file:')
|
|
409
|
404
|
return path
|
|
410
|
405
|
|
|
411
|
406
|
def sanitize_filename(filename: str) -> str:
|
| ... |
... |
@@ -437,7 +432,7 @@ def reduce_chars_newlines(text: str, max_chars: int = 0, |
|
437
|
432
|
If any of the params is not present (None or 0) the action on it is not
|
|
438
|
433
|
performed
|
|
439
|
434
|
"""
|
|
440
|
|
- def _cut_if_long(string_):
|
|
|
435
|
+ def _cut_if_long(string_: str) -> str:
|
|
441
|
436
|
if len(string_) > max_chars:
|
|
442
|
437
|
string_ = string_[:max_chars - 3] + '…'
|
|
443
|
438
|
return string_
|
| ... |
... |
@@ -458,12 +453,12 @@ def _cut_if_long(string_): |
|
458
|
453
|
return reduced_text
|
|
459
|
454
|
|
|
460
|
455
|
|
|
461
|
|
-def get_contact_dict_for_account(account: str) -> Dict[str, types.BareContact]:
|
|
|
456
|
+def get_contact_dict_for_account(account: str) -> dict[str, types.BareContact]:
|
|
462
|
457
|
"""
|
|
463
|
458
|
Creates a dict of jid -> contact with all contacts of account
|
|
464
|
459
|
Can be used for completion lists
|
|
465
|
460
|
"""
|
|
466
|
|
- contacts_dict = {}
|
|
|
461
|
+ contacts_dict: dict[str, types.BareContact] = {}
|
|
467
|
462
|
client = app.get_client(account)
|
|
468
|
463
|
for contact in client.get_module('Roster').iter_contacts():
|
|
469
|
464
|
contacts_dict[str(contact.jid)] = contact
|
| ... |
... |
@@ -572,6 +567,7 @@ def get_global_show() -> str: |
|
572
|
567
|
|
|
573
|
568
|
def get_global_status_message() -> str:
|
|
574
|
569
|
maxi = 0
|
|
|
570
|
+ status_message = ''
|
|
575
|
571
|
for account, con in app.connections.items():
|
|
576
|
572
|
if not app.settings.get_account_setting(account,
|
|
577
|
573
|
'sync_with_global_status'):
|
| ... |
... |
@@ -614,14 +610,14 @@ def get_jid_from_iq(iq_obj): |
|
614
|
610
|
jid = get_full_jid_from_iq(iq_obj)
|
|
615
|
611
|
return app.get_jid_without_resource(jid)
|
|
616
|
612
|
|
|
617
|
|
-def get_auth_sha(sid, initiator, target):
|
|
|
613
|
+def get_auth_sha(sid: str, initiator: str, target: str) -> str:
|
|
618
|
614
|
"""
|
|
619
|
615
|
Return sha of sid + initiator + target used for proxy auth
|
|
620
|
616
|
"""
|
|
621
|
617
|
return hashlib.sha1(
|
|
622
|
618
|
(f'{sid}{initiator}{target}').encode('utf-8')).hexdigest()
|
|
623
|
619
|
|
|
624
|
|
-def remove_invalid_xml_chars(string_):
|
|
|
620
|
+def remove_invalid_xml_chars(string_: str) -> str:
|
|
625
|
621
|
if string_:
|
|
626
|
622
|
string_ = re.sub(INVALID_XML_CHARS_REGEX, '', string_)
|
|
627
|
623
|
return string_
|
| ... |
... |
@@ -771,7 +767,7 @@ def get_proxy(proxy_name: str) -> Optional[ProxyData]: |
|
771
|
767
|
username=username,
|
|
772
|
768
|
password=password)
|
|
773
|
769
|
|
|
774
|
|
-def version_condition(current_version, required_version):
|
|
|
770
|
+def version_condition(current_version: str, required_version: str) -> bool:
|
|
775
|
771
|
if V(current_version) < V(required_version):
|
|
776
|
772
|
return False
|
|
777
|
773
|
return True
|
| ... |
... |
@@ -1327,9 +1323,8 @@ def _on_load_finished(file: Gio.File, |
|
1327
|
1323
|
file.load_contents_async(None, _on_load_finished)
|
|
1328
|
1324
|
|
|
1329
|
1325
|
|
|
1330
|
|
-def convert_gio_to_openssl_cert(cert):
|
|
1331
|
|
- cert = load_certificate(FILETYPE_PEM, cert.props.certificate_pem.encode())
|
|
1332
|
|
- return cert
|
|
|
1326
|
+def convert_gio_to_openssl_cert(cert: Gio.TlsCertificate) -> X509:
|
|
|
1327
|
+ return load_certificate(FILETYPE_PEM, cert.props.certificate_pem.encode())
|
|
1333
|
1328
|
|
|
1334
|
1329
|
|
|
1335
|
1330
|
def get_custom_host(account: str) -> Optional[Tuple[str,
|
| ... |
... |
@@ -1403,7 +1398,7 @@ def ask_for_status_message(status: str, signin: bool = False) -> bool: |
|
1403
|
1398
|
return app.settings.get('always_ask_for_status_message')
|
|
1404
|
1399
|
|
|
1405
|
1400
|
|
|
1406
|
|
-def get_group_chat_nick(account: str, room_jid: str) -> str:
|
|
|
1401
|
+def get_group_chat_nick(account: str, room_jid: Union[JID, str]) -> str:
|
|
1407
|
1402
|
nick = app.nicks[account]
|
|
1408
|
1403
|
|
|
1409
|
1404
|
client = app.get_client(account)
|
| ... |
... |
@@ -1416,7 +1411,7 @@ def get_group_chat_nick(account: str, room_jid: str) -> str: |
|
1416
|
1411
|
return nick
|
|
1417
|
1412
|
|
|
1418
|
1413
|
|
|
1419
|
|
-def get_muc_context(jid: str) -> str:
|
|
|
1414
|
+def get_muc_context(jid: Union[JID, str]) -> Optional[str]:
|
|
1420
|
1415
|
disco_info = app.storage.cache.get_last_disco_info(jid)
|
|
1421
|
1416
|
if disco_info is None:
|
|
1422
|
1417
|
return None
|