def find_cutom_pipeline_processor()

in tools/malware_research/custom_pipelines.py [0:0]

• 11 lines of code
• 6 McCabe index (conditional complexity)

def find_cutom_pipeline_processor(pipeline: dict[str, Any]) -> str:
    if not "_meta" in pipeline.keys():
        return ""
    if not pipeline["_meta"]["managed_by"] == "fleet":
        return ""
    for p in pipeline["processors"]:
        if not "pipeline" in p.keys():
            continue
        if p["pipeline"]["name"].endswith("@custom"):
           return p["pipeline"]["name"]
    return ""