PyLink/plugins/antispam.py

# antispam.py: Basic services-side spamfilters for IRC

from pylinkirc import conf, utils
from pylinkirc.log import log

mydesc = ("Provides anti-spam functionality.")
sbot = utils.register_service("antispam", default_nick="AntiSpam", desc=mydesc)

def die(irc=None):
    utils.unregister_service("antispam")

_UNICODE_CHARMAP = {
    'A': 'AΑАᎪᗅᴀ𝐀𝐴𝑨𝒜𝓐𝔄𝔸𝕬𝖠𝗔𝘈𝘼𝙰𝚨𝛢𝜜𝝖𝞐',
    'B': 'BʙΒВвᏴᗷᛒℬ𐌁𝐁𝐵𝑩𝓑𝔅𝔹𝕭𝖡𝗕𝘉𝘽𝙱𝚩𝛣𝜝𝝗𝞑',
    'C': 'CϹСᏟℂℭⅭⲤ𐌂𝐂𝐶𝑪𝒞𝓒𝕮𝖢𝗖𝘊𝘾𝙲',
    'D': 'DᎠᗞᗪᴅⅅⅮ𝐃𝐷𝑫𝒟𝓓𝔇𝔻𝕯𝖣𝗗𝘋𝘿𝙳',
    'E': 'EΕЕᎬᴇℰ⋿ⴹ𝐄𝐸𝑬𝓔𝔈𝔼𝕰𝖤𝗘𝘌𝙀𝙴𝚬𝛦𝜠𝝚𝞔',
    'F': 'FϜᖴℱ𝐅𝐹𝑭𝓕𝔉𝔽𝕱𝖥𝗙𝘍𝙁𝙵𝟊',
    'G': 'GɢԌԍᏀᏳ𝐆𝐺𝑮𝒢𝓖𝔊𝔾𝕲𝖦𝗚𝘎𝙂𝙶',
    'H': 'HʜΗНнᎻᕼℋℌℍⲎ𝐇𝐻𝑯𝓗𝕳𝖧𝗛𝘏𝙃𝙷𝚮𝛨𝜢𝝜𝞖',
    'J': 'JЈᎫᒍᴊ𝐉𝐽𝑱𝒥𝓙𝔍𝕁𝕵𝖩𝗝𝘑𝙅𝙹',
    'K': 'KΚКᏦᛕKⲔ𝐊𝐾𝑲𝒦𝓚𝔎𝕂𝕶𝖪𝗞𝘒𝙆𝙺𝚱𝛫𝜥𝝟𝞙',
    'L': 'LʟᏞᒪℒⅬ𝐋𝐿𝑳𝓛𝔏𝕃𝕷𝖫𝗟𝘓𝙇𝙻',
    'M': 'MΜϺМᎷᗰᛖℳⅯⲘ𐌑𝐌𝑀𝑴𝓜𝔐𝕄𝕸𝖬𝗠𝘔𝙈𝙼𝚳𝛭𝜧𝝡𝞛',
    'N': 'NɴΝℕⲚ𝐍𝑁𝑵𝒩𝓝𝔑𝕹𝖭𝗡𝘕𝙉𝙽𝚴𝛮𝜨𝝢𝞜',
    'P': 'PΡРᏢᑭᴘᴩℙⲢ𝐏𝑃𝑷𝒫𝓟𝔓𝕻𝖯𝗣𝘗𝙋𝙿𝚸𝛲𝜬𝝦𝞠',
    'Q': 'Qℚⵕ𝐐𝑄𝑸𝒬𝓠𝔔𝕼𝖰𝗤𝘘𝙌𝚀',
    'R': 'RƦʀᎡᏒᖇᚱℛℜℝ𝐑𝑅𝑹𝓡𝕽𝖱𝗥𝘙𝙍𝚁',
    'S': 'SЅՏᏕᏚ𝐒𝑆𝑺𝒮𝓢𝔖𝕊𝕾𝖲𝗦𝘚𝙎𝚂',
    'T': 'TΤτТтᎢᴛ⊤⟙Ⲧ𐌕𝐓𝑇𝑻𝒯𝓣𝔗𝕋𝕿𝖳𝗧𝘛𝙏𝚃𝚻𝛕𝛵𝜏𝜯𝝉𝝩𝞃𝞣𝞽',
    'U': 'UՍሀᑌ∪⋃𝐔𝑈𝑼𝒰𝓤𝔘𝕌𝖀𝖴𝗨𝘜𝙐𝚄',
    'V': 'VѴ٧۷ᏙᐯⅤⴸ𝐕𝑉𝑽𝒱𝓥𝔙𝕍𝖁𝖵𝗩𝘝𝙑𝚅',
    'W': 'WԜᎳᏔ𝐖𝑊𝑾𝒲𝓦𝔚𝕎𝖂𝖶𝗪𝘞𝙒𝚆',
    'X': 'XΧХ᙭ᚷⅩ╳Ⲭⵝ𐌗𐌢𝐗𝑋𝑿𝒳𝓧𝔛𝕏𝖃𝖷𝗫𝘟𝙓𝚇𝚾𝛸𝜲𝝬𝞦',
    'Y': 'YΥϒУҮᎩᎽⲨ𝐘𝑌𝒀𝒴𝓨𝔜𝕐𝖄𝖸𝗬𝘠𝙔𝚈𝚼𝛶𝜰𝝪𝞤',
    'Z': 'ZΖᏃℤℨ𝐙𝑍𝒁𝒵𝓩𝖅𝖹𝗭𝘡𝙕𝚉𝚭𝛧𝜡𝝛𝞕',
    'a': 'aɑαа⍺𝐚𝑎𝒂𝒶𝓪𝔞𝕒𝖆𝖺𝗮𝘢𝙖𝚊𝛂𝛼𝜶𝝰𝞪',
    'b': 'bƄЬᏏᖯ𝐛𝑏𝒃𝒷𝓫𝔟𝕓𝖇𝖻𝗯𝘣𝙗𝚋',
    'c': 'cϲсᴄⅽⲥ𝐜𝑐𝒄𝒸𝓬𝔠𝕔𝖈𝖼𝗰𝘤𝙘𝚌',
    'd': 'ⅾdԁᏧᑯⅆⅾ𝐝𝑑𝒅𝒹𝓭𝔡𝕕𝖉𝖽𝗱𝘥𝙙𝚍',
    'e': 'eеҽ℮ℯⅇ𝐞𝑒𝒆𝓮𝔢𝕖𝖊𝖾𝗲𝘦𝙚𝚎ᥱ',
    'f': 'fſϝքẝ𝐟𝑓𝒇𝒻𝓯𝔣𝕗𝖋𝖿𝗳𝘧𝙛𝚏𝟋',
    'g': 'gƍɡցᶃℊ𝐠𝑔𝒈𝓰𝔤𝕘𝖌𝗀𝗴𝘨𝙜𝚐',
    'h': 'hһհᏂℎ𝐡𝒉𝒽𝓱𝔥𝕙𝖍𝗁𝗵𝘩𝙝𝚑',
    'i': 'iıɩɪιіӏᎥℹⅈⅰ⍳ꙇ𝐢𝑖𝒊𝒾𝓲𝔦𝕚𝖎𝗂𝗶𝘪𝙞𝚒𝚤𝛊𝜄𝜾𝝸𝞲',
    'j': 'jϳјⅉ𝐣𝑗𝒋𝒿𝓳𝔧𝕛𝖏𝗃𝗷𝘫𝙟𝚓',
    'k': 'k𝐤𝑘𝒌𝓀𝓴𝔨𝕜𝖐𝗄𝗸𝘬𝙠𝚔',
    'l': 'ⅼ',
    'm': 'ⅿm',
    'n': 'nոռ𝐧𝑛𝒏𝓃𝓷𝔫𝕟𝖓𝗇𝗻𝘯𝙣𝚗ᥒ',
    'o': 'ⲟഠοо',
    'p': 'pρϱр⍴ⲣ𝐩𝑝𝒑𝓅𝓹𝔭𝕡𝖕𝗉𝗽𝘱𝙥𝚙𝛒𝛠𝜌𝜚𝝆𝝔𝞀𝞎𝞺𝟈',
    'q': 'qԛգզ𝐪𝑞𝒒𝓆𝓺𝔮𝕢𝖖𝗊𝗾𝘲𝙦𝚚',
    'r': 'rгᴦⲅ𝐫𝑟𝒓𝓇𝓻𝔯𝕣𝖗𝗋𝗿𝘳𝙧𝚛',
    's': 'sƽѕꜱ𝐬𝑠𝒔𝓈𝓼𝔰𝕤𝖘𝗌𝘀𝘴𝙨𝚜',
    't': 't𝐭𝑡𝒕𝓉𝓽𝔱𝕥𝖙𝗍𝘁𝘵𝙩𝚝',
    'u': 'uʋυսᴜ𝐮𝑢𝒖𝓊𝓾𝔲𝕦𝖚𝗎𝘂𝘶𝙪𝚞𝛖𝜐𝝊𝞄𝞾ᥙ',
    'v': 'vνѵטᴠⅴ∨⋁𝐯𝑣𝒗𝓋𝓿𝔳𝕧𝖛𝗏𝘃𝘷𝙫𝚟𝛎𝜈𝝂𝝼𝞶',
    'w': 'wɯѡԝաᴡ𝐰𝑤𝒘𝓌𝔀𝔴𝕨𝖜𝗐𝘄𝘸𝙬𝚠',
    'x': 'x×хᕁᕽ᙮ⅹ⤫⤬⨯𝐱𝑥𝒙𝓍𝔁𝔵𝕩𝖝𝗑𝘅𝘹𝙭𝚡',
    'y': 'yɣʏγуүყᶌỿℽ𝐲𝑦𝒚𝓎𝔂𝔶𝕪𝖞𝗒𝘆𝘺𝙮𝚢𝛄𝛾𝜸𝝲𝞬',
    'z': 'zᴢ𝐳𝑧𝒛𝓏𝔃𝔷𝕫𝖟𝗓𝘇𝘻𝙯𝚣',
    '/': '᜵⁄∕⧸／',
    '\\': '⧵﹨⧹＼',
    ' ': '\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\xa0\u202f\u205f',
    '.': '․．',
    '-': '˗╴﹣－−⎼',
    '!': '﹗！ǃⵑ︕',
    ':': ':˸։፡᛬⁚∶⠆︓﹕',
    '#': '＃﹟'
}

