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

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.util.types import AnyGateway 

11 

12 

13class MucAdminMixin(DispatcherMixin): 

14 __slots__: list[str] = [] 

15 

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

17 super().__init__(xmpp) 

18 self.xmpp.register_handler( 

19 CoroutineCallback( 

20 "RoomModerate", 

21 StanzaPath("iq/moderate"), 

22 self.on_user_moderation, 

23 ) 

24 ) 

25 self.xmpp.register_handler( 

26 CoroutineCallback( 

27 "RoomSetAffiliation", 

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

29 self.on_user_set_affiliation, 

30 ) 

31 ) 

32 self.xmpp.register_handler( 

33 CoroutineCallback( 

34 "RoomGetAffiliation", 

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(xmpp_id, muc) 

57 await muc.on_moderate(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.get_or_create_by_jid(JID(item["jid"])) 

68 part = muc.participants.get_or_create(contact=contact) 

69 else: 

70 await muc.participants.ensure_filled() 

71 part = muc.participants.get_by_resource(item["nick"]) 

72 if part is None: 

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

74 

75 if item["affiliation"]: 

76 await part.on_set_affiliation( 

77 item["affiliation"], 

78 item["reason"] or None, 

79 item["nick"] or None, 

80 ) 

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

82 await part.on_kick(item["reason"] or None) 

83 

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

85 

86 @exceptions_to_xmpp_errors 

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

88 assert isinstance(iq, Iq) 

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

90 

91 if not affiliation: 

92 raise XMPPError("bad-request") 

93 

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

95 muc = await session.rooms.get_or_create_by_jid(iq.get_to()) 

96 

97 reply = iq.reply() 

98 reply.enable("mucadmin_query") 

99 i = 0 

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

101 i += 1 

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

103 if i == 1000: 

104 # we might hit stanza size limits without that 

105 break 

106 reply.send()