splunk-security_content/bin/validate_manifests.py

#!/usr/bin/python
'''
Take the manifest files and build files for Enterprise Security
'''
import glob
import json
import jsonschema
import sys
import argparse
from os import path

# HIGH Level Fields
STORY_CATEGORIES = [
    "Abuse", "Adversary Tactics", "Best Practices",
    "Cloud Security", "Malware", "Vulnerability"
]

VALID_SEARCH_TYPES = [
    "detection", "investigative", "contextual", "support"
]

VALID_DATA_MODELS = [
    "Alerts",
    "Application_State",
    "Authentication",
    "Certificates",
    "Change_Analysis",
    "Change",
    "Malware",
    "Email",
    "Identity_Management",
    "Network_Resolution",
    "Network_Traffic",
    "Vulnerabilities",
    "Web",
    "Network_Sessions",
    "Updates",
    "Risk",
    "Endpoint"]

RISK_OBJECCT_TYPE = [
    "system", "user", "other"
]

PROVIDING_TECHNOLOGIES = [
    "Apache", "Bro", "Microsoft Windows", "Linux", "macOS",
    "Netbackup", "Splunk Enterprise", "Splunk Enterprise Security",
    "Splunk Stream", "Active Directory", "Bluecoat",
    "Carbon Black Response", "Carbon Black Protect", "CrowdStrike Falcon",
    "Microsoft Exchange", "Nessus", "Palo Alto Firewall", "Qualys",
    "Sysmon", "Tanium", "Ziften", "AWS", "OSquery"
]

KILL_CHAIN_PHASES = [
    "Reconnaissance", "Weaponization", "Delivery", "Exploitation",
    "Installation", "Command and Control", "Actions on Objectives"
]

CIS_CONTROLS = [
    "CIS 1", "CIS 2", "CIS 3", "CIS 4", "CIS 5", "CIS 6", "CIS 7", "CIS 8",
    "CIS 9", "CIS 10", "CIS 11", "CIS 12", "CIS 13", "CIS 14", "CIS 15",
    "CIS 16", "CIS 17", "CIS 18", "CIS 19", "CIS 20"
]

NIST_FIELDS = [
    "ID.AM", "ID.RA", "PR.DS", "PR.IP", "PR.AC", "PR.PT", "PR.AT", "PR.MA",
    "DE.CM", "DE.DP", "DE.AE", "RS.MI", "RS.AN", "RS.RP", "RS.IM", "RS.CO",
    "RC.IM", "RC.CO"
]

MITRE_ATTACK_FIELDS = [
    "Persistence",
    "Privilege Escalation",
    "Defense Evasion",
    "Credential Access",
    "Discovery",
    "Lateral Movement",
    "Authentication Package",
    "Execution",
    "Collection",
    "Exfiltration",
    "Command and Control",
    "Accessibility Features",
    "Account Discovery",
    "AppInit DLLs",
    "Application Deployment Software",
    "Application Shimming",
    "Application Window Discovery",
    "Audio Capture",
    "Automated Collection",
    "Automated Exfiltration",
    "Basic Input/Output System",
    "Binary Padding",
    "Bootkit",
    "Brute Force",
    "Bypass User Account Control",
    "Change Default File Association",
    "Clipboard Data",
    "Code Signing",
    "Command-Line Interface",
    "Commonly Used Port",
    "Communication Through Removable Media",
    "Component Firmware",
    "Component Object Model Hijacking",
    "Connection Proxy",
    "Credential Dumping",
    "Credential Manipulation",
    "Credentials in Files",
    "Custom Command and Control Protocol",
    "Custom Cryptographic Protocol",
    "DLL Injection",
    "DLL Search Order Hijacking",
    "DLL Side-Loading",
    "Data Compressed",
    "Data Encrypted",
    "Data Obfuscation",
    "Data Staged",
    "Data Transfer Size Limits",
    "Data from Local System",
    "Data from Network Shared Drive",
    "Data from Removable Media",
    "Disabling Security Tools",
    "Email Collection",
    "Execution through API",
    "Exfiltration Over Alternative Protocol",
    "Exfiltration Over Command and Control Channel",
    "Exfiltration Over Other Network Medium",
    "Exfiltration Over Physical Medium",
    "Exploitation of Vulnerability",
    "Fallback Channels",
    "File Deletion",
    "File System Logical Offsets",
    "File System Permissions Weakness",
    "File and Directory Discovery",
    "Graphical User Interface",
    "Hypervisor",
    "Indicator Blocking",
    "Indicator Removal from Tools",
    "Indicator Removal on Host",
    "Input Capture",
    "InstallUtil",
    "Legitimate Credentials",
    "Local Network Configuration Discovery",
    "Local Network Connections Discovery",
    "Local Port Monitor",
    "Logon Scripts",
    "MSBuild",
    "Masquerading",
    "Modify Existing Service",
    "Modify Registry",
    "Multi-Stage Channels",
    "Multiband Communication",
    "Multilayer Encryption",
    "NTFS Extended Attributes",
    "Network Service Scanning",
    "Network Share Connection Removal",
    "Network Sniffing",
    "New Service",
    "Obfuscated Files or Information",
    "Pass the Hash",
    "Pass the Ticket",
    "Path Interception",
    "Peripheral Device Discovery",
    "Permission Groups Discovery",
    "PowerShell",
    "Process Discovery",
    "Process Hollowing",
    "Query Registry",
    "Redundant Access",
    "Registry Run Keys / Start Folder",
    "Regsvcs/Regasm",
    "Regsvr32",
    "Remote Desktop Protocol",
    "Create Account",
    "Remote File Copy",
    "Remote Services",
    "Remote System Discovery",
    "Replication Through Removable Media",
    "Rootkit",
    "Rundll32",
    "Scheduled Task",
    "Scheduled Transfer",
    "Screen Capture",
    "Scripting",
    "Security Software Discovery",
    "Security Support Provider",
    "Service Execution",
    "Service Registry Permissions Weakness",
    "Shared Webroot",
    "Shortcut Modification",
    "Software Packing",
    "Standard Application Layer Protocol",
    "Standard Cryptographic Protocol",
    "Standard Non-Application Layer Protocol",
    "System Information Discovery",
    "System Owner/User Discovery",
    "System Service Discovery",
    "System Time Discovery",
    "Taint Shared Content",
    "Third-party Software",
    "Timestomp",
    "Two-Factor Authentication Interception",
    "Uncommonly Used Port",
    "Video Capture",
    "Valid Accounts",
    "Web Service",
    "Web Shell",
    "Windows Admin Shares",
    "Windows Management Instrumentation Event Subscription",
    "Windows Management Instrumentation",
    "Windows Remote Management",
    "Winlogon Helper DLL",
    "Initial Access",
    "Exploitation for Privilege Escalation"]

