Command
your security data

Collect

Fetch paginated API results over HTTP, follow the cursor links, unpack the response envelope, and publish what arrives for the next pipeline.

let $url = "https://api.example.com/security/events"
let $headers = {
  "Authorization": f"Bearer {secret("EVENTS_TOKEN")}",
}
from_http f"{$url}?limit=500",
  headers=$headers,
  paginate=(page => {
    url: f"{$url}?limit=500&cursor={page.next_cursor}",
  } if page.next_cursor != null),
  paginate_delay=250ms,
  max_retry_count=5 {
  read_json
}
unroll events
this = events
ingest = {source: "events-api", received_at: now()}
publish "events.raw"

Operators: from_http, read_json, unroll, set, publish

Functions: secret, now

Parse

Receive Syslog over UDP and turn the message body into typed fields with key-value and request-line parsing.

accept_udp "0.0.0.0:514"
syslog = data.parse_syslog()
fields = syslog.message.trim().parse_kv(unflatten_separator=".")
fields.http.request = fields.http.request.parse_ssv(
  header=["method", "url", "version"],
)
fields.src.ip = ip(fields.src.ip)
fields.dst.ip = ip(fields.dst.ip)
fields.src.port = int(fields.src.port)
fields.dst.port = int(fields.dst.port)
this = {
  timestamp: syslog.timestamp,
  hostname: syslog.hostname,
  app: syslog.app_name,
  ...fields,
}

Operators: accept_udp, set

Functions: parse_syslog, trim, parse_kv, parse_ssv, ip, int

Shape

Restructure events without changing what they mean. Spread records and lists, round timestamps, test subnet membership, derive totals, and sort tags.

let $corp = 10.0.0.0/8
timestamp = timestamp.round(1s)
duration = duration.round(1ms)
client = {
  ...client,
  internal: client.ip in $corp,
}
server = {
  ...server,
  internal: server.ip in $corp,
}
request.path = request.url.split("?")[0]
totals = {
  bytes: bytes_in + bytes_out,
  packets: packets_in + packets_out,
}
tags = [...tags?, request.method.to_lower()].sort()
drop bytes_in, bytes_out, packets_in, packets_out

Operators: set, drop

Functions: round, split, to_lower, sort

Normalize

Map vendor fields into OCSF, ASIM, ECS, CIM, UDM, or your own schema, and keep whatever has no home yet. Every downstream tool then sees the same semantics.

unmapped = this
@name = "ocsf.network_activity"
metadata = {
  version: "1.9.0",
  product: {name: "PAN-OS", vendor_name: "Palo Alto Networks"},
}
category_uid = 4
class_uid = 4001
activity_id = 6
severity_id = 1
type_uid = class_uid * 100 + activity_id
time = move timestamp
src_endpoint = {ip: move src.ip, port: move src.port}
dst_endpoint = {ip: move dst.ip, port: move dst.port}
connection_info.protocol_name = (move proto).to_lower()
ocsf_derive
ocsf_cast

Operators: set, ocsf_derive, ocsf_cast

Functions: to_lower

Optimize

Drop what nobody queries and collapse repeats while the data is still in flight, so storage cost and analyst noise stay down.

where class_uid == 4001 and activity_id == 6
severity_id = severity_id? else 1
deduplicate src_endpoint.ip, dst_endpoint.ip,
  connection_info.protocol_name,
  create_timeout=1h,
  count_field=suppressed
where severity_id >= 4 or suppressed > 10

Operators: where, set, deduplicate

Protect

Protect sensitive fields with encryption, pseudonymization, anonymization, masking, salted hashes, redaction, and drops. Keep useful correlations without exposing raw identities.

let $ip_seed = secret("cryptopan-key").decode_hex()
let $hash_salt = env("HASH_SALT")
src_endpoint.ip = encrypt_cryptopan(src_endpoint.ip, seed=$ip_seed)
user.email_addr = f"*****@{user.email_addr.split("@")[1]}"
user.uid = user.uid.hash_sha256(seed=$hash_salt)
if card_number?.is_luhn_valid() {
  card_number = card_number.slice(begin=-4).pad_start(16, "*")
} else if card_number? != null {
  card_number = "******"
}
unmapped.dob = unmapped.dob.format_time("%Y-01-01").time()
unmapped.api_key = "******"
drop unmapped.password, unmapped.token

Operators: set, drop

Functions: secret, decode_hex, env, encrypt_cryptopan, split, hash_sha256, is_luhn_valid, slice, pad_start, format_time, time

