Coverage for slidge/core/dispatcher/caps.py: 93%

1from __future__ import annotations 

2 

3import logging 

4from typing import TYPE_CHECKING 

5 

6from slixmpp import Presence 

7from slixmpp.exceptions import XMPPError 

8from slixmpp.xmlstream import StanzaBase 

9 

10from ...util.types import ClientType 

11from .util import DispatcherMixin 

12 

13if TYPE_CHECKING: 

14 from slidge.contact import LegacyContact 

15 from slidge.util.types import AnyGateway 

16 

17 

18class CapsMixin(DispatcherMixin): 

19 __slots__: list[str] = [] 

20 __contact_vers: dict[ClientType, str] 

21 __gateway_ver: str | None 

22 

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

24 super().__init__(xmpp) 

25 xmpp.del_filter("out", xmpp.plugin["xep_0115"]._filter_add_caps) # type:ignore[arg-type] 

26 xmpp.add_filter("out", self._filter_add_caps) 

27 self.__contact_vers: dict[ClientType, str] = {} # type:ignore[misc] 

28 self.__gateway_ver: str | None = None # type:ignore[misc] 

29 

30 async def _filter_add_caps(self, stanza: StanzaBase) -> StanzaBase: 

31 # we rolled our own "add caps on presences" filter because 

32 # there is too much magic happening in slixmpp 

33 # anyway, we probably want to roll our own "dynamic disco"/caps 

34 # module in the long run, so it's a step in this direction 

35 if not isinstance(stanza, Presence): 

36 return stanza 

37 

38 if "caps" in stanza: 

39 return stanza 

40 

41 if stanza["type"] not in ("available", "chat", "away", "dnd", "xa"): 

42 return stanza 

43 

44 pfrom = stanza.get_from() 

45 

46 caps = self.xmpp.plugin["xep_0115"] 

47 

48 if pfrom == self.xmpp.boundjid.bare: 

49 if self.__gateway_ver is None: 

50 ver = self.__gateway_ver = await caps.get_verstring(pfrom) # type:ignore[misc] 

51 else: 

52 ver = self.__gateway_ver 

53 else: 

54 try: 

55 session = self.xmpp.get_session_from_jid(stanza.get_to()) 

56 except XMPPError: 

57 log.debug("not adding caps 1") 

58 return stanza 

59 

60 if session is None: 

61 return stanza 

62 

63 await session.ready 

64 

65 try: 

66 contact: LegacyContact = await session.contacts.by_jid(pfrom) 

67 except XMPPError: 

68 return stanza 

69 

70 ver = self.__contact_vers.get(contact.client_type) 

71 

72 if ver is None: 

73 ver = self.__contact_vers[ 

74 contact.client_type 

75 ] = await contact.get_caps_ver(pfrom) 

76 

77 log.debug("Ver: %s", ver) 

78 

79 if ver: 

80 stanza["caps"]["node"] = caps.caps_node 

81 stanza["caps"]["hash"] = caps.hash 

82 stanza["caps"]["ver"] = ver 

83 return stanza 

84 

85 def get_caps_ver(self, client_type: ClientType) -> str | None: 

86 return self.__contact_vers.get(client_type) 

87 

88 

89log = logging.getLogger(__name__)