ALL_UUIDS = []


def validate_search_manifest(search):
    '''Confirmsearch has the required fields for savedsearches.conf'''
    errors = []
    if search['search_id'] == '':
        errors.append('ERROR: Blank Search ID')

    if search['search_id'] in ALL_UUIDS:
        errors.append('ERROR: Duplicate UUID found: %s' % search['search_id'])
    else:
        ALL_UUIDS.append(search['search_id'])

    if search['search_name'].endswith(" "):
        errors.append(
            "ERROR: Search name has trailing spaces: '%s'" %
            search['search_name'])

    if search['search_type'] not in VALID_SEARCH_TYPES:
        errors.append("ERROR: Invalid search type: %s" % search['search_type'])

    if 'mappings' in search:
        if 'kill_chain_phases' in search['mappings']:
            for kill_chain in search['mappings']['kill_chain_phases']:
                if kill_chain not in KILL_CHAIN_PHASES:
                    errors.append(
                        'ERROR: Invalid kill chain phase: %s' %
                        kill_chain)

        if 'cis20' in search['mappings']:
            for cis in search['mappings']['cis20']:
                if cis not in CIS_CONTROLS:
                    errors.append('ERROR: Invalid CIS field: %s' % cis)

        if 'mitre_attack' in search['mappings']:
            for attack in search['mappings']['mitre_attack']:
                if attack not in MITRE_ATTACK_FIELDS:
                    errors.append('ERROR: Invalid ATT&CK label %s' % attack)

        if 'nist' in search['mappings']:
            for nist in search['mappings']['nist']:
                if nist not in NIST_FIELDS:
                    errors.append('ERROR: Invalid nist label %s' % nist)

    if 'data_models' in search['data_metadata']:
        for data_model in search['data_metadata']['data_models']:
            if data_model not in VALID_DATA_MODELS:
                errors.append('ERROR: Invalid data model: %s' % data_model)

            if data_model not in search['search']:
                errors.append(
                    'WARNING: Data model listed but is not in search: %s' %
                    data_model)

    if 'providing_technologies' in search['data_metadata']:
        for providing_technology in \
                search['data_metadata']['providing_technologies']:
            if providing_technology not in PROVIDING_TECHNOLOGIES:
                errors.append(
                    'ERRORS: Unknown product in providing technologies: %s' %
                    providing_technology)

    if search['search_type'] == 'detection':
        try:
            if search['correlation_rule']['risk']:
                for risk_object_type in \
                        search['correlation_rule']['risk']['risk_object_type']:
                    if risk_object_type not in RISK_OBJECCT_TYPE:
                        errors.append(
                            'ERRORS: Unknown risk object type, must be system, \
                                    user, or other got: %s' %
                            risk_object_type)
        except BaseException:
            errors.append('WARNING: correlation_rule is missing risk object')

    if 'providing_technologies' in search['data_metadata']:
        for providing_technology in \
                search['data_metadata']['providing_technologies']:
            if providing_technology not in PROVIDING_TECHNOLOGIES:
                errors.append(
                    'ERRORS: Unknown product in providing technologies: %s' %
                    providing_technology)
    # Check to see if datamodel or tstats is in the search.  If so, make sure
    # data_models is defined
    if '| tstats' in search['search'] or 'datamodel' in search['search']:
        if 'data_models' not in search['data_metadata']:
            errors.append(
                "The search uses a data model but 'data_models' \
                        field is not set")

        if 'data_models' in search and not \
                search['data_metadata']['data_models']:
            errors.append(
                "The search uses a data model but 'data_models' is empty")

    if 'sourcetype' in search['search']:
        if 'data_sourcetypes' not in search['data_metadata']:
            errors.append(
                "The search specifies a sourcetype but 'data_sourcetypes' \
                        field is not set")

        if 'data_sourcetypes' in search and not \
                search['data_metadata']['data_sourcetypes']:
            errors.append(
                "The search specifies a sourcetype but \
                        'data_sourcetypes' is empty")

    try:
        search['search_description'].encode('ascii')
    except UnicodeEncodeError:
        errors.append("search_description not ascii")

    if 'how_to_implement' in search:
        try:
            search['how_to_implement'].encode('ascii')
        except UnicodeEncodeError:
            errors.append("how_to_implement not ascii")

    if 'eli5' in search:
        try:
            search['eli5'].encode('ascii')
        except UnicodeEncodeError:
            errors.append("eli5 not ascii")

    if 'known_false_positives' in search:
        try:
            search['known_false_positives'].encode('ascii')
        except UnicodeEncodeError:
            errors.append("known_false_positives not ascii")

    if 'correlation_rule' in search and 'notable' in \
            search['correlation_rule']:
        try:
            search['correlation_rule']['notable']['rule_title'].encode('ascii')
        except UnicodeEncodeError:
            errors.append("rule_title not ascii")

        try:
            search['correlation_rule']['notable']['rule_description'].encode(
                'ascii')
        except UnicodeEncodeError:
            errors.append("rule_description not ascii")

    return errors


