sid/rtbl.py

   1 # -*- coding: utf-8 -*-
   2 # SPDX-FileCopyrightText: 2023 kaliko <kaliko@azylum.org>
   3 # SPDX-License-Identifier: GPL-3.0-or-later
   4 """A Real Time Block List plugin"""
   5
   6 from hashlib import sha256
   7 from typing import Dict, Optional
   8
   9 from slixmpp import JID
  10 from slixmpp.exceptions import XMPPError
  11 from slixmpp.xmlstream import tostring
  12
  13 from .plugin import Plugin
  14
  15
  16 def jid_to_sha256(jid: JID) -> str:
  17     """Convert Bare JID to sha256 hexdigest"""
  18     return sha256(jid.bare.encode('utf-8')).hexdigest()
  19
  20
  21 class BL:
  22     """Plain object to keep track of block list items.
  23     Only used in RTBL plugin."""
  24
  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)
  29
  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
  34
  35     def retract_item(self, item):
  36         """remove bl item"""
  37         self.sha256_jids.pop(item[id], None)
  38
  39     def insert_item(self, item):
  40         """insert bl item"""
  41         text = None
  42         for i in item['payload']:
  43             try:
  44                 text = i.text
  45                 break
  46             except AttributeError:
  47                 continue
  48         self.sha256_jids[item['id']] = text
  49
  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]
  55
  56     def __len__(self):
  57         """Implement the built-in function len(), use for boolean evaluation"""
  58         return len(self.sha256_jids)
  59
  60
  61 class RTBL(Plugin):
  62     """Spam guard plugin for MUC.
  63     """
  64     #: Pubsub server
  65     pubsub_server = 'example.org'
  66     #: Pubsub server node to subscribe to
  67     node = 'muc_bans_sha256'
  68
  69     def __init__(self, bot):
  70         Plugin.__init__(self, bot)
  71         bot.register_plugin('xep_0059')  # Result Set Management
  72         bot.register_plugin('xep_0060')  # Publish-Subscribe
  73         self.handlers = [
  74                 ('session_start', self._subscribe),
  75                 ('pubsub_retract', self._retract),
  76                 ('pubsub_publish', self._publish),
  77                 (f'muc::{self.bot.room}::presence', self.got_presence),
  78                 (f'muc::{self.bot.room}::got_offline', self.got_offline),
  79                 (f'muc::{self.bot.room}::got_online', self.got_online)
  80         ]
  81         self.add_handlers()
  82         self.bot = bot
  83         self.participants = set()
  84         self.moderator = False
  85         self.blocklist: BL = None
  86
  87     def _exit(self):
  88         self.rm_handlers()
  89         self.bot.unregister_bot_plugin(self)
  90
  91     async def _subscribe(self, *args):
  92         try:
  93             nodes = await self.bot['xep_0060'].get_nodes(self.pubsub_server)
  94             nodes_av = [_.get('node') for _ in nodes['disco_items']]
  95             self.log.debug(f'nodes available: {nodes_av}')
  96             if self.node not in nodes_av:
  97                 self.log.error(f'{self.node} node not available on {self.pubsub_server}')
  98                 await self._create()
  99             iq = await self.bot['xep_0060'].subscribe(self.pubsub_server, self.node)
 100             subscription = iq['pubsub']['subscription']
 101             self.log.info('Subscribed %s to node %s', subscription['jid'], subscription['node'])
 102         except XMPPError as error:
 103             self.log.error('Could not subscribe %s to node %s: %s',
 104                            self.bot.boundjid.full, self.node, error.format())
 105             self._exit()
 106             return
 107         node_blocklist = await self.bot['xep_0060'].get_items(self.pubsub_server, self.node)
 108         self.blocklist = BL(node_blocklist['pubsub']['items'])
 109         mess = f'Got {len(self.blocklist)} items in block list'
 110         self.log.info(mess)
 111
 112     async def _create(self):
 113         """Try to create node"""
 114         try:
 115             await self.bot['xep_0060'].create_node(self.pubsub_server, self.node)
 116             self.log.info('Created node %s', self.node)
 117         except XMPPError as err:
 118             self.log.error('Could not create node %s: %s', self.node, err.format())
 119             raise XMPPError(f'Could not create node {self.node}') from err
 120
 121     def _retract(self, msg):
 122         """Handler receiving a retract item event."""
 123         self.log.debug('Retracted item %s from %s' % (
 124             msg['pubsub_event']['items']['retract']['id'],
 125             msg['pubsub_event']['items']['node']))
 126         self.blocklist.retract_item(msg['pubsub_event']['items']['retract']['id'])
 127
 128     async def _publish(self, msg):
 129         """Handler receiving a publish item event."""
 130         self.log.debug('Published item %s to %s:' % (
 131               msg['pubsub_event']['items']['item']['id'],
 132               msg['pubsub_event']['items']['node']))
 133         data = msg['pubsub_event']['items']['item']['payload']
 134         if data is not None:
 135             self.log.debug(tostring(data))
 136         else:
 137             self.log.debug('No item content')
 138             return
 139         self.blocklist.insert_item(msg['pubsub_event']['items']['item']['id'])
 140         # Are current participants in the block list
 141         for jid in self.participants:
 142             self.rtbl_ban(jid)
 143
 144     async def rtbl_ban(self, jid):
 145         """Ban jid in RTBL"""
 146         if not self.moderator:
 147             return
 148         if not self.blocklist:
 149             self.log.debug("block list not populated yet")
 150             return
 151         if self.blocklist.check(jid):
 152             self.log.debug(f'About to ban {jid}')
 153             reason = self.blocklist.get_reason(jid)
 154             if reason is not None:
 155                 reason = f'rtbl {reason}'
 156             await self.ban(jid.bare, reason=reason)
 157
 158     def got_offline(self, pres):
 159         """Handler method for leaving MUC participants"""
 160         fjid = pres['muc']['jid']
 161         user = fjid if fjid.full else pres['muc']['nick']
 162         try:
 163             self.participants.remove(user)
 164         except KeyError:
 165             self.log.error('KeyError removing participant: "%s"', user)
 166         self.log.debug(f'participants: -{user} (len:{len(self.participants)})')
 167
 168     def got_presence(self, pres):
 169         """Does bot have required permissions"""
 170         if 110 in pres['muc']['status_codes']:
 171             if pres['muc']['role'] != 'moderator':
 172                 self.log.error(
 173                     'Please give the bot moderator permissions. Will only log actions.'
 174                 )
 175                 self.moderator = False
 176                 return
 177             else:
 178                 self.log.info('Got moderator permissions.')
 179                 self.moderator = True
 180
 181     async def got_online(self, pres):
 182         """Handler method for new MUC participants"""
 183         fjid = pres['muc']['jid']
 184         user = fjid if fjid.full else pres['muc']['nick']
 185         self.participants.add(user)
 186         self.log.debug(f'participants: +{user} (len:{len(self.participants)})')
 187         await self.rtbl_ban(user)
 188
 189 # VIM MODLINE
 190 # vim: ai ts=4 sw=4 sts=4 expandtab