feat(soc-optimization-unified): soc-block-domain containment action

Packs/soc-optimization-unified/Lists/SOCActionClassMap_V3/SOCActionClassMap_V3_data.json

@@ -30,5 +30,6 @@
   "soc-enrich-ip": "indicator",
   "soc-enrich-file": "indicator",
   "soc-detonate-file": "indicator",
-  "soc-notify-user": "comms"
+  "soc-notify-user": "comms",
+  "soc-block-domain": "indicator"
 }

Packs/soc-optimization-unified/Lists/SOCActionTimeMap_V3/SOCActionTimeMap_V3_data.json

@@ -122,5 +122,9 @@
   "soc-unblock-sender": {
     "time_minutes": 10,
     "action_category": "recovery"
+  },
+  "soc-block-domain": {
+    "time_minutes": 15,
+    "action_category": "containment"
   }
 }

Packs/soc-optimization-unified/Lists/SOCFrameworkActions_V3/SOCFrameworkActions_V3_data.json

@@ -1370,5 +1370,45 @@
     "phase": "recovery",
     "entity_scope": "artifact",
     "capability": "Email recovery \u2014 release held message"
+  },
+  "soc-block-domain": {
+    "responses": {
+      "ThreatStopv2": {
+        "command": "threatstopv2-update-domain-list",
+        "target_arg": "ioc_domain",
+        "required_args": [
+          "list_id"
+        ],
+        "inline_args": {
+          "action": "add",
+          "ioc_domain": "${SOCFramework.Artifacts.Network.DNS.Query}",
+          "list_id": "",
+          "comments": "Blocked by SOC Framework containment"
+        }
+      },
+      "Zscaler": {
+        "command": "zscaler-category-add-url",
+        "target_arg": "url",
+        "required_args": [
+          "category-id"
+        ],
+        "inline_args": {
+          "url": "${SOCFramework.Artifacts.Network.DNS.Query}",
+          "category-id": "",
+          "using": ""
+        }
+      }
+    },
+    "shadow_mode": true,
+    "phase": "containment",
+    "entity_scope": "artifact",
+    "sequence_rank": 21,
+    "capability": "Network containment \u2014 block domain at DNS and web proxy",
+    "dispatch": "all_applicable",
+    "shadow_mode_by_actor": {
+      "analyst": true,
+      "issue_emergency": true,
+      "case_lifecycle": true
+    }
   }
 }

Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper.py

@@ -405,6 +405,28 @@ def get_or_create_run_id(ctx):
     return run_id
 
 
