Coverage for slidge/core/dispatcher/vcard.py: 89%

1from typing import TYPE_CHECKING 

2 

3from slixmpp import CoroutineCallback, Iq, StanzaPath, register_stanza_plugin 

4from slixmpp.exceptions import XMPPError 

5from slixmpp.plugins.xep_0084.stanza import MetaData 

6from slixmpp.plugins.xep_0292.stanza import NS as VCard4NS 

7 

8from slidge.util.util import fix_namespaces 

9 

10from ...contact import LegacyContact 

11from ...group import LegacyParticipant 

12from ...util.types import AnySession, Avatar 

13from .util import DispatcherMixin, exceptions_to_xmpp_errors 

14 

15if TYPE_CHECKING: 

16 from slidge.util.types import AnyGateway 

17 

18 

19class VCardMixin(DispatcherMixin): 

20 __slots__: list[str] = [] 

21 

22 def __init__(self, xmpp: "AnyGateway") -> None: 

23 super().__init__(xmpp) 

24 xmpp.register_handler( 

25 CoroutineCallback( 

26 "get_vcard", StanzaPath("iq@type=get/vcard"), self.on_get_vcard 

27 ) 

28 ) 

29 xmpp.remove_handler("VCardTemp") 

30 xmpp.register_handler( 

31 CoroutineCallback( 

32 "VCardTemp", 

33 StanzaPath("iq/vcard_temp"), 

34 self.__vcard_temp_handler, 

35 ) 

36 ) 

37 # TODO: MR to slixmpp adding this to XEP-0084 

38 register_stanza_plugin( 

39 self.xmpp.plugin["xep_0060"].stanza.Item, 

40 MetaData, 

41 ) 

42 

43 @exceptions_to_xmpp_errors 

44 async def on_get_vcard(self, iq: Iq) -> None: 

45 session = await self._get_session(iq, logged=True) 

46 contact = await session.contacts.by_jid(iq.get_to()) 

47 vcard = await contact.get_vcard() 

48 reply = iq.reply() 

49 if vcard: 

50 reply.append(vcard) 

51 else: 

52 reply.enable("vcard") 

53 reply.send() 

54 

55 @exceptions_to_xmpp_errors 

56 async def __vcard_temp_handler(self, iq: Iq) -> None: 

57 if iq["type"] == "get": 

58 return await self.__handle_get_vcard_temp(iq) 

59 

60 if iq["type"] == "set": 

61 return await self.__handle_set_vcard_temp(iq) 

62 

63 async def __fetch_user_avatar(self, session: AnySession) -> tuple[bytes, str]: 

64 hash_ = session.user.avatar_hash 

65 if not hash_: 

66 raise XMPPError( 

67 "item-not-found", "The slidge user does not have any avatar set" 

68 ) 

69 meta_iq = await self.xmpp.plugin["xep_0060"].get_item( 

70 session.user_jid, 

71 MetaData.namespace, 

72 hash_, 

73 ifrom=self.xmpp.boundjid.bare, 

74 ) 

75 info = meta_iq["pubsub"]["items"]["item"]["avatar_metadata"]["info"] 

76 type_ = info["type"] 

77 data_iq = await self.xmpp.plugin["xep_0084"].retrieve_avatar( 

78 session.user_jid, hash_, ifrom=self.xmpp.boundjid.bare 

79 ) 

80 bytes_ = data_iq["pubsub"]["items"]["item"]["avatar_data"]["value"] 

81 return bytes_, type_ 

82 

83 async def __handle_get_vcard_temp(self, iq: Iq) -> None: 

84 session = self.xmpp.get_session_from_stanza(iq) 

85 entity = await session.get_contact_or_group_or_participant(iq.get_to(), False) 

86 if not entity: 

87 raise XMPPError("item-not-found") 

88 

89 bytes_ = None 

90 if isinstance(entity, LegacyParticipant): 

91 if entity.is_user: 

92 bytes_, type_ = await self.__fetch_user_avatar(session) 

93 if not bytes_: 

94 raise XMPPError( 

95 "internal-server-error", 

96 "Could not fetch the slidge user's avatar", 

97 ) 

98 avatar = None 

99 vcard = None 

100 elif not (contact := entity.contact): 

101 raise XMPPError("item-not-found", "This participant has no contact") 

102 else: 

103 vcard = await contact.get_vcard() 

104 avatar = contact.get_avatar() 

105 type_ = "image/png" 

106 else: 

107 avatar = entity.get_avatar() 

108 type_ = "image/png" 

109 if isinstance(entity, LegacyContact): 

110 vcard = await entity.get_vcard() 

111 else: 

112 vcard = None 

113 v = self.xmpp.plugin["xep_0054"].make_vcard() 

114 if avatar is not None and avatar.data: 

115 bytes_ = avatar.data.get_value() 

116 if bytes_: 

117 v["PHOTO"]["BINVAL"] = bytes_ 

118 v["PHOTO"]["TYPE"] = type_ 

119 if vcard: 

120 fix_namespaces( 

121 vcard.xml, 

122 VCard4NS, 

123 self.xmpp.plugin["xep_0054"].stanza.VCardTemp.namespace, 

124 ) 

125 for el in vcard.xml: 

126 v.append(el) 

127 reply = iq.reply() 

128 reply.append(v) 

129 reply.send() 

130 

131 async def __handle_set_vcard_temp(self, iq: Iq) -> None: 

132 muc = await self.get_muc_from_stanza(iq) 

133 to = iq.get_to() 

134 

135 if to.resource: 

136 raise XMPPError("bad-request", "You cannot set participants avatars") 

137 

138 data = iq["vcard_temp"]["PHOTO"]["BINVAL"] or None 

139 try: 

140 legacy_id = await muc.on_avatar( 

141 data, iq["vcard_temp"]["PHOTO"]["TYPE"] or None 

142 ) 

143 except XMPPError: 

144 raise 

145 except Exception as e: 

146 raise XMPPError("internal-server-error", str(e)) 

147 reply = iq.reply(clear=True) 

148 reply.enable("vcard_temp") 

149 reply.send() 

150 

151 if not data: 

152 await muc.set_avatar(None) 

153 return 

154 

155 if legacy_id: 

156 await muc.set_avatar(Avatar(data=data, unique_id=legacy_id))