def _prep_maketrans(data):
    from_s = ''
    to_s = ''
    for target, chars in data.items():
        from_s += chars
        to_s += target * len(chars)

    return str.maketrans(from_s, to_s)

UNICODE_CHARMAP = _prep_maketrans(_UNICODE_CHARMAP)

PUNISH_OPTIONS = ['kill', 'ban', 'quiet', 'kick', 'block']
EXEMPT_OPTIONS = ['voice', 'halfop', 'op']
DEFAULT_EXEMPT_OPTION = 'halfop'
def _punish(irc, target, channel, punishment, reason):
    """Punishes the target user. This function returns True if the user was successfully punished."""
    if target not in irc.users:
        log.warning("(%s) antispam: got target %r that isn't a user?", irc.name, target)
        return False
    elif irc.is_oper(target):
        log.debug("(%s) antispam: refusing to punish oper %s/%s", irc.name, target, irc.get_friendly_name(target))
        return False

    target_nick = irc.get_friendly_name(target)

    if channel:
        c = irc.channels[channel]
        exempt_level = irc.get_service_option('antispam', 'exempt_level', DEFAULT_EXEMPT_OPTION).lower()

        if exempt_level not in EXEMPT_OPTIONS:
            log.error('(%s) Antispam exempt %r is not a valid setting, '
                      'falling back to defaults; accepted settings include: %s',
                      irc.name, exempt_level, ', '.join(EXEMPT_OPTIONS))
            exempt_level = DEFAULT_EXEMPT_OPTION

        if exempt_level == 'voice' and c.is_voice_plus(target):
            log.debug("(%s) antispam: refusing to punish voiced and above %s/%s", irc.name, target, target_nick)
            return False
        elif exempt_level == 'halfop' and c.is_halfop_plus(target):
            log.debug("(%s) antispam: refusing to punish halfop and above %s/%s", irc.name, target, target_nick)
            return False
        elif exempt_level == 'op' and c.is_op_plus(target):
            log.debug("(%s) antispam: refusing to punish op and above %s/%s", irc.name, target, target_nick)
            return False

    my_uid = sbot.uids.get(irc.name)
    # XXX workaround for single-bot protocols like Clientbot
    if irc.pseudoclient and not irc.has_cap('can-spawn-clients'):
        my_uid = irc.pseudoclient.uid

    bans = set()
    log.debug('(%s) antispam: got %r as punishment for %s/%s', irc.name, punishment,
              target, irc.get_friendly_name(target))

    def _ban():
        bans.add(irc.make_channel_ban(target))
    def _quiet():
        bans.add(irc.make_channel_ban(target, ban_type='quiet'))
    def _kick():
        irc.kick(my_uid, channel, target, reason)
        irc.call_hooks([my_uid, 'ANTISPAM_KICK', {'channel': channel, 'text': reason, 'target': target,
                                                  'parse_as': 'KICK'}])
    def _kill():
        if target not in irc.users:
            log.debug('(%s) antispam: not killing %s/%s; they already left', irc.name, target,
                      irc.get_friendly_name(target))
            return
        userdata = irc.users[target]
        irc.kill(my_uid, target, reason)
        irc.call_hooks([my_uid, 'ANTISPAM_KILL', {'target': target, 'text': reason,
                                                  'userdata': userdata, 'parse_as': 'KILL'}])

    kill = False
    successful_punishments = 0
    for action in set(punishment.split('+')):
        if action not in PUNISH_OPTIONS:
            log.error('(%s) Antispam punishment %r is not a valid setting; '
                      'accepted settings include: %s OR any combination of '
                      'these joined together with a "+".',
                      irc.name, punishment, ', '.join(PUNISH_OPTIONS))
            return
        elif action == 'block':
            # We only need to increment this for this function to return True
            successful_punishments += 1
        elif action == 'kill':
            kill = True  # Delay kills so that the user data doesn't disappear.
        # XXX factorize these blocks
        elif action == 'kick' and channel:
            try:
                _kick()
            except NotImplementedError:
                log.warning("(%s) antispam: Kicks are not supported on this network, skipping; "
                            "target was %s/%s", irc.name, target_nick, channel)
            else:
                successful_punishments += 1
        elif action == 'ban' and channel:
            try:
                _ban()
            except (ValueError, NotImplementedError):
                log.warning("(%s) antispam: Bans are not supported on this network, skipping; "
                            "target was %s/%s", irc.name, target_nick, channel)
            else:
                successful_punishments += 1
        elif action == 'quiet' and channel:
            try:
                _quiet()
            except (ValueError, NotImplementedError):
                log.warning("(%s) antispam: Quiet is not supported on this network, skipping; "
                            "target was %s/%s", irc.name, target_nick, channel)
            else:
                successful_punishments += 1

    if bans:  # Set all bans at once to prevent spam
        irc.mode(my_uid, channel, bans)
        irc.call_hooks([my_uid, 'ANTISPAM_BAN',
                        {'target': channel, 'modes': bans, 'parse_as': 'MODE'}])
    if kill:
        try:
            _kill()
        except NotImplementedError:
            log.warning("(%s) antispam: Kills are not supported on this network, skipping; "
                        "target was %s/%s", irc.name, target_nick, channel)
        else:
            successful_punishments += 1

    if not successful_punishments:
        log.warning('(%s) antispam: Failed to punish %s with %r, target was %s', irc.name,
                    target_nick, punishment, channel or 'a PM')

    return bool(successful_punishments)

