Coverage for slidge/core/dispatcher/pubsub.py: 95%

1from slixmpp import CoroutineCallback, Iq, StanzaPath 

2from slixmpp.exceptions import XMPPError 

3from slixmpp.plugins.xep_0060.stanza import Affiliation, Subscription 

4from slixmpp.plugins.xep_0084.stanza import Data as AvatarData 

5from slixmpp.plugins.xep_0084.stanza import MetaData as AvatarMetadata 

6from slixmpp.xmlstream import StanzaBase 

7 

8from slidge.db.models import Space 

9 

10from ...util.types import AnyGateway, AnySession 

11from .util import DispatcherMixin, exceptions_to_xmpp_errors 

12 

13 

14class PubSubMixin(DispatcherMixin): 

15 __slots__: list[str] = [] 

16 xmpp: AnyGateway 

17 

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

19 super().__init__(xmpp) 

20 

21 for path, func in [ 

22 ("get/pubsub/items", self.__get_items), 

23 ("set/pubsub/subscribe", self.__set_subscribe), 

24 ("set/pubsub/unsubscribe", self.__set_unsubscribe), 

25 ("get/pubsub/affiliations", self.__get_affiliations), 

26 ("get/pubsub_owner/affiliations", self.__owner_get_affiliations), 

27 ("get/pubsub/subscriptions", self.__get_subscriptions), 

28 ("get/pubsub_owner/subscriptions", self.__owner_get_subscriptions), 

29 ]: 

30 self.xmpp.register_handler( 

31 CoroutineCallback( 

32 func.__name__, 

33 StanzaPath(f"iq@to={self.xmpp.boundjid.bare}@type={path}"), 

34 func, # type:ignore[arg-type] # ty:ignore[invalid-argument-type] 

35 ) 

36 ) 

37 

38 self.xmpp.register_handler( 

39 CoroutineCallback( 

40 "get_items_any", 

41 StanzaPath("iq@type=get/pubsub/items"), 

42 self.__get_items_any, 

43 ) 

44 ) 

45 

46 async def __get_items_any(self, iq: StanzaBase) -> None: 

47 ito = iq.get_to() 

48 if not ito.local: 

49 return # handled elsewhere (all over the place actually, refactor needed!) 

50 node = iq["pubsub"]["items"]["node"] 

51 if node not in _NODES_HANDLED_ELSEWHERE: 

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

53 

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

55 node = iq["pubsub"]["items"]["node"] 

56 

57 if node in _NODES_HANDLED_ELSEWHERE: 

58 # handled in slidge/core/pubsub.py 

59 # TODO: have a single entrypoint for all get/pubsub/items 

60 return 

61 

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

63 legacy_id = await self.__get_legacy_id(session, node) 

64 item_ids: list[str] = [item["id"] for item in iq["pubsub"]["items"]["items"]] 

65 

66 room_item_ids = list(filter(_is_not_space_image, item_ids)) 

67 invalid = list(filter(self.__is_not_local_jid, room_item_ids)) 

68 

69 if invalid: 

70 raise XMPPError( 

71 "item-not-found", 

72 f"These items are not part of this space: {invalid}.", 

73 ) 

74 

75 room_legacy_ids = ( 

76 [self.__item_to_room_legacy_id(item_id) for item_id in room_item_ids] 

77 if room_item_ids 

78 else None 

79 ) 

80 

81 space = await session.bookmarks.get_updated_space( 

82 legacy_id, room_legacy_id_filter=room_legacy_ids 

83 ) 

84 if space is None: 

85 raise XMPPError("item-not-found", f"No space '{legacy_id}'") 

86 

87 with self.xmpp.store.session() as orm: 

88 orm.add(space) 

89 if room_item_ids and len(space.rooms) != len(room_item_ids): 

90 raise XMPPError( 

91 "item-not-found", 

92 f"Could not find rooms: {set(room_item_ids) - {str(r.jid) for r in space.rooms}}", 

93 ) 

94 

95 reply = iq.reply() 

96 reply["pubsub"]["items"]["node"] = node 

97 self.xmpp.pubsub.set_space_items(space, reply["pubsub"]["items"], item_ids) 

98 

99 if len(reply["pubsub"]["items"]) == 0: 

100 if len(item_ids) == 1: 

101 raise XMPPError( 

102 "item-not-found", f"{item_ids[0]!r} could not be found." 

103 ) 

104 else: 

105 raise XMPPError( 

106 "item-not-found", 

107 f"Some of these items could not be found: {item_ids}.", 

108 ) 

109 

110 reply.send() 

111 

112 def __item_to_room_legacy_id(self, item_id: str) -> str: 

113 _type, legacy_id = self.xmpp.localpart_to_legacy_id( 

114 item_id.removesuffix(f"@{self.xmpp.boundjid.bare}") 

115 ) 

116 return legacy_id 

117 

118 def __is_not_local_jid(self, item_id: str) -> bool: 

119 return not item_id.endswith(f"@{self.xmpp.boundjid.bare}") 

120 

121 async def __get_legacy_id(self, session: AnySession, node: str) -> str: 

122 try: 

123 legacy_id = session.bookmarks.space_node_to_legacy_id(node) 

124 except Exception as e: # noqa: BLE001 

125 raise XMPPError("item-not-found", str(e)) 

126 with self.xmpp.store.session() as orm: 

127 if not self.xmpp.store.spaces.exists(orm, session.user_pk, legacy_id): 

