splunk-malware-filter/bin/getopendbl.py

#!/usr/bin/env python

"""
Get IP blocklists from OpenDBL
Usage: "| getopendbl | outputlookup override_if_empty=false opendbl_ip.csv"
Recommend to update the lookup file every 15 minutes (cron "*/15 * * * *")
"""

import sys
from datetime import datetime, timezone
from os import path
from re import search

sys.path.insert(0, path.join(path.dirname(__file__), "..", "lib"))

from splunklib.searchcommands import Configuration, GeneratingCommand, Option, dispatch
from utils import Utility

OPENDBL_LIST = {
    "Emerging Threats: Known Compromised Hosts": "etknown.list",
    "TOR exit nodes": "tor-exit.list",
    "BruteforceBlocker": "bruteforce.list",
    "Blocklist.de All": "blocklistde-all.list",
    "Talos": "talos.list",
    "Dshield": "dshield.list",
    "SSL Abuse IP list": "sslblock.list",
}


@Configuration()
class GetOpenDBL(Utility, GeneratingCommand):
    """Defines a search command that generates event records"""

    custom_message = Option(name="message")

    def generate(self):
        for name, dl_path in OPENDBL_LIST.items():
            blocklist = self.download(f"https://opendbl.net/lists/{dl_path}")
            updated = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
            # parse updated time from header comment
            for line in filter(lambda row: row[0] == "#", blocklist.splitlines()):
                if "Last updated" in line:
                    last_updated = search(
                        r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}", line
                    ).group()
                    # Assume UTC timezone
                    updated = (
                        datetime.strptime(last_updated, "%Y-%m-%d %H:%M")
                        .replace(tzinfo=timezone.utc)
                        .strftime("%Y-%m-%dT%H:%M:%SZ")
                    )
                    break

            for line in filter(lambda row: row[0] != "#", blocklist.splitlines()):
                row = {
                    "start": line,
                    "end": line,
                    "netmask": "32",
                    "cidr_range": f"{line}/32",
                    "name": name,
                    "updated": updated,
                }

                if "-" in line:
                    row["start"], row["end"] = line.split("-")
                    row["netmask"] = 24
                    row["cidr_range"] = f"{row['start']}/{row['netmask']}"

                if (
                    isinstance(self.custom_message, str)
                    and len(self.custom_message) >= 1
                ):
                    row["custom_message"] = self.custom_message

                yield self.gen_record(**row)


if __name__ == "__main__":
    dispatch(GetOpenDBL, sys.argv, sys.stdin, sys.stdout, __name__)