MASSHIGHLIGHT_DEFAULTS = {
    'min_length': 50,
    'min_nicks': 5,
    'reason': "Mass highlight spam is prohibited",
    'punishment': 'kick+ban',
    'enabled': False
}
def handle_masshighlight(irc, source, command, args):
    """Handles mass highlight attacks."""
    channel = args['target']
    text = args['text']
    mhl_settings = irc.get_service_option('antispam', 'masshighlight',
                                          MASSHIGHLIGHT_DEFAULTS)

    if not mhl_settings.get('enabled', False):
        return

    my_uid = sbot.uids.get(irc.name)

    # XXX workaround for single-bot protocols like Clientbot
    if irc.pseudoclient and not irc.has_cap('can-spawn-clients'):
        my_uid = irc.pseudoclient.uid

    if (not irc.connected.is_set()) or (not my_uid):
        # Break if the network isn't ready.
        log.debug("(%s) antispam.masshighlight: skipping processing; network isn't ready", irc.name)
        return
    elif not irc.is_channel(channel):
        # Not a channel - mass highlight blocking only makes sense within channels
        log.debug("(%s) antispam.masshighlight: skipping processing; %r is not a channel", irc.name, channel)
        return
    elif irc.is_internal_client(source):
        # Ignore messages from our own clients.
        log.debug("(%s) antispam.masshighlight: skipping processing message from internal client %s", irc.name, source)
        return
    elif source not in irc.users:
        log.debug("(%s) antispam.masshighlight: ignoring message from non-user %s", irc.name, source)
        return
    elif channel not in irc.channels or my_uid not in irc.channels[channel].users:
        # We're not monitoring this channel.
        log.debug("(%s) antispam.masshighlight: skipping processing message from channel %r we're not in", irc.name, channel)
        return
    elif len(text) < mhl_settings.get('min_length', MASSHIGHLIGHT_DEFAULTS['min_length']):
        log.debug("(%s) antispam.masshighlight: skipping processing message %r; it's too short", irc.name, text)
        return

    if irc.get_service_option('antispam', 'strip_formatting', True):
        text = utils.strip_irc_formatting(text)

    # Strip :, from potential nicks
    words = [word.rstrip(':,') for word in text.split()]

    userlist = [irc.users[uid].nick for uid in irc.channels[channel].users.copy()]
    min_nicks = mhl_settings.get('min_nicks', MASSHIGHLIGHT_DEFAULTS['min_nicks'])

    # Don't allow repeating the same nick to trigger punishment
    nicks_caught = set()

    punished = False
    for word in words:
        if word in userlist:
            nicks_caught.add(word)
        if len(nicks_caught) >= min_nicks:
            # Get the punishment and reason.
            punishment = mhl_settings.get('punishment', MASSHIGHLIGHT_DEFAULTS['punishment']).lower()
            reason = mhl_settings.get('reason', MASSHIGHLIGHT_DEFAULTS['reason'])

            log.info("(%s) antispam: punishing %s => %s for mass highlight spam",
                     irc.name,
                     irc.get_friendly_name(source),
                     channel)
            punished = _punish(irc, source, channel, punishment, reason)
            break

    log.debug('(%s) antispam.masshighlight: got %s/%s nicks on message to %r', irc.name,
              len(nicks_caught), min_nicks, channel)
    return not punished  # Filter this message from relay, etc. if it triggered protection

