Query OCSF data

This page shows how to query an OCSF dataset with APL. It covers the end of the OCSF flow in Axiom: the dataset is prepared and receives conformed events, as described in Send OCSF data to Axiom. To search the same data from Splunk, see Search OCSF data from Splunk.

The examples run against the 100 synthetic events in the OCSF for Axiom pack’s axiom_normalize_post_processor sample after conforming. The outputs show what each query returns for that sample. Replace ocsf with the name of your dataset.

Choose the right reference

Every OCSF attribute lives in exactly one place. The place determines how you reference it in APL:

Column names that contain dots must be quoted with brackets, for example ['src_endpoint.ip']. For more information, see Entity names. Inside map fields, use index notation for each level. For more information, see Map fields.

To see which columns your dataset actually contains, use getschema. If an attribute isn’t a column, it’s either inside a map field or under unmapped:

['ocsf']
| getschema

Filter and group on numeric *_id columns rather than their string siblings. For example, use status_id == 2 rather than status == 'Failure'. Producers spell the strings differently, but the IDs are fixed by OCSF. The OCSF schema browser lists the meaning of every ID.

Count events by class

Start with an inventory of the classes in the dataset:

['ocsf']
| summarize events = count() by class_uid, class_name
| sort by events desc

Filter on promoted columns

Failed logons by user and source address. In the Authentication class (3002), activity_id 1 is a logon, status_id 2 is a failure, and user is the account that tried to log on:

['ocsf']
| where class_uid == 3002 and activity_id == 1 and status_id == 2
| summarize failures = count() by ['user.name'], ['src_endpoint.ip'], status_detail
| sort by failures desc

Promoted columns are typed, so numeric comparisons and sums need no conversion. Top destinations by bytes in Network Activity (4001):

['ocsf']
| where class_uid == 4001
| summarize bytes = sum(['traffic.bytes']) by ['dst_endpoint.ip']
| top 3 by bytes

Read fields from unmapped

unmapped keeps the original OCSF nesting, so you read a field with one index per level. Values inside a map field are dynamic, so convert them with tostring, toint, or a similar function before you group or compare. The operating system of the device that recorded each logon is under unmapped because device.os is an object inside an object:

['ocsf']
| where class_uid == 3002
| extend os = tostring(['unmapped']['device']['os']['name'])
| summarize events = count() by os

To find which keys each class carries under unmapped, list the top-level keys with bag_keys and expand them into rows:

['ocsf']
| where isnotnull(unmapped)
| extend keys = bag_keys(unmapped)
| mv-expand keys
| summarize events = count() by class_name, key = tostring(keys)
| sort by class_name asc, key asc

The Detection Finding events also show an array of objects under unmapped. Index into it like any other array:

['ocsf']
| where class_uid == 2004
| project ['device.hostname'],
    rule = tostring(['unmapped']['finding_info']['analytic']['name']),
    process = tostring(['unmapped']['evidences'][0]['process']['name'])

Querying a map field uses more query-hours than querying a column. If you query an unmapped field often, define a virtual field for it so your team can reuse it by name.

Query arrays of objects

The six promoted arrays, observables, vulnerabilities, answers, attacks, file.hashes, and process.file.hashes, are each stored whole in their own map field. Every element keeps its own attributes, so an observable’s value stays next to its type and a technique stays next to its tactic.

To read one element, index it:

['ocsf']
| where class_uid == 2004
| project ['finding_info.title'], ['device.hostname'],
    technique = tostring(['attacks'][0]['technique']['uid']),
    tactic = tostring(['attacks'][0]['tactic']['name'])

To work with every element, expand the array into rows with mv-expand. For example, list every IP address observable in Detection Findings. In observables, type_id 2 is an IP address:

['ocsf']
| where class_uid == 2004
| mv-expand observables
| where toint(observables['type_id']) == 2
| summarize findings = count() by ip = tostring(observables['value'])

To pull one attribute from every element without expanding rows, use array_extract. It returns an array per event:

['ocsf']
| where class_uid == 2004
| extend techniques = array_extract(['attacks'], @'$[*].technique.uid')
| project ['device.hostname'], techniques

Investigate an entity across classes

Because all classes share one dataset and one vocabulary, one query can follow an entity across event types. Everything the sample recorded about one workstation:

['ocsf']
| where ['device.hostname'] == 'wks-014.corp.example.com'
    or ['src_endpoint.hostname'] == 'wks-014.corp.example.com'
    or ['dst_endpoint.hostname'] == 'wks-014.corp.example.com'
| summarize events = count() by class_name

Events are sparse: each class fills only its own columns. In cross-class queries, expect empty values in columns that a class doesn’t use, and guard with isnotnull or isnotempty where it matters.

What’s next

• Search OCSF data from Splunk
• Map fields
• APL overview