surimisp

#!/usr/bin/python3
"""
Copyright(C) 2014-2018, Stamus Networks
Written by Eric Leblond <eleblond@stamus-networks.com>

"""

import argparse
import logging
import time
import json
from pygtail import Pygtail
from subprocess import call
import requests
import os
import sys
import yaml
from elasticsearch import Elasticsearch
import redis
from datetime import datetime, timedelta
import signal

from threading import *
from queue import *

MISP_URLS = { 'hostname': 'attributes/text/download/hostname', 'domain': 'attributes/text/download/domain',
         'url': 'attributes/text/download/url' }

q = Queue()

must_exit = False

have_daemon = True
try:
    import daemon
except:
    logging.warning("No daemon support available, install python-daemon if feature is needed")
    have_daemon = False

conf_parser = argparse.ArgumentParser(add_help = False)
conf_parser.add_argument("-c", "--conf_file",
                         metavar="FILE")
args, remaining_argv = conf_parser.parse_known_args()


config = {}
if args.conf_file:
    with open(args.conf_file, 'r') as conffile:
        config = yaml.load(conffile)

def get_from_conf(config, key, def_value):
    if key in config:
        return config[key]
    else:
        return def_value

PROXY_PARAMS = get_from_conf(config, 'proxy_params', None)
WHITELIST = get_from_conf(config, 'whitelist', [])

parser = argparse.ArgumentParser(description='Suricata MISP IOC script')
parser.add_argument("-c", "--conf_file", help='Set config file to use')
parser.add_argument('-f', '--files', default=get_from_conf(config, 'files', '/var/log/suricata/eve.json'), help='JSON files to monitor', nargs='+')
parser.add_argument('-a', '--alerts', default=get_from_conf(config, 'alerts', '/var/log/suricata/ioc.json'), help='JSON file to store events to')
parser.add_argument('-v', '--verbose', default=get_from_conf(config, 'verbose', False), action="count", help="Show verbose output, use multiple times increase verbosity")
parser.add_argument('-l', '--log', default=get_from_conf(config, 'log', None), help='File to log output to (default to stdout)')
parser.add_argument('-b', '--batch', default=get_from_conf(config, 'batch', False), action="store_true", help="Read file and exit at end")
parser.add_argument('-w', '--workers', default=get_from_conf(config, 'workers', 1), type=int, help='Number of alert workers to start')
parser.add_argument('-u', '--url', default=get_from_conf(config, 'url', None), help='Set option to url where JSON file to monitor, if unset no refresh')
parser.add_argument('-e', '--elasticsearch', default=get_from_conf(config, 'elasticsearch', None), help='Set elasticsearch server and use it as input, if unset use file')
parser.add_argument('-r', '--redis', default=get_from_conf(config, 'redis', None), help='Set redis server and use it as input, if unset use file')
parser.add_argument('-i', '--interval', default=get_from_conf(config, 'interval', 3600), type=int, help='Interval between file update in second')
parser.add_argument('-d', '--basedir', default=get_from_conf(config, 'basedir', '/var/lib/surimisp/'), help='Directory where data will stay')
parser.add_argument('-k', '--apikey', default=get_from_conf(config, 'apikey', None), help='API key to use')
parser.add_argument('-S', '--strict', default=get_from_conf(config, 'strict', False), action="store_true", help='Be strict on TLS checks')

ALERT_SUBOBJECT = { "hostname": { "action": "allowed", "category": "Misc Attack", "gid": 1, "rev": 4, "severity": 3, "signature": "IOC alert on HTTP/TLS hostname", "signature_id": 1 },
"domain": { "action": "allowed", "category": "Misc Attack", "gid": 1, "rev": 4, "severity": 3, "signature": "IOC alert on DNS request name", "signature_id": 2 },
"url": { "action": "allowed", "category": "Misc Attack", "gid": 1, "rev": 4, "severity": 3, "signature": "IOC alert on HTTP url", "signature_id": 3 },
"ip": { "action": "allowed", "category": "Misc Attack", "gid": 1, "rev": 4, "severity": 3, "signature": "IOC alert on IP address", "signature_id": 4 } }

if have_daemon:
    parser.add_argument('-D', '--daemon', default=False, action="store_true", help="Run as unix daemon")

args = parser.parse_args(remaining_argv)

if args.url and not args.apikey:
    print("URL specified and no API key aborting.")
    sys.exit(1)

if args.verbose >= 3:
    loglevel=logging.DEBUG
