|
9 | 9 | python trigger_search.py --type "On demand" # YAML structure for a type |
10 | 10 | python trigger_search.py --events # All Signal event values (API) |
11 | 11 | python trigger_search.py --events detection # Filter event values by text |
| 12 | + python trigger_search.py --fields Investigatable/EPP # Payload field paths for a trigger |
12 | 13 | python trigger_search.py --list --json # Machine-readable output |
13 | 14 | """ |
14 | 15 |
|
|
60 | 61 | "case, identity event, etc.). trigger.type is always 'Signal' and the " |
61 | 62 | "trigger MUST carry an 'event' field naming the event source " |
62 | 63 | "(the trigger category, e.g. 'Investigatable/NGSIEM'). Find event " |
63 | | - "values with --events." |
| 64 | + "values with --events, and the payload field paths a trigger delivers " |
| 65 | + "with --fields <category>." |
64 | 66 | ), |
65 | 67 | "yaml_example": """\ |
66 | 68 | trigger: |
@@ -155,6 +157,72 @@ def search_event_triggers(query=None): |
155 | 157 | return triggers |
156 | 158 |
|
157 | 159 |
|
| 160 | +def _flatten_trigger_fields(fields, prefix=""): |
| 161 | + """Flatten a trigger's recursive fields[] tree into (path, type, display) rows. |
| 162 | +
|
| 163 | + Top-level field `name` values are already fully-qualified dotted paths |
| 164 | + (e.g. ``Trigger.Detection.DetectionID``). Nested `fields[]` children carry |
| 165 | + only a relative `name` (e.g. ``Tactic`` under ``Trigger.Detection.MitreAttack``), |
| 166 | + so those are joined onto the parent path with a dot. |
| 167 | + """ |
| 168 | + rows = [] |
| 169 | + for field in fields or []: |
| 170 | + name = field.get("name", "") |
| 171 | + path = f"{prefix}.{name}" if prefix else name |
| 172 | + children = field.get("fields") |
| 173 | + if children: |
| 174 | + rows.extend(_flatten_trigger_fields(children, path)) |
| 175 | + else: |
| 176 | + rows.append((path, field.get("type", ""), field.get("display", ""))) |
| 177 | + return rows |
| 178 | + |
| 179 | + |
| 180 | +def search_trigger_fields(category): |
| 181 | + """Return the payload field paths a trigger delivers, for a given category. |
| 182 | +
|
| 183 | + `category` is a Signal `event:` value (e.g. ``Investigatable/EPP``). Returns |
| 184 | + a list of {path, type, display} dicts describing every leaf field in the |
| 185 | + trigger payload — the exact ``${data['Trigger....']}`` references available |
| 186 | + to downstream actions. Returns an empty list if the category is unknown. |
| 187 | + """ |
| 188 | + try: |
| 189 | + client = get_client() |
| 190 | + resp = client.search_triggers(filter=f"category:'{category}'") |
| 191 | + if not isinstance(resp, dict): |
| 192 | + return [] |
| 193 | + resources = resp.get("body", {}).get("resources", []) |
| 194 | + except (ConnectionError, RuntimeError, OSError): |
| 195 | + return [] |
| 196 | + |
| 197 | + if not resources: |
| 198 | + return [] |
| 199 | + rows = _flatten_trigger_fields(resources[0].get("fields", [])) |
| 200 | + rows.sort(key=lambda r: r[0]) |
| 201 | + return [{"path": p, "type": t, "display": d} for p, t, d in rows] |
| 202 | + |
| 203 | + |
| 204 | +def _print_fields(category, as_json): |
| 205 | + """Print the payload field paths for a trigger category (from the API).""" |
| 206 | + fields = search_trigger_fields(category) |
| 207 | + if as_json: |
| 208 | + print(json.dumps(fields, indent=2)) |
| 209 | + return |
| 210 | + if not fields: |
| 211 | + print( |
| 212 | + f"No fields found for category '{category}'. Check the value with " |
| 213 | + "--events, or verify credentials." |
| 214 | + ) |
| 215 | + return |
| 216 | + print(f"\nPayload fields for '{category}' ({len(fields)}):\n") |
| 217 | + print(" Reference any of these downstream as ${data['<path>']}.\n") |
| 218 | + for field in fields: |
| 219 | + print(f" ${{data['{field['path']}']}}") |
| 220 | + meta = field["type"] + (f" — {field['display']}" if field["display"] else "") |
| 221 | + if meta.strip(): |
| 222 | + print(f" {meta}") |
| 223 | + print() |
| 224 | + |
| 225 | + |
158 | 226 | def list_all_triggers(): |
159 | 227 | """Return the built-in catalog of the four trigger types.""" |
160 | 228 | return {name: info.copy() for name, info in TRIGGER_CATALOG.items()} |
@@ -232,11 +300,21 @@ def main(): |
232 | 300 | help="List Signal event sources (name -> event value) from the API, " |
233 | 301 | "optionally filtered by QUERY (e.g. --events detection)", |
234 | 302 | ) |
| 303 | + group.add_argument( |
| 304 | + "--fields", |
| 305 | + "-f", |
| 306 | + metavar="CATEGORY", |
| 307 | + help="List the payload field paths a trigger delivers, for a Signal " |
| 308 | + "category (e.g. --fields Investigatable/EPP). Prints ready-to-use " |
| 309 | + "${data['Trigger....']} references so you never guess a field path.", |
| 310 | + ) |
235 | 311 | parser.add_argument("--json", action="store_true", help="Machine-readable JSON output") |
236 | 312 | args = parser.parse_args() |
237 | 313 |
|
238 | 314 | if args.events is not None: |
239 | 315 | _print_events(args.events, args.json) |
| 316 | + elif args.fields: |
| 317 | + _print_fields(args.fields, args.json) |
240 | 318 | elif args.list: |
241 | 319 | _print_list(list_all_triggers(), args.json) |
242 | 320 | elif args.type: |
|
0 commit comments