1 # -*- coding: utf-8 -*-
2 # SPDX-FileCopyrightText: 2014, 2020, 2023 kaliko <kaliko@azylum.org>
3 # SPDX-FileCopyrightText: 2007-2012 Thomas Perl <thp.io/about>
4 # SPDX-License-Identifier: GPL-3.0-or-later
5 """A Real Time Block List plugin"""
7 from hashlib import sha256
8 from typing import Dict, Optional
10 from slixmpp import JID
11 from slixmpp.exceptions import XMPPError
12 from slixmpp.xmlstream import tostring
14 from .plugin import Plugin
17 def jid_to_sha256(jid: JID) -> str:
18 """Convert Bare JID to sha256 hexdigest"""
19 return sha256(jid.bare.encode('utf-8')).hexdigest()
23 """Plain object to keep track of block list items"""
25 def __init__(self, initial_bl):
26 self.sha256_jids: Dict[str, Optional[str]] = {}
27 for item in initial_bl:
28 self.insert_item(item)
30 def check(self, jid: JID) -> bool:
31 """Check the presence of the JID in the blocklist"""
32 jidhash = jid_to_sha256(jid)
33 return jidhash in self.sha256_jids
35 def retract_item(self, item):
37 self.sha256_jids.pop(item[id], None)
39 def insert_item(self, item):
42 for i in item['payload']:
46 except AttributeError:
48 self.sha256_jids[item['id']] = text
50 def get_reason(self, jid: JID) -> Optional[str]:
51 """Check the presence of the JID in the blocklist"""
52 jidhash = jid_to_sha256(jid)
53 # Raises if item does not exist
54 return self.sha256_jids[jidhash]
57 """Implement the built-in function len(), use for boolean evaluation"""
58 return len(self.sha256_jids)
64 pubsub_server = 'example.org'
65 node = 'muc_bans_sha256'
67 def __init__(self, bot):
68 Plugin.__init__(self, bot)
69 bot.register_plugin('xep_0059') # Result Set Management
70 bot.register_plugin('xep_0060') # Publish-Subscribe
72 ('session_start', self._subscribe),
73 ('pubsub_retract', self._retract),
74 ('pubsub_publish', self._publish),
75 (f'muc::{self.bot.room}::presence', self.got_presence),
76 (f'muc::{self.bot.room}::got_offline', self.got_offline),
77 (f'muc::{self.bot.room}::got_online', self.got_online)
81 self.participants = set()
82 self.moderator = False
83 self.blocklist: BL = None
87 self.bot.unregister_bot_plugin(self)
89 async def _subscribe(self, *args):
91 nodes = await self.bot['xep_0060'].get_nodes(self.pubsub_server)
92 nodes_av = [_.get('node') for _ in nodes['disco_items']]
93 self.log.debug(f'nodes available: {nodes_av}')
94 if self.node not in nodes_av:
95 self.log.error(f'{self.node} node not available on {self.pubsub_server}')
97 iq = await self.bot['xep_0060'].subscribe(self.pubsub_server, self.node)
98 subscription = iq['pubsub']['subscription']
99 self.log.info('Subscribed %s to node %s', subscription['jid'], subscription['node'])
100 except XMPPError as error:
101 self.log.error('Could not subscribe %s to node %s: %s',
102 self.bot.boundjid.full, self.node, error.format())
105 node_blocklist = await self.bot['xep_0060'].get_items(self.pubsub_server, self.node)
106 self.blocklist = BL(node_blocklist['pubsub']['items'])
107 mess = f'Got {len(self.blocklist)} items in block list'
110 async def _create(self):
111 """Try to create node"""
113 await self.bot['xep_0060'].create_node(self.pubsub_server, self.node)
114 self.log.info('Created node %s', self.node)
115 except XMPPError as err:
116 self.log.error('Could not create node %s: %s', self.node, err.format())
117 raise XMPPError(f'Could not create node {self.node}') from err
119 def _retract(self, msg):
120 """Handler receiving a retract item event."""
121 self.log.debug('Retracted item %s from %s' % (
122 msg['pubsub_event']['items']['retract']['id'],
123 msg['pubsub_event']['items']['node']))
124 self.blocklist.retract_item(msg['pubsub_event']['items']['retract']['id'])
126 async def _publish(self, msg):
127 """Handler receiving a publish item event."""
128 self.log.debug('Published item %s to %s:' % (
129 msg['pubsub_event']['items']['item']['id'],
130 msg['pubsub_event']['items']['node']))
131 data = msg['pubsub_event']['items']['item']['payload']
133 self.log.debug(tostring(data))
135 self.log.debug('No item content')
137 self.blocklist.insert_item(msg['pubsub_event']['items']['item']['id'])
138 # Are current participants in the block list
139 for jid in self.participants:
142 async def rtbl_ban(self, jid):
143 """Ban jid in RTBL"""
144 if not self.moderator:
146 if not self.blocklist:
147 self.log.debug("block list not populated yet")
149 if self.blocklist.check(jid):
150 self.log.debug(f'About to ban {jid}')
151 reason = self.blocklist.get_reason(jid)
152 if reason is not None:
153 reason = f'rtbl {reason}'
154 await self.ban(jid.bare, reason=reason)
156 def got_offline(self, pres):
157 """Handler method for laving MUC participants"""
158 fjid = pres['muc']['jid']
159 user = fjid if fjid.full else pres['muc']['nick']
161 self.participants.remove(user)
163 self.log.error('KeyError removing participant: "%s"', user)
164 self.log.debug(f'participants: -{user} (len:{len(self.participants)})')
166 def got_presence(self, pres):
167 """Does bot have required permissions"""
168 if 110 in pres['muc']['status_codes']:
169 if pres['muc']['role'] != 'moderator':
171 'Please give the bot moderator permissions. Will only log actions.'
173 self.moderator = False
176 self.log.info('Got moderator permissions.')
177 self.moderator = True
179 async def got_online(self, pres):
180 """Handler method for new MUC participants"""
181 fjid = pres['muc']['jid']
182 user = fjid if fjid.full else pres['muc']['nick']
183 self.participants.add(user)
184 self.log.debug(f'participants: +{user} (len:{len(self.participants)})')
185 await self.rtbl_ban(user)
188 # vim: ai ts=4 sw=4 sts=4 expandtab