SIEM Tools
Security Information and Event Management (SIEM) tools, as a whole, allow for the ingestion of data from various sources, and serve as a point of collection so that data may be queried (searched), aggregated, correlated, and acted upon later. Common SIEM platforms include Splunk, IBM QRadar, Microsoft Sentinel, GrayLog, and others, some of which you may or may not be familiar with.
All SIEM tools have some mechanism to collect this data; they may be called collectors, listeners, sources, or other product-specific terminology, but their purpose is to allow data from outside the SIEM to be moved into its repositories. SIEM platforms may use a push or a pull model, meaning data can be sent to the SIEM initiated by the outside (third-party) data source (a push model), or it may be requested from a third-party source by the consumer, in this case, the SIEM (a pull model). With this in mind, the basis for a SIEM is simple, but the problem comes from attempting to store and review data in very different formats.
Parsing
What is Parsing?
When data arrives at the SIEM it is typically in a structured format; while unstructured formats do exist, they are generally uncommon and discouraged. The format of the logs depends on an agreed upon setting for both the log shipper (source) and the log consumer or collector (the SIEM, or standalone log collector). Generally, this format may be Syslog, Common Event Format (CEF), Log Event Extended Format (LEEF), JavaScript Object Notation (JSON), or a variety of other formats. However, data from different sources often uses a different format, and as such, to be able to store and retrieve this data in a consistent manner, the data must be translated to a consistent format. The process by which this data is made homogenous is called parsing. In parsing, data is passed through the parser – a programmatic sieve that translates one format of data into another.
Fields
Fields are something I will refer to a lot in this article. In essence, a field is simple a key-value pair. The key, or name of the field denotes what kind of data we are storing or referring to, for example, a field name may be ‘user.name’. Field names have many different formats, which we will explore later on, but some common ones include ‘source.x’, ‘Vendor.x’, ‘user.x’, ‘source_ip’, and so on. Some fields separate segments with a period, others with an underscore, some mix both like so ‘Vendor.Source_IP’. The second piece of the field, our value stores the corresponding data we associate with the field name for any particular event. For instance, if we had a field called ‘user.name’, a valid corresponding value could be something like “Bob”. In a SIEM, searches for particular sets of data or events are typically conducted using these fields.
Fields are commonly segmented using prefixes, such as ‘Vendor.’ or ‘user.’ as seen above, another common one is ‘event.’. The appropriate usage for fields depends largely on the data it maps to. There are also standard fields with unique formatting. For instance, CrowdStrike uses a universal ‘@timestamp’ field to denote a standard timestamp format, these fields that start with ‘@’ are called metadata fields by CrowdStrike. Another example may be ‘#event.dataset’, which would be considered a ‘tag’ field due to the ‘#’ character at the start of the field name. All other fields that do not fall into these two categories are considered ‘user’ fields by CrowdStrike. While this distinction is not paricularly important, understanding the standard terminology used in SIEMs and parsing is useful when working on custom implementations.
Stages of Parsing
Parsing typically consists of several stages, which depend on the parsing language used, but this article will only be covering the CrowdStrike Parsing Standard (CPS). At a high level, parsing aims to extract data from known fields in the original structured data, and store it in new fields in the homogeneous format which all data will share. Fields are the identifier for a particular piece of data, for instance, in the example
Jan 18 11:07:53 host CEF:0|SecurityVendor|Product|1.0|100|LoginSuccess|5|src=192.168.1.10 dst=192.168.1.20 suser=Bob msg=success
“suser” is the name of the field, and “Bob” is the value stored in the field. Generally, these fields will be separated by a known character or identifier, which will allow us to differentiate between all the separate pieces of our data. In this example, header fields are delimited by pipes ‘|’, and the extension (payload) fields are key-value pairs separated by an equals sign ‘=’, with individual pairs delineated by spaces; this is an example of a CEF log.
Preparser
The first job of the parser is to extract these values and map them to a new structure, such as ‘dst: 192.167.1.20’ and ‘msg: success’. This process can be called pre-parsing, or field extraction. Typically, most log formats will have header or metadata fields, such as what device generated the log, a generation timestamp, severity, etc. After these initial fields come the payload fields, which are formatted a bit differently than the header fields, and may be a key-value pair, JSON object format, or another data structure for the respective payload fields. Pre-parsing ends when the header data has been pulled out of the original message as fields. Preparsing does not work yet on unstructured payloads such as ‘message’ fields, as they require a bit more nuance to extract.
Metadata
The next stage of parsing is generally referred to as the metadata stage. When parsing logs for a SIEM, all logs are expected to have some fields in common, these fields are called metadata fields, and adding them is part of a process called standardization.
Metadata fields adhere to whatever parsing standard, or schema you are utilizing. One of the most common schemas to use is the Elastic Common Schema (ECS), alongside this, CrowdStrike uses their own CrowdStrike Parsing Standard (CPS) to add their own required metadata fields to event logs, so standards can be combined as the need arises, as long as your standard is, well, standard for your logs. Examples of metadata fields may be a parsing standard version, a common name for the module which shipped the event log, the version of the parser being used, and so on. There are also non-metadata fields related to log standardization, but this is covered in the next stage of parsing.
Normalization
After standardizing the log entry, we now can begin to alter the data format to match how we expect to view the logs in our SIEM. This stage is called normalization, and generally consists of extracting, then renaming payload fields to a standard naming convention, dropping unused fields, and adding event-specific standard payload fields.
Extracting fields from our general ‘message’ payload data is largely the same process as extracting header data, but can be a bit more complex due to the structure of the data, or lack thereof. This will be covered in more depth later in the CrowdStrike-specific parsing section.
Renaming fields is as easy as mapping a field name to a new name, and keeping the value as-is, allowing for easier searching across different log sources. Dropping fields may be useful for cutting out excess data, or ensuring data minimization for data collection limitation. Data collection limitation may be done for several reasons, including regulatory requirements, privacy expectations, relevancy, or efficiency. However, it is the opinion of this author that field dropping is a tuning activity, and should be conducted (if not in violation of data privacy regulations) after a parser is built and ingested data can be observed to determine what fields are useful and what are not.
At the end of this stage of parsing, standard event fields are added, which are similar to metadata fields, but contain event-specific data, rather than source-specific data, such as what action was taken for the event to be generated, what category the event may fit into, and other such data.
Post-Normalization
After normalization, some parsing flows dedicate a section to post-normalization activities for other custom logic that does not fit directly into normalization, but we will not cover that here, as it is not a standard stage of parsing, but rather done as needed for specific use-cases. Some examples may be enriching data with third-party feeds or sources, or performing basic analysis or correlation within your parser itself to have enriched data passed directly to the repository. Finally, at the end of your initial parser creation, your logs will begin to flow through the parser and be stored in whatever repository you have sent your logs to, a SIEM in this case, and can begin being queried with all your fancy new fields and format.
Once a parser has been created the work is not completely over. It is vital during parser-building to use multiple sample logs as test-cases for your parser to ensure complete log coverage and extensibility. As you ingest and parse logs at production volume you may find that your parser needs adjustment, tweaks, or other work, which is a very normal part of the process, and is almost always necessary to some degree.
The CrowdStrike Next-Generation SIEM
Now that we understand what a SIEM is, and how parsing operates, we can begin to apply this directly to the CrowdStrike platform.
How a SIEM Ingests Data
The CrowdStrike Next-Gen SIEM platform is enabled by a series of steps. First, data sources generate some kind of log in interest, this could be a Windows DHCP server log, firewall traffic policy log, or any other kind of log.
Next, the data source, through a log shipping configuration, sends this log to a collector service. Sometimes this collector service is directly integrated into the CrowdStrike console, which uses API connections to cut out a standalone connector service. Otherwise, CrowdStrike offers a LogScale Log Collector service which can be installed on a server to be used as a data collector box. This machine receives logs from the various data sources, then pipes that data to the CrowdStrike console as specified over a dedicated connection, typically using an API endpoint.
With the data passed to the CrowdStrike cloud, it first hits the parser, which will extract, populate, and normalize fields for events. Finally, these parsed events will be sent to whatever data storage method is used, this can be long-term storage, or more likely, the NG-SIEM repository for third-party data, which is where your data can be queried, aggregated, correlated, and more. The following diagram gives a basic outline of the process.
Console Navigation
Within the CrowdStrike console side menu, select the ‘Data Connectors’ section towards the bottom of the menu. Within the Data Connectors popout select ‘Parsers’
CrowdStrike supplies a wealth of pre-built parsers for your use with various sources you may or may not leverage with your NG-SIEM instance. However, for the sake of this article we will be creating our own custom parser. Select ‘Add new parser’ from the right side of the Parsers menu.
This option will open a dialogue to create a new parser, in this menu we will enter the name of our parser. CrowdStrike parser name standards generally exclude any capitalization and leverage hyphens ‘-’ to separate words. For example, a good parser name would be ‘nutanix-flow-security-hit-logs’
The first word is generally the company name, followed by the product or service name, and then after that you can add whatever you want, which in this case will be the specific type of log we are expecting.
Enter your parser name, and then in the dropdown generally you are going to keep the ‘Blank Template’ option, however, you can import parsers, or use a pre-existing template to base yours off of. For the sake of this article we will use a blank template.
For this article, I will be using a parser for the microsegmentation tool called Nutanix Flow, so your parser may require different tuning, different extraction methods, and will involve different fields.
Building Your Parser
Test Cases
When you first open your new parser you will not actually have a fully ‘blank template’, as there are basic sections outlined for you using comments, as well as some metadata fields already populated for your convenience. You will also notice on the right side of the screen there are test data cases already populated.
Test cases are samples of event data that have not been parsed at all yet, and are structured in a specific (hopefully standard) log format. They serve as sets of data that we can apply our parser to and see if we achieve the parsing we so desire from the parser itself. For instance, we can view all of our extracted fields, if the test cases parse properly to include all the necessary metadata fields, if field standardization and normalization was correct, and so on.
Creating Test Cases
Adding your own tests cases is simple, however, it can be a pain to collect sample logs that reflect what your data will look like. Thankfully, there is a simple solution to this problem: ingest real data from your source into the NG-SIEM. While it will be completely unparsed data, we can collect the @rawstring value from those events and use them as our test cases to validate. Once a good sample size of data is collected, turn the connection off again for your log source, to leverage your ingest limits efficiently and not flood your repository with unparsed data. It is not going to hurt anything to have a few minutes worth of unparsed data in your repo.
The step I like to begin with is removing the default test cases and adding our own. This can be done by clicking on the individual test case, and clicking the ‘Delete test’ button above the list of tests. Once you confirm the deletion of all test cases we can begin to add our own test cases.
With your test rawstrings in hand, click on the ‘Add test’ button and paste in the entire rawstring value as-is to the corresponding test field. How many you add is up to you, but I generally try to find ones with different structures to their data (if your data is all similar looking that is fine, and certainly not a bad thing), but the most varied test cases will provide you with the most accurate requirements for your parser to cover all potential data received.
Due to the fact that various data sources will use different formats of data, your test cases will look different than mine unless you are also parsing for Nutanix Flow, in which case, lucky you!
Leveraging Test Cases
With our test cases created, clicking ‘Run tests’ will attempt to parse your test cases using your parser. I recommend you do this immediately as it does not make any changes to any of your tests or your parser. In the output of each test case you have three tabs:
Fields – Contains all of the fields being extracted, added, or otherwise included in some way to your final event log. Field names and values are seen here, and this section can help you very quickly debug your regex.
Assertions – Assertions are a tool which allows you to declare that a specific field must have a specific value, else your test will fail. Assertions are not required, but can help ensure that a field extracts to the correct format and expected output.
Schema violations – Shows any errors caused by non-adherence to the schema requirements for the CrowdStrike Parsing Standard. Specific error details are listed in the validation error column, and can assist in troubleshooting test case failures.
Now that you understand test cases, and how they can be leveraged, we will begin the actual writing of the parser query itself.
Planning Your Parse
Before you actually start typing in the parser script field, you want to have a general outline of your parser process. As described in the Parsing section of this article, there are several steps that a parser should take according to best practices. Keep in mind that a parser script is like any other code or query, there are a few solutions to any given problem, and while there are best practices, subjectivity often plays a part in how your parser is built, so feel free to explore new functions, organizational structures, and other methodologies and find what works best for your use case.
The Parser Script
With all of these disclaimers out of the way we can finally move on to the bread and butter of this article, how to actually write the parser script.
All CrowdStrike parsers are written using the CrowdStrike Query Language (CQL), which is based on the LogScale Query Language, which is further derived from the Splunk Query Language. Knowing this, if you are familiar with building queries in the advanced event search menu then this will be much easier. If you are not familiar with the query language, most of the documentation is, as of writing this, still located in the LogScale documentation and may require slight adjustment, but I highly recommend reading up on the basic syntax and common functions for queries and parsing alike to get a feel for the way things are written. Click here to access the LogScale Documentation. Alertnatively, CrowdStrike also has a dedicated page for creating a parser with some of the basic steps and functions required, but does not get very granular. That page can be found here.
Out of the box you are given a pre-set list of metadata definitions, and a lot of comments to help figure out how to write your script. Lower down there are also other fields pre-populated to assist with script creation. Here is where things start to get more interesting.
While the standard parsing script format may change over time there are some conventions I will cover in this article that may or may not change as the years pass. In this article I will use commented regions in the code to easily tell where in the parser we are.
Preparse
The first region we will define is the preparse region. For an example of how I, along with many other CrowdStrike parsers, define parser regions, see below.
// #region PREPARSE
/************************************************************
****** Parse timestamp and log headers
****** Extract message field for parsing
****** Parse structured data
************************************************************/As stated, we will begin by parsing our timestamp and header fields. To do so, we get a single field to begin working with, called ‘@rawstring’, which contains the entire unedited message from the log source. To actually extract our fields, we can use several techniques, but the most adaptable is simple regex field extraction. You don’t need to be a regex pro to build a parser, but there are some important concepts that should be applied to properly parse fields using it.
Preparse Tips
CrowdStrike provides a lot of sample code in the comments of the default parser script, use that as a resource for learning. Leveraging AI to assist in these scripts has sped things up dramatically in my experience, just be sure to heavily validate all AI-supplied code as any developer should. CrowdStrike also supplies some basic pre-written extractors for common log formats to pull header fields and format timestamps.
It is also worth noting that CrowdStrike has their own AI assistant known as Charlotte AI, and I have been informed that creating parsers for the NG-SIEM is made very easy with Charlotte, but as this time have not been able to confirm this claim with my own hands.
Log Formats
My example uses a Syslog format, specifically RFC 3164, the BSD Syslog standard. CrowdStrike provides the exact regex I need to extract my header fields and pull my ‘message’ field. I will break down the regex for this extraction to provide an example below.
@rawstring = /<(?<log.syslog.priority>d+?)>d+ (?<ts>S+?) (?<log.syslog.hostname>S*?) (-|(?<log.syslog.appname>S+)) (-|(?<log.syslog.procid>S+)) (-|(?<log.syslog.msgid>S+)) (-|(?<log.syslog.structured_data>S+)) (?<message>.*)?/;
The field @rawstring contains our whole log, and we want to separate header fields from our payload fields. The way we do this is by first taking all of our header fields, which are a set of known fields that come before the payload, then everything after these are extracted is a payload, simple enough. Our goal with preparsing is to extract all header data as fields, then extract the general ‘message’ field for later extraction of specific message fields.
Regex Extraction
To actually extract fields we use the syntax (?<field.name>) which is positioned between two markers of where we want our field value to begin and end. By positioning the parenthesis we can be inclusive or exclusive of these markers in our field values. For example, if the log starts with a syslog priority field as the Syslog format does, the log entry may look like this:
<134>2025-…(rest of log excluded)
By using the regex /<(?<log.syslog.priority>d+?)>(?<ts>(S+?))/ we take the ‘<’ symbol, begin our field extraction, search for at least a single digit (still within the parenthesis, indicating we are including these digits in our field value), and then stop collecting the field value as soon as we hit a ‘>’ character, which will extract a field called ‘log.syslog.priority’ and add it to our fields. Afterwards we do this process again, just immediately after our ‘>’ character we begin pulling for a new field called ‘ts’, short for timestamp, and include all non-whitespace characters (S+?) until we hit a space, at which point we stop collecting the field value and save the field.
With this basic principle, it becomes a matter of understanding what format of data you need to scrape for your field, and how to delineate between the start and stop of field values, then build the regex for it. If you want to ever test your regex to see what exactly each step of it does I highly recommend regex101.com while testing to retain your sanity as you spend hours missing a single character interrupting your regular expression.
Extracting Header Fields
Now that we have the not-so-simple process of doing your first field extraction, it gets much more simple from here on out. Your logs hopefully will be all the same format, however, if that is not the case, your parser can use ‘case’ statements when extracting header fields to match the correct regex to your supplied log and extract the proper header fields accordingly. Below you will see an example of this with several supported formats that you may steal and use for your own script.
case {
// syslog, format IETF (RFC 5424) example: <14>1 2018-11-30T16:09:10Z PA-220 - - - - 1,2018/11/30 16:09:09,012801096514,TRAFFIC,end,2049....
@rawstring = /<(?<log.syslog.priority>d+?)>d+ (?<ts>S+?) (?<log.syslog.hostname>S*?) (-|(?<log.syslog.appname>S+)) (-|(?<log.syslog.procid>S+)) (-|(?<log.syslog.msgid>S+)) (-|(?<log.syslog.structured_data>S+)) (?<message>.*)?/;
// syslog, format BSD (RFC 3164) example: <14>Nov 30 16:09:08 PA-220 1,2021/10/26 14:49:02,,SYSTEM,general.....
@rawstring = /<(?<log.syslog.priority>d+?)>(?<ts>(S+?)) (?<log.syslog.hostname>S*) (?<log.syslog.appname>S*): (?<message>.*)?/;
// raw events, example: 1,2021/10/25 20:25:39,,CONFIG,0,2561,2021/10/25 20:25:39,81.2.69.193,,set,admi......
@rawstring = /^d*,/ | message := @rawstring;
// Events picked up from a file example: Oct 05 09:07:32 PA-VM 1,2020/05/07 02:40:08,44A1B3FC68F5304,TRAFFIC,end,2049,.....
@rawstring = /^(?<ts>S+?s+?S+?s+?S+?)s+?(?<log.syslog.hostname>S+?)s+?(?<message>.*)/;
@rawstring = /^{/
| parseJson(field=@rawstring, prefix="Vendor.", handleNull=discard, excludeEmpty=true);
*;
}You may or may not use this same technique in your parser, but being aware it is possible is still useful. There may be techniques not covered in here that are needed in your parser, but being creative is just part of the process as with any script!
Handling Timestamps
The final step in the preparse region is to format our timestamp. To do this, CrowdStrike has built-in functions for parsing timestamps with any format, as you get to define the format yourself. This function can be a bit tricky to figure out if you have not used it before, but it can use both the Java DateTimeFormatter format, or special format specifier words for various epoch times. To read more about the parseTimestamp function and specifics you can read its documentation page here.
Different log sources may log events using different timestamp formats, but they may also log in a different timezone. The above function has a parameter to help define what time zone your log should reflect, and will automatically adjust the time to reflect the accurate time in the relative timezone. Below is an example from the CrowdStrike documentation for parsing a timestamp with the format “2015-12-18T20:39:35”.
parseTimestamp("yyyy-MM-dd'T'HH:mm:ss", field=timestamp, timezone="America/Chicago")
Now that we have completed the preparse steps we can finally move on to the metadata region.
Metadata
The metadata region is a very small region, consisting mostly of pre-made fields prepared for you by CrowdStrike. According to the ECS and CPS standards, we only have a few fields required for every log, some of which are metadata fields, unrelated to the individual events, and some are event-specific which we will cover in normalization.
The fields most parser scripts will require in the metadata region are as follows:
| Vendor := "REQUIRED_INPUT"
| Parser.version := "1.0.0"
| ecs.version := "8.17.0"
| Cps.version := "2.0.0"
| event.module := "REQUIRED_INPUT"
| observer.type := "REQUIRED_INPUT"That covers it for the metadata fields, if you are still getting schema violations refer to this section and normalization to ensure all of the required fields are populated, as the standards are subject to change, but unless you update the version of the standard you are using you should be fine. Now we can move to the last section covered in this article, normalization.
As you can see, some of these fields are pre-set and will not need to be changed, while others require you to set them yourself. As long as they are consistent the values of these are more for your own reference than anything. You may also notice the syntax for assigning a field requires a ‘:=’ assignment operator, so be sure to not forget the colon when making these, otherwise you will just be evaluating instead of assigning.
Normalization
Normalization is an important process, the first goal of this stage is to extract all of our unstructured message/payload data into fields. The second goal is to add, rename, and drop fields to adhere to our schema requirements and desired event output format. it just takes a bit of research into your parsing standard and schema, as well as some effort to extract our payload data.
When it comes to the Elastic Common Standard and the CrowdStrike Parsing Standard you will find that certain conventions are required, as expected. Some event attributes are required such as event.kind, and event.category, among others, but refer to the documentation or your schema violations to see if you are missing anything. Fields like event.category or event.type may require a case statement to set appropriately, as some events from the same source may differ and require a different category or type, so be on the lookout for when that may be needed.
Alongside these fields are some other standard field definitions such as user fields, source and destination fields, client fields, and so on. Depending on what exactly your logs contain you may or may not use some or all of these, but by assigning these fields we can ensure that when correlating disparate log sources we can find values by the same field name. For instance, if you wanted to search for the user “Bob” between all logs, you may need to search different fields like Vendor.username, Vendor.source.user, Vendor.Username (capitalization DOES matter!), or so on.
With standardization, if all events have the same field name for the primary user being referenced in the log, you only have to search for ‘user.name’ and you will find all instances across all sources, which enables log aggregation and correlation, as intended with a SIEM.
Message Fields
With that out of the way, we can handle the meat of our event, our message/payload fields. The ‘message’ field is a clump of our entire payload waiting to be parsed. To parse the payload may take a little more nuance, but don’t fear, there are several ways to approach this. Payloads may come in several formats, or may even switch between formats at points. Being able to identify a specific format is helpful, but good ‘ol regex is always an option to parse fields. However, be warned, if you decide to hard-code your regex extraction, you may find that your logs actually don’t all have the same payload fields, which is why having varied test cases is extremely important to ensure full coverage.
When handling payloads, the CrowdStrike Query Language has several built-in functions to handle set formats, as we see parseJson in the above script, this is just one of several that exist. A full list can be found in the documentation, but one useful one is kvParse, which is used for key-value pairs.
Extracting Message Fields
In the parser I made for this article all test cases had several fields that always appeared, and several fields that may or may not exist. If the log is well-formatted it may be easy to see these differences, or it may be difficult, so your experience may vary. My recommendation is to use regex on any fields you cannot find a pre-built function for parsing, then use the functions anywhere you can to simplify your expressions. You can even extract groups of fields as a single field and extract them later using a pre-built function if they sit between two regex-able fields. See below for my example where I do just this.
| case {
message = /(?<Vendor.Timestamp>S+?) (?<Vendor.ReceiveTime>S+?) (?<Vendor.Policy_UUID>.*?]) (?<Vendor.Policy_Name>.*?)[(?<Vendor.Session_Information>.*?)] (?<Vendor.StartFields>.*?) ORIG:(?<Vendor.Origin>.*?) REPLY:(?<Vendor.Reply>.*?BYTES=d+?)(?<Vendor.EndFields>.*)/
| kvParse(field=Vendor.StartFields, prefix="Vendor")
| kvParse(field=Vendor.Origin, prefix="Vendor.Origin")
| kvParse(field=Vendor.Reply, prefix="Vendor.Reply")
| kvParse(field=Vendor.EndFields, prefix="Vendor");
*
}This case statement is only handling a single case, but allows me to extract our payload fields in a few sets. First I use a regular expression to pull out some fields like the Timestamp and ReceiveTime among others, then we get to my field for ‘Vendor.StartFields’, for which I have a harsh end delineation which identifies our next field group ‘Vendor.Origin’, and so on. With this case statement I extract several groupings of key-value pair fields that I can always expect in my logs, and then afterwards use kvParse to pull the fields from them without needing to make my regex more complicated.
Normalization Functions
When it comes to normalizing field names you have two main options, using the rename function to rename fields in-place, meaning the old field simple vanishes in favor of the renamed new field, or you can assign the new field value to equal the old field using the assignment operator, but if you do this, you must also decide if you want to retain the old fields. For cases where you want to keep both vendor-specific fields you can just leave it as is, but if you want to only keep standardized field names you can use the drop function to remove the fields from your event, cleaning up what you see when searching for events.
// Normalization using 'rename'
| rename([["original_field_1", "Renamed Field 1"],["original_field_2", "Renamed Field 2"]])
// Alternatively, from the CrowdStrike Docs
| rename(field=badName, as=goodName)
// Dropping Fields, can be an array, or a single value
| drop(["original_field_1", "original_field_2", badName])Note that you can add quotes around field names, and while this is optional for fields with no spaces in their name, fields with spaces must have quotes to be properly referenced. Additionally, many of these functions support single values, as well as arrays using comma separated values inside of brackets ‘[ ]’.
When a field is dropped, it will no longer be included in the final event field list as viewed in the SIEM. it is vital to note that dropping fields anywhere in your parser DOES NOT reduce your overall ingestion volume. The CrowdStrike NG-SIEM ingest amount is determined by anything sent to the cloud, which means anything that hits your parser at all. If your aim is to reduce your ingestion to the SIEM look at filtering at the collector, or at the log source itself if possible.
The Finished Parser
Congratulations! You have all the components of a custom parser for your CrowdStrike NG-SIEM data source. I will add below a finished version of a simple custom parser. Your parser may not match this exactly, in fact, it shouldn’t. This article hopefully gives you the tools and resources to go off and create your own parser to suit your needs for your unique logs. While there can be even more intricate methods to parser creation that this article does not cover, I encourage you to scour the documentation to learn for yourself and find creative solutions for your problems.
// #region PREPARSE
/************************************************************
****** Parse timestamp and log headers
****** Extract message field for parsing
****** Parse structured data
************************************************************/
case {
// syslog, format BSD (RFC 3164) example: <14>Nov 30 16:09:08 PA-220 1,2021/10/26 14:49:02,,SYSTEM,general.....
@rawstring = /<(?<log.syslog.priority>\d+?)>(?<ts>(\S+?)) (?<log.syslog.hostname>\S*) (?<log.syslog.appname>\S*): (?<message>.*)?/;
*;
}
| parseTimestamp(format="yyyy-MM-dd'T'HH:mm:ss.SSSSSSXXX", field="ts", as=@timestamp, timezone="UTC")
// #endregion
// #region METADATA
/************************************************************
****** Static Metadata Definitions
************************************************************/
| Parser.version := "1.0.0"
| Vendor := "nutanix"
| event.module := "flow"
| ecs.version := "8.17.0"
| Cps.version := "1.0.0"
// #endregion
// #region NORMALIZATION
/************************************************************
****** Parse unstructured data (i.e. message field)
****** Normalize fields to data model
************************************************************/
// Extract all message/payload fields, some directly, some into temporary field assignments for further extraction
| case {
message = /(?<Vendor.Timestamp>S+?) (?<Vendor.ReceiveTime>S+?) (?<Vendor.Policy_UUID>.*?]) (?<Vendor.Policy_Name>.*?)[(?<Vendor.Session_Information>.*?)] (?<Vendor.StartFields>.*?) ORIG:(?<Vendor.Origin>.*?) REPLY:(?<Vendor.Reply>.*?BYTES=d+?)(?<Vendor.EndFields>.*)/
| kvParse(field=Vendor.StartFields, prefix="Vendor")
| kvParse(field=Vendor.Origin, prefix="Vendor.Origin")
| kvParse(field=Vendor.Reply, prefix="Vendor.Reply")
| kvParse(field=Vendor.EndFields, prefix="Vendor");
*
}
// Drop extracted structured fields used for parsing
| drop([Vendor.StartFields, Vendor.Origin, Vendor.Reply, Vendor.EndFields, message, ts])
// Standardize naming convention for more readable field names
| rename([[Vendor.ACTION, Vendor.Action], [Vendor.DIRECTION, Vendor.Direction], [Vendor.DPORT, Vendor.Destination_Port], [Vendor.DST, Vendor.Destination_IP], [Vendor.Origin.BYTES, Vendor.Origin.Bytes], [Vendor.Origin.PKTS, Vendor.Origin.Packets], [Vendor.PROTO, Vendor.Protocol], [Vendor.Reply.BYTES, Vendor.Reply.Bytes], [Vendor.Reply.PKTS, Vendor.Reply.Packets], [Vendor.SPORT, Vendor.Source_Port], [Vendor.SRC, Vendor.Source_IP]])
// drop empty fields
| case { Vendor.Timestamp = "" | drop([Vendor.Timestamp]); * }
| case { Vendor.ReceiveTime = "" | drop([Vendor.ReceiveTime]); * }
| case { Vendor.Policy_UUID = "" | drop([Vendor.Policy_UUID]); * }
| case { Vendor.Policy_Name = "" | drop([Vendor.Policy_Name]); * }
| case { Vendor.Session_Information = "" | drop([Vendor.Session_Information]); * }
| case { Vendor.Source_IP = "" | drop(Vendor.Source_IP); * }
| case { Vendor.Destination_IP = "" | drop(Vendor.Destination_IP); * }
| case { Vendor.Protocol = "" | drop(Vendor.Protocol); * }
| case { Vendor.Action = "" | drop(Vendor.Action); * }
| case { Vendor.PACKET_STATUS = "" | drop(Vendor.PACKET_STATUS); * }
// *** event fields
| event.dataset := "nutanix.flow"
| event.kind := "event"
| array:append("event.category[]", values="network")
| case {
Vendor.Action = "ALLOW" | event.outcome := "success" | array:append("event.type[]", values="allowed");
Vendor.Action = "DROP" | event.outcome := "success" | array:append("event.type[]", values="denied");
Vendor.Action = "MONITOR" | event.outcome := "success" | array:append("event.type[]", values=["allowed", "info"]);
* | event.outcome := "failure" | array:append("event.type[]", values=["end", "error"]);
}
| event.duration := Vendor.ElapsedTime
| event.start := Vendor.StartTime
| destination.ip := Vendor.Destination_IP
| destination.port := Vendor.Destination_Port
| event.action := lower(Vendor.Action)
| event.direction := lower(Vendor.Direction)
| event.description := Vendor.DESCRIPTION
| network.bytes := Vendor.Origin.Bytes
| network.packets := Vendor.Origin.Packets
| network.protocol := Vendor.Protocol
| rule.name := Vendor.Policy_Name
| rule.id := Vendor.Policy_UUID
| source.ip := Vendor.Source_IP
| source.port := Vendor.Source_Port
| event.created := Vendor.ReceiveTime
| event.original := Vendor.Timestamp
| network.bytes_in := Vendor.Reply.Bytes
| network.packets_in := Vendor.Reply.Packets
// #endregionIt is very worth noting that while this guide does not cover the entirety of a Next-Gen SIEM implementation. That would be a much longer article, but I may create such articles broken up by stages of implementing the CrowdStrike Next-Gen SIEM, so stay tuned!
Finalization
The work does not necessarily end here. Once you save your parser and apply it to your connector you should monitor for errors that appear in the events and react accordingly. If you get an error on specific logs try adding them to your test cases and repeat the parser building process to account for the new test case. This is an iterative process, so don’t be afraid to take a step backwards if you find an issue in your original script.
The default ‘blank’ custom parser provided by CrowdStrike provides a wealth of knowledge, and you should review it closely to understand what may be missing if you run into issues. I also go into other parsers for different sources and scour the functions and techniques used in them to find solutions to my issues.
By the end of this process you should be able to build your parser from scratch for any data source you want to pull into your NG-SIEM instance and have all the fields required for whatever use-case you may have.
On a side note, this is my first blog article upload, and I want to continue making these articles around similar topics, and even less technical topics. Please feel free to leave comments below with feedback or thoughts on this page so I can improve my writing! If you feel this helped feel free to let me know, or if you think I missed something crucial be sure to tell me as well!
Leave a Reply