When to Use Universal Event Sources
While InsightIDR supports a variety of Event Sources, there are also times where ingesting events from an unsupported source may be necessary. Historically this could be accomplished with the use of Generic Syslog as an event source type; however, a generic, unstructured log makes it difficult for user attribution or to leverage some in-product functionality. Instead, if one of the following types of data is being ingested from an unsupported tool, leveraging the built-in Universal Event Sources is the best way to go:
- DHCP
- Antivirus
- Ingress Authentication
- VPN
In the case of Ingress Authentication and using the Universal Event Format, InsightIDR will continue to use this formatted activity for incident detection, visualization on the Ingress Locations map and dashboards, as well as investigations in Log Search. No need to lose functionality you expect from InsightIDR just because a source isn’t natively supported.
Getting Started
There are many ways to get started with ingesting data from unsupported sources. Python is a quick go-to for scripting, and NXLog is a utility available with a community edition if scripting isn’t in your wheelhouse. While you don’t have to limit yourself to just those, getting started is quick and easy.
NXLog
A previous write-up exists for using Universal Event Formats with NXLog to transform an ingress authentication log and send it to the respective Universal Event Source. The thorough walk-through is a good starting point if NXLog is your tool of choice!
Python
When writing a script to transform and send data to an Universal Event Source, there are three main components of the script:
- Receive RAW third-party events
- Transform events to Universal Event Format
- Send events to InsightIDR collector
In the example script provided, the RAW source could just be a generic JSON response from a third-party endpoint. In our example, we will hardcode some data that is returned from a REST API:
def get_events():
raw_events = """
[
{
"id":"b01b1726-0147-425e-a7f7-21f252050400",
"createdDateTime":"2018-11-06T18:48:33.8527147Z",
"userDisplayName":"Jon Doe",
"userPrincipalName":"jdoe@www.contoso.com",
"userId":"d7cc485d-2c1b-422c-98fd-5ce52859a4a3",
"appId":"c44b4083-3bb0-49c1-b47d-974e53cbdf3c",
"appDisplayName":"Azure Portal",
"ipAddress":"207.254.19.10",
"clientAppUsed":"Browser",
"status":{}
},
"""
return json.loads(raw_events)
Then once we read and parse the json, we will need to pull out the elements and map them to our InsightIDR format. As documented on the help pages, a very specific format is necessary when ingesting authentication events:
def transform_events(incoming_events):
tevents = []
for e in incoming_events:
tevents.append(
{
"version": "v1",
"event_type": "INGRESS_AUTHENTICATION",
"time": e.get('createdDateTime'),
"account": e.get('userPrincipalName'),
"source_ip": e.get('ipAddress'),
"authentication_result": "SUCCESS" if e.get('status').get('errorCode') == 0 else "FAILURE",
"authentication_target": e.get('appDisplayName'),
}
)
return(tevents)
And finally, with the use of a syslog client you can send the message to an InsightIDR collector:
for event in events:
syslog_client.forward_to_collector(json.dumps(event))
A full script to use as a starting point has been included. Check it out and let us know how else you’ve been successful with using Universal Event Sources in your organization.
Script
import socket
import json
# Syslog client for sending events to InsightIDR collector
class SyslogClient:
def __init__(self, host, port):
self.socket = self.setup()
self.host = host
self.port = port
@staticmethod
def setup():
# Setup socket for sending syslog UDP packet
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
return sock
def forward_to_collector(self, syslog_message):
# Send message to listener
self.socket.sendto(syslog_message.encode(), (self.host, self.port))
return
# Placeholder for mocking events from third-party source
def get_events():
# Payload based off of Microsoft Graph API Example:
# https://docs.microsoft.com/en-us/graph/api/signin-list?view=graph-rest-beta&tabs=http#example-2-user-signs-in-with-only-primary-authentication-primary-authentication-is-through-cloud-password
raw_events = """
[
{
"id":"b01b1726-0147-425e-a7f7-21f252050400",
"createdDateTime":"2018-11-06T18:48:33.8527147Z",
"userDisplayName":"Jon Doe",
"userPrincipalName":"jdoe@www.contoso.com",
"userId":"d7cc485d-2c1b-422c-98fd-5ce52859a4a3",
"appId":"c44b4083-3bb0-49c1-b47d-974e53cbdf3c",
"appDisplayName":"Azure Portal",
"ipAddress":"207.254.19.10",
"clientAppUsed":"Browser",
"status":{}
},
{
"id":"b01b1726-0147-425e-a7f7-21f252050400",
"createdDateTime":"2018-11-07T18:48:33.8527147Z",
"userDisplayName":"Jon Doe",
"userPrincipalName":"jdoe@www.contoso.com",
"userId":"d7cc485d-2c1b-422c-98fd-5ce52859a4a3",
"appId":"c44b4083-3bb0-49c1-b47d-974e53cbdf3c",
"appDisplayName":"Azure Portal",
"ipAddress":"207.254.19.10",
"clientAppUsed":"Browser",
"status":{}
},
{
"id":"b01b1726-0147-425e-a7f7-21f252050400",
"createdDateTime":"2018-11-07T18:48:33.8527147Z",
"userDisplayName":"Jon Doe",
"userPrincipalName":"jdoe@www.contoso.com",
"userId":"d7cc485d-2c1b-422c-98fd-5ce52859a4a3",
"appId":"c44b4083-3bb0-49c1-b47d-974e53cbdf3c",
"appDisplayName":"Azure Portal",
"ipAddress":"207.254.19.10",
"clientAppUsed":"Browser",
"status":{}
}
]
"""
return json.loads(raw_events)
# Put event into InsightIDR Universal Event Format for Ingress Authentication
def transform_events(incoming_events):
tevents = []
for e in incoming_events:
tevents.append(
{
"version": "v1",
"event_type": "INGRESS_AUTHENTICATION",
"time": e.get('createdDateTime'),
"account": e.get('userPrincipalName'),
"source_ip": e.get('ipAddress'),
"authentication_result": "SUCCESS" if e.get('status').get('errorCode') == 0 else "FAILURE",
"authentication_target": e.get('appDisplayName'),
}
)
return(tevents)
if __name__ == "__main__":
# Initialize client vars
syslog_client = SyslogClient('localhost', 6514)
events = get_events()
events = transform_events(events)
for event in events:
syslog_client.forward_to_collector(json.dumps(event))