Coverage for slidge/core/dispatcher/disco.py: 91%

1import logging 

2from typing import TYPE_CHECKING, Any 

3 

4import sqlalchemy as sa 

5import sqlalchemy.orm 

6from slixmpp import JID 

7from slixmpp.exceptions import XMPPError 

8from slixmpp.plugins.xep_0004.stanza import Form 

9from slixmpp.plugins.xep_0030.stanza.info import DiscoInfo 

10from slixmpp.plugins.xep_0030.stanza.items import DiscoItems 

11from slixmpp.types import OptJid 

12 

13from ...db.models import Room, Space 

14from ...util.types import AnySession 

15from .util import DispatcherMixin 

16 

17if TYPE_CHECKING: 

18 from slidge.util.types import AnyGateway 

19 

20 

21class DiscoMixin(DispatcherMixin): 

22 __slots__: list[str] = [] 

23 

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

25 super().__init__(xmpp) 

26 

27 xmpp.plugin["xep_0030"].set_node_handler( 

28 "get_info", 

29 jid=None, 

30 node=None, 

31 handler=self.get_info, 

32 ) 

33 

34 xmpp.plugin["xep_0030"].set_node_handler( 

35 "get_items", 

36 jid=None, 

37 node=None, 

38 handler=self.get_items, 

39 ) 

40 

41 async def get_info( 

42 self, 

43 jid: OptJid, 

44 node: str | None, 

45 ifrom: OptJid, 

46 data: Any, # noqa:ANN401 

47 ) -> DiscoInfo | None: 

48 # TODO: OMG the control flow in this method 🤮 

49 if ifrom == self.xmpp.boundjid.bare or ( 

50 jid in (self.xmpp.boundjid.bare, None) 

51 and (not node or node.startswith("http://slixmpp.com/ver/")) 

52 ): 

53 return self.xmpp.plugin["xep_0030"].static.get_info(jid, node, ifrom, data) 

54 

55 if ifrom is None: 

56 raise XMPPError("subscription-required") 

57 

58 assert jid is not None 

59 session = await self._get_session_from_jid(jid=ifrom) 

60 

61 if not jid.username: 

62 assert node is not None # 🤮 

63 return await self.__spaces_info(session, node) 

64 

65 log.debug("Looking for entity: %s", jid) 

66 

67 entity = await session.get_contact_or_group_or_participant(jid) 

68 

69 if entity is None: 

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

71 

72 return await entity.get_disco_info(jid, node) 

73 

74 async def __spaces_info(self, session: AnySession, node: str) -> DiscoInfo: 

75 legacy_id = session.spaces.node_to_legacy_id(node) 

76 space = await session.spaces.get_or_create(legacy_id) 

77 form = Form() 

78 form["type"] = "result" 

79 form.add_field( 

80 var="FORM_TYPE", 

81 ftype="hidden", 

82 value="http://jabber.org/protocol/pubsub#meta-data", 

83 ) 

84 form.add_field( 

85 var="pubsub#type", 

86 value="urn:xmpp:spaces:0", 

87 ) 

88 form.add_field( 

89 var="pubsub#creator", 

90 type="jid-single", 

91 value=self.xmpp.boundjid.bare 

92 if space.creator is None 

93 else space.creator.jid.bare, 

94 ) 

95 form.add_field(var="pubsub#title", value=space.name) 

96 form.add_field( 

97 var="pubsub#owner", 

98 type="jid-multi", 

99 value=[owner.jid.bare for owner in space.owners], 

100 ) 

101 if space.member_count is not None: 

102 form.add_field(var="pubsub#num_subscribers", value=str(space.member_count)) 

103 if space.description is not None: 

104 form.add_field(var="pubsub#description", value=space.description) 

105 info = DiscoInfo() 

106 info.add_identity(category="pubsub", itype="leaf") 

107 info.add_feature("http://jabber.org/protocol/pubsub") 

108 for feat in ( 

109 "meta-data", 

110 "item-ids", 

111 "manage-subscriptions", 

112 "modify-affiliations", 

113 "outcast-affiliation", 

114 "retract-items", 

115 "retrieve-affiliations", 

116 "retrieve-items", 

117 "retrieve-subscriptions", 

118 "subscribe", 

119 "subscription-notifications", 

120 ): 

121 info.add_feature(f"http://jabber.org/protocol/pubsub#{feat}") 

122 info.append(form) 

123 return info 

124 

125 async def get_items( 

126 self, 

127 jid: OptJid, 

128 node: str | None, 

129 ifrom: OptJid, 

130 data: Any, # noqa:ANN401 

131 ) -> DiscoItems: 

132 if ifrom is None: 

133 raise XMPPError("bad-request") 

134 

135 assert ifrom is not None 

136 session = await self._get_session_from_jid(ifrom) 

137 

138 if jid == self.xmpp.boundjid.bare or not jid: 

139 if node: 

140 return DiscoItems() 

141 else: 

142 return await self.__list_rooms(session, data) 

143 

144 entity = await session.get_contact_or_group_or_participant(jid) 

145 

146 if entity is None: 

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

148 

149 return await entity.get_disco_items(node) 

150 

151 async def __list_rooms( 

152 self, 

153 session: "AnySession", 

154 data: dict, # type:ignore[type-arg] 

155 ) -> DiscoItems: 

156 try: 

157 included_types = data["disco_items"]["filter"]["included_types"] 

158 except KeyError: 

159 included_types = [] 

160 

161 d = DiscoItems() 

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

163 if included_types == ["urn:xmpp:spaces:0"]: 

164 if not self.xmpp.SPACES: 

165 raise XMPPError( 

166 "feature-not-implemented", 

167 "This gateway does not support XEP-0503", 

168 ) 

169 await self.__spaces(session, d) 

170 elif not included_types: 

171 self.__rooms(orm, session, d) 

172 else: 

173 raise XMPPError("item-not-found", f"No items for {included_types}") 

174 

175 return d 

176 

177 async def __spaces(self, session: AnySession, d: DiscoItems) -> None: 

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

179 for space in orm.execute( 

180 sa.select(Space) 

181 .options(sa.orm.load_only(Space.legacy_id, Space.name)) 

182 .filter_by(user=session.user) 

183 ).scalars(): 

184 d.add_item( 

185 JID(self.xmpp.boundjid.bare), 

186 name=space.name, 

187 node=session.spaces.legacy_id_to_node(space.legacy_id), 

188 ) 

189 

190 @staticmethod 

191 def __rooms( 

192 orm: sqlalchemy.orm.Session, session: AnySession, d: DiscoItems 

193 ) -> None: 

194 for room in orm.execute( 

195 sa.select(Room) 

196 .options(sa.orm.load_only(Room.jid_localpart, Room.name)) 

197 .filter_by(user=session.user) 

198 .order_by(Room.name) 

199 ).scalars(): 

200 d.add_item(room.jid, name=room.name) 

201 

202 

203log = logging.getLogger(__name__)