elif args.verbose >= 2:
    loglevel=logging.INFO
elif args.verbose >= 1:
    loglevel=logging.WARNING
else:
    loglevel=logging.ERROR

hostname_list = None
domain_list = None
url_list = None
count = { 'hostname': 0, 'url': 0, 'domain': 0 }

def setup_logging(args):
    if args.log:
        logging.basicConfig(filename=args.log,
                            format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                            level=loglevel)
    else:
        logging.basicConfig(level=loglevel)

def fetch_data(baseurl, basedir):
    session = requests.Session()
    session.headers.update(
            {'Authorization': args.apikey }
            )
    for url in MISP_URLS:
        resp = session.get(baseurl + MISP_URLS[url], proxies = PROXY_PARAMS, verify = args.strict)
        fpath = os.path.join(basedir, url)
        fdata = open(fpath, 'w')
        fdata.write(resp.content.decode('utf-8'))
        fdata.close()

def load_data(dfile):
    iocfile = open(dfile, 'r')
    entries = []
    for line in iocfile:
        entry = line.rstrip('\n')
        skip = False
        for domain in WHITELIST:
            if entry.endswith(domain):
                skip = True
                break
        if not skip:
            entries.append(entry)
    return set(entries)

def load_all_data(basedir):
    global hostname_list
    global domain_list
    global url_list
    hostname_list = load_data(os.path.join(basedir, 'hostname'))
    domain_list = load_data(os.path.join(basedir, 'domain'))
    url_list = load_data(os.path.join(basedir, 'url'))
 
def check_http(event, queue = None):
    global count
    if queue == None:
        queue = q
    try:
        if event['http']['hostname'] in hostname_list:
           event['ioc'] = 'hostname'
           event['alert'] = ALERT_SUBOBJECT['hostname']
           queue.put(event)
           count['hostname'] = count['hostname'] + 1
        if event['http']['url'] in url_list:
           event['ioc'] = 'url'
           event['alert'] = ALERT_SUBOBJECT['url']
           queue.put(event)
           count['url'] = count['url'] + 1
    except:
        pass

def check_dns(event, queue = None):
    if queue == None:
        queue = q
    try:
        if event['dns']['rrname'] in domain_list:
           event['ioc'] = 'domain'
           event['alert'] = ALERT_SUBOBJECT['domain']
           queue.put(event)
           count['domain'] = count['domain'] + 1
    except:
        pass

def check_tls(event, queue = None):
    if queue == None:
        queue = q
    try:
        if event['tls']['sni'] in hostname_list:
           event['ioc'] = 'hostname'
           event['alert'] = ALERT_SUBOBJECT['hostname']
           queue.put(event)
           count['hostname'] = count['hostname'] + 1
    except:
        pass

def AlertSender(mode = 'file', alerts = None, queue = None):
    global must_exit
    while True:
        event = queue.get()
        # Switch event to alert
        event['event_type'] = 'alert'
        if mode == 'file':
            alerts.write(json.dumps(event) + '\n')
            alerts.flush()
        queue.task_done()
        if must_exit:
            return


def FetchData(interval = 3600, url = None, basedir = None):
    while 1:
        time.sleep(float(interval))
        logging.info("Updating IOC lists")
        fetch_data(url, basedir)
        load_all_data(basedir)

def parse_source_lines(source, queue = None):
   for line in source:
       try:
           event = json.loads(line)
       except json.decoder.JSONDecodeError:
           continue
       if 'event_type' in event:
           if event['event_type'] == 'http':
               check_http(event, queue = queue)
           elif event['event_type'] == 'dns':
               check_dns(event, queue = queue)
           elif event['event_type'] == 'tls':
               check_tls(event, queue = queue)

def TreatJsonFile(args = None, source = None, queue = None):
    global must_exit
    if args.batch:
        for logfile in args.files:
            source = open(logfile, 'r')
            start_time = time.clock()
            parse_source_lines(source, queue = queue)
            end_time = time.clock()
            logging.info("Matching on '%s' took %fs" % (logfile, end_time - start_time))
            logging.info("Count: " + repr(count))
    else:
        while 1:
            logfile = Pygtail(source)
            parse_source_lines(logfile, queue = queue)
            if must_exit:
                return
            time.sleep(0.1)

