Coverage for slidge / core / dispatcher / muc / admin.py: 93%

1import typing 

2 

3from slixmpp import JID, CoroutineCallback, Iq, StanzaPath 

4from slixmpp.exceptions import XMPPError 

5from slixmpp.xmlstream import StanzaBase 

6 

7from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

8 

9if typing.TYPE_CHECKING: 

10 from slidge.core.gateway import BaseGateway 

11 

12 

13class MucAdminMixin(DispatcherMixin): 

14 __slots__: list[str] = [] 

15 

16 def __init__(self, xmpp: "BaseGateway") -> None: 

17 super().__init__(xmpp) 

18 self.xmpp.register_handler( 

19 CoroutineCallback( 

20 "MUCModerate", 

21 StanzaPath("iq/moderate"), 

22 self.on_user_moderation, 

23 ) 

24 ) 

25 self.xmpp.register_handler( 

26 CoroutineCallback( 

27 "MUCSetAffiliation", 

28 StanzaPath("iq@type=set/mucadmin_query"), 

29 self.on_user_set_affiliation, 

30 ) 

31 ) 

32 self.xmpp.register_handler( 

33 CoroutineCallback( 

34 "MUCGetAffiliation", 

35 StanzaPath("iq@type=get/mucadmin_query"), 

36 self.on_muc_admin_query_get, 

37 ) 

38 ) 

39 

40 @exceptions_to_xmpp_errors 

41 async def on_user_moderation(self, iq: StanzaBase) -> None: 

42 assert isinstance(iq, Iq) 

43 muc = await self.get_muc_from_stanza(iq) 

44 

45 moderate = iq["moderate"] 

46 xmpp_id = iq["moderate"]["id"] 

47 if not xmpp_id: 

48 raise XMPPError("bad-request", "Missing moderated message ID") 

49 

50 if not moderate["retract"]: 

51 raise XMPPError( 

52 "feature-not-implemented", 

53 "Slidge only implements moderation/retraction", 

54 ) 

55 

56 legacy_id = self._xmpp_msg_id_to_legacy(muc.session, xmpp_id, muc) 

57 await muc.session.on_moderate(muc, legacy_id, moderate["reason"] or None) 

58 iq.reply(clear=True).send() 

59 

60 @exceptions_to_xmpp_errors 

61 async def on_user_set_affiliation(self, iq: StanzaBase) -> None: 

62 assert isinstance(iq, Iq) 

63 muc = await self.get_muc_from_stanza(iq) 

64 

65 item = iq["mucadmin_query"]["item"] 

66 if item["jid"]: 

67 contact = await muc.session.contacts.by_jid(JID(item["jid"])) 

68 else: 

69 part = await muc.get_participant( 

70 item["nick"], fill_first=True, create=False 

71 ) 

72 if part is None: 

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

74 assert part.contact is not None 

75 contact = part.contact 

76 

77 if item["affiliation"]: 

78 await muc.on_set_affiliation( 

79 contact, 

80 item["affiliation"], 

81 item["reason"] or None, 

82 item["nick"] or None, 

83 ) 

84 elif item["role"] == "none": 

85 await muc.on_kick(contact, item["reason"] or None) 

86 

87 iq.reply(clear=True).send() 

88 

89 @exceptions_to_xmpp_errors 

90 async def on_muc_admin_query_get(self, iq: StanzaBase) -> None: 

91 assert isinstance(iq, Iq) 

92 affiliation = iq["mucadmin_query"]["item"]["affiliation"] 

93 

94 if not affiliation: 

95 raise XMPPError("bad-request") 

96 

97 session = await self._get_session(iq, 1, logged=True) 

98 muc = await session.bookmarks.by_jid(iq.get_to()) 

99 

100 reply = iq.reply() 

101 reply.enable("mucadmin_query") 

102 async for participant in muc.get_participants(affiliation): 

103 reply["mucadmin_query"].append(participant.mucadmin_item()) 

104 reply.send()