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
|
"""
Slixmpp: The Slick XMPP Library
Copyright (C) 2020 Mathieu Pasquet <mathieui@mathieui.net>
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
from typing import (
Dict,
Optional,
Set,
Tuple,
)
from slixmpp import JID, Message, Iq
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp.plugins import BasePlugin
from slixmpp.xmlstream import register_stanza_plugin
from slixmpp.xmlstream.matcher import MatchXPath
from slixmpp.xmlstream.handler import Callback
from slixmpp.plugins.xep_0404 import stanza
from slixmpp.plugins.xep_0004.stanza import Form
NODES = [
'urn:xmpp:mix:nodes:jidmap',
]
class XEP_0404(BasePlugin):
'''XEP-0404: MIX JID Hidden Channels'''
name = 'xep_0404'
description = 'MIX-ANON'
dependencies = {'xep_0369'}
stanza = stanza
namespace = stanza.NS
def plugin_init(self) -> None:
stanza.register_plugins()
async def get_anon_raw(self, channel: JID, *,
ifrom: Optional[JID] = None, **pubsubkwargs) -> Iq:
"""
Get the jid-participant mapping result (raw).
:param JID channel: MIX channel JID
"""
return await self.xmpp['xep_0030'].get_items(
channel.bare,
ifrom=ifrom,
**pubsubkwargs
)
async def get_anon_by_jid(self, channel: JID, *,
ifrom: Optional[JID] = None, **pubsubkwargs) -> Dict[JID, str]:
"""
Get the jid-participant mapping, by JID
:param JID channel: MIX channel JID
"""
raw = await self.get_anon_raw(channel, ifrom=ifrom, **pubsubkwargs)
mapping = {}
for item in raw['pubsub']['items']:
mapping[item['anon_participant']['jid']] = item['id']
return mapping
async def get_anon_by_id(self, channel: JID, *,
ifrom: Optional[JID] = None, **pubsubkwargs) -> Dict[str, JID]:
"""
Get the jid-participant mapping, by participant id
:param JID channel: MIX channel JID
"""
raw = await self.get_anon_raw(channel, ifrom=ifrom, **pubsubkwargs)
mapping = {}
for item in raw['pubsub']['items']:
mapping[item['id']] = item['anon_participant']['jid']
return mapping
async def get_preferences(self, channel: JID, *,
ifrom: Optional[JID] = None, **iqkwargs) -> Form:
"""
Get channel preferences with default values.
:param JID channel: MIX channel JID
"""
iq = self.xmpp.make_iq_get(ito=channel.bare, ifrom=ifrom)
iq.enable('user_preference')
prefs_stanza = await iq.send(**iqkwargs)
return prefs_stanza['user_preference']['form']
async def set_preferences(self, channel: JID, form: Form, *,
ifrom: Optional[JID] = None, **iqkwargs) -> Form:
"""
Set channel preferences
:param JID channel: MIX channel JID
:param Form form: A 0004 form with updated preferences
"""
iq = self.xmpp.make_iq_set(ito=channel.bare, ifrom=ifrom)
iq['user_preference']['form'] = form
prefs_result = await iq.send(**iqkwargs)
return prefs_result['user_preference']['form']
|
