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

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 

6 

7from ...group import LegacyParticipant 

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

9from .util import DispatcherMixin, exceptions_to_xmpp_errors 

10 

11if TYPE_CHECKING: 

12 from slidge.util.types import AnyGateway 

13 

14 

15class VCardMixin(DispatcherMixin): 

16 __slots__: list[str] = [] 

17 

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

19 super().__init__(xmpp) 

20 xmpp.register_handler( 

21 CoroutineCallback( 

22 "get_vcard", 

23 StanzaPath("iq@type=get/vcard"), 

24 self.on_get_vcard, # type:ignore[arg-type] 

25 ) 

26 ) 

27 xmpp.remove_handler("VCardTemp") 

28 xmpp.register_handler( 

29 CoroutineCallback( 

30 "VCardTemp", 

31 StanzaPath("iq/vcard_temp"), 

32 self.__vcard_temp_handler, # type:ignore[arg-type] 

33 ) 

34 ) 

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

36 register_stanza_plugin( 

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

38 MetaData, 

39 ) 

40 

41 @exceptions_to_xmpp_errors 

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

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

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

45 vcard = await contact.get_vcard() 

46 reply = iq.reply() 

47 if vcard: 

48 reply.append(vcard) 

49 else: 

50 reply.enable("vcard") 

51 reply.send() 

52 

53 @exceptions_to_xmpp_errors 

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

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

56 return await self.__handle_get_vcard_temp(iq) 

57 

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

59 return await self.__handle_set_vcard_temp(iq) 

60 

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

62 hash_ = session.user.avatar_hash 

63 if not hash_: 

64 raise XMPPError( 

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

66 ) 

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

68 session.user_jid, 

69 MetaData.namespace, 

70 hash_, 

71 ifrom=self.xmpp.boundjid.bare, 

72 ) 

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

74 type_ = info["type"] 

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

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

77 ) 

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

79 return bytes_, type_ 

80 

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

82 session = self.xmpp.get_session_from_stanza(iq) 

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

84 if not entity: 

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

86 

87 bytes_ = None 

88 if isinstance(entity, LegacyParticipant): 

89 if entity.is_user: 

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

91 if not bytes_: 

92 raise XMPPError( 

93 "internal-server-error", 

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

95 ) 

96 avatar = None 

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

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

99 else: 

100 avatar = contact.get_avatar() 

101 type_ = "image/png" 

102 else: 

103 avatar = entity.get_avatar() 

104 type_ = "image/png" 

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

106 if avatar is not None and avatar.data: 

107 bytes_ = avatar.data.get_value() 

108 if bytes_: 

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

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

111 

112 reply = iq.reply() 

113 reply.append(v) 

114 reply.send() 

115 

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

117 muc = await self.get_muc_from_stanza(iq) 

118 to = iq.get_to() 

119 

120 if to.resource: 

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

122 

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

124 try: 

125 legacy_id = await muc.on_avatar( 

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

127 ) 

128 except XMPPError: 

129 raise 

130 except Exception as e: # noqa: BLE001 

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

132 reply = iq.reply(clear=True) 

133 reply.enable("vcard_temp") 

134 reply.send() 

135 

136 if not data: 

137 await muc.set_avatar(None) 

138 return 

139 

140 if legacy_id: 

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