Coverage for slidge/slixfix/__init__.py: 84%

1# This module contains patches for slixmpp; some have pending requests upstream 

2# and should be removed on the next slixmpp release. 

3 

4# ruff: noqa: F401 

5 

6import uuid 

7 

8import slixmpp.plugins 

9import slixmpp.stanza.roster 

10from slixmpp import Message, register_stanza_plugin 

11from slixmpp.exceptions import IqError 

12from slixmpp.plugins.xep_0050 import XEP_0050, Command 

13from slixmpp.plugins.xep_0356.permissions import IqPermission 

14from slixmpp.plugins.xep_0356.privilege import XEP_0356 

15from slixmpp.plugins.xep_0385.sims import XEP_0385 

16from slixmpp.plugins.xep_0385.sims import stanza as stanza_sims 

17from slixmpp.plugins.xep_0469.stanza import NS as PINNED_NS 

18from slixmpp.plugins.xep_0469.stanza import Pinned 

19from slixmpp.xmlstream import StanzaBase 

20 

21from . import ( 

22 link_preview, 

23 xep_0077, 

24 xep_0100, 

25 xep_0153, 

26 xep_0292, 

27) 

28 

29 

30def plugin_init(self): 

31 register_stanza_plugin(self.xmpp["xep_0372"].stanza.Reference, stanza_sims.Sims) 

32 register_stanza_plugin(Message, stanza_sims.Sims) 

33 

34 register_stanza_plugin(stanza_sims.Sims, stanza_sims.Sources) 

35 register_stanza_plugin(stanza_sims.Sims, self.xmpp["xep_0234"].stanza.File) 

36 register_stanza_plugin( 

37 stanza_sims.Sources, self.xmpp["xep_0372"].stanza.Reference, iterable=True 

38 ) 

39 

40 

41XEP_0385.plugin_init = plugin_init 

42 

43 

44def set_pinned(self, val: bool) -> None: 

45 extensions = self.parent() 

46 if val: 

47 extensions.enable("pinned") 

48 else: 

49 extensions._del_sub(f"{{{PINNED_NS}}}pinned") 

50 

51 

52Pinned.set_pinned = set_pinned 

53 

54 

55def session_bind(self, jid) -> None: 

56 self.xmpp["xep_0030"].add_feature(Command.namespace) 

57 # awful hack to for the disco items: we need to comment this line 

58 # related issue: https://todo.sr.ht/~nicoco/slidge/131 

59 # self.xmpp['xep_0030'].set_items(node=Command.namespace, items=tuple()) 

60 

61 

62XEP_0050.session_bind = session_bind # type:ignore 

63 

64 

65def reply(self, body=None, clear: bool = True): 

66 """ 

67 Overrides slixmpp's Message.reply(), since it strips to sender's resource 

68 for mtype=groupchat, and we do not want that, because when we raise an XMPPError, 

69 we actually want to preserve the resource. 

70 (this is called in RootStanza.exception() to handle XMPPErrors) 

71 """ 

72 new_message = StanzaBase.reply(self, clear) 

73 new_message["thread"] = self["thread"] 

74 new_message["parent_thread"] = self["parent_thread"] 

75 

76 del new_message["id"] 

77 if self.stream is not None and self.stream.use_message_ids: 

78 new_message["id"] = self.stream.new_id() 

79 

80 if body is not None: 

81 new_message["body"] = body 

82 return new_message 

83 

84 

85Message.reply = reply # type: ignore 

86 

87# TODO: remove me when https://codeberg.org/poezio/slixmpp/pulls/3622 is merged 

88 

89 

90class PrivilegedIqError(IqError): 

91 """ 

92 Exception raised when sending a privileged IQ stanza fails. 

93 """ 

94 

95 def nested_error(self) -> IqError | None: 

96 """ 

97 Return the IQError generated from the inner IQ stanza, if present. 

98 """ 

99 if "privilege" in self.iq: 

100 if "forwarded" in self.iq["privilege"]: 

101 if "iq" in self.iq["privilege"]["forwarded"]: 

102 return IqError(self.iq["privilege"]["forwarded"]["iq"]) 

103 return None 

104 

105 

106async def send_privileged_iq(self, encapsulated_iq, iq_id=None): 

107 """ 

108 Send an IQ on behalf of a user 

109 

110 Caution: the IQ *must* have the jabber:client namespace 

111 

112 Raises :class:`PrivilegedIqError` on failure. 

113 """ 

114 iq_id = iq_id or str(uuid.uuid4()) 

115 encapsulated_iq["id"] = iq_id 

116 server = encapsulated_iq.get_to().domain 

117 perms = self.granted_privileges.get(server) 

118 if not perms: 

119 raise PermissionError(f"{server} has not granted us any privilege") 

120 itype = encapsulated_iq["type"] 

121 for ns in encapsulated_iq.plugins.values(): 

122 type_ = perms.iq[ns.namespace] 

123 if type_ == IqPermission.NONE: 

124 raise PermissionError( 

125 f"{server} has not granted any IQ privilege for namespace {ns.namespace}" 

126 ) 

127 elif type_ == IqPermission.BOTH: 

128 pass 

129 elif type_ != itype: 

130 raise PermissionError( 

131 f"{server} has not granted IQ {itype} privilege for namespace {ns.namespace}" 

132 ) 

133 iq = self.xmpp.make_iq( 

134 itype=itype, 

135 ifrom=self.xmpp.boundjid.bare, 

136 ito=encapsulated_iq.get_from(), 

137 id=iq_id, 

138 ) 

139 iq["privileged_iq"].append(encapsulated_iq) 

140 

141 try: 

142 resp = await iq.send() 

143 except IqError as exc: 

144 raise PrivilegedIqError(exc.iq) 

145 

146 return resp["privilege"]["forwarded"]["iq"] 

147 

148 

149XEP_0356.send_privileged_iq = send_privileged_iq 

150 

151 

152slixmpp.plugins.PLUGINS.extend( 

153 [ 

154 "link_preview", 

155 "xep_0292_provider", 

156 ] 

157)