utils.add_hook(handle_masshighlight, 'PRIVMSG', priority=1000)
utils.add_hook(handle_masshighlight, 'NOTICE', priority=1000)

TEXTFILTER_DEFAULTS = {
    'reason': "Spam is prohibited",
    'punishment': 'kick+ban+block',
    'watch_pms': False,
    'enabled': False,
    'munge_unicode': True,
}
def handle_textfilter(irc, source, command, args):
    """Antispam text filter handler."""
    target = args['target']
    text = args['text']
    txf_settings = irc.get_service_option('antispam', 'textfilter',
                                          TEXTFILTER_DEFAULTS)

    if not txf_settings.get('enabled', False):
        return

    my_uid = sbot.uids.get(irc.name)

    # XXX workaround for single-bot protocols like Clientbot
    if irc.pseudoclient and not irc.has_cap('can-spawn-clients'):
        my_uid = irc.pseudoclient.uid

    if (not irc.connected.is_set()) or (not my_uid):
        # Break if the network isn't ready.
        log.debug("(%s) antispam.textfilters: skipping processing; network isn't ready", irc.name)
        return
    elif irc.is_internal_client(source):
        # Ignore messages from our own clients.
        log.debug("(%s) antispam.textfilters: skipping processing message from internal client %s", irc.name, source)
        return
    elif source not in irc.users:
        log.debug("(%s) antispam.textfilters: ignoring message from non-user %s", irc.name, source)
        return

    if irc.is_channel(target):
        channel_or_none = target
        if target not in irc.channels or my_uid not in irc.channels[target].users:
            # We're not monitoring this channel.
            log.debug("(%s) antispam.textfilters: skipping processing message from channel %r we're not in", irc.name, target)
            return
    else:
        channel_or_none = None
        watch_pms = txf_settings.get('watch_pms', TEXTFILTER_DEFAULTS['watch_pms'])

        if watch_pms == 'services':
            if not irc.get_service_bot(target):
                log.debug("(%s) antispam.textfilters: skipping processing; %r is not a service bot (watch_pms='services')", irc.name, target)
                return
        elif watch_pms == 'all':
            log.debug("(%s) antispam.textfilters: checking all PMs (watch_pms='all')", irc.name)
            pass
        else:
            # Not a channel.
            log.debug("(%s) antispam.textfilters: skipping processing; %r is not a channel and watch_pms is disabled", irc.name, target)
            return

    # Merge together global and local textfilter lists.
    txf_globs = set(conf.conf.get('antispam', {}).get('textfilter_globs', [])) | \
                set(irc.serverdata.get('antispam_textfilter_globs', []))

    punishment = txf_settings.get('punishment', TEXTFILTER_DEFAULTS['punishment']).lower()
    reason = txf_settings.get('reason', TEXTFILTER_DEFAULTS['reason'])

    if irc.get_service_option('antispam', 'strip_formatting', True):
        text = utils.strip_irc_formatting(text)
    if txf_settings.get('munge_unicode', TEXTFILTER_DEFAULTS['munge_unicode']):
        text = str.translate(text, UNICODE_CHARMAP)

    punished = False
    for filterglob in txf_globs:
        if utils.match_text(filterglob, text):
            log.info("(%s) antispam: punishing %s => %s for text filter %r",
                     irc.name,
                     irc.get_friendly_name(source),
                     irc.get_friendly_name(target),
                     filterglob)
            punished = _punish(irc, source, channel_or_none, punishment, reason)
            break

    return not punished  # Filter this message from relay, etc. if it triggered protection

