splunk-security_content/bin/newcontent.py

#!/usr/bin/python

'''
Helps you create new Splunk Security Content.
'''

from pathlib import Path
from PyInquirer import prompt, Separator
import os
import getpass
from jinja2 import Environment, FileSystemLoader
import uuid
from datetime import date
from os import path
import sys


def detection_wizard(security_content_path,type,TEMPLATE_PATH):
    questions = [
        {
            # get provider
            'type': 'list',
            'message': 'what kind of detection is this',
            'name': 'detection_kind',
            'choices': [
                {
                    'name': 'endpoint'
                },
                {
                    'name': 'cloud'
                },
                {
                    'name': 'application'
                },
                {
                    'name': 'network'
                },
                {
                    'name': 'web'
                },
                {
                    'name': 'experimental'
                },

            ],
            'default': 'endpoint'
        },
        {
            'type': 'input',
            'message': 'enter detection name',
            'name': 'detection_name',
            'default': 'Powershell Encoded Command',
        },
        {
            'type': 'input',
            'message': 'enter author name',
            'name': 'detection_author',
        },
        {
            # get provider
            'type': 'list',
            'message': 'select a detection type',
            'name': 'detection_type',
            'choices': [
                {
                    'name': 'batch'
                },
                {
                    'name': 'streaming'
                },
            ],
            'default': 'batch'
        },
        {
            # get provider
            'type': 'checkbox',
            'message': 'select the datamodels used in the detection',
            'name': 'datamodels',
            'choices': [
                {
                    'name': 'Endpoint',
                    'checked': True
                },
                {
                    'name': 'Network_Traffic'
                },
                {
                    'name': 'Authentication'
                },
                {
                    'name': 'Change'
                },
                {
                    'name': 'Change_Analysis'
                },
                {
                    'name': 'Email'
                },
                {
                    'name': 'Network_Resolution'
                },
                {
                    'name': 'Network_Traffic'
                },
                {
                    'name': 'Network_Sessions'
                },
                {
                    'name': 'Updates'
                },
                {
                    'name': 'Vulnerabilities'
                },
                {
                    'name': 'Web'
                },
            ],
        },
        {
            # get api_key
            'type': 'input',
            'message': 'enter search (spl)',
            'name': 'detection_search',
            'default': '| UPDATE_SPL'
        },
        {
            # get api_key
            'type': 'input',
            'message': 'enter MITRE ATT&CK Technique IDs related to the detection, comma delimited for multiple',
            'name': 'mitre_attack_ids',
            'default': 'T1003.002'
        },
        {
            # get provider
            'type': 'checkbox',
            'message': 'select kill chain phases related to the detection',
            'name': 'kill_chain_phases',
            'choices': [

                {
                    'name': 'Reconnaissance'
                },
                {
                    'name': 'Intrusion'
                },
                {
                    'name': 'Exploitation',
                    'checked': True
                },
                {
                    'name': 'Privilege Escalation'
                },
                {
                    'name': 'Lateral Movement'
                },
                {
                    'name': 'Obfuscation'
                },
                {
                    'name': 'Denial of Service'
                },
                {
                    'name': 'Exfiltration'
                },
            ],
        },
        {
            # get provider
            'type': 'list',
            'message': 'security_domain for detection',
            'name': 'security_domain',
            'choices': [
                {
                    'name': 'access'
                },
                {
                    'name': 'endpoint'
                },
                {
                    'name': 'network'
                },
                {
                    'name': 'threat'
                },
                {
                    'name': 'identity'
                },
                {
                    'name': 'audit'
                },

            ],
            'default': 'endpoint'
        },
    ]

    answers = prompt(questions)

    mitre_attack_id = [x.strip() for x in answers['mitre_attack_ids'].split(',')]

    print(mitre_attack_id)

    j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
                     trim_blocks=True)

    if answers['detection_type'] == 'batch':
        answers['products'] = ['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud']
    elif answers['detection_type'] == 'streaming':
        answers['products'] = ['UEBA for Security Cloud']

    # grab some vars for the test
    detection_kind = answers['detection_kind']


    # write a detection example
    template = j2_env.get_template('detection.j2')
    detection_name = answers['detection_name']
    detection_file_name =  detection_name.replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
    output_path = path.join(security_content_path, 'detections/' + detection_kind + '/' + detection_file_name + '.yml')
    output = template.render(uuid=uuid.uuid1(), date=date.today().strftime('%Y-%m-%d'),
    author=answers['detection_author'], name=answers['detection_name'],
    description='UPDATE_DESCRIPTION', how_to_implement='UPDATE_HOW_TO_IMPLEMENT', known_false_positives='UPDATE_KNOWN_FALSE_POSITIVES',
    references='',datamodels=answers['datamodels'],
    search= answers['detection_search'] + ' | `' + detection_file_name + '_filter`',
    type=answers['detection_type'], analytic_story_name='UPDATE_STORY_NAME', mitre_attack_id=mitre_attack_id,
    kill_chain_phases=answers['kill_chain_phases'], dataset_url='UPDATE_DATASET_URL',
    products=answers['products'], security_domain=answers['security_domain'])
    with open(output_path, 'w', encoding="utf-8") as f:
        f.write(output)

    print("\n> contentctl wrote the detection to: {0}\n".format(output_path))

    questions = [
        {
            'type': 'confirm',
            'message': 'would you like to configure the test file for detection: {0}'.format(answers['detection_name']),
            'name': 'continue',
            'default': True,
        },
        {
            'type': 'input',
            'message': 'enter pass condition for the test of detection: {0}'.format(answers['detection_name']),
            'name': 'pass_condition',
            'default': '| stats count | where count > 0',
            'when': lambda answers: answers['continue'],
        },
        {
            'type': 'input',
            'message': 'enter earliest_time for the test of detection: {0}'.format(answers['detection_name']),
            'name': 'earliest_time',
            'default': '-24h',
            'when': lambda answers: answers['continue'],
        },
        {
            'type': 'input',
            'message': 'enter latest_time for the test of detection: {0}'.format(answers['detection_name']),
            'name': 'latest_time',
            'default': 'now',
            'when': lambda answers: answers['continue'],
        },
    ]


    answers = prompt(questions)
    if answers['continue']:
        # and a corresponding test files
        template = j2_env.get_template('test.j2')
        test_name = detection_file_name + '.test.yml'
        output_path = path.join(security_content_path, 'tests/' + detection_kind + '/' + test_name)
        output = template.render(name=detection_name + ' Unit Test',
        detection_name=detection_name,
        detection_path='detections/' + detection_kind + '/' + detection_file_name + '.yml', pass_condition=answers['pass_condition'],
        earliest_time=answers['earliest_time'], latest_time=answers['latest_time'], file_name='UPDATE_FILE_NAME',
        splunk_source='UPDATE_SPLUNK_SOURCE',splunk_sourcetype='UPDATE_SPLUNK_SOURCETYPE',dataset_url='UPDATE_DATASET_URL')
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
    else:
        # and a corresponding test files
        template = j2_env.get_template('test.j2')
        test_name = detection_file_name + '.test.yml'
        output_path = path.join(security_content_path, 'tests/' + detection_kind + '/' + test_name)
        output = template.render(name=detection_name + ' Unit Test',
        detection_name=detection_name,
        detection_path=detection_kind + '/' + detection_file_name + '.yml', pass_condition='| stats count | where count > 0',
        earliest_time='-24h', latest_time='now',file_name='UPDATE_FILE_NAME', splunk_source='UPDATE_SPLUNK_SOURCE',
        splunk_sourcetype='UPDATE_SPLUNK_SOURCETYPE', dataset_url='UPDATE_DATASET_URL' )
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
    print("\n> contentctl wrote the test for this detection to: {0}\n".format(output_path))

