I have been searching for a way to enrich EventID 3 Sysmon logs with CommunityID for a long time, and most of the solutions I found propose complex solutions. To achieve that, we will harness the power of Apache NiFi to manipulate, route, and filter out Sysmon logs.
Apache NiFi provides a simple graphical interface and many plugins(processors based on NiFi terminology) to achieve almost everything.
In our proposed solution, we are using built-in Processors to create a workflow that will read the Sysmon logs, extract the EventID as an attribute, and if the EventID has the value of 3, then it will try to enrich them with a CommunityID value by executing a simple Python script.
The workflow in NiFi is depicted below
We will analyze in depth the 3 NiFi processors that we have to utilize in order to enrich our logs.
EventID Attribute ( Processor: ExtractText)
Our first step is to read the Sysmon Logs as they come into our NiFi Flow and extract the EventID number. This will help us in the next steps to distinguish which of them are network connections, because to calculate the CommunityID, you need five mandatory fields (and one optional):
– Source IP
– Source Port
– Destination IP
– Destination Port
– Protocol
– Seed (Default: 0)
We will use the ExtractText Processor to read the XML logs, and with a simple regular expression, we will export the EventID number in a NiFi Flow Attribute.
The Property in our case will have the name “EventID” and value “<EventID>([^.<]+)”. This will read the raw log and try to match the text. If we have a match, it will create a new attribute with the EventID and the value of the EventID.
Route based on EventID ( Processor: RouteOnAttribute)
The second step in the process is to create a routing rule that will check the EventID attribute, and if it has the value of “3”, it will send it to a processor that will calculate the CommunityID; otherwise, it will skip the calculation process, and it will follow the remaining flow.
We choose the Routing Strategy as “Route to Property name” so NiFi will know where to find the attribute for routing. Then we have to create a New Property with the name “isEventID3” which will evaluate if the value is 3.
With this step, an additional Relationship will be created with the name “isEventID3” which will forward to the next linked processor only the logs that have EventID equal to 3. All the other logs will follow the “unmatched” Relationship.
Calculate the CommunityID ( Processor: ExecuteStreamCommand)
Finally, we have to calculate the CommunityID. The Python script must be uploaded to our NiFi server.
The script can be found at https://github.com/chrisanag1985/calculate_communityid/blob/main/calculate_communityid.py
The “communityid” dependency must be installed. More information can be found at https://pypi.org/project/communityid/
Our Python script must be accessible from the NiFi process.
In my case, “python3.12” will be used, which is already installed in my NiFi server. In the “Command Arguments” property, we have to specify the absolute path of the Python script.
FYI: In the Python script, there is also a variable named “seed”. This must be changed accordingly to match the configuration that you have in the other tools that you use, like Suricata, Zeek, and Snort. The seed helps to avoid conflicts in CommunityID calculation in cases where you have overlapping network ranges.
If everything works, we will be able to see in the EventID 3 logs that a new field has been added that contains the calculated CommunityID.
Now, we are ready to correlate Sysmon EventID 3 logs with logs from other open source network tools like Suricata.
