# -*- coding: utf-8 -*-
-# SPDX-FileCopyrightText: 2014, 2020, 2023 kaliko <kaliko@azylum.org>
-# SPDX-FileCopyrightText: 2007-2012 Thomas Perl <thp.io/about>
+# SPDX-FileCopyrightText: 2023 kaliko <kaliko@azylum.org>
# SPDX-License-Identifier: GPL-3.0-or-later
"""A Real Time Block List plugin"""
class BL:
- """Plain object to keep track of block list items"""
+ """Plain object to keep track of block list items.
+ Only used in RTBL plugin."""
+ #: Initial seed to ease testing
+ init = {}
def __init__(self, initial_bl):
- self.sha256_jids: Dict[str, Optional[str]] = {}
+ self.sha256_jids: Dict[str, Optional[str]] = dict(BL.init)
for item in initial_bl:
self.insert_item(item)
class RTBL(Plugin):
- """Spam guard for MUC
+ """Spam guard plugin for MUC.
"""
+ #: Pubsub server
pubsub_server = 'example.org'
+ #: Pubsub server node to subscribe to
node = 'muc_bans_sha256'
def __init__(self, bot):
self.blocklist = BL(node_blocklist['pubsub']['items'])
mess = f'Got {len(self.blocklist)} items in block list'
self.log.info(mess)
+ # Are current participants in the block list
+ for jid in list(self.participants):
+ await self.rtbl_ban(jid)
async def _create(self):
"""Try to create node"""
self.log.debug('Retracted item %s from %s' % (
msg['pubsub_event']['items']['retract']['id'],
msg['pubsub_event']['items']['node']))
- self.blocklist.retract_item(msg['pubsub_event']['items']['retract']['id'])
+ self.blocklist.retract_item(msg['pubsub_event']['items']['retract'])
async def _publish(self, msg):
"""Handler receiving a publish item event."""
else:
self.log.debug('No item content')
return
- self.blocklist.insert_item(msg['pubsub_event']['items']['item']['id'])
+ self.blocklist.insert_item(msg['pubsub_event']['items']['item'])
# Are current participants in the block list
- for jid in self.participants:
- self.rtbl_ban(jid)
+ for jid in list(self.participants):
+ await self.rtbl_ban(jid)
async def rtbl_ban(self, jid):
"""Ban jid in RTBL"""
await self.ban(jid.bare, reason=reason)
def got_offline(self, pres):
- """Handler method for laving MUC participants"""
+ """Handler method for leaving MUC participants"""
fjid = pres['muc']['jid']
user = fjid if fjid.full else pres['muc']['nick']
try: