Coverage for slidge/room/bookmarks.py: 88%
217 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1import copy
2import logging
3import warnings
4from collections import defaultdict
5from collections.abc import Iterable
6from typing import ClassVar, Unpack
8import sqlalchemy as sa
9import sqlalchemy.orm
10from slixmpp import JID, Iq
11from slixmpp.exceptions import IqError, IqTimeout
12from slixmpp.plugins.xep_0004.stanza.form import Form
13from slixmpp.plugins.xep_0060.stanza import Item as PubsubItem
14from slixmpp.plugins.xep_0060.stanza import Items as PubsubItems
16from ..core.collection import BaseLocalPart
17from ..db.models import Room, Space
18from ..util.types import AnyRoom, AnySession, RoomType
19from ..util.util import derive_wired_class
20from .room import LegacyRoom, _RoomProps
23class LegacyBookmarks[RoomT: AnyRoom = LegacyRoom](BaseLocalPart[Room, RoomT]):
24 """
25 This is instantiated once per :class:`~slidge.BaseSession`
26 """
28 _muc_cls: ClassVar[type[LegacyRoom]] = LegacyRoom # type:ignore[type-abstract]
29 """
30 The concrete :class:`.LegacyRoom` subclass these bookmarks produce.
32 Derived automatically from the generic parameter, e.g.,
33 ``class Bookmarks(LegacyBookmarks[Room])`` produces ``Room`` instances.
34 """
36 def __init_subclass__(cls, **kwargs: object) -> None:
37 super().__init_subclass__(**kwargs)
38 derive_wired_class(cls, LegacyBookmarks, "_muc_cls")
40 def __init__(self, session: AnySession) -> None:
41 super().__init__(
42 session=session,
43 model=Room,
44 child_cls=self._muc_cls, # type:ignore[arg-type]
45 store=session.xmpp.store.rooms,
46 legacy_type="room",
47 )
49 self.user_jid = session.user_jid
50 self._user_nick: str = self._session.user_jid.node
52 self.log = logging.getLogger(f"{self.user_jid.bare}:bookmarks")
53 self.ready = self._session.xmpp.loop.create_future()
54 if not self.xmpp.ROOMS:
55 self.ready.set_result(True)
57 def _create(self, legacy_id: str, **kwargs: Unpack[_RoomProps]) -> RoomT: # type:ignore[override]
58 return super()._create(legacy_id, **kwargs)
60 @property
61 def user_nick(self) -> str:
62 return self._user_nick
64 @user_nick.setter
65 def user_nick(self, nick: str) -> None:
66 self._user_nick = nick
68 async def on_creation_request(self, name: str, legacy_ids: list[str]) -> str:
69 """
70 Called when the user requests the creation of a room via the dedicated
71 :term:`Command`.
73 :param name: Name of the room.
74 :param legacy_ids: list of contacts legacy ids that should be members
75 of the new room.
77 :return: Legacy ID of the newly created room.
78 """
79 raise NotImplementedError
81 async def _fill(self) -> None:
82 rooms: list[LegacyRoom] = []
83 updates = defaultdict[int, dict[str, object]](dict)
84 with self._session.spaces._preload():
85 async for muc in self._batch_fill():
86 rooms.append(muc)
87 if muc._stored._is_loaded("space") and muc._stored.space:
88 space = self._session.spaces.get(muc._stored.space.legacy_id)
89 assert space
90 updates[muc._stored.space.id].update(space._pending_updates)
91 with self._orm() as orm:
92 orm.execute(
93 sa.update(Space), [{"id": k, **v} for k, v in updates.items()]
94 )
95 orm.commit()
96 if self.__privileged:
97 await self._pep_sync(rooms)
99 @property
100 def __privileged(self) -> bool:
101 priv = (
102 self.xmpp.plugin["xep_0356"]
103 .granted_privileges[self._user.jid.server]
104 .iq["http://jabber.org/protocol/pubsub"]
105 == "both"
106 )
107 if not priv:
108 warnings.warn(
109 f"Slidge is unprivileged for {self._user.jid.server}, bookmark sync is not possible",
110 UserWarning,
111 )
112 return priv
114 def schedule_pep_add(self, room: LegacyRoom, why: str) -> None:
115 if self.is_filling:
116 return
117 self._session.create_task(
118 self._session.rooms._pep_add(room),
119 f"update bookmark of {room} after {why}",
120 )
122 async def _pep_add(self, room: LegacyRoom) -> None:
123 if not self.__privileged:
124 return
125 old = await self.__pep_get(room)
126 new = await self.__pep_make(room)
127 change = self.__pep_update(old, new)
128 if change:
129 await self.__pep_set(new)
130 else:
131 self.log.debug("User bookmark for %s, nothing to do", room)
133 async def _pep_del(self, room: LegacyRoom) -> None:
134 """Remove the Room from the user's XMPP bookmarks (:xep:`0402`).
136 Does not actually leave the room on the legacy network, nor apply any other
137 side-effect.
139 Requires that slidge has the IQ privileged set correctly on the XMPP server.
140 """
141 if not self.__privileged:
142 return
143 iq = Iq(stype="set", sfrom=self.user_jid, sto=self.user_jid)
144 iq["pubsub"]["retract"]["node"] = self.xmpp.plugin["xep_0402"].stanza.NS
145 iq["pubsub"]["retract"]["notify"] = "true"
146 iq["pubsub"]["retract"]["item"]["id"] = str(room.jid)
147 try:
148 await self.xmpp.plugin["xep_0356"].send_privileged_iq(iq)
149 except IqError as e:
150 warnings.warn(
151 f"Something went wrong while trying to remove a bookmark: {e}"
152 )
154 async def _pep_sync(self, rooms: Iterable[LegacyRoom] = ()) -> None:
155 if not self.__privileged:
156 return
157 if not self._session.preferences.get("bookmarks_push", True):
158 return
159 old_items = await self.__pep_get_all()
161 if old_items is None:
162 self.log.debug("Could not get bookmarks")
163 old_items_map: dict[str, PubsubItem] = {}
164 else:
165 old_items_map = {
166 item["id"]: item # type:ignore[misc]
167 for item in old_items
168 }
170 for room in rooms or self:
171 if room.type != RoomType.GROUP and not self._session.preferences.get(
172 "bookmarks_push_public"
173 ):
174 continue
175 old = old_items_map.get(room.jid.bare)
176 new = await self.__pep_make(room)
177 change = self.__pep_update(old, new)
178 if change:
179 self._session.log.debug("bookmark change for %s", room)
180 await self.__pep_set(new)
181 self._session.log.debug("bookmark updated")
183 async def __pep_get_all(self) -> PubsubItems | None:
184 iq = Iq(stype="get", sfrom=self.user_jid, sto=self.user_jid)
185 iq["pubsub"]["items"]["node"] = self.xmpp.plugin["xep_0402"].stanza.NS
186 return await self.__pep_fetch(iq)
188 async def __pep_get(self, room: LegacyRoom) -> PubsubItem | None:
189 item = PubsubItem()
190 item["id"] = room.jid
191 iq = Iq(stype="get", sfrom=self.user_jid, sto=self.user_jid)
192 iq["pubsub"]["items"]["node"] = self.xmpp.plugin["xep_0402"].stanza.NS
193 iq["pubsub"]["items"].append(item)
194 items = await self.__pep_fetch(iq)
195 if not items or len(items) == 0:
196 return None
197 return items["item"] # type:ignore[no-any-return]
199 async def __pep_fetch(self, iq: Iq) -> PubsubItems | None:
200 try:
201 ans = await self.xmpp.plugin["xep_0356"].send_privileged_iq(iq)
202 except IqTimeout:
203 warnings.warn(f"Cannot fetch bookmark for {self.user_jid}: timeout")
204 except IqError as exc:
205 warnings.warn(f"Cannot fetch bookmark for {self.user_jid}: {exc}")
206 except PermissionError:
207 warnings.warn(
208 f"IQ privileges (XEP0356) not granted for {self.user_jid}, we cannot fetch the user bookmarks"
209 )
210 else:
211 return ans["pubsub"]["items"] # type:ignore[no-any-return]
212 return None
214 async def __pep_set(self, item: PubsubItem) -> None:
215 iq = Iq(stype="set", sfrom=self.user_jid, sto=self.user_jid)
216 iq["pubsub"]["publish"]["node"] = self.xmpp.plugin["xep_0402"].stanza.NS
217 iq["pubsub"]["publish"].append(item)
218 iq["pubsub"]["publish_options"] = _bookmarks_form()
219 try:
220 await self.xmpp.plugin["xep_0356"].send_privileged_iq(iq)
221 except (PermissionError, IqError) as e:
222 warnings.warn(
223 f"Something went wrong while trying to set the bookmarks: {e}"
224 )
225 self._session.send_gateway_invite(
226 item["id"],
227 reason="This group could not be added automatically for you, most "
228 "likely because this gateway is not configured as a privileged entity. "
229 "Contact your administrator.",
230 )
232 async def __pep_make(self, room: LegacyRoom) -> PubsubItem:
233 item = PubsubItem()
234 item["id"] = room.jid.bare
236 conf = item["conference"]
238 conf["nick"] = room.participants.get_user().jid.resource
240 if room.type == RoomType.GROUP or self._session.preferences.get(
241 "autojoin_public", False
242 ):
243 conf["autojoin"] = True
245 ext = None
247 if room.notify_settings is not None:
248 ext = ext or conf["extensions"]
249 ext["notify"].configure(room.notify_settings)
251 if room.pinned:
252 ext = ext or conf["extensions"]
253 ext.enable("pinned")
255 return item
257 def __pep_update(self, old_item: PubsubItem | None, new_item: PubsubItem) -> bool:
258 if old_item is None:
259 return True
261 old = old_item["conference"]
262 new = new_item["conference"]
263 change: bool = old["nick"] != new["nick"]
265 if old._get_attr("autojoin"):
266 new["autojoin"] = old["autojoin"]
267 elif new._get_attr("autojoin"):
268 change = True
270 if "extensions" not in old_item and "extensions" in new_item:
271 return change
273 old_ext = old["extensions"]
274 new_ext = new["extensions"]
276 if "pinned" in old_ext:
277 new_ext["pinned"] = old_ext["pinned"]
278 del old_ext["pinned"]
279 elif "pinned" in new_ext:
280 change = True
282 if "notify" in old_ext:
283 new_ext["notify"] = old_ext["notify"]
284 del old_ext["notify"]
285 elif "notify" in new_ext:
286 change = True
288 if change:
289 for child in old_ext.xml:
290 new_ext.xml.append(copy.deepcopy(child))
292 return change
294 async def remove(
295 self,
296 muc: AnyRoom,
297 reason: str = "You left this group from the official client.",
298 kick: bool = True,
299 ) -> None:
300 """
301 Delete everything about a specific group.
303 This should be called when the user leaves the group from the official
304 app.
306 :param muc: The Room to remove.
307 :param reason: Optionally, a reason why this group was removed.
308 :param kick: Whether the user should be kicked from this group. Set this
309 to False in case you do this somewhere else in your code, eg, on
310 receiving the confirmation that the group was deleted.
311 """
312 if kick:
313 user_participant = muc.participants.get_user()
314 user_participant.kick(reason)
315 with self.xmpp.store.session() as orm:
316 orm.add(muc._stored)
317 orm.refresh(muc._stored)
318 orm.delete(muc._stored)
319 orm.commit()
321 def _batch_load(self) -> sa.Select[Room]:
322 return (
323 super()
324 ._batch_load()
325 .options(
326 sa.orm.undefer_group("subject"),
327 sa.orm.undefer_group("disco"),
328 sa.orm.undefer_group("bookmarks"),
329 sa.orm.joinedload(Room.space).joinedload(Space.avatar),
330 sa.orm.joinedload(Room.space).joinedload(Space.banner),
331 )
332 )
334 def _shutdown(self, orm: sa.orm.Session) -> None:
335 user_part_jid = JID(self._session.user.jid)
336 for (
337 user_resource,
338 connected_resources,
339 muc_localpart,
340 ) in self.xmpp.store.participants.get_user_participants(
341 orm, self._session.user_pk
342 ):
343 muc_jid = JID(f"{muc_localpart}@{self.xmpp.boundjid.bare}")
344 for connected_resource in connected_resources:
345 user_full_jid = JID(user_part_jid)
346 user_full_jid.resource = connected_resource
347 muc_jid.resource = user_resource
348 presence = self.xmpp.make_presence(
349 pfrom=muc_jid, pto=user_full_jid, ptype="unavailable"
350 )
351 presence["muc"]["affiliation"] = "none"
352 presence["muc"]["role"] = "none"
353 presence["muc"]["status_codes"] = {110, 332}
354 presence.send()
357def _bookmarks_form() -> Form:
358 form = Form()
359 form["type"] = "submit"
360 form.add_field(
361 "FORM_TYPE",
362 value="http://jabber.org/protocol/pubsub#publish-options",
363 ftype="hidden",
364 )
365 form.add_field("pubsub#persist_items", value="1")
366 form.add_field("pubsub#max_items", value="max")
367 form.add_field("pubsub#send_last_published_item", value="never")
368 form.add_field("pubsub#access_model", value="whitelist")
369 return form