128 raise XMPPError( 

129 "item-not-found", f"This is not a known space: '{legacy_id}'" 

130 ) 

131 return legacy_id 

132 

133 async def __set_subscribe(self, iq: Iq) -> None: 

134 node = iq["pubsub"]["subscribe"]["node"] 

135 

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

137 await self.__get_legacy_id(session, node) 

138 

139 reply = iq.reply(clear=True) 

140 sub = reply["pubsub"]["subscription"] 

141 sub["node"] = node 

142 sub["jid"] = session.user_jid 

143 sub["subscription"] = "subscribed" 

144 reply.send() 

145 

146 @exceptions_to_xmpp_errors 

147 async def __set_unsubscribe(self, iq: Iq) -> None: 

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

149 

150 jid = iq["pubsub"]["unsubscribe"]["jid"] 

151 if jid != session.user_jid.bare: 

152 raise XMPPError("bad-request", f"Cannot unsubscribe JID {jid}") 

153 

154 node = iq["pubsub"]["unsubscribe"]["node"] 

155 legacy_id = session.bookmarks.space_node_to_legacy_id(node) 

156 

157 await session.on_leave_space(legacy_id) 

158 reply = iq.reply(clear=True) 

159 

160 sub = reply["pubsub"]["subscription"] 

161 sub["node"] = node 

162 sub["jid"] = session.user_jid 

163 sub["subscription"] = "none" 

164 reply.send() 

165 

166 async def __get_affiliations(self, iq: Iq) -> None: 

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

168 

169 reply = iq.reply() 

170 node = iq["pubsub"]["affiliations"]["node"] 

171 if node: 

172 legacy_id = await self.__get_legacy_id(session, node) 

173 with self.xmpp.store.session() as orm: 

174 if not self.xmpp.store.spaces.exists( 

175 orm, session.user_pk, str(legacy_id) 

176 ): 

177 raise XMPPError("item-not-found", f"Space '{legacy_id}' not found") 

178 affiliation = reply["pubsub"]["affiliations"]["affiliation"] 

179 affiliation["node"] = node 

180 affiliation["affiliation"] = "subscriber" 

181 else: 

182 with self.xmpp.store.session() as orm: 

183 spaces = list(self.xmpp.store.spaces.get_all(orm, session.user_pk)) 

184 for space in spaces: 

185 reply["pubsub"]["affiliations"].append( 

186 await self.__make_affiliation(session, space) 

187 ) 

188 

189 reply.send() 

190 

191 @staticmethod 

192 async def __make_affiliation(session: AnySession, space: Space) -> Affiliation: 

193 affiliation = Affiliation() 

194 node = session.bookmarks.space_legacy_id_to_node(space.legacy_id) 

195 affiliation["node"] = node 

196 affiliation["affiliation"] = "subscriber" 

197 return affiliation 

198 

199 async def __owner_get_affiliations(self, iq: Iq) -> None: 

200 raise XMPPError( 

201 "forbidden", "Slidge does not implement managing space affiliations." 

202 ) 

203 

204 async def __owner_get_subscriptions(self, iq: Iq) -> None: 

205 node = iq["pubsub_owner"]["subscriptions"]["node"] 

206 if not node: 

207 raise XMPPError("bad-request", "No node was specified") 

208 

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

210 await self.__get_legacy_id(session, node) 

211 

212 reply = iq.reply(clear=False) 

213 sub = reply["pubsub_owner"]["subscriptions"]["subscription"] 

214 sub["jid"] = session.user_jid 

215 sub["subscription"] = "subscribed" 

216 reply.send() 

217 

218 async def __get_subscriptions(self, iq: Iq) -> None: 

219 _node = iq["pubsub"]["subscriptions"]["node"] 

220 

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

222 

223 reply = iq.reply(clear=False) 

224 subscriptions = reply["pubsub"]["subscriptions"] 

225 if node := iq["pubsub"]["subscriptions"]["node"]: 

226 await self.__get_legacy_id(session, node) 

227 subscription = subscriptions["subscription"] 

228 subscription["node"] = node 

229 subscription["jid"] = session.user_jid 

230 subscription["subscription"] = "subscribed" 

231 else: 

232 with self.xmpp.store.session() as orm: 

233 spaces = list(self.xmpp.store.spaces.get_all(orm, session.user_pk)) 

234 for space in spaces: 

235 subscriptions.append(await self.__make_subscription(session, space)) 

236 

237 reply.send() 

238 

239 @staticmethod 

240 async def __make_subscription(session: AnySession, space: Space) -> Subscription: 

241 subscription = Subscription() 

242 node = session.bookmarks.space_legacy_id_to_node(space.legacy_id) 

243 subscription["node"] = node 

244 subscription["jid"] = session.user_jid 

245 subscription["subscription"] = "subscribed" 

246 return subscription 

247 

248 

249def _is_not_space_image(item_id: str) -> bool: 

250 return item_id not in _SPACE_IMAGES_NAMESPACES 

251 

252 

253_SPACE_IMAGES_NAMESPACES = ( 

254 "urn:xmpp:spaces:avatar:metadata:0", 

255 "urn:xmpp:spaces:banner:metadata:0", 

256) 

257 

258_NODES_HANDLED_ELSEWHERE = ( 

259 AvatarData.namespace, 

260 AvatarMetadata.namespace, 

261 "urn:xmpp:vcard4", 

262 *_SPACE_IMAGES_NAMESPACES, 

263)