path: root/bugzilla-junker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# A helper script to go through the latest comments added to a bugzilla
# to see if any of them link to external sites. If the reviewer deems them
# spammy, the script will tag them as such.
#
# Caution: work in progress
#
__author__ = 'Konstantin Ryabitsev <konstantin@linuxfoundation.org>'

import sys
import requests
import argparse
import logging
import re
import shelve
import datetime
import notify2
import time

from urllib.parse import urlparse
from configparser import ConfigParser

logger = logging.getLogger('default')

APIKEY = None
BZURL = None

REQSESSION = None

CACHEDATA = None


def notify_desktop(message):
    notify2.init('bugjunker')
    n = notify2.Notification('bugjunker', message)
    n.set_timeout(notify2.EXPIRES_NEVER)
    n.show()


def get_session():
    global REQSESSION
    if REQSESSION is None:
        REQSESSION = requests.session()
        REQSESSION.headers.update({'User-Agent': 'bugjunker'})
    return REQSESSION


def ban_hammer(spammers):
    params = {}
    for spammer in set(spammers):
        path = 'user/{spammer}'.format(spammer=spammer)
        logger.info('Banning %s', spammer)
        payload = {
            'email_enabled': False,
            'login_denied_text': 'Spammer',
        }
        bz_put(path, params, payload)


def tag_hammer(spamcids, spamtag):
    params = {}
    for cid in set(spamcids):
        logger.info('Tagging comment %s', cid)
        path = 'bug/comment/{cid}/tags'.format(cid=cid)
        payload = {
            'comment_id': cid,
            'add': [spamtag],
        }
        bz_put(path, params, payload)


def bug_hammer(spambugs, args):
    params = {}
    for bugid in set(spambugs):
        logger.info('Junking bug %s', bugid)
        path = 'bug/{bugid}'.format(bugid=bugid)
        payload = {
            'groups': {'add': [args.group]},
            'status': args.status,
            'resolution': args.resolution,
        }
        bz_put(path, params, payload)


def bz_get(path, params):
    url = '{BZURL}/rest/{path}'.format(BZURL=BZURL, path=path)
    params['api_key'] = APIKEY
    ses = get_session()
    res = ses.get(url, params=params)
    return res.json()


def bz_put(path, params, payload):
    url = '{BZURL}/rest/{path}'.format(BZURL=BZURL, path=path)
    params['api_key'] = APIKEY
    ses = get_session()
    res = ses.put(url, params=params, json=payload)
    return res.json()


def load_cache(cachefile):
    global CACHEDATA
    if CACHEDATA is not None:
        return CACHEDATA

    # noinspection PyBroadException
    try:
        with shelve.open(cachefile, 'r') as wc:
            logger.info('Loading cache from %s', cachefile)
            CACHEDATA = dict(wc)
    except:
        CACHEDATA = {
            'seencids': list(),
            'seenaids': list(),
            'okdomains': list(),
            'okfolks': list(),
        }
        pass

    if 'lastrun' in CACHEDATA:
        lastrun = CACHEDATA['lastrun']
    else:
        lastrun = '24h'

    return lastrun, CACHEDATA


def save_cache(cachefile, cachedata):
    with shelve.open(cachefile, 'c') as wc:
        for key, val in cachedata.items():
            wc[key] = val
        wc.sync()


def check_bad_urls(urls, okdomains):
    for url in urls:
        try:
            up = urlparse(url)
        except ValueError:
            return url, None
        isok = False
        for okd in okdomains:
            if okd == up.netloc:
                isok = True
                break
        if not isok:
            return url, up.netloc

    return None, None


def is_junk_attachment(attid):
    attid = str(attid)
    logger.info('    checking attachment %s', attid)
    path = 'bug/attachment/{attid}'.format(attid=attid)
    attinfo = bz_get(path, {})
    if attid not in attinfo['attachments']:
        return False
    attdata = attinfo['attachments'][attid]
    if attdata['is_patch'] or attdata['content_type'] in ('text/plain',):
        return False
    if attdata['content_type'] == 'text/html':
        # Almost certainly junk
        logger.info('    junking attachment %s', attid)
        payload = {
            'content_type': 'text/plain',
            'filename': 'caution.txt',
            'is_private': True,
        }
        bz_put(path, {}, payload)
        return True
    return False