Enrich

Add domain reputation and asset inventory from live lookup tables, then ask a local model to summarize what came back.

where class_uid == 4003
// Lookup tables add reputation and asset context.
context_enrich "domain_reputation",
  key=query.hostname,
  into=observables,
  mode="append"
context_enrich "endpoint_assets",
  key=src_endpoint.ip,
  into=src_endpoint
// GLM5.2 runs on local Ollama.
ai_prompt model="GLM5.2",
  endpoint="http://127.0.0.1:11434/v1",
  system="Summarize this enriched DNS event.",
  data={
    query: query.hostname,
    reputation: observables,
    asset: src_endpoint,
  },
  into=ai.summary

Operators: where, context_enrich, ai_prompt

Analyze

Model coding-agent behavior as operation distributions, duration thresholds, and distinct-session estimates.

where "ai_operation" in metadata.profiles?
window size=15min, on=time {
  summarize agent=ai_agent.name,
    span_duration=tdigest(trace?.span?.duration?),
    operations=frequency_table(api?.operation?),
    sessions=hll(ai_agent.instance_uid)
  p95_duration_ms = tdigest_quantile(span_duration, 0.95)
  active_sessions = hll_cardinality(sessions)
  start = $window.start
  end = $window.end
}
publish "ocsf.agent-insights"

Operators: where, window, summarize, set, publish

Functions: tdigest, frequency_table, hll, tdigest_quantile, hll_cardinality

Detect

Spot SMB traffic spikes with event-time windows and a statistical baseline, while the events are still moving.

where class_uid == 4006 and traffic.bytes? != null
window size=70min, every=10min, on=time {
  summarize src=src_endpoint.hostname, samples=count(),
    avg_bytes=mean(traffic.bytes),
    stdev_bytes=stddev(traffic.bytes),
    current_bytes=max(traffic.bytes)
  upper_bound = avg_bytes + stdev_bytes * 2
  where samples >= 5 and current_bytes > upper_bound
  start = $window.start
  end = $window.end
}

Operators: where, window, summarize, set

Functions: count, mean, stddev, max

Store

Persist compressed, encrypted Parquet in object storage. Rotation and partitioning keep the lake predictable and cheap to query.

let $key = decode_base64(secret("PARQUET_KEY"))
where class_uid == 4001
year = time.year()
month = time.month()
day = time.day()
to_s3 "s3://security-lake/ocsf/network/**/data_{uuid}.parquet",
  partition_by=[class_uid, metadata.product.name,
    year, month, day],
  max_size=128M {
  write_parquet compression_type="zstd", encryption_key=$key
}

Operators: where, set, to_s3, write_parquet

Functions: decode_base64, secret, year, month, day

Search

Query events where they live with federated search in TQL. Tenzir pushes supported filters and field selections down to ClickHouse as SQL.

let $yesterday = now() - 24h
from_clickhouse host="clickhouse", table="security.network_activity"
// Tenzir transpiles these filters and field selections to SQL.
where time >= $yesterday and severity_id >= 4
select time, src_endpoint, dst_endpoint, traffic, severity_id
// Sort and limit the results in Tenzir.
sort -time
head 100
where src_endpoint.ip in 10.0.0.0/8

Operators: from_clickhouse, where, select, sort, head

Functions: now

Route

Split, publish, fork, and load-balance streams to the right destinations. Send alerts, archives, and SIEM feeds without duplicating collection work.

let $splunk = [
  {url: "https://splunk-a:8088", token: secret("HEC_A")},
  {url: "https://splunk-b:8088", token: secret("HEC_B")},
]
subscribe "ocsf.normalized"
fork {
  where severity_id >= 4
  publish "ocsf.alerts.high"
}
load_balance $splunk {
  to_splunk $splunk.url, hec_token=$splunk.token
}

Operators: subscribe, fork, where, publish, load_balance, to_splunk

Functions: secret

Replay

Send stored events back to the start of the pipeline. Repair bounded timestamp disorder, shift the timeline, and recreate an incident at twenty times its original speed.

from_file "s3://security-lake/ocsf/network/**/day=17/*.parquet" {
  read_parquet
}
where time >= 2024-01-17T09:00:00Z and time < 2024-01-17T10:00:00Z
reorder on=time, tolerance=2min
timeshift time, start=now()
delay time, speed=20.0
publish "ocsf.replay"

Operators: from_file, read_parquet, where, reorder, timeshift, delay, publish

Functions: now