misp-modules/misp_modules/modules/import_mod/cof2misp.py

""" PassiveDNS Common Output Format (COF) MISP importer.

Takes as input a valid COF file or the output of the dnsdbflex utility
and creates MISP objects for the input.

Copyright 2021: Farsight Security (https://www.farsightsecurity.com/)

Author: Aaron Kaplan <aaron@lo-res.org>

Released under the Apache 2.0 license.
See: https://www.apache.org/licenses/LICENSE-2.0.txt

"""

import sys
import json
import base64


import ndjson

# from pymisp import MISPObject, MISPEvent, PyMISP
from pymisp import MISPObject

from cof2misp.cof import validate_cof


create_specific_attributes = False          # this is for https://github.com/MISP/misp-objects/pull/314


misperrors = {'error': 'Error'}
userConfig = {}

inputSource = ['file']

mispattributes = {'inputSource': ['file'], 'output': ['MISP objects'],
                  'format': 'misp_standard'}


moduleinfo = {'version': '0.2', 'author': 'Aaron Kaplan',
              'description': 'Module to import the passive DNS Common Output Format (COF) and merge as a MISP objet into a MISP event.',
              'module-type': ['import']}

moduleconfig = []


# misp = PyMISP()


def parse_and_insert_cof(data: str) -> dict:
    """Parse and validate the COF data.

    Parameters
    ----------
      data as a string

    Returns
    -------
      A dict with either the error message or the data which may be sent off the the caller of handler()

    Raises
    --------
      none. All Exceptions will be handled here. On error, a misperror is returned.
    """

    objects = []
    try:
        entries = ndjson.loads(data)
        for entry in entries:           # iterate over all ndjson lines

            # validate here (simple validation or full JSON Schema validation)
            if not validate_cof(entry):
                return {"error": "Could not validate the COF input '%s'" % entry}

            # Next, extract some fields
            rrtype = entry['rrtype'].upper()
            rrname = entry['rrname'].rstrip('.')
            rdata = [x.rstrip('.') for x in entry['rdata']]

            # create a new MISP object, based on the passive-dns object for each nd-JSON line
            o = MISPObject(name='passive-dns', standalone=False, comment='created by cof2misp')

            # o.add_tag('tlp:amber')                                    # FIXME: we'll want to add a tlp: tag to the object
            if 'bailiwick' in entry:
                o.add_attribute('bailiwick', value=entry['bailiwick'].rstrip('.'))

            #
            # handle the combinations of rrtype (domain, ip) on both left and right side
            #

            if create_specific_attributes:
                if rrtype in ['A', 'AAAA', 'A6']:                           # address type
                    # address type
                    o.add_attribute('rrname_domain', value=rrname)
                    for r in rdata:
                        o.add_attribute('rdata_ip', value=r)
                elif rrtype in ['CNAME', 'DNAME', 'NS']:                    # both sides are domains
                    o.add_attribute('rrname_domain', value=rrname)
                    for r in rdata:
                        o.add_attribute('rdata_domain', value=r)
                elif rrtype in ['SOA']:                                     # left side is a domain, right side is text
                    o.add_attribute('rrname_domain', value=rrname)

            #
            # now do the regular filling up of rrname, rrtype, time_first, etc.
            #
            o.add_attribute('rrname', value=rrname)
            o.add_attribute('rrtype', value=rrtype)
            for r in rdata:
                o.add_attribute('rdata', value=r)
            o.add_attribute('raw_rdata', value=json.dumps(rdata))       # FIXME: do we need to hex encode it?
            o.add_attribute('time_first', value=entry['time_first'])
            o.add_attribute('time_last', value=entry['time_last'])
            o.first_seen = entry['time_first']      # is this redundant?
            o.last_seen = entry['time_last']

            #
            # Now add the other optional values.                        # FIXME: how about a map() other function. DNRY
            #
            for k in ['count', 'sensor_id', 'origin', 'text', 'time_first_ms', 'time_last_ms', 'zone_time_first', 'zone_time_last']:
                if k in entry and entry[k]:
                    o.add_attribute(k, value=entry[k])

            #
            # add COF entry to MISP object
            #
            objects.append(o.to_json())

        r = {'results': {'Object': [json.loads(o) for o in objects]}}
    except Exception as ex:
        misperrors["error"] = "An error occured during parsing of input: '%s'" % (str(ex),)
        return misperrors
    return r


def parse_and_insert_dnsdbflex(data: str):
    """Parse and validate the more simplier dndsdbflex output data.

    Parameters
    ----------
      data as a string

    Returns
    -------
      A dict with either the error message or the data which may be sent off the the caller of handler()

    Raises
    --------
      none
    """
    return {"error": "NOT IMPLEMENTED YET"}            # XXX FIXME: need a MISP object for dnsdbflex


def is_dnsdbflex(data: str) -> bool:
    """Check if the supplied data conforms to the dnsdbflex output (which only contains rrname and rrtype)

    Parameters
    ----------
      ndjson data as a string

    Returns
    -------
      True or False

    Raises
    --------
      none
    """

    try:
        j = ndjson.loads(data)
        for line in j:
            if not set(line.keys()) == {'rrname', 'rrtype'}:
                return False            # shortcut. We assume it's not if a single line does not conform
        return True
    except Exception as ex:
        print("oops, this should not have happened. Maybe not an ndjson file? Reason: %s" % (str(ex),), file=sys.stderr)
        return False


def is_cof(data: str) -> bool:
    return True


def handler(q=False):
    if q is False:
        return False

    request = json.loads(q)
    # Parse the json, determine which type of JSON it is (dnsdbflex or COF?)
    # Validate it
    # transform into MISP object
    # push to MISP
    # event_id = request['event_id']
    # event = misp.get_event(event_id)
    # print("event_id = %s" % event_id, file=sys.stderr)
    try:
        data = base64.b64decode(request["data"]).decode('utf-8')
        if not data:
            return json.dumps({'success': 0})       # empty file is ok
        if is_dnsdbflex(data):
            return parse_and_insert_dnsdbflex(data)
        elif is_cof(data):
            # check if it's valid COF format
            return parse_and_insert_cof(data)
        else:
            return {'error': 'Could not find any valid COF input nor dnsdbflex input. Please have a loot at: https://datatracker.ietf.org/doc/draft-dulaunoy-dnsop-passive-dns-cof/'}
    except Exception as ex:
        print("oops, got exception %s" % str(ex), file=sys.stderr)
        return {'error': "Got exception %s" % str(ex)}


def introspection():
    return mispattributes


def version():
    moduleinfo['config'] = moduleconfig
    return moduleinfo


if __name__ == '__main__':
    x = open('test.json', 'r')
    r = handler(q=x.read())
    print(json.dumps(r))