def process_bugs(cmdargs, cachefile, c, bugs, spamtag):
    spammers = list()
    spamcids = list()
    spambugs = list()

    for bug in bugs:
        logger.info('Analyzing [%s]: %s', bug['id'], bug['summary'])
        params = {}
        bugid = bug['id']
        path = 'bug/{bugid}/comment'.format(bugid=bugid)
        comments = bz_get(path, params)
        for f1, f2 in comments['bugs'].items():
            c_count = -1
            for comment in f2['comments']:
                c_count += 1
                cid = comment['id']
                if cid in c['seencids']:
                    # already seen, skip
                    continue

                c['seencids'].append(cid)

                creator = comment['creator']
                if creator in c['okfolks']:
                    # Known good person
                    continue

                tags = comment['tags']
                if spamtag in tags:
                    # already marked as spammy
                    continue

                if creator in spammers:
                    # Made by a known spammer, banit
                    spamcids.append(cid)
                    logger.info('    auto-tagging comment by %s: %s', creator, cid)
                    continue

                if cmdargs.checkatt:
                    attid = comment['attachment_id']
                    if attid is not None and attid not in c['seenaids']:
                        c['seenaids'].append(attid)
                        if is_junk_attachment(attid):
                            logger.info('    check if spammer: %s', creator)

                # Look for remote URLs in the comment
                if bug['url'].find('http') > -1 or comment['text'].find('http') > -1:
                    urls = re.findall(r'(https?://\S+)', comment['text'])
                    if len(bug['url']):
                        urls.append(bug['url'])
                    badurl, baddomain = check_bad_urls(urls, c['okdomains'])

                    if badurl is not None:
                        if cmdargs.noninteractive:
                            notify_desktop('Spam in %s: %s' % (bug['id'], baddomain))
                            when = datetime.datetime.strptime(bug['last_change_time'], '%Y-%m-%dT%H:%M:%SZ')
                            c['seencids'].remove(cid)
                            c['lastrun'] = when.strftime('%Y-%m-%d %H:%M:%S')
                            c['needsinput'] = True
                            return spammers, spamcids, spambugs

                        logger.info('    ---')
                        logger.info('    suspish URL: %s', badurl)
                        logger.info('    checkit out: %s/show_bug.cgi?id=%s#c%s',
                                    BZURL, bugid, c_count)
                        baw = input('    (b)an, (a)llow, (w)hitelist %s: ' % baddomain)

                        if baw == 'a':
                            c['okfolks'].append(creator)
                            save_cache(cachefile, c)
                            continue

                        if baw == 'w':
                            logger.info('    whitelisted %s', baddomain)
                            c['okdomains'].append(baddomain)
                            c['okfolks'].append(creator)
                            save_cache(cachefile, c)
                            continue

                        logger.info('    spamcid: %s', cid)
                        spamcids.append(cid)

                        # If it's a comment #0, then the whole bug needs junking
                        if c_count == 0:
                            spambugs.append(bug['id'])

                        if creator not in spammers:
                            logger.info('    spammer: %s', creator)
                            spammers.append(creator)

    return spammers, spamcids, spambugs


def main(args):
    global BZURL
    global APIKEY

    logger.setLevel(logging.DEBUG)

    ch = logging.StreamHandler()
    formatter = logging.Formatter('%(message)s')
    ch.setFormatter(formatter)

    if args.quiet:
        ch.setLevel(logging.CRITICAL)
    elif args.debug:
        ch.setLevel(logging.DEBUG)
    else:
        ch.setLevel(logging.INFO)

    logger.addHandler(ch)

    logger.info('Loading configuration file %s', args.config)
    config = ConfigParser()
    config.read(args.config)
    BZURL = config.get('main', 'url')
    APIKEY = config.get('main', 'apikey')

    spamtag = config.get('main', 'spamtag')

    if config.get('main', 'logfile'):
        ch = logging.FileHandler(config.get('main', 'logfile'))
        fmt = '[%(process)d] %(asctime)s - %(message)s'
        ch.setFormatter(logging.Formatter(fmt))
        ch.setLevel(logging.INFO)
        logger.addHandler(ch)

    cachefile = config.get('main', 'cache')
    lastrun, c = load_cache(cachefile)


    if args.lookback is not None:
        lastrun = args.lookback

    while True:
        if args.noninteractive and 'needsinput' in c and c['needsinput']:
            logger.info('Need to run interactively to make some decisions')
            sys.exit(0)

        params = {
            'chfieldfrom': lastrun,
            'include_fields': 'id,summary,last_change_time,url',
        }
        logger.info('Querying %s for changes since %s', BZURL, lastrun)

        unow = datetime.datetime.utcnow()
        json = bz_get('bug', params)
        c['lastrun'] = unow.strftime('%Y-%m-%d %H:%M:%S')
        c['needsinput'] = False
        if len(json['bugs']):
            spammers, spamcids, spambugs = process_bugs(args, cachefile, c, json['bugs'], spamtag)

            if len(spammers) or len(spamcids) or len(spambugs):
                ban_hammer(spammers)
                tag_hammer(spamcids, spamtag)
                bug_hammer(spambugs, args)
            else:
                logger.info('No new spam found')
        else:
            logger.info('No changes since %s', lastrun)

        save_cache(cachefile, c)

        if not args.sleep:
            sys.exit(0)

        logger.info('Sleeping %d seconds', args.sleep)
        time.sleep(args.sleep)


def cmd():
    description = 'Junk spammy bugzilla comments and ban their authors'
    parser = argparse.ArgumentParser(description=description, prog='bz-comment-junker.py')
    parser.add_argument('-c', '--config', required=True,
                        help='Configuration file')
    parser.add_argument('-q', '--quiet', action='store_true', default=False,
                        help='Output only errors')
    parser.add_argument('-d', '--debug', action='store_true', default=False,
                        help='Add debugging info')
    parser.add_argument('-l', '--lookback', default=None,
                        help='How far back to look (default: since last run, or 24h if no cached data)')
    parser.add_argument('-n', '--noninteractive', action='store_true', default=False,
                        help='Run non-interactively and send an alert when potential spam is found')
    parser.add_argument('-a', '--check-attachments', action='store_true', dest='checkatt', default=False,
                        help='Check attachments for junkiness')
    parser.add_argument('--sleep', type=int, default=0,
                        help='After the run, sleep N seconds and then run again')
    parser.add_argument('--status', default='RESOLVED',
                        help='Status value for junked bugs')
    parser.add_argument('--resolution', default='INVALID',
                        help='Resolution value for junked bugs')
    parser.add_argument('--product', default='Other',
                        help='Product value for junked bugs')
    parser.add_argument('--component', default='Spam',
                        help='Component value for junked bugs')
    parser.add_argument('--group', default='Junk',
                        help='Private group name for junked bugs')

    args = parser.parse_args()
    main(args)


if __name__ == '__main__':
    cmd()