def validate_story_manifest(story):
    ''' Validate that the analytic story manifest is in the proper format '''
    errors = []

    if story['id'] == '':
        errors.append('ERROR: Blank Content ID')

    if story['id'] in ALL_UUIDS:
        errors.append('ERROR: Duplicate UUID found: %s' % story['id'])
    else:
        ALL_UUIDS.append(story['id'])

    try:
        story['description'].encode('ascii')
    except UnicodeEncodeError:
        errors.append("description not ascii")

    try:
        story['narrative'].encode('ascii')
    except UnicodeEncodeError:
        errors.append("narrative not ascii")

    if story['category'] not in STORY_CATEGORIES:
        errors.append('ERROR: Invalid category found: %s' % story['category'])
    return errors


def main():
    ''' Open manifest file, validate required fields '''

    errors = False
    story_manifests = {}
    story_schema_file = path.join(
        path.expanduser(MANIFEST_DIRECTORY),
        'spec/analytic_story.json.spec')
    story_schema = json.loads(open(story_schema_file, 'rb').read())
    story_manifest_files = path.join(path.expanduser(MANIFEST_DIRECTORY),
                                     "*/stories/*.json")
    for story_manifest_file in glob.glob(story_manifest_files):
        try:
            story_manifest_data = json.loads(
                open(story_manifest_file, 'r').read())
        except Exception as e:
            print "Error reading %s" % story_manifest_file
            errors = True
            continue

        try:
            jsonschema.validate(story_manifest_data, story_schema)
        except jsonschema.exceptions.ValidationError as json_ve:
            print story_manifest_file
            print json_ve.message
            errors = True
            continue

        story_errors = validate_story_manifest(story_manifest_data)
        if story_errors:
            errors = True
            for err in story_errors:
                print path.basename(story_manifest_file)
                print "\t%s" % err

        story_manifests[story_manifest_data['name']] = story_manifest_data

    detection_search_schema_file = path.join(
        path.expanduser(MANIFEST_DIRECTORY), 'spec/detection_search.json.spec')
    detection_search_schema = json.loads(
        open(detection_search_schema_file, 'rb').read())
    contextual_search_schema_file = path.join(
        path.expanduser(MANIFEST_DIRECTORY),
        'spec/contextual_search.json.spec')
    contextual_search_schema = json.loads(
        open(contextual_search_schema_file, 'rb').read())
    investigative_search_schema_file = path.join(
        path.expanduser(MANIFEST_DIRECTORY),
        'spec/investigative_search.json.spec')
    investigative_search_schema = json.loads(
        open(investigative_search_schema_file, 'rb').read())
    support_search_schema_file = path.join(
        path.expanduser(MANIFEST_DIRECTORY), 'spec/support_search.json.spec')
    support_search_schema = json.loads(
        open(support_search_schema_file, 'rb').read())

    search_manifests = {}
    search_manifest_files = path.join(
        path.expanduser(MANIFEST_DIRECTORY), "*/searches/*.json")
    for search_manifest_file in glob.glob(search_manifest_files):
        try:
            search_manifest_data = json.loads(
                open(search_manifest_file, 'r').read())
        except Exception as e:
            errors = True
            print "Error reading %s" % search_manifest_file
            print e
            continue

        if 'status' in search_manifest_data and \
                search_manifest_data['status'] == 'development':
            continue

        if 'search_type' not in search_manifest_data:
            print "Error in %s, no search_type found" % search_manifest_file
            errors = True
            continue

        if search_manifest_data['search_type'] not in [
                'support', 'detection', 'contextual', 'investigative']:
            print "Error in %s, invalid search_type, %s, found" % (
                search_manifest_file, search_manifest_data)
            errors = True
            continue

        try:
            if search_manifest_data['search_type'] == 'detection':
                search_schema = detection_search_schema
            elif search_manifest_data['search_type'] == 'contextual':
                search_schema = contextual_search_schema
            elif search_manifest_data['search_type'] == 'investigative':
                search_schema = investigative_search_schema
            else:
                search_schema = support_search_schema

            jsonschema.validate(search_manifest_data, search_schema)
        except jsonschema.exceptions.ValidationError as json_ve:
            errors = True
            print search_manifest_file
            print json_ve.message
        except Exception as broad_exception:
            errors = True
            print search_manifest_file
            print broad_exception

        sm_errors = validate_search_manifest(search_manifest_data)
        if sm_errors:
            errors = True
            for err in sm_errors:
                print path.basename(search_manifest_file)
                print "\t%s" % err

        search_manifests[search_manifest_data['search_name']
                         ] = search_manifest_data

    for story_name, story_data in story_manifests.iteritems():
        if 'detection_searches' in story_data['searches']:
            for ds in story_data['searches']['detection_searches']:
                if ds not in search_manifests:
                    print "INFO: %s has a detection search, \
                        %s, not in local repo. Verify name of search provided \
                        in Enterprise Security Content Updates" % (
                        story_name, ds)
                    continue

                if search_manifests[ds]['search_type'] != 'detection':
                    errors = True
                    print "ERROR: \"%s\" mislabeled \"%s\" \
                        as a detection search" % (
                        story_name, ds)

        if 'investigative_searches' in story_data['searches']:
            for invs in story_data['searches']['investigative_searches']:
                if invs not in search_manifests:
                    print "INFO: %s has an investigative search, \
                    %s, not in local repo. \
                    Verify name of search provided in \
                    Enterprise Security Content Updates" % (
                        story_name, invs)
                    continue

                if search_manifests[invs]['search_type'] != 'investigative':
                    errors = True
                    print "ERROR: \"%s\" mislabeled \"%s\" \
                            as a investigative search" % (
                        story_name, invs)

        if 'contextual_searches' in story_data['searches']:
            for cs in story_data['searches']['contextual_searches']:
                if cs not in search_manifests:
                    print "INFO: %s has a contextual search, \
                        %s, not in local repo. Verify name of search provided \
                        in Enterprise Security Content Updates" \
                        % (story_name, cs)
                    continue

                if search_manifests[cs]['search_type'] != 'contextual':
                    errors = True
                    print "ERROR: \"%s\" mislabeled \"%s\" \
                           as a contextual search" % (
                        story_name, cs)

        if 'support_searches' in story_data['searches']:
            for ss in story_data['searches']['support_searches']:
                if ss not in search_manifests:
                    print "INFO: %s has a support search, %s, \
                    not in local repo. Verify name of search provided \
                    in Enterprise Security Content Updates" % (
                        story_name, ss)
                    continue

                if search_manifests[ss]['search_type'] != 'support':
                    errors = True
                    print "ERROR: \"%s\" mislabeled \"%s\" \
                        as a support search" % (
                        story_name, ss)

    print "%d story manifests checked" % len(story_manifests)
    print "%d search manifests checked" % len(search_manifests)

    if errors:
        sys.exit("Errors found")
    else:
        print "No errors found"


if __name__ == "__main__":
    # grab arguments
    parser = argparse.ArgumentParser(description="validates security content manifests", epilog="""
        Validates security manifest for correctness, adhering to spec and other common items.""")
    parser.add_argument("-p", "--path", required=True, help="path to security-security content repo")

    # parse them
    args = parser.parse_args()
    MANIFEST_DIRECTORY = args.path
    main()