skills/workflows/references/cel-expressions.md

# CEL Expressions in Fusion Workflows

CrowdStrike Fusion uses Google's Common Expression Language (CEL) for conditional logic
and data transformation. This reference covers syntax, CrowdStrike-specific extensions,
and YAML quoting rules learned from building our 30 production workflows.

---

## Where CEL is used

- **`cel_expression`** key in `conditions` blocks — routing logic
- **HTTP Action bodies** — data transformation before sending to external APIs
- **Property templates** — complex computed values in `${...}` expressions

---

## Basic operators

| Operator | Description | Example |
|----------|-------------|---------|
| `==` | Equality | `data['type'] == 'ip'` |
| `!=` | Inequality | `data['status'] != 'closed'` |
| `<`, `>`, `<=`, `>=` | Comparison | `data['score'] >= 80` |
| `&&` | Logical AND | `a == 'x' && b == 'y'` |
| `\|\|` | Logical OR | `a == 'x' \|\| a == 'y'` |
| `!` | Logical NOT | `!cs.cidr.valid(data['ip'])` |
| `? :` | Ternary | `(x != null ? x : 'default')` |
| `in` | Membership | `'admin' in data['roles']` |
| `+` | String concat / list concat | `data['a'] + data['b']` |

---

## List comprehensions

```
list.map(item, expr)         # Transform: [1,2,3].map(x, x * 2) → [2,4,6]
list.filter(item, cond)      # Filter:    [1,2,3].filter(x, x > 1) → [2,3]
list.exists(item, cond)      # Any match: [1,2,3].exists(x, x > 2) → true
list.all(item, cond)         # All match: [1,2,3].all(x, x > 0) → true
list.exists_one(item, cond)  # Exactly one match
```

**Real example** (from RAN-004 — extracting host IDs from ThreatGraph results):
```
${data['GetDevicesIPv4.Connections'].map(item, item.HostID)}
```

---

## String functions

```
string.contains('sub')       # Substring check
string.startsWith('prefix')
string.endsWith('suffix')
string.matches('regex')      # RE2 regex match
size(string)                 # Length
```

---

## CrowdStrike custom CEL functions

These are extensions not in standard CEL. Discovered from our production workflows:

### `cs.cidr.valid(string)` → bool
Check if a string is a valid IPv4/IPv6 address or CIDR.
```
cs.cidr.valid(data['iocs.#'])
```

### `cs.string.find(string, regex)` → string
Find the first regex match in a string. Returns the match or empty string.
```
cs.string.find(data['iocs.#'], '[A-Fa-f0-9]{64}') == data['iocs.#']
```
This pattern checks if the entire string is a 64-char hex (SHA256).

### `cs.map.merge(list_of_maps)` → map
Merge multiple maps into one. Later entries overwrite earlier ones.
```
cs.map.merge([
    {"searchReason": data['searchReason']},
    (data['from'] != null ? {"from": data['from']} : {})
])
```
Combined with ternary operators, this builds conditional JSON bodies.

### `has(field)` — standard CEL
Test if a field exists (not null).
```
has(data['optional_field'])
```

### `cs.timestamp.parse(string, layout)` → timestamp
Parse a time string into a timestamp. Use `'RFC3339'` for the ISO-8601 times Fusion emits, such as the detection time in `Trigger.ObservedTime`.
```
cs.timestamp.parse(data['Trigger.ObservedTime'], 'RFC3339')
```

Combine it with standard CEL `timestamp()` and `duration()` for time math. To get **Unix epoch milliseconds** — what CQL's `setTimeInterval()` and many external APIs expect — subtract the epoch and take the total milliseconds of the resulting duration, then cast to `int`:
```
int((cs.timestamp.parse(data['Trigger.ObservedTime'], 'RFC3339') - timestamp('1970-01-01T00:00:00Z')).getMilliseconds())
```
Subtract a `duration(...)` to shift the window — e.g. one hour before the detection:
```
int((cs.timestamp.parse(data['Trigger.ObservedTime'], 'RFC3339') - duration('1h') - timestamp('1970-01-01T00:00:00Z')).getMilliseconds())
```
`.getMilliseconds()` on a duration returns the **total** milliseconds, not a 0–999 component. Verified live: an `ObservedTime` of `2026-08-21T12:00:00Z` yields `1787313600000` (end) and `1787310000000` (start, one hour earlier).