def story_wizard(security_content_path,type, TEMPLATE_PATH):
    questions = [
        {
            'type': 'input',
            'message': 'enter story name',
            'name': 'story_name',
            'default': 'Suspicious Powershell Behavior',
        },
        {
            'type': 'input',
            'message': 'enter author name',
            'name': 'story_author',
        },
        {
            'type': 'list',
            'message': 'select a story type',
            'name': 'story_type',
            'choices': [
                {
                    'name': 'batch'
                },
                {
                    'name': 'streaming'
                },
            ],
            'default': 'batch'
        },
        {
            'type': 'checkbox',
            'message': 'select a category',
            'name': 'category',
            'choices': [
                {
                    'name': 'Adversary Tactics',
                    'checked': True
                },
                {
                    'name': 'Account Compromise'
                },
                {
                    'name': 'Unauthorized Software'
                },
                {
                    'name': 'Best Practices'
                },
                {
                    'name': 'Cloud Security'
                },
                {
                    'name': 'Command and Control'
                },
                {
                    'name': 'Lateral Movement'
                },
                {
                    'name': 'Ransomware'
                },
                {
                    'name': 'Privilege Escalation'
                },
                ],
            },
            {
                # get provider
                'type': 'list',
                'message': 'select a use case',
                'name': 'usecase',
                'choices': [
                    {
                        'name': 'Advanced Threat Detection',
                        'checked': True
                    },
                    {
                        'name': 'Security Monitoring'
                    },
                    {
                        'name': 'Compliance'
                    },
                    {
                        'name': 'Insider Threat'
                    },
                    {
                        'name': 'Application Security'
                    },
                    {
                        'name': 'Other'
                    },
            ],
        },
    ]
    answers = prompt(questions)
    j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
                     trim_blocks=True)
    if answers['story_type'] == 'batch':
        answers['products'] = ['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud']
    elif answers['story_type'] == 'streaming':
        answers['products'] = ['UEBA for Security Cloud']

    template = j2_env.get_template('story.j2')
    story_name = answers['story_name']
    story_file_name =  story_name.replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
    output_path = path.join(security_content_path, 'stories/' + story_file_name + '.yml')
    output = template.render(uuid=uuid.uuid1(), date=date.today().strftime('%Y-%m-%d'),
    author=answers['story_author'], name=answers['story_name'], description='UPDATE_DESCRIPTION',
    narrative='UPDATE_NARRATIVE', references=['https://www.destroyallsoftware.com/talks/wat'],
    type=answers['story_type'], analytic_story_name=answers['story_name'],
    categories=answers['category'], usecase=answers['usecase'], products=answers['products'])
    with open(output_path, 'w', encoding="utf-8") as f:
        f.write(output)
    print("contentctl wrote a example story to: {0}".format(output_path))