+TENANT_CONFIG_LIST = "SOCFrameworkTenantConfig_V3"
+
+
+def load_tenant_config():
+    """The customer's vendor values: {vendor: {arg: value}}.
+
+    Created on the tenant, never shipped in a pack, so an upgrade cannot
+    overwrite it. Missing or unreadable means nothing is configured, and any
+    action that needs a value skips with a reason naming it.
+    """
+    try:
+        res = demisto.executeCommand("getList", {"listName": TENANT_CONFIG_LIST})
+        raw = res[0].get("Contents") if res and isinstance(res, list) else None
+    except Exception as e:
+        demisto.debug(f"SOCCommandWrapper: {TENANT_CONFIG_LIST} unreadable ({e})")
+        return {}
+    if not raw or raw == "Item not found":
+        return {}
+    cfg = _try_json_loads(raw) if isinstance(raw, str) else raw
+    return cfg if isinstance(cfg, dict) else {}
+
+
 def brand_available(brand):
     """Whether the vendor's brand is registered and has an active instance.
 
@@ -707,6 +729,10 @@ def enrich_payload(payload, ctx, issue, wrapper_values, args):
     payload["has_error"] = wrapper_values["has_error"]
     payload["error_type"] = wrapper_values["error_type"]
     payload["error_message"] = wrapper_values["error_message"]
+    # Present only when the run fanned out over a list of targets, so each
+    # row says which domain, host or user it acted on.
+    if wrapper_values.get("action_target") is not None:
+        payload["action_target"] = wrapper_values["action_target"]
 
     set_if_missing(payload, "lifecycle", args.get("LifeCycle"))
     set_if_missing(payload, "lifecycle", demisto.get(ctx, "SOCFramework.lifecycle"))
@@ -1164,6 +1190,16 @@ def _dispatch_vendor(shared, vendor, vendor_data, target_arg, single_vendor):
     _inline_args_template = vendor_data.get("inline_args", {})
     inline_args = _resolve_templates(_inline_args_template, ctx)
 
+    # Customer-specific values (a block-list ID, a proxy category) come from
+    # SOCFrameworkTenantConfig_V3, keyed by vendor then arg name. They fill
+    # only args the registry ships as empty literals - never an arg bound to
+    # evidence on the issue - so the registry stays upgradeable and the
+    # customer's values survive every pack install.
+    for k, v in (shared.get("tenant_config", {}).get(vendor) or {}).items():
+        if (k in _inline_args_template and _inline_args_template.get(k) in (None, "")
+                and not _is_ctx_template(_inline_args_template.get(k))):
+            inline_args[k] = v
+
     # 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
@@ -1212,6 +1248,39 @@ def _dispatch_vendor(shared, vendor, vendor_data, target_arg, single_vendor):
                  "outcome": "skipped_no_artifact",
                  "reason": "all context-derived command args resolved empty"}]
 
+    # Guard: a response branch may name args that are tenant facts rather than
+    # evidence - a block-list ID, a proxy category. Shipped empty, they are set
+    # per customer in the tenant config list. Until they are, the branch skips and says
+    # which value is missing, in shadow as well, because a simulation without
+    # them would show a call that could never succeed.
+    _missing_cfg = [
+        k for k in (vendor_data.get("required_args") or [])
+        if inline_args.get(k) in (None, "", [], {}, "null", "None")
+    ]
+    if _missing_cfg:
+        reason = (
+            "not configured: set " + ", ".join(_missing_cfg)
+            + f" for {vendor} in {TENANT_CONFIG_LIST}"
+        )
+        skip_record = {
+            "action": action, "vendor": vendor, "command": command,
+            "args": inline_args, "shadow_mode": shadow_mode, "success": False,
+            "skipped": True, "skip_reason": reason,
+            "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 (vendor not configured)",
+            {"reason": reason, "action": action, "vendor": vendor, "command": command},
+            tags,
+        )
+        if single_vendor:
+            return_results(f"{action} skipped - {reason}")
+        return [{"vendor": vendor, "target": None,
+                 "outcome": "skipped_not_configured", "reason": reason}]
+
     # 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
@@ -1532,13 +1601,20 @@ def main():
         "schema_entries": schema_entries, "issue": issue,
         "action_time_minutes": action_time_minutes,
         "action_time_category": action_time_category,
+        # Read only for actions that declare customer values, so every other
+        # action makes exactly the list reads it made before.
+        "tenant_config": (load_tenant_config()
+                          if any((d or {}).get("required_args") for _, d in plan) else {}),
     }
 
     single_vendor = len(plan) == 1
     outcomes = []
     for pair_vendor, pair_data in plan:
+        # Vendors name the same target differently (ioc_domain vs url), so a
+        # response branch may carry its own target_arg; the action's is the default.
+        pair_target_arg = (pair_data or {}).get("target_arg") or target_arg
         outcomes.extend(
-            _dispatch_vendor(shared, pair_vendor, pair_data, target_arg, single_vendor)
+            _dispatch_vendor(shared, pair_vendor, pair_data, pair_target_arg, single_vendor)
         )
 
     # One summary entry whenever the run fanned out - per-pair entries above

Packs/soc-optimization-unified/Scripts/SOCCommandWrapper/SOCCommandWrapper_test_block_domain.py

@@ -0,0 +1,149 @@
+"""
+SOCCommandWrapper_test_block_domain.py
+
+soc-block-domain against the registry row as it ships in
+Lists/SOCFrameworkActions_V3, not a hand-copied fixture, so a change to the
+row is tested here. Covers per-vendor target args (ThreatStop ioc_domain,
+Zscaler url), customer values from SOCFrameworkTenantConfig_V3, the
+not-configured skip, shadow, and the
+action_target column on the dataset row.
+
+Run: pytest Scripts/SOCCommandWrapper/SOCCommandWrapper_test_block_domain.py -v
+"""
+
+import copy
+import json
+import os
+
+from SOCCommandWrapper_test import load_script
+
+_LIST = os.path.join(os.path.dirname(__file__), "..", "..", "Lists",
+                     "SOCFrameworkActions_V3", "SOCFrameworkActions_V3_data.json")
+SHIPPED = json.load(open(_LIST))["soc-block-domain"]
+
+
+def _row(shadow=False):
+    """The shipped row, unedited except for shadow - customer values never go here."""
+    row = copy.deepcopy(SHIPPED)
+    row["shadow_mode"] = shadow
+    row["shadow_mode_by_actor"] = {}
+    return {"soc-block-domain": row}
+
+
+def _tenant(configured=("ThreatStopv2", "Zscaler")):
+    cfg = {}
+    if "ThreatStopv2" in configured:
+        cfg["ThreatStopv2"] = {"list_id": "test-list-guid"}
+    if "Zscaler" in configured:
+        cfg["Zscaler"] = {"category-id": "CUSTOM_99"}
+    return cfg
+
+
+def _run(action_map, domains, tenant=None, actor="analyst"):
+    script, demisto = load_script()
+    demisto._args = {"action": "soc-block-domain", "list_name": "SOCFrameworkActions_V3",
+                     "output_key": "UCTest.Executions", "Action_Actor": actor}
+    demisto._context = {"SOCFramework": {
+        "Artifacts": {"Network": {"DNS": {"Query": domains}}}, "RunID": "run-1"}}
+
+    def _route(args):
+        name = (args or {}).get("listName")
+        if name == "SOCFrameworkActions_V3":
+            return [{"Contents": json.dumps(action_map)}]
+        if name == "SOCFrameworkTenantConfig_V3" and tenant is not None:
+            return [{"Contents": json.dumps(tenant)}]
+        return [{"Contents": "Item not found"}]
+
+    demisto._command_responses = {
+        "getList": _route,
+        "getIssues": [{"Contents": {"data": [{"id": "100"}]}}],
+        "threatstopv2-update-domain-list": [{"Type": 1, "Contents": "ok"}],
+        "zscaler-category-add-url": [{"Type": 1, "Contents": "ok"}],
+        "socfw-post-to-dataset": [{"Type": 1, "Contents": "ok"}],
+    }
+    script.main()
+    return demisto
+
+
+def _calls(demisto, name):
+    return [a for c, a in demisto._commands if c == name]
+
+
+def _rows(demisto):
+    return [json.loads(a["JSON"]) for a in _calls(demisto, "socfw-post-to-dataset")]
+
+
+def test_shipped_row_is_containment_shadowed_fan_out():
+    assert SHIPPED["phase"] == "containment"
+    assert SHIPPED["shadow_mode"] is True
+    assert all(v is True for v in SHIPPED["shadow_mode_by_actor"].values())
+    assert SHIPPED["dispatch"] == "all_applicable"
+    assert set(SHIPPED["responses"]) == {"ThreatStopv2", "Zscaler"}
+
+
+def test_configured_blocks_every_domain_in_both_vendors_with_their_own_arg():
+    d = _run(_row(), ["evil-a.example", "evil-b.example"], _tenant())
+    ts = _calls(d, "threatstopv2-update-domain-list")
+    zs = _calls(d, "zscaler-category-add-url")
+    assert sorted(a["ioc_domain"] for a in ts) == ["evil-a.example", "evil-b.example"]
+    assert sorted(a["url"] for a in zs) == ["evil-a.example", "evil-b.example"]
+    assert all(a["list_id"] == "test-list-guid" and a["action"] == "add" for a in ts)
+    assert all(a["category-id"] == "CUSTOM_99" for a in zs)
+    # empty optional instance is dropped, never sent as using=""
+    assert all("using" not in a for a in zs)
+
+
+def test_customer_values_come_from_tenant_config_not_the_registry_row():
+    # the shipped row carries no customer values at all
+    assert SHIPPED["responses"]["ThreatStopv2"]["inline_args"]["list_id"] == ""
+    assert SHIPPED["responses"]["Zscaler"]["inline_args"]["category-id"] == ""
+    d = _run(_row(), ["evil-a.example"],
+             {"Zscaler": {"category-id": "CUSTOM_99", "using": "Zscaler_PROD"}})
+    zs = _calls(d, "zscaler-category-add-url")
+    assert zs and zs[0]["category-id"] == "CUSTOM_99" and zs[0]["using"] == "Zscaler_PROD"
+
+
+def test_tenant_config_never_overrides_evidence_from_the_issue():
+    d = _run(_row(), ["evil-a.example"], {"Zscaler": {"category-id": "CUSTOM_99",
+                                                       "url": "attacker-chosen.example"}})
+    assert [a["url"] for a in _calls(d, "zscaler-category-add-url")] == ["evil-a.example"]
+
+
+def test_dataset_rows_name_the_target():
+    d = _run(_row(), ["evil-a.example", "evil-b.example"], _tenant())
+    rows = [r for r in _rows(d) if r.get("universal_command") == "soc-block-domain"]
+    assert len(rows) == 4
+    assert sorted((r["vendor"], r["action_target"]) for r in rows) == [
+        ("ThreatStopv2", "evil-a.example"), ("ThreatStopv2", "evil-b.example"),
+        ("Zscaler", "evil-a.example"), ("Zscaler", "evil-b.example")]
+
+
+def test_no_tenant_config_list_skips_with_the_value_to_set():
+    d = _run(_row(), ["evil-a.example"], None)
+    assert not _calls(d, "threatstopv2-update-domain-list")
+    assert not _calls(d, "zscaler-category-add-url")
+    recs = d.get(d._context, "UCTest.Executions")
+    reasons = {r["vendor"]: r["skip_reason"] for r in recs}
+    assert len(recs) == 2 and all(r["skipped"] for r in recs)
+    assert "list_id" in reasons["ThreatStopv2"] and "SOCFrameworkTenantConfig_V3" in reasons["ThreatStopv2"]
+    assert "category-id" in reasons["Zscaler"]
+
+
+def test_one_vendor_configured_other_still_blocks():
+    d = _run(_row(), ["evil-a.example", "evil-b.example"], _tenant(("ThreatStopv2",)))
+    assert len(_calls(d, "threatstopv2-update-domain-list")) == 2
+    assert not _calls(d, "zscaler-category-add-url")
+
+
+def test_shipped_shadow_simulates_every_pair_and_calls_nothing():
+    d = _run(_row(shadow=True), ["evil-a.example", "evil-b.example"], _tenant())
+    assert not _calls(d, "threatstopv2-update-domain-list")
+    assert not _calls(d, "zscaler-category-add-url")
+    rows = [r for r in _rows(d) if r.get("universal_command") == "soc-block-domain"]
+    assert len(rows) == 4 and all(r["action_status"] == "simulated" for r in rows)
+
+
+def test_no_domain_on_issue_skips_without_calling():
+    d = _run(_row(), None, _tenant())
+    assert not _calls(d, "threatstopv2-update-domain-list")
+    assert not _calls(d, "zscaler-category-add-url")