---

## YAML quoting rules — critical gotchas

### Single quotes in `cel_expression` values

CEL expressions use `data['field']` with single quotes. In YAML, this requires careful quoting:

**Option 1 — Unquoted (works when no YAML special chars present)**:
```yaml
cel_expression: data['iocs.#.ioc_type'] == 'ip'
```

**Option 2 — Single-quoted YAML string (double the inner single quotes)**:
```yaml
cel_expression: '!cs.cidr.valid(data[''iocs.#'']) && cs.string.find(data[''iocs.#''], ''[A-Fa-f0-9]{64}'') != data[''iocs.#'']'
```

**Option 3 — Block scalar (safest for complex expressions)**:
```yaml
cel_expression: >-
    !cs.cidr.valid(data['iocs.#'])
    && cs.string.find(data['iocs.#'], '[A-Fa-f0-9]{64}') != data['iocs.#']
```

### When quoting is required

You MUST quote the CEL expression if it contains:
- `!` at the start (YAML interprets as tag)
- `:` followed by space (YAML key-value separator)
- `#` not inside quotes (YAML comment)
- `{` or `}` at the start (YAML flow mapping)
- `[` or `]` at the start (YAML flow sequence)

### `display` array quoting

The `display` field often contains HTML-escaped versions of the expression.
CrowdStrike's export uses `&#39;` for single quotes and `&amp;` for ampersands:

```yaml
display:
    - '!cs.cidr.valid(data[&#39;iocs.#&#39;]) &amp;&amp; ...'
```

When authoring manually, use plain human-readable text instead:
```yaml
display:
    - IOC type is IP address
```

---

## CEL vs FQL expressions in conditions

Fusion workflows support two condition syntaxes:

| YAML key | Language | Use case |
|----------|----------|----------|
| `cel_expression` | CEL | Data comparisons, type checks, string matching |
| `expression` | FQL-style | Field membership checks (e.g., group inclusion) |

Prefer `cel_expression` (CEL) for new workflows. FQL-style `expression` is legacy; keep it only when reading older exported workflows.

**FQL-style example** (legacy):
```yaml
expression: GetUserIdentityContext.Groups:!['SkipCrowdStrikeWorkflows']
```
This checks that the Groups array does NOT contain "SkipCrowdStrikeWorkflows".

**`else` and `else_if` with CEL**: CEL conditions support both `else_if` (chain to the next condition node) and `else` (default fallthrough). Build if / else-if / else like this:

```yaml
conditions:
    is_bar:                          # IF
        cel_expression: data['foo'] == "bar"
        next:
            - PrintBar
        else_if: is_tea
    is_tea:                          # ELSE IF
        cel_expression: data['foo'] == "tea"
        next:
            - PrintTea
        else:                        # ELSE (default fallthrough)
            - PrintDefault
```

Under the hood this is an exclusive gateway in the workflow JSON, where the `else` branch is the gateway's `default` flow. The YAML is a conversion of that JSON; the backend processes the JSON.

---

## Common patterns from our workflows

### Type routing (RAN-004)
```yaml
conditions:
    is_ip:
        cel_expression: data['iocs.#.ioc_type'] == 'ip'
        next: [ProcessIP]
    is_sha256:
        cel_expression: data['iocs.#.ioc_type'] == 'sha256'
        next: [ProcessSHA256]
    is_domain:
        cel_expression: data['iocs.#.ioc_type'] == 'domain'
        next: [ProcessDomain]
```

### IOC type detection without enum (PHI-004)
```yaml
conditions:
    is_ip:
        cel_expression: cs.cidr.valid(data['iocs.#'])
    is_sha256:
        cel_expression: cs.string.find(data['iocs.#'], '[A-Fa-f0-9]{64}') == data['iocs.#']
    is_domain:
        cel_expression: "!cs.cidr.valid(data['iocs.#']) && cs.string.find(data['iocs.#'], '[A-Fa-f0-9]{64}') != data['iocs.#']"
```

### Conditional JSON body building (PHI-001)
```yaml
json:
    data:
        - |-
          ${
              cs.map.merge([
                  {"searchReason": data['searchReason']},
                  (data['from'] != null && data['from'] != "" ? {"from": data['from']} : {})
              ])
          }
```