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

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 ifrom = iq.get_from() 

48 if not ifrom.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 [ 

77 await session.bookmarks.jid_local_part_to_legacy_id( 

78 x.removesuffix(f"@{self.xmpp.boundjid.bare}") 

79 ) 

80 for x in room_item_ids 

81 ] 

82 if room_item_ids 

83 else None 

84 ) 

85 

86 space = await session.bookmarks.get_updated_space( 

87 legacy_id, room_legacy_id_filter=room_legacy_ids 

88 ) 

89 if space is None: 

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

91 

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

93 orm.add(space) 

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

95 raise XMPPError( 

96 "item-not-found", 

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

98 ) 

99 

100 reply = iq.reply() 

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

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

103 

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

105 if len(item_ids) == 1: 

106 raise XMPPError( 

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

108 ) 

109 else: 

110 raise XMPPError( 

111 "item-not-found", 

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

113 ) 

114 

115 reply.send() 

116 

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

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

119 

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

121 try: 

122 legacy_id = await session.bookmarks.space_node_to_legacy_id(node) 

123 except Exception as e: # noqa: BLE001 

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

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

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

127 raise XMPPError( 

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

129 ) 

130 return legacy_id 

131 

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

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

134 

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

136 await self.__get_legacy_id(session, node) 

137 

138 reply = iq.reply(clear=True) 

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

140 sub["node"] = node 

141 sub["jid"] = session.user_jid 

142 sub["subscription"] = "subscribed" 

143 reply.send() 

144 

145 @exceptions_to_xmpp_errors 

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

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

148 

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

150 if jid != session.user_jid.bare: 

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

152 

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

154 legacy_id = await session.bookmarks.space_node_to_legacy_id(node) 

155 

156 await session.on_leave_space(legacy_id) 

157 reply = iq.reply(clear=True) 

158 

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

160 sub["node"] = node 

161 sub["jid"] = session.user_jid 

162 sub["subscription"] = "none" 

163 reply.send() 

164 

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

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

167 

168 reply = iq.reply() 

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

170 if node: 

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

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

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

174 orm, session.user_pk, str(legacy_id) 

175 ): 

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

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

178 affiliation["node"] = node 

179 affiliation["affiliation"] = "subscriber" 

180 else: 

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

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

183 for space in spaces: 

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

185 await self.__make_affiliation(session, space) 

186 ) 

187 

188 reply.send() 

189 

190 @staticmethod 

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

192 affiliation = Affiliation() 

193 node = await session.bookmarks.space_legacy_id_to_node(space.legacy_id) 

194 affiliation["node"] = node 

195 affiliation["affiliation"] = "subscriber" 

196 return affiliation 

197 

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

199 raise XMPPError( 

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

201 ) 

202 

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

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

205 if not node: 

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

207 

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

209 await self.__get_legacy_id(session, node) 

210 

211 reply = iq.reply(clear=False) 

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

213 sub["jid"] = session.user_jid 

214 sub["subscription"] = "subscribed" 

215 reply.send() 

216 

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

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

219 

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

221 

222 reply = iq.reply(clear=False) 

223 subscriptions = reply["pubsub"]["subscriptions"] 

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

225 await self.__get_legacy_id(session, node) 

226 subscription = subscriptions["subscription"] 

227 subscription["node"] = node 

228 subscription["jid"] = session.user_jid 

229 subscription["subscription"] = "subscribed" 

230 else: 

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

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

233 for space in spaces: 

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

235 

236 reply.send() 

237 

238 @staticmethod 

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

240 subscription = Subscription() 

241 node = await session.bookmarks.space_legacy_id_to_node(space.legacy_id) 

242 subscription["node"] = node 

243 subscription["jid"] = session.user_jid 

244 subscription["subscription"] = "subscribed" 

245 return subscription 

246 

247 

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

249 return item_id not in _SPACE_IMAGES_NAMESPACES 

250 

251 

252_SPACE_IMAGES_NAMESPACES = ( 

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

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

255) 

256 

257_NODES_HANDLED_ELSEWHERE = ( 

258 AvatarData.namespace, 

259 AvatarMetadata.namespace, 

260 "urn:xmpp:vcard4", 

261)