Sigma Rule Conversion Test

sigma-test.py

from sigma.processing.pipeline import ProcessingPipeline
from sigma.rule import SigmaRule
from sigma.collection import SigmaCollection
from sigma.conversion.backends import splunk
from sigma.processing.pipeline import ProcessingPipeline, ProcessingItem
from sigma.processing.transformations import FieldMappingTransformation, QueryExpressionPlaceholderTransformation
from glob import glob
from pprint import pprint

from sigma.processing.pipelines.crowdstrike import crowdstrike_splunk_pipeline
from sigma.processing.pipelines.sysmon import sysmon_pipeline
from sigma.processing.resolver import ProcessingPipelineResolver
	

files = glob('../sigma-internal/rules/windows/**/*.yml', recursive=True)
for file in files:
    try:
        # open rule
        conf = "./crowdstrike.yml"
        with open(file, 'r') as f:
            ff = f
            # rule = SigmaRule.from_yaml(ff)
            # if rule.status == "stable":
            #     print("foo")
            # pprint(rule.logsource)
            # pprint(rule.detection.parsed_condition)

            with open(conf, 'r') as c:

                # pipeline = ProcessingPipeline([
                #     ProcessingItem(FieldMappingTransformation({
                #         "CommandLine": "mappedA",
                #     })),
                #     ProcessingItem(FieldMappingTransformation({
                #         "fieldB": "mappedB",
                #     })),
                # ])

                pipeline = crowdstrike_splunk_pipeline() 

                backend = splunk.SplunkBackend(pipeline)
                rules = SigmaCollection.from_yaml(ff)
                res = backend.convert(rules)
                if res != []:
                    print("[+] processing file: {}".format(file))
                    print()
                    print(res)
                    print()
    except Exception as e:
        print(e)