Coverage for slidge/core/pubsub.py: 87%
281 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1from __future__ import annotations
3import logging
4from collections.abc import Iterable
5from copy import copy
6from pathlib import Path
7from typing import TYPE_CHECKING, ClassVar
9from slixmpp import (
10 JID,
11 CoroutineCallback,
12 ElementBase,
13 Iq,
14 Presence,
15 StanzaPath,
16 register_stanza_plugin,
17)
18from slixmpp.exceptions import IqError, IqTimeout, XMPPError
19from slixmpp.plugins.base import BasePlugin, register_plugin
20from slixmpp.plugins.xep_0060.stanza import Event, EventItem, EventItems, Item, Items
21from slixmpp.plugins.xep_0084.stanza import Data as AvatarData
22from slixmpp.plugins.xep_0084.stanza import Info
23from slixmpp.plugins.xep_0084.stanza import MetaData as AvatarMetadata
24from slixmpp.plugins.xep_0172.stanza import UserNick
25from slixmpp.plugins.xep_0292.stanza import VCard4
26from slixmpp.plugins.xep_0402.stanza import Conference
27from slixmpp.types import JidStr, OptJidStr
28from sqlalchemy.orm.exc import DetachedInstanceError
30from ..contact import LegacyContact
31from ..db.avatar import CachedAvatar
32from ..db.models import GatewayUser, Space
33from ..util.lock import NamedLockMixin
34from ..util.types import AnySession
35from .dispatcher.util import exceptions_to_xmpp_errors
37if TYPE_CHECKING:
38 from ..core.gateway import BaseGateway
40VCARD4_NAMESPACE = "urn:xmpp:vcard4"
43class PepAvatar:
44 def __init__(self) -> None:
45 self.metadata: AvatarMetadata | None = None
46 self.id: str | None = None
47 self.url: str | None = None
48 self._avatar_data_path: Path | None = None
50 @staticmethod
51 def from_cache(cached_avatar: CachedAvatar | None) -> PepAvatar:
52 a = PepAvatar()
53 if cached_avatar:
54 a.set_avatar_from_cache(cached_avatar)
55 return a
57 @property
58 def data(self) -> AvatarData | None:
59 if self._avatar_data_path is None:
60 return None
61 data = AvatarData()
62 data.set_value(self._avatar_data_path.read_bytes())
63 return data
65 def set_avatar_from_cache(self, cached_avatar: CachedAvatar) -> None:
66 metadata = AvatarMetadata()
67 self.id = cached_avatar.hash
68 if cached_avatar.path.exists():
69 metadata.add_info(
70 id=cached_avatar.hash,
71 itype="image/png",
72 ibytes=cached_avatar.path.stat().st_size,
73 height=str(cached_avatar.height),
74 width=str(cached_avatar.width),
75 )
76 if cached_avatar._stored.http_url:
77 self.url = cached_avatar._stored.http_url
78 metadata.add_info(
79 id=cached_avatar._stored.http_id,
80 itype=cached_avatar._stored.http_type,
81 ibytes=str(cached_avatar._stored.http_bytes),
82 url=cached_avatar._stored.http_url,
83 height=str(cached_avatar._stored.http_height),
84 width=str(cached_avatar._stored.http_width),
85 )
86 self.metadata = metadata
87 if cached_avatar.hash:
88 self._avatar_data_path = cached_avatar.path
90 @property
91 def http_metadata(self) -> Info | None:
92 if not self.metadata:
93 return None
94 items = self.metadata["items"]
95 if len(items) == 2:
96 return items[-1] # type:ignore[no-any-return]
97 return None
100class PubSubComponent(NamedLockMixin, BasePlugin):
101 xmpp: BaseGateway
103 name = "pubsub"
104 description = "Pubsub component"
105 dependencies: ClassVar[set[str]] = {
106 "xep_0030",
107 "xep_0060",
108 "xep_0115",
109 "xep_0163",
110 }
111 default_config: ClassVar[dict[str, str | None]] = {"component_name": None}
112 component_name: str
114 def __init__(self, *a: object, **kw: object) -> None:
115 super().__init__(*a, **kw)
116 # slixmpp usually does this via self.xmpp["xep_0163"].register_pep(),
117 # in the session_bind hook of plugins. However, for components,
118 # session_bind is never called.
119 register_stanza_plugin(EventItem, UserNick)
120 register_stanza_plugin(EventItem, Conference)
122 def plugin_init(self) -> None:
123 self.xmpp.register_handler(
124 CoroutineCallback(
125 "pubsub_get_avatar_data",
126 StanzaPath(f"iq@type=get/pubsub/items@node={AvatarData.namespace}"),
127 self._get_avatar_data, # type:ignore[arg-type] # ty:ignore
128 )
129 )
130 self.xmpp.register_handler(
131 CoroutineCallback(
132 "pubsub_get_avatar_metadata",
133 StanzaPath(f"iq@type=get/pubsub/items@node={AvatarMetadata.namespace}"),
134 self._get_avatar_metadata, # type:ignore[arg-type] # ty:ignore
135 )
136 )
137 self.xmpp.register_handler(
138 CoroutineCallback(
139 "pubsub_get_vcard",
140 StanzaPath(f"iq@type=get/pubsub/items@node={VCARD4_NAMESPACE}"),
141 self._get_vcard, # type:ignore[arg-type] # ty:ignore
142 )
143 )
145 disco = self.xmpp.plugin["xep_0030"]
146 disco.add_identity("pubsub", "pep", self.component_name)
147 disco.add_identity("account", "registered", self.component_name)
148 disco.add_feature("http://jabber.org/protocol/pubsub#event")
149 disco.add_feature("http://jabber.org/protocol/pubsub#retrieve-items")
150 disco.add_feature("http://jabber.org/protocol/pubsub#persistent-items")
152 async def __get_features(self, from_: JID, ver_string: str | None) -> list[str]:
153 if ver_string:
154 info = await self.xmpp.plugin["xep_0115"].get_caps(from_, ver_string)
155 else:
156 info = None
157 if info is None:
158 async with self.lock(from_):
159 try:
160 iq_or_info = await self.xmpp.plugin["xep_0030"].get_info(
161 from_, cached=True
162 )
163 except (IqError, IqTimeout):
164 log.debug("Could not get disco#info of %s, ignoring", from_)
165 return []
166 if isinstance(iq_or_info, Iq):
167 info = iq_or_info["disco_info"]
168 else:
169 info = iq_or_info
170 return info["features"] # type:ignore[no-any-return]
172 async def on_get_online(self, session: AnySession, presence: Presence) -> None:
173 pfrom = presence.get_from()
174 await self._broadcast_component(pfrom)
175 features = await self.__get_features(pfrom, presence["caps"]["ver"])
176 for contact in session.contacts.known_contacts(only_friends=True).values():
177 await self.xmpp.pubsub._broadcast_contact(pfrom, contact, features)
179 async def _broadcast_component(self, user_full_jid: JID) -> None:
180 pep_nick = get_user_nick(self.xmpp.COMPONENT_NAME)
181 await self.__broadcast(
182 data=pep_nick, from_=self.xmpp.boundjid.bare, to=user_full_jid
183 )
184 if self.xmpp.avatar is not None:
185 item = PepAvatar.from_cache(self.xmpp.avatar)
186 assert item.metadata is not None
187 await self.__broadcast(
188 data=item.metadata,
189 from_=self.xmpp.boundjid.bare,
190 to=user_full_jid,
191 id=item.metadata["info"]["id"],
192 )
194 async def _broadcast_contact(
195 self,
196 user_full_jid: JID,
197 contact: LegacyContact,
198 features: list[str],
199 ) -> None:
200 if AvatarMetadata.namespace + "+notify" in features:
201 pep_avatar = PepAvatar.from_cache(contact.get_cached_avatar())
202 if pep_avatar.metadata is not None:
203 await self.__broadcast(
204 data=pep_avatar.metadata,
205 from_=contact.jid.bare,
206 to=user_full_jid,
207 id=pep_avatar.metadata["info"]["id"],
208 )
209 if UserNick.namespace + "+notify" in features and contact.name is not None:
210 await self.__broadcast(
211 data=get_user_nick(contact.name),
212 from_=contact.jid.bare,
213 to=user_full_jid,
214 )
215 if VCARD4_NAMESPACE + "+notify" in features:
216 vcard = await contact.get_vcard()
217 if vcard is not None:
218 await self.broadcast_vcard_event(contact.jid.bare, user_full_jid, vcard)
220 async def broadcast_vcard_event(
221 self, from_: JID | str, to: JID, vcard: VCard4
222 ) -> None:
223 item = Item()
224 item.namespace = VCARD4_NAMESPACE
225 item["id"] = "current"
226 # vcard: VCard4 = await self.xmpp["xep_0292_provider"].get_vcard(from_, to)
227 # The vcard content should NOT be in this event according to the spec:
228 # https://xmpp.org/extensions/xep-0292.html#sect-idm45669698174224
229 # but movim expects it to be here, and I guess it does not hurt
231 log.debug("Broadcast vcard4 event: %s", vcard)
232 await self.__broadcast(
233 data=vcard,
234 from_=JID(from_).bare,
235 to=to,
236 id="current",
237 node=VCARD4_NAMESPACE,
238 )
240 async def __get_contact(self, stanza: Iq | Presence) -> LegacyContact:
241 session = self.xmpp.get_session_from_stanza(stanza)
242 return await session.contacts.get_or_create_by_jid(stanza.get_to())
244 async def _get_authorized_avatar(self, stanza: Iq | Presence) -> PepAvatar:
245 if stanza.get_to() == self.xmpp.boundjid.bare:
246 return PepAvatar.from_cache(self.xmpp.avatar)
247 contact = await self.__get_contact(stanza)
248 item = PepAvatar()
249 cached_avatar = contact.get_cached_avatar()
250 if cached_avatar is not None:
251 item.set_avatar_from_cache(cached_avatar)
252 return item
254 async def _get_authorized_nick(self, stanza: Iq | Presence) -> UserNick:
255 if stanza.get_to() == self.xmpp.boundjid.bare:
256 return get_user_nick(self.xmpp.COMPONENT_NAME)
257 contact = await self.__get_contact(stanza)
258 return get_user_nick(contact.name)
260 def __reply_with(
261 self, iq: Iq, content: AvatarData | AvatarMetadata | None, item_id: str | None
262 ) -> None:
263 requested_items = iq["pubsub"]["items"]
265 if len(requested_items) == 0:
266 self._reply_with_payload(iq, content, item_id)
267 else:
268 for item in requested_items:
269 if item["id"] == item_id:
270 self._reply_with_payload(iq, content, item_id)
271 return
272 raise XMPPError("item-not-found")
274 @exceptions_to_xmpp_errors
275 async def _get_avatar_data(self, iq: Iq) -> None:
276 pep_avatar = await self._get_authorized_avatar(iq)
277 self.__reply_with(iq, pep_avatar.data, pep_avatar.id)
279 @exceptions_to_xmpp_errors
280 async def _get_avatar_metadata(self, iq: Iq) -> None:
281 pep_avatar = await self._get_authorized_avatar(iq)
282 self.__reply_with(iq, pep_avatar.metadata, pep_avatar.id)
284 @exceptions_to_xmpp_errors
285 async def _get_vcard(self, iq: Iq) -> None:
286 # this is not the proper way that clients should retrieve VCards, but
287 # gajim does it this way.
288 # https://xmpp.org/extensions/xep-0292.html#sect-idm45669698174224
289 session = self.xmpp.get_session_from_stanza(iq)
290 contact = await session.contacts.get_or_create_by_jid(iq.get_to())
291 vcard = await contact.get_vcard()
292 if vcard is None:
293 raise XMPPError("item-not-found")
294 self._reply_with_payload(iq, vcard, "current", VCARD4_NAMESPACE)
296 @staticmethod
297 def _reply_with_payload(
298 iq: Iq,
299 payload: AvatarMetadata | AvatarData | VCard4 | None,
300 id_: str | None,
301 namespace: str | None = None,
302 ) -> None:
303 result = iq.reply()
304 item = Item()
305 if payload:
306 item.set_payload(payload.xml)
307 item["id"] = id_
308 result["pubsub"]["items"]["node"] = (
309 namespace if namespace else payload.namespace
310 )
311 result["pubsub"]["items"].append(item)
312 result.send()
314 async def __broadcast(
315 self,
316 data: ElementBase | None,
317 from_: JidStr,
318 to: OptJidStr = None,
319 items: EventItems | None = None,
320 **kwargs: object,
321 ) -> None:
322 from_ = JID(from_)
323 if from_ != self.xmpp.boundjid.bare and to is not None:
324 to = JID(to)
325 session = self.xmpp.get_session_from_jid(to)
326 if session is None:
327 return
328 await session.ready
330 item = EventItem()
331 if data:
332 item.set_payload(data.xml)
333 for k, v in kwargs.items():
334 item[k] = v
336 if items is None:
337 items = EventItems()
338 items.append(item)
339 assert data is not None
340 items["node"] = kwargs.get("node") or data.namespace
342 event = Event()
343 event.append(items)
345 msg = self.xmpp.Message()
346 msg.set_type("headline")
347 msg.set_from(from_)
348 msg.append(event)
350 if to is None:
351 with self.xmpp.store.session() as orm:
352 for u in orm.query(GatewayUser).all():
353 new_msg = copy(msg)
354 new_msg.set_to(u.jid.bare)
355 new_msg.send()
356 else:
357 msg.set_to(to)
358 msg.send()
360 async def broadcast_avatar(
361 self, from_: JidStr, to: JidStr, cached_avatar: CachedAvatar | None
362 ) -> None:
363 if cached_avatar is None:
364 await self.__broadcast(AvatarMetadata(), from_, to)
365 else:
366 pep_avatar = PepAvatar()
367 pep_avatar.set_avatar_from_cache(cached_avatar)
368 assert pep_avatar.metadata is not None
369 await self.__broadcast(pep_avatar.metadata, from_, to, id=pep_avatar.id)
371 def broadcast_nick(
372 self,
373 user_jid: JID,
374 jid: JidStr,
375 nick: str | None = None,
376 ) -> None:
377 jid = JID(jid)
378 nickname = get_user_nick(nick)
379 log.debug("New nickname: %s", nickname)
380 self.xmpp.create_task(
381 self.__broadcast(nickname, jid, user_jid.bare),
382 name=f"broadcast nick of {jid}",
383 )
385 def broadcast_space(
386 self, session: AnySession, space: Space, node: str, item_ids: Iterable[str] = ()
387 ) -> None:
388 items = EventItems()
389 items["node"] = node
390 self.set_space_items(space, items, item_ids)
391 session.create_task(
392 self.__broadcast(
393 None,
394 self.xmpp.boundjid.bare,
395 session.user_jid,
396 items=items,
397 ),
398 name=f"broadcast space {node!r}",
399 )
401 def set_space_items(
402 self,
403 space: Space,
404 collection: Items | EventItems,
405 item_ids: Iterable[str] = (),
406 ) -> None:
407 item_cls = Item if isinstance(collection, Items) else EventItem
409 try:
410 rooms = space.rooms
411 except DetachedInstanceError:
412 # if the caller does not care about rooms, it should ensure this
413 # has been loaded
414 rooms = []
416 for room in rooms:
417 if item_ids and str(room.jid) not in item_ids:
418 continue
419 item = item_cls()
420 item["id"] = room.jid
421 item.enable("conference")
422 if room.name:
423 item["conference"]["name"] = room.name
424 collection.append(item)
426 for attr, avatar in [("avatar", space.avatar), ("banner", space.banner)]:
427 ns = f"urn:xmpp:spaces:{attr}:metadata:0"
428 if avatar is None or avatar.http_id is None:
429 continue
430 if item_ids and ns not in item_ids:
431 continue
432 item = item_cls()
433 item["id"] = ns
434 meta = self.xmpp.plugin["xep_0084"].stanza.MetaData()
435 meta.add_info(
436 id=avatar.http_id,
437 itype=avatar.http_type,
438 ibytes=str(avatar.http_bytes),
439 height=str(avatar.http_height),
440 width=str(avatar.http_width),
441 url=avatar.http_url,
442 )
443 item.append(meta)
444 collection.append(item)
446 def broadcast_space_metadata(
447 self, session: AnySession, space: Space, node: str
448 ) -> None:
449 msg = self.xmpp.make_message(
450 mtype="headline", mto=session.user_jid, mfrom=self.xmpp.boundjid.bare
451 )
452 msg["pubsub_event"]["configuration"]["node"] = node
454 form = msg["pubsub_event"]["configuration"]["form"]
455 form["type"] = "result"
457 form.add_field(
458 var="FORM_TYPE",
459 ftype="hidden",
460 value="http://jabber.org/protocol/pubsub#node_config",
461 )
462 form.add_field(var="pubsub#title", value=space.name)
464 # MUSTs
465 form.add_field(var="pubsub#type", value="urn:xmpp:spaces:0")
466 form.add_field(var="pubsub#notify_retract", value="true")
467 form.add_field(var="pubsub#persist_items", value="true")
468 form.add_field(var="pubsub#purge_offline", value="false")
469 # SHOULDs
470 form.add_field(var="pubsub#notify_sub", value="true")
471 form.add_field(var="pubsub#notify_config", value="true")
472 form.add_field(var="pubsub#notify_delete", value="true")
473 form.add_field(var="pubsub#publish_model", value="publishers")
474 # MUST (private spaces)
475 form.add_field(var="pubsub#access_model", value="authorize")
477 msg.send()
480def get_user_nick(nick: str | None = None) -> UserNick:
481 user_nick = UserNick()
482 if nick is not None:
483 user_nick["nick"] = nick
484 return user_nick
487log = logging.getLogger(__name__)
488register_plugin(PubSubComponent)