utils.add_hook(handle_textfilter, 'PRIVMSG', priority=999)
utils.add_hook(handle_textfilter, 'NOTICE', priority=999)

PARTQUIT_DEFAULTS = {
    'watch_quits': True,
    'watch_parts': True,
    'part_filter_message': "Reason filtered",
    'quit_filter_message': "Reason filtered",
}
def handle_partquit(irc, source, command, args):
    """Antispam part/quit message filter."""
    text = args.get('text')
    pq_settings = irc.get_service_option('antispam', 'partquit',
                                         PARTQUIT_DEFAULTS)

    if not text:
        return  # No text to match against
    elif command == 'QUIT' and not pq_settings.get('watch_quits', True):
        return  # Not enabled
    elif command == 'PART' and not pq_settings.get('watch_parts', True):
        return

    # Merge together global and local partquit filter lists.
    pq_globs = set(conf.conf.get('antispam', {}).get('partquit_globs', [])) | \
               set(irc.serverdata.get('antispam_partquit_globs', []))
    if not pq_globs:
        return

    for filterglob in pq_globs:
        if utils.match_text(filterglob, text):
            # For parts, also log the affected channels
            if command == 'PART':
                filtered_message = pq_settings.get('part_filter_message', PARTQUIT_DEFAULTS['part_filter_message'])
                log.info('(%s) antispam: filtered part message from %s on %s due to part/quit filter glob %s',
                         irc.name, irc.get_hostmask(source), ','.join(args['channels']), filterglob)
            else:
                filtered_message = pq_settings.get('quit_filter_message', PARTQUIT_DEFAULTS['quit_filter_message'])
                log.info('(%s) antispam: filtered quit message from %s due to part/quit filter glob %s',
                         irc.name, args['userdata'].nick, filterglob)
            args['text'] = filtered_message
            break
utils.add_hook(handle_partquit, 'PART', priority=999)
utils.add_hook(handle_partquit, 'QUIT', priority=999)