From 941ab18643fbd94a5031830894c948646c1c8211 Mon Sep 17 00:00:00 2001 From: Scott Brumley Date: Fri, 16 Oct 2026 09:47:28 -0400 Subject: [PATCH] feat(soc-optimization-unified): targets[] expansion and per-action dispatch mode in SOCCommandWrapper Two registry-row fields, both optional, neither an operator setting: - target_arg names the inline arg that carries the thing being acted on. When its context reference resolves to a list, the wrapper dispatches once per element instead of stringifying the list into one garbage value - the multi-domain bug. One approval upstream covers the run. - dispatch: all_applicable dispatches every vendor branch the action declares, each checked and recorded on its own; one vendor failing cannot kill the rest. Absent means routed: the original ladder, unchanged. The single-vendor single-target path is the original main() body moved into _dispatch_one with terminal semantics preserved - same war-room entries, same records, same return_error/raise - so every existing action behaves byte-for-byte as before. Per-(vendor,target) execution records and dataset rows carry target/action_target; a dispatch summary war-room entry appears only when a run fans out. Tests: 9 new cases (fan-out matrix, dedupe, failure isolation, shadow per-pair, legacy equivalence pins); 6 fail on the pre-change code, 3 pin equivalence. Existing suites pass unchanged; the 9 pre-existing detonate-file failures are identical on main and here. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01X6z5STKWCdm3aN5XwxWpgz --- .../SOCCommandWrapper/SOCCommandWrapper.py | 801 ++++++++++++------ .../SOCCommandWrapper_test_dispatch.py | 293 +++++++ 2 files changed, 828 insertions(+), 266 deletions(-) create mode 100644 Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper_test_dispatch.py diff --git a/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper.py b/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper.py index 9c805791..59d56e77 100644 --- a/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper.py +++ b/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper.py @@ -162,6 +162,42 @@ def _resolve_ctx_path(ctx, path): return node +def _resolve_ctx_path_list(ctx, path): + """Resolve a dotted context path keeping the WHOLE leaf list. + + _resolve_ctx_path deliberately returns the first non-empty scalar so that + single-valued args stay scalars. Target expansion needs the opposite: when + an issue carries four domains, all four are the work. Same traversal, + different leaf rule — a list comes back filtered of empties, a scalar + comes back as itself. + """ + node = ctx + + for segment in str(path).split("."): + if isinstance(node, list): + collected = [] + for item in node: + if isinstance(item, dict) and segment in item: + v = item[segment] + if isinstance(v, list): + collected.extend(v) + else: + collected.append(v) + node = collected + elif isinstance(node, dict): + node = node.get(segment) + else: + return None + + if node in (None, "", [], {}): + return None + + if isinstance(node, list): + return [v for v in node if v not in (None, "", [], {})] + + return node + + def _resolve_templates(obj, ctx): if isinstance(obj, dict): return {k: _resolve_templates(v, ctx) for k, v in obj.items()} @@ -754,273 +790,52 @@ def post_dataset_payload(payload, tags=None): ) -def main(): - args = demisto.args() - ctx = demisto.context() - - demisto.debug(f"RAW TAGS: {args.get('tags')} | TYPE: {type(args.get('tags'))}") - - action = args.get("action") - using = args.get("using") - list_name = args.get("list_name") - output_key = args.get("output_key") - # shadow_mode is now read from the action entry in SOCFrameworkActions_V3. - # Analysis/Enrichment: shadow_mode: false — always execute. - # C/E/R: shadow_mode: true — suppressed until PS flips to false. - # To go live: edit shadow_mode on the specific action in the list. - # The args shadow_mode field is retained for backward compat but ignored. - - raw_tags = args.get("tags") - tags = parse_tags(raw_tags) - - if not action: - return_error("Missing action") - - if not list_name: - return_error("Missing list_name") - - list_data = demisto.executeCommand("getList", {"listName": list_name}) - - if not list_data or "Contents" not in list_data[0]: - return_error("Failed to load action list") - - action_map = _try_json_loads(list_data[0]["Contents"]) - - if not action_map: - return_error("Invalid JSON in action list") - - action_entry = action_map.get(action) - - if not action_entry: - return_error(f"Action not found: {action}") - - # Load action time map — soft fail if list not present or action not listed. - # SOCActionTimeMap_V3 is customer-editable: one entry per SOC Framework - # action name with time_minutes and action_category fields. - # Both values are written to the dataset so dashboards need no join. - _time_map_result = demisto.executeCommand("getList", {"listName": "SOCActionTimeMap_V3"}) - _time_map: dict = {} - if _time_map_result and isinstance(_time_map_result, list) and "Contents" in _time_map_result[0]: - _time_map = _try_json_loads(_time_map_result[0].get("Contents", "")) or {} - _time_entry = _time_map.get(action, {}) if isinstance(_time_map, dict) else {} - action_time_minutes = int(_time_entry.get("time_minutes", 0)) if isinstance(_time_entry, dict) else 0 - action_time_category = str(_time_entry.get("action_category", "")) if isinstance(_time_entry, dict) else "" - demisto.debug( - f"SOCActionTimeMap_V3: action={action} " - f"time_minutes={action_time_minutes} " - f"action_category={action_time_category}" - ) - - # Read shadow_mode from the action entry — single source of truth - shadow_mode = resolve_shadow_mode(action_entry, args.get("Action_Actor")) - - responses = action_entry.get("responses", {}) - - # ── Multi-vendor response routing ───────────────────────────────────────── - # Resolution order: - # 1. Look up the action's class (endpoint/identity/email/indicator/network) - # from SOCActionClassMap_V3. - # 2. Look up responses.{class} in the alert's SOCProductCategoryMap entry. - # This is the correct integration for this action type on this source. - # 3. Fall back to SOCFramework.Product.response (legacy single-vendor field). - # 4. Fall back to first available vendor in the action's responses dict. - # - # This allows a CrowdStrike Endpoint alert to use CrowdStrike for endpoint - # actions but Active Directory for identity actions — configured in one place - # (SOCProductCategoryMap_V3) without touching playbooks. - # ───────────────────────────────────────────────────────────────────────── - - vendor = None - vendor_data = None - - # Step 1: get action class from SOCActionClassMap_V3 - action_class_map = demisto.executeCommand( - "getList", {"listName": "SOCActionClassMap_V3"} - ) - action_class = None - if action_class_map and isinstance(action_class_map, list): - acm_raw = action_class_map[0].get("Contents", "") - if acm_raw and acm_raw != "Item not found": - import json as _json - try: - acm = _json.loads(acm_raw) - action_class = acm.get(action) - except Exception: - pass - - # Step 2: look up responses.{action_class} from the category map entry - if action_class: - category_responses = demisto.get(ctx, "SOCFramework.Product.responses") or {} - if isinstance(category_responses, str): - try: - category_responses = _json.loads(category_responses) - except Exception: - category_responses = {} - class_vendor = category_responses.get(action_class) - if class_vendor: - vendor_data = responses.get(class_vendor) - if vendor_data: - vendor = class_vendor - demisto.debug( - f"SOCCommandWrapper: multi-vendor routing — action={action} " - f"class={action_class} vendor={vendor}" - ) - - # Step 2b: platform actions are not vendor actions. - # - # Starring a case, raising its severity and closing an issue are done by the - # platform itself, so they carry a single "Builtin" implementation and no - # source vendor can satisfy them. Routing them by vendor skipped them on every - # alert: soc-star-case was skipped as "no implementation for source vendor - # 'CrowdStrike Falcon'", and soc-close-issue has the same shape. - # - # This is narrow on purpose. It applies ONLY when Builtin is the action's one - # and only implementation, so there is no vendor ambiguity to resolve wrongly - - # which is what the "never fall back to first available vendor" rule below - # exists to prevent. An action with Builtin plus vendor branches still routes - # by vendor. - if not vendor_data and set(responses.keys()) == {"Builtin"}: - vendor_data = responses.get("Builtin") - if vendor_data: - vendor = "Builtin" - demisto.debug( - f"SOCCommandWrapper: platform action — action={action} " - f"dispatched as Builtin, independent of source vendor" - ) - - # Step 3: fall back to legacy SOCFramework.Product.response - if not vendor_data: - vendor = demisto.get(ctx, "SOCFramework.Product.response") - vendor_data = responses.get(vendor) if vendor else None - if vendor_data: - demisto.debug( - f"SOCCommandWrapper: legacy response routing — vendor={vendor}" - ) - - # Step 4: no matching vendor — skip cleanly with a recorded reason. - # Do NOT fall back to "first available vendor": that silently dispatches a - # non-matching vendor's command (e.g. core-quarantine-files against a - # CrowdStrike host because soc-quarantine-files lacks a CrowdStrike branch). - # Instead, write a skipped execution record so SOCSummarizePhase and the - # case audit trail show the action was attempted but had no vendor - # implementation for the alert's source. - if not vendor_data: - legacy_vendor = demisto.get(ctx, "SOCFramework.Product.response") - available = sorted(responses.keys()) - skip_reason = ( - f"no implementation for source vendor {legacy_vendor!r} in " - f"{action}.responses; available: {available}" - ) - demisto.debug( - f"SOCCommandWrapper: skipping {action} — {skip_reason} " - f"(action_class={action_class})" - ) - - skip_record = { - "action": action, - "vendor": legacy_vendor or "(unresolved)", - "command": None, - "args": {}, - "shadow_mode": shadow_mode, - "success": False, - "skipped": True, - "skip_reason": skip_reason, - "tags": tags or [], - "run_id": get_or_create_run_id(ctx), - "timestamp": utc_now(), - } - if output_key: - append_context(output_key, skip_record) - - warroom_log( - f"SOC Framework — {action}: skipped (no vendor implementation)", - { - "reason": "no implementation for source vendor", - "source_vendor": legacy_vendor or "(unresolved)", - "action_class": action_class, - "action": action, - "available_vendors": available, - }, - tags, - ) - return_results( - f"{action} skipped — no implementation for source vendor " - f"{legacy_vendor or '(unresolved)'} in this action's responses" - ) - return - command = vendor_data.get("command") - _inline_args_template = vendor_data.get("inline_args", {}) - inline_args = _resolve_templates(_inline_args_template, ctx) +def _dispatch_one(shared, vendor, vendor_data, command, inline_args, target, terminal): + """Dispatch one (vendor, target) pair: shadow or execute, record, dataset row. - # Guard: the action declares argument templates but every CONTEXT-DERIVED - # one resolved empty (e.g. soc-enrich-file's sha256=${SOCFramework.Artifacts.Hash} - # when this alert carries no file hash). Firing a command with a missing required - # arg fails and hangs the lifecycle, so skip cleanly with a recorded reason -- - # mirroring the Upon-Trigger enrichment, which only fires when the artifact is - # non-empty. - # - # Only context references count toward "is there anything to act on". A static - # literal carries no evidence, and letting it vote here keeps the dispatch alive - # after the identifying arg has already been dropped below. Observed on - # soc-enrich-user/Microsoft Graph User, where properties=displayName,... is static - # and user=${...Identity.User.UPN} resolved empty: the call fired with only - # properties, and Graph rejected the malformed path with - # "Unexpected segment DynamicPathSegment. Expected property/$value". - _ctx_arg_keys = [ - k for k, raw in _inline_args_template.items() if _is_ctx_template(raw) - ] - _empty_ctx_args = [ - k for k in _ctx_arg_keys - if inline_args.get(k) in (None, "", [], {}, "null", "None") - ] - - if _ctx_arg_keys and len(_empty_ctx_args) == len(_ctx_arg_keys): - skip_record = { - "action": action, "vendor": vendor, "command": command, - "args": inline_args, "shadow_mode": shadow_mode, "success": False, - "skipped": True, - "skip_reason": ( - "no non-empty artifact for command args: " - + ", ".join(sorted(_empty_ctx_args)) - ), - "tags": tags or [], "run_id": get_or_create_run_id(ctx), - "timestamp": utc_now(), - } - if output_key: - append_context(output_key, skip_record) - warroom_log( - f"SOC Framework - {action}: skipped (no artifact to act on)", - {"reason": "all context-derived command args resolved empty", - "empty_args": ", ".join(sorted(_empty_ctx_args)), - "action": action, "vendor": vendor, "command": command}, - tags, - ) - return_results(f"{action} skipped - no non-empty artifact for {command}") - return - - timestamp = utc_now() - run_id = get_or_create_run_id(ctx) - - schema_entries = load_schema_entries(ctx, args.get("LifeCycle")) - if not schema_entries: - schema_entries = default_schema_entries() + Behavior for the terminal case (single vendor, single target — every action + before dispatch modes existed) is the original main() body unchanged: the + same war-room entries, the same records, return_error on command failure + and a bare raise on unexpected exceptions. Non-terminal calls record the + failure and report it in the dispatch summary instead, so one vendor or + target cannot kill the rest of the fan-out. - issue = get_issue_data() + Returns {"vendor", "target", "outcome", "reason"} with outcome one of + simulated | success | failed | skipped_no_artifact | integration_unavailable. + """ + action = shared["action"] + args = shared["args"] + ctx = shared["ctx"] + tags = shared["tags"] + raw_tags = shared["raw_tags"] + output_key = shared["output_key"] + using = shared["using"] + shadow_mode = shared["shadow_mode"] + run_id = shared["run_id"] + timestamp = shared["timestamp"] + schema_entries = shared["schema_entries"] + issue = shared["issue"] + action_time_minutes = shared["action_time_minutes"] + action_time_category = shared["action_time_category"] + + resolved_payload = { + "run_id": run_id, + "action": action, + "vendor": vendor, + "command": command, + "args": inline_args, + "shadow_mode": shadow_mode, + "raw_tags": raw_tags, + "raw_tags_type": str(type(raw_tags)), + "tags": tags + } + if target is not None: + resolved_payload["target"] = target warroom_log( "SOC Framework - Universal Command Resolved", - { - "run_id": run_id, - "action": action, - "vendor": vendor, - "command": command, - "args": inline_args, - "shadow_mode": shadow_mode, - "raw_tags": raw_tags, - "raw_tags_type": str(type(raw_tags)), - "tags": tags - }, + resolved_payload, tags ) @@ -1036,6 +851,8 @@ def main(): "tags": tags, "timestamp": timestamp } + if target is not None: + record["target"] = target if output_key: append_context(output_key, record) @@ -1064,6 +881,8 @@ def main(): "action_time_minutes": action_time_minutes, "action_time_category": action_time_category } + if target is not None: + wrapper_values["action_target"] = target dataset_payload = enrich_payload(base_payload, ctx, issue, wrapper_values, args) post_dataset_payload(dataset_payload, tags) @@ -1080,8 +899,9 @@ def main(): footer=shadow_footer(action, args.get("Action_Actor")) ) - return_results("Shadow Mode: command not executed") - return + if terminal: + return_results("Shadow Mode: command not executed") + return {"vendor": vendor, "target": target, "outcome": "simulated"} try: warroom_log( @@ -1142,6 +962,8 @@ def main(): "tags": tags, "timestamp": timestamp } + if target is not None: + record["target"] = target if output_key: append_context(output_key, record) @@ -1165,6 +987,8 @@ def main(): "action_time_minutes": action_time_minutes, "action_time_category": action_time_category } + if target is not None: + wrapper_values["action_target"] = target dataset_payload = enrich_payload(base_payload, ctx, issue, wrapper_values, args) post_dataset_payload(dataset_payload, tags) @@ -1196,9 +1020,13 @@ def main(): outputs_prefix="UC", outputs={"action": action, "status": "integration_unavailable", "command": command} )) - return + return {"vendor": vendor, "target": target, + "outcome": "integration_unavailable", "reason": error_msg} - return_error(error_msg) + if terminal: + return_error(error_msg) + return {"vendor": vendor, "target": target, + "outcome": "failed", "reason": error_msg} record = { "run_id": run_id, @@ -1211,6 +1039,8 @@ def main(): "tags": tags, "timestamp": timestamp } + if target is not None: + record["target"] = target if output_key: append_context(output_key, record) @@ -1234,6 +1064,8 @@ def main(): "action_time_minutes": action_time_minutes, "action_time_category": action_time_category } + if target is not None: + wrapper_values["action_target"] = target dataset_payload = enrich_payload(base_payload, ctx, issue, wrapper_values, args) post_dataset_payload(dataset_payload, tags) @@ -1251,6 +1083,7 @@ def main(): ) return_results(result) + return {"vendor": vendor, "target": target, "outcome": "success"} except Exception as e: record = { @@ -1265,6 +1098,8 @@ def main(): "tags": tags, "timestamp": timestamp } + if target is not None: + record["target"] = target if output_key: append_context(output_key, record) @@ -1289,6 +1124,8 @@ def main(): "action_time_minutes": action_time_minutes, "action_time_category": action_time_category } + if target is not None: + wrapper_values["action_target"] = target dataset_payload = enrich_payload(base_payload, ctx, issue, wrapper_values, args) post_dataset_payload(dataset_payload, tags) @@ -1305,8 +1142,440 @@ def main(): tags ) - raise + if terminal: + raise + return {"vendor": vendor, "target": target, "outcome": "failed", "reason": str(e)} + + +def _dispatch_vendor(shared, vendor, vendor_data, target_arg, single_vendor): + """Resolve one vendor branch's args and dispatch it once per target. + + target_arg names the inline arg that carries the thing being acted on; + when its context reference resolves to a list, each usable element gets + its own dispatch. Scalars behave exactly as before. + """ + action = shared["action"] + ctx = shared["ctx"] + tags = shared["tags"] + shadow_mode = shared["shadow_mode"] + output_key = shared["output_key"] + + command = vendor_data.get("command") + _inline_args_template = vendor_data.get("inline_args", {}) + inline_args = _resolve_templates(_inline_args_template, ctx) + + # Guard: the action declares argument templates but every CONTEXT-DERIVED + # one resolved empty (e.g. soc-enrich-file's sha256=${SOCFramework.Artifacts.Hash} + # when this alert carries no file hash). Firing a command with a missing required + # arg fails and hangs the lifecycle, so skip cleanly with a recorded reason -- + # mirroring the Upon-Trigger enrichment, which only fires when the artifact is + # non-empty. + # + # Only context references count toward "is there anything to act on". A static + # literal carries no evidence, and letting it vote here keeps the dispatch alive + # after the identifying arg has already been dropped below. Observed on + # soc-enrich-user/Microsoft Graph User, where properties=displayName,... is static + # and user=${...Identity.User.UPN} resolved empty: the call fired with only + # properties, and Graph rejected the malformed path with + # "Unexpected segment DynamicPathSegment. Expected property/$value". + _ctx_arg_keys = [ + k for k, raw in _inline_args_template.items() if _is_ctx_template(raw) + ] + _empty_ctx_args = [ + k for k in _ctx_arg_keys + if inline_args.get(k) in (None, "", [], {}, "null", "None") + ] + + if _ctx_arg_keys and len(_empty_ctx_args) == len(_ctx_arg_keys): + skip_record = { + "action": action, "vendor": vendor, "command": command, + "args": inline_args, "shadow_mode": shadow_mode, "success": False, + "skipped": True, + "skip_reason": ( + "no non-empty artifact for command args: " + + ", ".join(sorted(_empty_ctx_args)) + ), + "tags": tags or [], "run_id": shared["run_id"], + "timestamp": shared["timestamp"], + } + if output_key: + append_context(output_key, skip_record) + warroom_log( + f"SOC Framework - {action}: skipped (no artifact to act on)", + {"reason": "all context-derived command args resolved empty", + "empty_args": ", ".join(sorted(_empty_ctx_args)), + "action": action, "vendor": vendor, "command": command}, + tags, + ) + return_results(f"{action} skipped - no non-empty artifact for {command}") + return [{"vendor": vendor, "target": None, + "outcome": "skipped_no_artifact", + "reason": "all context-derived command args resolved empty"}] + + # Resolve the target arg's RAW context value. Template resolution above + # deliberately collapses a leaf list to its first non-empty scalar (the + # right rule for ordinary args) — which is exactly how multi-value issues + # lost all but one target. The target arg alone gets the list-preserving + # walk. + target_values = None + if target_arg: + raw_tmpl = _inline_args_template.get(target_arg) + if _is_ctx_template(raw_tmpl): + s = str(raw_tmpl).strip() + m = CTX_REF_RE.match(s) + target_values = _resolve_ctx_path_list(ctx, m.group(1) if m else s) + elif isinstance(inline_args.get(target_arg), list): + target_values = inline_args.get(target_arg) + + targets = [None] + if target_arg and isinstance(target_values, list): + usable, seen = [], set() + for t in target_values: + key = str(t) + if t in (None, "") or key in seen: + continue + seen.add(key) + usable.append(t) + if not usable: + skip_record = { + "action": action, "vendor": vendor, "command": command, + "args": inline_args, "shadow_mode": shadow_mode, "success": False, + "skipped": True, + "skip_reason": f"target list {target_arg} resolved to no usable values", + "tags": tags or [], "run_id": shared["run_id"], + "timestamp": shared["timestamp"], + } + if output_key: + append_context(output_key, skip_record) + warroom_log( + f"SOC Framework - {action}: skipped (no usable targets)", + {"reason": f"target list {target_arg} resolved to no usable values", + "action": action, "vendor": vendor, "command": command}, + tags, + ) + return_results(f"{action} skipped - no usable targets for {command}") + return [{"vendor": vendor, "target": None, + "outcome": "skipped_no_artifact", + "reason": f"target list {target_arg} resolved to no usable values"}] + targets = usable + + results = [] + for t in targets: + call_args = dict(inline_args) + if t is not None: + call_args[target_arg] = t + terminal = single_vendor and len(targets) == 1 + results.append( + _dispatch_one(shared, vendor, vendor_data, command, call_args, t, terminal) + ) + return results + + + +def main(): + args = demisto.args() + ctx = demisto.context() + + demisto.debug(f"RAW TAGS: {args.get('tags')} | TYPE: {type(args.get('tags'))}") + + action = args.get("action") + using = args.get("using") + list_name = args.get("list_name") + output_key = args.get("output_key") + # shadow_mode is now read from the action entry in SOCFrameworkActions_V3. + # Analysis/Enrichment: shadow_mode: false — always execute. + # C/E/R: shadow_mode: true — suppressed until PS flips to false. + # To go live: edit shadow_mode on the specific action in the list. + # The args shadow_mode field is retained for backward compat but ignored. + + raw_tags = args.get("tags") + tags = parse_tags(raw_tags) + + if not action: + return_error("Missing action") + + if not list_name: + return_error("Missing list_name") + + list_data = demisto.executeCommand("getList", {"listName": list_name}) + + if not list_data or "Contents" not in list_data[0]: + return_error("Failed to load action list") + + action_map = _try_json_loads(list_data[0]["Contents"]) + + if not action_map: + return_error("Invalid JSON in action list") + + action_entry = action_map.get(action) + + if not action_entry: + return_error(f"Action not found: {action}") + + # Load action time map — soft fail if list not present or action not listed. + # SOCActionTimeMap_V3 is customer-editable: one entry per SOC Framework + # action name with time_minutes and action_category fields. + # Both values are written to the dataset so dashboards need no join. + _time_map_result = demisto.executeCommand("getList", {"listName": "SOCActionTimeMap_V3"}) + _time_map: dict = {} + if _time_map_result and isinstance(_time_map_result, list) and "Contents" in _time_map_result[0]: + _time_map = _try_json_loads(_time_map_result[0].get("Contents", "")) or {} + _time_entry = _time_map.get(action, {}) if isinstance(_time_map, dict) else {} + action_time_minutes = int(_time_entry.get("time_minutes", 0)) if isinstance(_time_entry, dict) else 0 + action_time_category = str(_time_entry.get("action_category", "")) if isinstance(_time_entry, dict) else "" + demisto.debug( + f"SOCActionTimeMap_V3: action={action} " + f"time_minutes={action_time_minutes} " + f"action_category={action_time_category}" + ) + + # Read shadow_mode from the action entry — single source of truth + shadow_mode = resolve_shadow_mode(action_entry, args.get("Action_Actor")) + + responses = action_entry.get("responses", {}) + + # ── Dispatch mode ──────────────────────────────────────────────────────── + # dispatch on the action entry decides fan-out. Absent means "routed": + # exactly one vendor, resolved by the routing ladder below — the original + # behavior, unchanged. "all_applicable" dispatches every vendor branch the + # action declares, each checked and recorded on its own; the routing ladder + # is skipped because the answer is all of them. The field lives on registry + # rows authored with the framework; it is not an operator setting. + dispatch_mode = str(action_entry.get("dispatch") or "routed").strip().lower() + if dispatch_mode not in ("routed", "all_applicable"): + demisto.debug( + f"SOCCommandWrapper: unknown dispatch {dispatch_mode!r} on {action} - treating as routed" + ) + dispatch_mode = "routed" + + vendor = None + vendor_data = None + + if dispatch_mode == "routed": + # ── Multi-vendor response routing ───────────────────────────────────────── + # Resolution order: + # 1. Look up the action's class (endpoint/identity/email/indicator/network) + # from SOCActionClassMap_V3. + # 2. Look up responses.{class} in the alert's SOCProductCategoryMap entry. + # This is the correct integration for this action type on this source. + # 3. Fall back to SOCFramework.Product.response (legacy single-vendor field). + # 4. Fall back to first available vendor in the action's responses dict. + # + # This allows a CrowdStrike Endpoint alert to use CrowdStrike for endpoint + # actions but Active Directory for identity actions — configured in one place + # (SOCProductCategoryMap_V3) without touching playbooks. + # ───────────────────────────────────────────────────────────────────────── + + vendor = None + vendor_data = None + + # Step 1: get action class from SOCActionClassMap_V3 + action_class_map = demisto.executeCommand( + "getList", {"listName": "SOCActionClassMap_V3"} + ) + action_class = None + if action_class_map and isinstance(action_class_map, list): + acm_raw = action_class_map[0].get("Contents", "") + if acm_raw and acm_raw != "Item not found": + import json as _json + try: + acm = _json.loads(acm_raw) + action_class = acm.get(action) + except Exception: + pass + + # Step 2: look up responses.{action_class} from the category map entry + if action_class: + category_responses = demisto.get(ctx, "SOCFramework.Product.responses") or {} + if isinstance(category_responses, str): + try: + category_responses = _json.loads(category_responses) + except Exception: + category_responses = {} + class_vendor = category_responses.get(action_class) + if class_vendor: + vendor_data = responses.get(class_vendor) + if vendor_data: + vendor = class_vendor + demisto.debug( + f"SOCCommandWrapper: multi-vendor routing — action={action} " + f"class={action_class} vendor={vendor}" + ) + + # Step 2b: platform actions are not vendor actions. + # + # Starring a case, raising its severity and closing an issue are done by the + # platform itself, so they carry a single "Builtin" implementation and no + # source vendor can satisfy them. Routing them by vendor skipped them on every + # alert: soc-star-case was skipped as "no implementation for source vendor + # 'CrowdStrike Falcon'", and soc-close-issue has the same shape. + # + # This is narrow on purpose. It applies ONLY when Builtin is the action's one + # and only implementation, so there is no vendor ambiguity to resolve wrongly - + # which is what the "never fall back to first available vendor" rule below + # exists to prevent. An action with Builtin plus vendor branches still routes + # by vendor. + if not vendor_data and set(responses.keys()) == {"Builtin"}: + vendor_data = responses.get("Builtin") + if vendor_data: + vendor = "Builtin" + demisto.debug( + f"SOCCommandWrapper: platform action — action={action} " + f"dispatched as Builtin, independent of source vendor" + ) + + # Step 3: fall back to legacy SOCFramework.Product.response + if not vendor_data: + vendor = demisto.get(ctx, "SOCFramework.Product.response") + vendor_data = responses.get(vendor) if vendor else None + if vendor_data: + demisto.debug( + f"SOCCommandWrapper: legacy response routing — vendor={vendor}" + ) + + # Step 4: no matching vendor — skip cleanly with a recorded reason. + # Do NOT fall back to "first available vendor": that silently dispatches a + # non-matching vendor's command (e.g. core-quarantine-files against a + # CrowdStrike host because soc-quarantine-files lacks a CrowdStrike branch). + # Instead, write a skipped execution record so SOCSummarizePhase and the + # case audit trail show the action was attempted but had no vendor + # implementation for the alert's source. + if not vendor_data: + legacy_vendor = demisto.get(ctx, "SOCFramework.Product.response") + available = sorted(responses.keys()) + skip_reason = ( + f"no implementation for source vendor {legacy_vendor!r} in " + f"{action}.responses; available: {available}" + ) + demisto.debug( + f"SOCCommandWrapper: skipping {action} — {skip_reason} " + f"(action_class={action_class})" + ) + + skip_record = { + "action": action, + "vendor": legacy_vendor or "(unresolved)", + "command": None, + "args": {}, + "shadow_mode": shadow_mode, + "success": False, + "skipped": True, + "skip_reason": skip_reason, + "tags": tags or [], + "run_id": get_or_create_run_id(ctx), + "timestamp": utc_now(), + } + if output_key: + append_context(output_key, skip_record) + + warroom_log( + f"SOC Framework — {action}: skipped (no vendor implementation)", + { + "reason": "no implementation for source vendor", + "source_vendor": legacy_vendor or "(unresolved)", + "action_class": action_class, + "action": action, + "available_vendors": available, + }, + tags, + ) + return_results( + f"{action} skipped — no implementation for source vendor " + f"{legacy_vendor or '(unresolved)'} in this action's responses" + ) + return + + plan = [] + if dispatch_mode == "all_applicable": + plan = [(v, responses[v]) for v in sorted(responses) + if isinstance(responses.get(v), dict)] + if not plan: + skip_reason = f"{action} declares dispatch all_applicable but no response branches" + demisto.debug(f"SOCCommandWrapper: skipping {action} - {skip_reason}") + skip_record = { + "action": action, "vendor": "(none)", "command": None, "args": {}, + "shadow_mode": shadow_mode, "success": False, "skipped": True, + "skip_reason": skip_reason, "tags": tags or [], + "run_id": get_or_create_run_id(ctx), "timestamp": utc_now(), + } + if output_key: + append_context(output_key, skip_record) + warroom_log( + f"SOC Framework - {action}: skipped (no response branches)", + {"reason": skip_reason, "action": action}, + tags, + ) + return_results(f"{action} skipped - no response branches declared") + return + else: + plan = [(vendor, vendor_data)] + + # target_arg names the one inline arg that carries the thing being acted + # on. When its context reference resolves to a list, the wrapper + # dispatches once per element - one approval upstream still covers the + # whole run. + target_arg = action_entry.get("target_arg") + + timestamp = utc_now() + run_id = get_or_create_run_id(ctx) + + schema_entries = load_schema_entries(ctx, args.get("LifeCycle")) + if not schema_entries: + schema_entries = default_schema_entries() + + issue = get_issue_data() + + shared = { + "action": action, "args": args, "ctx": ctx, "using": using, + "output_key": output_key, "tags": tags, "raw_tags": raw_tags, + "shadow_mode": shadow_mode, "run_id": run_id, "timestamp": timestamp, + "schema_entries": schema_entries, "issue": issue, + "action_time_minutes": action_time_minutes, + "action_time_category": action_time_category, + } + + single_vendor = len(plan) == 1 + outcomes = [] + for pair_vendor, pair_data in plan: + outcomes.extend( + _dispatch_vendor(shared, pair_vendor, pair_data, target_arg, single_vendor) + ) + + # One summary entry whenever the run fanned out - per-pair entries above + # carry the detail; this is the at-a-glance line. Legacy single dispatch + # emits nothing extra, so existing actions read exactly as before. + if len(outcomes) > 1: + counts = {} + for o in outcomes: + counts[o["outcome"]] = counts.get(o["outcome"], 0) + 1 + warroom_log( + f"SOC Framework - {action}: dispatch summary", + { + "dispatch": dispatch_mode, + "vendors": sorted({str(o["vendor"]) for o in outcomes}), + "results": counts, + "detail": [ + {k: o.get(k) for k in ("vendor", "target", "outcome", "reason") + if o.get(k) is not None} + for o in outcomes + ], + }, + tags, + ) + hard_failures = [o for o in outcomes if o["outcome"] == "failed"] + if hard_failures and len(hard_failures) == len(outcomes): + return_error( + f"{action}: every dispatch failed - " + + "; ".join( + f"{o['vendor']}: {o.get('reason', 'error')}" for o in hard_failures + ) + ) + return_results( + f"{action}: " + + ", ".join(f"{k} {v}" for k, v in sorted(counts.items())) + ) if __name__ in ("__builtin__", "builtins", "__main__"): main() + diff --git a/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper_test_dispatch.py b/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper_test_dispatch.py new file mode 100644 index 00000000..1c7f1dee --- /dev/null +++ b/Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper_test_dispatch.py @@ -0,0 +1,293 @@ +""" +SOCCommandWrapper_test_dispatch.py + +Unit tests for targets[] expansion and the per-action dispatch mode +(routed | all_applicable). Follows the exact harness pattern from +SOCCommandWrapper_test.py — demistomock in builtins, no tenant required. + +Run: pytest Scripts/SOCCommandWrapper/SOCCommandWrapper_test_dispatch.py -v +""" + +import json + +import pytest + +from SOCCommandWrapper_test import load_script + + +BLOCK_DOMAIN_TWO_VENDORS = { + "soc-block-domain": { + "shadow_mode": False, + "dispatch": "all_applicable", + "target_arg": "domain", + "responses": { + "ThreatStop": { + "command": "threatstop-block", + "inline_args": {"domain": "${SOCFramework.Artifacts.Network.Domain}"}, + }, + "Zscaler": { + "command": "zscaler-block", + "inline_args": {"domain": "${SOCFramework.Artifacts.Network.Domain}"}, + }, + }, + } +} + + +def _args(action="soc-block-domain"): + return { + "action": action, + "list_name": "SOCFrameworkActions_V3", + "output_key": "UCTest.Executions", + "tags": "cer", + } + + +def _ctx(domains): + return {"SOCFramework": {"Artifacts": {"Network": {"Domain": domains}}, + "RunID": "run-1"}} + + +def _action_list_response(action_map): + """Route getList by listName: the action list gets the fixture, every + other list (SOCActionTimeMap_V3, SOCActionClassMap_V3, schema lists) + gets 'Item not found' — matching a tenant where only the action list + is populated.""" + def _route(args): + if (args or {}).get("listName") == "SOCFrameworkActions_V3": + return [{"Contents": json.dumps(action_map)}] + return [{"Contents": "Item not found"}] + return _route + + +def _commands_named(demisto, name): + return [c for c in demisto._commands if c[0] == name] + + +def test_all_applicable_dispatches_every_vendor_per_target(): + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx(["evil-a.example", "evil-b.example"]) + demisto._command_responses = { + "getList": _action_list_response(BLOCK_DOMAIN_TWO_VENDORS), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 1, "Contents": "ok"}], + "zscaler-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + # 2 vendors x 2 targets = 4 vendor calls, each with a scalar domain + ts_calls = _commands_named(demisto, "threatstop-block") + zs_calls = _commands_named(demisto, "zscaler-block") + assert len(ts_calls) == 2 and len(zs_calls) == 2 + sent = sorted(a["domain"] for _, a in ts_calls + zs_calls) + assert sent == ["evil-a.example", "evil-a.example", + "evil-b.example", "evil-b.example"] + for _, call_args in ts_calls + zs_calls: + assert isinstance(call_args["domain"], str) + + # one execution record per (vendor, target), each carrying its target + recs = demisto.get(demisto._context, "UCTest.Executions") + assert len(recs) == 4 + assert sorted(r["target"] for r in recs) == sorted(sent) + assert all(r["success"] for r in recs) + + +def test_targets_deduped_and_empties_dropped(): + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx(["evil.example", "evil.example", "", None, "other.example"]) + demisto._command_responses = { + "getList": _action_list_response(BLOCK_DOMAIN_TWO_VENDORS), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 1, "Contents": "ok"}], + "zscaler-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + assert len(_commands_named(demisto, "threatstop-block")) == 2 + assert len(_commands_named(demisto, "zscaler-block")) == 2 + + +def test_all_applicable_one_vendor_fails_other_still_dispatches(): + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx("single.example") # scalar context value + demisto._command_responses = { + "getList": _action_list_response(BLOCK_DOMAIN_TWO_VENDORS), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 4, "Contents": "boom from vendor"}], + "zscaler-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + # One hard failure out of two must NOT raise — fan-out survives. + script.main() + + assert len(_commands_named(demisto, "zscaler-block")) == 1 + recs = demisto.get(demisto._context, "UCTest.Executions") + assert len(recs) == 2 + outcomes = sorted((r["vendor"], bool(r.get("success"))) for r in recs) + assert outcomes == [("ThreatStop", False), ("Zscaler", True)] + + +def test_all_applicable_every_vendor_failing_raises(): + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx("single.example") + demisto._command_responses = { + "getList": _action_list_response(BLOCK_DOMAIN_TWO_VENDORS), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 4, "Contents": "boom ts"}], + "zscaler-block": [{"Type": 4, "Contents": "boom zs"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + with pytest.raises(RuntimeError, match="every dispatch failed"): + script.main() + + +def test_all_applicable_shadow_simulates_each_pair(): + action_map = json.loads(json.dumps(BLOCK_DOMAIN_TWO_VENDORS)) + action_map["soc-block-domain"]["shadow_mode"] = True + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx(["evil-a.example", "evil-b.example"]) + demisto._command_responses = { + "getList": _action_list_response(action_map), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + # no vendor command ever fires in shadow + assert not _commands_named(demisto, "threatstop-block") + assert not _commands_named(demisto, "zscaler-block") + # but each (vendor, target) pair is simulated and recorded + recs = demisto.get(demisto._context, "UCTest.Executions") + assert len(recs) == 4 + assert all(r["shadow_mode"] for r in recs) + # one dataset row per pair + posts = _commands_named(demisto, "socfw-post-to-dataset") + assert len(posts) == 4 + + +def test_routed_with_list_target_fans_out_on_single_vendor(): + action_map = { + "soc-block-domain": { + "shadow_mode": False, + "target_arg": "domain", + "responses": { + "ThreatStop": { + "command": "threatstop-block", + "inline_args": {"domain": "${SOCFramework.Artifacts.Network.Domain}"}, + }, + }, + } + } + script, demisto = load_script() + demisto._args = _args() + ctx = _ctx(["a.example", "b.example", "c.example"]) + ctx["SOCFramework"]["Product"] = {"response": "ThreatStop"} + demisto._context = ctx + demisto._command_responses = { + "getList": _action_list_response(action_map), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + calls = _commands_named(demisto, "threatstop-block") + assert [a["domain"] for _, a in calls] == ["a.example", "b.example", "c.example"] + recs = demisto.get(demisto._context, "UCTest.Executions") + assert len(recs) == 3 + + +def test_routed_scalar_without_new_fields_is_unchanged_legacy(): + action_map = { + "soc-block-domain": { + "shadow_mode": False, + "responses": { + "ThreatStop": { + "command": "threatstop-block", + "inline_args": {"domain": "${SOCFramework.Artifacts.Network.Domain}"}, + }, + }, + } + } + script, demisto = load_script() + demisto._args = _args() + ctx = _ctx("one.example") + ctx["SOCFramework"]["Product"] = {"response": "ThreatStop"} + demisto._context = ctx + demisto._command_responses = { + "getList": _action_list_response(action_map), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + calls = _commands_named(demisto, "threatstop-block") + assert len(calls) == 1 + recs = demisto.get(demisto._context, "UCTest.Executions") + assert len(recs) == 1 + # legacy records carry no target key and no dispatch summary is emitted + assert "target" not in recs[0] + assert not any("dispatch summary" in str(r) for r in demisto._results) + + +def test_unknown_dispatch_value_treated_as_routed(): + action_map = { + "soc-block-domain": { + "shadow_mode": False, + "dispatch": "everything-everywhere", + "responses": { + "ThreatStop": { + "command": "threatstop-block", + "inline_args": {"domain": "${SOCFramework.Artifacts.Network.Domain}"}, + }, + }, + } + } + script, demisto = load_script() + demisto._args = _args() + ctx = _ctx("one.example") + ctx["SOCFramework"]["Product"] = {"response": "ThreatStop"} + demisto._context = ctx + demisto._command_responses = { + "getList": _action_list_response(action_map), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "threatstop-block": [{"Type": 1, "Contents": "ok"}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + assert len(_commands_named(demisto, "threatstop-block")) == 1 + + +def test_target_list_resolving_empty_skips_cleanly(): + script, demisto = load_script() + demisto._args = _args() + demisto._context = _ctx(["", None]) + demisto._command_responses = { + "getList": _action_list_response(BLOCK_DOMAIN_TWO_VENDORS), + "getIssues": [{"Contents": {"data": [{"id": "100"}]}}], + "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}], + } + + script.main() + + assert not _commands_named(demisto, "threatstop-block") + assert not _commands_named(demisto, "zscaler-block") + recs = demisto.get(demisto._context, "UCTest.Executions") + assert recs and all(r.get("skipped") for r in recs)