def create_example(security_content_path,type, TEMPLATE_PATH):
    getpass.getuser()
    j2_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH),
                         trim_blocks=True)

    if type == 'detection':


        # write a detection example
        template = j2_env.get_template('detection.j2')
        detection_name = getpass.getuser() + '_' + type + '.yml.example'
        output_path = path.join(security_content_path, 'detections/endpoint/' + detection_name)
        output = template.render(uuid=uuid.uuid1(), date=date.today().strftime('%Y-%m-%d'),
        author='UPDATE_AUTHOR', name=getpass.getuser().capitalize() + ' ' + type.capitalize(),
        description='UPDATE_DESCRIPTION',
        how_to_implement='UPDATE_HOW_TO_IMPLENT',
        known_false_positives='UPDATE_KNOWN_FALSE_POSITIVES',
        references=['https://html5zombo.com/'],
        datamodels=['Endpoint'], search='| UPDATE_SPL | `' + getpass.getuser() + '_' + type + '_filter`',
        type='batch', analytic_story_name=' UPDATE_STORY_NAME', mitre_attack_id = 'T1003.01',
        kill_chain_phases=['Exploitation'], dataset_url='UPDATE_DATASET_URL',
        products=['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud'])
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
        print("contentctl wrote a example detection to: {0}".format(output_path))

        # and a corresponding test files
        template = j2_env.get_template('test.j2')
        test_name = getpass.getuser() + '_' + type + '.test.yml.example'
        output_path = path.join(security_content_path, 'tests/endpoint/' + test_name)
        output = template.render(name=getpass.getuser().capitalize() + ' ' + type.capitalize() + ' Unit Test',
        detection_name=getpass.getuser().capitalize() + ' ' + type.capitalize(),
        detection_path='endpoint/' + detection_name, pass_condition='| stats count | where count > 0',
        earliest_time='-24h', latest_time='now', file_name='UPDATE_FILE_NAME', splunk_source='UPDATE_SPLUNK_SOURCE',
        splunk_sourcetype='UPDATE_SPLUNK_SOURCETYPE',dataset_url='UPDATE_DATASET_URL')
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
        print("contentctl wrote a example test for this detection to: {0}".format(output_path))

    elif type == 'story':
        # write a story example
        template = j2_env.get_template('story.j2')
        story_name = getpass.getuser() + '_' + type + '.yml.example'
        output_path = path.join(security_content_path, 'stories/' + story_name)
        output = template.render(uuid=uuid.uuid1(), date=date.today().strftime('%Y-%m-%d'),
        author='UPDATE_AUTHOR', name=getpass.getuser().capitalize() + ' ' + type.capitalize(),
        description='UPDATE_DESCRIPTION',
        narrative='UPDATE_NARRATIVE',
        references=['https://www.destroyallsoftware.com/talks/wat'],
        type='batch', analytic_story_name=getpass.getuser().capitalize() + ' ' + type.capitalize(),
        categories=['Adversary Tactics'], usecase='Advanced Threat Detection', products=['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud'])
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
        print("contentctl wrote a example story to: {0}".format(output_path))

    elif type == 'baseline':
        # write a baseline example
        template = j2_env.get_template('baseline.j2')
        baseline_name = getpass.getuser() + '_' + type + '.yml.example'
        output_path = path.join(security_content_path, 'baselines/' + baseline_name)
        output = template.render(uuid=uuid.uuid1(), date=date.today().strftime('%Y-%m-%d'),
        author='UPDATE_AUTHOR', name=getpass.getuser().capitalize() + ' ' + type.capitalize(),
        description='UPDATE_DESCRIPTION',
        how_to_implement='UPDATE_HOW_TO_IMPLENT',
        known_false_positives='UPDATE_KNOWN_FALSE_POSITIVES',
        references=['https://html5zombo.com/'],
        datamodels=['Endpoint'], search='| UPDATE_SPL',
        type='batch', analytic_story_name='UPDATE_STORY_NAME',
        detection_name = 'UPDATE_DETECTION_NAME', dataset_url='UPDATE_DATASET_URL',
        products=['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud'])
        with open(output_path, 'w', encoding="utf-8") as f:
            f.write(output)
        print("contentctl wrote a example baseline to: {0}".format(output_path))


def new(security_content_path, VERBOSE, type, example_only):

    valid_content_objects = ['detection','story', 'baseline']
    if type not in valid_content_objects:
        print("ERROR: content type: {0} is not valid, please use: {1}".format(type, str(valid_content_objects)))
        sys.exit(1)

    TEMPLATE_PATH = path.join(security_content_path, 'bin/jinja2_templates')

    if example_only:
        create_example(security_content_path,type, TEMPLATE_PATH)
        sys.exit(0)

    if type == 'detection':
        detection_wizard(security_content_path, type, TEMPLATE_PATH)
    elif type == 'story':
        story_wizard(security_content_path, type, TEMPLATE_PATH)

    print("WARNING do not forget to replace the UPDATE_* values with the correct information on the files!\ncompleted..")