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
|
"""
Slixmpp: The Slick XMPP Library
Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
import hashlib
import logging
from asyncio import Future, ensure_future
from typing import (
Dict,
Optional,
)
from slixmpp import JID
from slixmpp.stanza import Presence
from slixmpp.exceptions import XMPPError, IqTimeout
from slixmpp.xmlstream import register_stanza_plugin, ElementBase
from slixmpp.plugins.base import BasePlugin
from slixmpp.plugins.xep_0153 import stanza, VCardTempUpdate
from slixmpp import future_wrapper
log = logging.getLogger(__name__)
class XEP_0153(BasePlugin):
name = 'xep_0153'
description = 'XEP-0153: vCard-Based Avatars'
dependencies = {'xep_0054'}
stanza = stanza
def plugin_init(self):
self._hashes = {}
register_stanza_plugin(Presence, VCardTempUpdate)
self.xmpp.add_filter('out', self._update_presence)
self.xmpp.add_event_handler('session_start', self._start)
self.xmpp.add_event_handler('presence_available', self._recv_presence)
self.xmpp.add_event_handler('presence_dnd', self._recv_presence)
self.xmpp.add_event_handler('presence_xa', self._recv_presence)
self.xmpp.add_event_handler('presence_chat', self._recv_presence)
self.xmpp.add_event_handler('presence_away', self._recv_presence)
self.api.register(self._set_hash, 'set_hash', default=True)
self.api.register(self._get_hash, 'get_hash', default=True)
self.api.register(self._reset_hash, 'reset_hash', default=True)
def plugin_end(self):
self.xmpp.del_filter('out', self._update_presence)
self.xmpp.del_event_handler('session_start', self._start)
self.xmpp.del_event_handler('session_end', self._end)
self.xmpp.del_event_handler('presence_available', self._recv_presence)
self.xmpp.del_event_handler('presence_dnd', self._recv_presence)
self.xmpp.del_event_handler('presence_xa', self._recv_presence)
self.xmpp.del_event_handler('presence_chat', self._recv_presence)
self.xmpp.del_event_handler('presence_away', self._recv_presence)
@future_wrapper
def set_avatar(self, jid: Optional[JID] = None,
avatar: Optional[bytes] = None,
mtype: Optional[str] = None, **iqkwargs) -> Future:
"""Set a VCard avatar.
:param jid: The JID to set the avatar for.
:param avatar: Avatar content.
:param mtype: Avatar file type (e.g. image/jpeg).
"""
if jid is None:
jid = self.xmpp.boundjid.bare
async def get_and_set_avatar():
timeout = iqkwargs.get('timeout', None)
timeout_cb = iqkwargs.get('timeout_callback', None)
try:
result = await self.xmpp['xep_0054'].get_vcard(
jid,
cached=False,
timeout=timeout
)
except IqTimeout as exc:
if timeout_cb is not None:
timeout_cb(exc)
raise
vcard = result['vcard_temp']
vcard['PHOTO']['TYPE'] = mtype
vcard['PHOTO']['BINVAL'] = avatar
try:
result = await self.xmpp['xep_0054'].publish_vcard(
jid=jid,
vcard=vcard,
**iqkwargs
)
except IqTimeout as exc:
timeout_cb(exc)
raise
self.api['reset_hash'](jid)
self.xmpp.roster[jid].send_last_presence()
return ensure_future(get_and_set_avatar(), loop=self.xmpp.loop)
async def _start(self, event):
try:
vcard = await self.xmpp['xep_0054'].get_vcard(self.xmpp.boundjid.bare)
data = vcard['vcard_temp']['PHOTO']['BINVAL']
if not data:
new_hash = ''
else:
new_hash = hashlib.sha1(data).hexdigest()
self.api['set_hash'](self.xmpp.boundjid, args=new_hash)
except XMPPError:
log.debug('Could not retrieve vCard for %s', self.xmpp.boundjid.bare)
def _update_presence(self, stanza: ElementBase) -> ElementBase:
if not isinstance(stanza, Presence):
return stanza
if stanza['type'] not in ('available', 'dnd', 'chat', 'away', 'xa'):
return stanza
current_hash = self.api['get_hash'](stanza['from'])
stanza['vcard_temp_update']['photo'] = current_hash
return stanza
def _recv_presence(self, pres: Presence):
try:
if pres.get_plugin('muc', check=True):
# Don't process vCard avatars for MUC occupants
# since they all share the same bare JID.
return
except:
pass
if not pres.match('presence/vcard_temp_update'):
self.api['set_hash'](pres['from'], args=None)
return
data = pres['vcard_temp_update']['photo']
if data is None:
return
self.xmpp.event('vcard_avatar_update', pres)
# =================================================================
def _reset_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
own_jid = (jid.bare == self.xmpp.boundjid.bare)
if self.xmpp.is_component:
own_jid = (jid.domain == self.xmpp.boundjid.domain)
self.api['set_hash'](jid, args=None)
if own_jid:
self.xmpp.roster[jid].send_last_presence()
def callback(iq):
if iq['type'] == 'error':
log.debug('Could not retrieve vCard for %s', jid)
return
try:
data = iq['vcard_temp']['PHOTO']['BINVAL']
except ValueError:
log.debug('Invalid BINVAL in vCard’s PHOTO for %s:', jid, exc_info=True)
data = None
if not data:
new_hash = ''
else:
new_hash = hashlib.sha1(data).hexdigest()
self.api['set_hash'](jid, args=new_hash)
self.xmpp['xep_0054'].get_vcard(jid=jid.bare, ifrom=ifrom,
callback=callback)
def _get_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
return self._hashes.get(jid.bare, None)
def _set_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
self._hashes[jid.bare] = args
|