def treat_redis_publisher(args):
    r = redis.StrictRedis(host=args.redis)
    p = r.pubsub()
    # FIXME hardcoded ....
    p.psubscribe('logstash-http', 'logstash-dns')
    while 1:
        msg = p.get_message()
        if msg:
            if msg['type'] == 'pmessage':
                event = json.loads(msg['data'])
                if 'event_type' in event:
                    if event['event_type'] == 'http':
                        check_http(event)
                    elif event['event_type'] == 'dns':
                        check_dns(event)
        else:
            time.sleep(0.1)


def treat_redis(args):
    r = redis.StrictRedis(host=args.redis)
    while 1:
        # FIXME hardcoded ....
        msg = r.rpoplpush('logstash-events', 'logstash')
        if msg:
            event = json.loads(msg)
            if 'event_type' in event:
                if event['event_type'] == 'http':
                    check_http(event)
                elif event['event_type'] == 'dns':
                    check_dns(event)
        else:
            time.sleep(0.1)

def treat_elasticsearch(args):
    es = Elasticsearch([args.elasticsearch])
    # FIXME real date
    orig_timestamp_str = '2014-03-08T13:43:22.551756'
    orig_timestamp = datetime.strptime(orig_timestamp_str,'%Y-%m-%dT%H:%M:%S.%f')
    end_timestamp = orig_timestamp + timedelta(hours = 1)

    query = '(event_type:http OR event_type:dns OR event_type:tls) AND timestamp:["%s" TO "%s"}' % (orig_timestamp.strftime('%Y-%m-%dT%H:%M:%S.%f'), end_timestamp.strftime('%Y-%m-%dT%H:%M:%S.%f'))
    results = es.search(q = query, index='_all', ignore_unavailable = True, scroll = 30, size = 20, search_type = 'scan')

    scroll_id = results['_scroll_id']

    while 1:
        for entry in results['hits']['hits']:
            event = entry['_source']
            if 'event_type' in event:
                if event['event_type'] == 'http':
                    check_http(event)
                elif event['event_type'] == 'dns':
                    check_dns(event)
        print("Scrolling once more")
        results = es.scroll(scroll_id = scroll_id, scroll = 10)
        print((results['hits']['hits'][0]))

def sigterm_handler(_signo, _stack_frame):
    global must_exit
    logging.info("Exiting program due to signal")
    must_exit = True

def main_task(args):
    global must_exit
    setup_logging(args)

    if args.url:
        fetch_data(args.url,args.basedir)
        if args.interval:
            t = Thread(target=FetchData, kwargs = {'interval': args.interval, 'url': args.url, 'basedir': args.basedir })
            t.daemon = True
            t.start()

    load_all_data(args.basedir)

    signal.signal(signal.SIGTERM, sigterm_handler)
    signal.signal(signal.SIGINT, sigterm_handler)

    if 'instances' in config:
        for instance in list(config['instances'].keys()):
            inst_config = config['instances'][instance]
            alerts = open(inst_config['alerts'], 'a+')
            queue = Queue()
            for i in range(args.workers):
                t = Thread(target=AlertSender, kwargs = {'mode': 'file', 'alerts': alerts, 'queue': queue })
                t.daemon = True
                t.start()
            for logfile in inst_config['files']:
                logging.info("Add reader for file '%s' with output '%s'" % ( logfile, inst_config['alerts'] ))
                t = Thread(target=TreatJsonFile, kwargs = {'source': logfile, 'args': args, 'queue': queue})
                t.daemon = True
                t.start()
        while 1:
            if must_exit:
                break
            try:
                time.sleep(0.1)
            except KeyboardInterrupt:
                must_exit = True
                return
        return

    alerts = open(args.alerts, 'a+')

    for i in range(args.workers):
        t = Thread(target=AlertSender, kwargs = {'mode': 'file', 'alerts': alerts, 'queue': q })
        t.daemon = True
        t.start()

    if args.batch:
        start_time = time.clock()

    if args.batch:
        end_time = time.clock()
        logging.info("Building sets took %fs" % (end_time - start_time))

    if args.redis:
        treat_redis(args)
    elif args.elasticsearch:
        treat_elasticsearch(args)
    else:
        if args.batch:
            TreatJsonFile(args = args)
        else:
            for logfile in args.files:
                t = Thread(target=TreatJsonFile, kwargs = {'source': logfile, 'args': args})
                t.daemon = True
                t.start()
            while 1:
                if must_exit:
                    break
                try:
                    time.sleep(0.1)
                except KeyboardInterrupt:
                    must_exit = True
                    return

if have_daemon and args.daemon:
    with daemon.DaemonContext():
        main_task(args)
else:
    main_task(args)