Testing Logstash Data Ingest

 By Tony Lee

When setting up an Elasticsearch Logstash Kibana (ELK) stack, you may discover the need to test your Logstash filters. There are probably many ways to accomplish this task, but we will cover some techniques and potential pitfalls in this article. 

Note: This article can be used to simulate any syslog replay for manual data ingest testing, but our example specifically covers Logstash as the receiver and a fabricated, but realistic, event.

Real-time Visibility

The first thing that will aid us in this ingest testing is real-time feedback. This could come from at least two potential places:

1)  "Real-time" discovery search within the Kibana UI

This is accomplished by using the discovery dashboard, setting a search term you know to be in your test message, and a short refresh rate as shown in the screenshot below.

Figure 1:  Kibana "Real-time" search using refresh interval

2) Monitoring the logstash log file for warnings or errors

For this, we will use the tail command with the follow option (-f) to watch it in real-time.  The location of your Logstash log may differ so adjust the path as necessary.

tail -f /var/log/logstash/logstash-plain.log

We are looking for clues that may have prevented proper ingest such as:

[2020-08-31T20:25:21,997][WARN ][logstash.filters.mutate][main][b14e-snip-a422] Exception caught while applying mutate filter {:exception=>"Could not set field 'name' on object 'compute-this' to value 'see-this'.This is probably due to trying to set a field like [foo][bar] = someValuewhen [foo] is not either a map or a string"}

Encrypted or Unencrypted Listener

Once we have real-time visibility in place, we need to determine if the listening port is expecting encrypted data since this will determine how we replay traffic to it. There are a few ways to determine this:

1) Check the logstash filter config file

The example below, is an example of an encrypted port.  We can see that is the case because we are defining the SSL information and have ssl_enable set to true.

input {

  tcp {

    port => 6514

    ssl_enable => true

    ssl_cert => "/etc/logstash/logstash.crt"

    ssl_key => "/etc/logstash/logstash.key"

    ssl_verify => false

  }

}

2) Check the logstash logs for SSL errors

If you have an encrypted listener and you send data using an unencrypted transport method (such as telnet), you will see SSL errors such as the following:

[2020-09-01T13:46:20,758][ERROR][logstash.inputs.tcp][main][359d9-snip-c5038d] Error in Netty pipeline: io.netty.handler.codec.DecoderException: javax.net.ssl.SSLHandshakeException: error:100000f7:SSL routines:OPENSSL_internal:WRONG_VERSION_NUMBER

3) Use a tool (such as openssl) to verify the SSL connection

Below is an example of checking the SSL connection using openssl, but other tools can be used.

openssl s_client -connect <logstash_host>:6514

If the listener is expected encrypted data, you will see details such as certificate subject, issuer, cipher suites and more.

--snip--

subject=C = AU, ST = Some-State, O = Internet Widgits Pty Ltd

issuer=C = AU, ST = Some-State, O = Internet Widgits Pty Ltd

--snip--

SSL handshake has read 1392 bytes and written 437 bytes

Verification error: self signed certificate

--snip--

Methods of Replay

Now that we have real-time visibility and we know if the listener is expecting encrypted data or not, we can look at different techniques to replay the traffic. We will start with unencrypted methods first because we can later tunnel the unencrypted data to an encrypted listener. We will also examine replaying an entire packet (including the header) vs. replaying just the data and having the header added.

1) Unencrypted replay of an exact packet (Included specified time and no addition of a header)

If you have example logs that you can expect that include the date/time format, you can replay the exact message using netcat/ncat.  Keep in mind that you are most likely sending a static time, so you will need set your Kibana time range appropriately.

First, place the contents of your event in a test file, we created a file called testevent.txt with the following contents (notice the included date and time):

<46>1 2020-08-31T02:08:08.988000Z sysloghost ExtraField - - [Location] Event Type: OurTest, Event Name: Blocked, Device Name: DESKTOPTESTLOGGER2, File Path: C:\Windows\system32\foo.ps1,  Interpreter: Powershell, Interpreter Version: 10.0.18362.1 (WinBuild.160101.0800), User Name: SYSTEM, Device Id: 4eaf3350a984

Second, use netcat or ncat to send the data to your listening port. The example shown below is sending to an unencrypted listener (be sure to replace logstash_host with the IP or hostname of your logstash server):

ncat <logstash_host> 6514 < testevent.txt

Then just monitor the logstash log and real-time search in Kibana to see the event and/or potential errors.

2) Encrypted replay of an exact packet (Included specified time and no addition of a header)

For this we will use the same testevent.txt file from above and nearly the same command, but we will add --ssl to force ncat to perform the encrypted handshake.

ncat --ssl <logstash_host> 6514 < testevent.txt

3) Unencrypted replay of an exact packet contents with an updated time

If you have packet contents, but want the header updated with the current time, you might be able to use the logger command in Linux.  The trick here is to get logger to reproduce your expected header. Use the following steps to attempt this:

Understand the logger options:

logger --help

Read in our test event and output to stderr for troubleshooting:

logger -s -f testevent.txt

Use logger options to alter the header (in our case, it was --rfc5424=notq) to match what we need and then create a new file with only the content and no header.  Ex:  testevent_noheader.txt

Figure 2:  Reproducing the event with an updated header

Send the event to the unencrypted listener and check for it in Kibana and Logstash logs:

logger --rfc5424=notq -s -f testevent_noheader.txt --server <logstash_host> --tcp --port 6515

4) Encrypted replay of an exact packet contents with an updated time

Unfortunately our version of logger does not have an option to enable encryption. So, if you were able to get logger to reproduce the header + content in the step above, but need to send it to an encrypted listener, you could once again use ncat to assist. The following command creates an unencrypted listener on your local host on port 6515--then anything written to that local port will be sent on in an encrypted state to port 6514.

Figure 3:  ncat listener to send data onto Logstash using SSL

Step 1) Create the listening wrapper:

ncat -l 6515 -c "ncat --ssl <logstash_host> 6514"

Step 2)  Send the packet to the wrapper using logger:

logger --rfc5424=notq -s -f testevent_noheader.txt --server localhost --tcp --port 6515

Conclusion

We are just scratching the surface in ways to test data ingest components such as Logstash. For instance, this could be expanded to include scripting with variables that are replaced with random data to generate more robust values. But that will be an exercise left to the user (or maybe a future article). We do hope this article proved useful and would love to know what you use for testing data ingest. Feel free to leave comments in the section below.  Thanks for reading.

rsyslog fun - Basic Splunk Log Collection and Forwarding - Part II

By Tony Lee

Welcome to part II in our series covering how to use rsyslog to route and forward logs to Splunk. Please see Part I of the series (http://securitysynapse.blogspot.com/2019/01/rsyslog-fun-basic-splunk-log-collection-part-i.html) for the basics in opening ports, routing traffic by IP address or hostname, and monitoring files to send the data on to Splunk Indexers. As a reminder, choosing between rsyslog, syslog-ng, or other software is entirely up to the reader and may depend on their environment and approved/available software. We also realize that this is not the only option for log architecture or collection, but it may help those faced with this task—especially if rsyslog is the standard in their environment. That said, let's look at some more advanced scenarios concerning file permissions, routing logs via regex, and routing logs via ports. We will wrap up with some helpful hints on a possible method to synchronize the rsyslog and Splunk configuration files.

File Permissions

There are times where you may need to adjust the file permissions for the files that rsyslog is writing to disk. For example, if following best practice and running the Splunk Universal Forwarder as a lower privileged account, it will need access to the logs files.  Using the following rsyslogd.conf directives at the top of the configuration file will change the permissions on the directories and files created.  The following example creates directories with permissions of 755 and files with a permission of 644:

$umask 0000 $DirCreateMode 0755 $FileCreateMode 0644

Routing logs via Regex

Another more advanced rsyslog option is the ability to drop or route data at the event level via regex. For example, maybe you want to drop certain packets -- such as Cisco Teardown packets generated from ASA's. Note: this rsyslog ability is useful since we are using Splunk Universal Forwarders in our example and not Splunk Heavy Forwarders.

Or maybe you have thousands of hosts and don't want to maintain a giant list of IP addresses in an if-statement. For example, maybe you want to route thousands of Cisco Meraki host packets to a particular file via a regex pattern.

Possibly even more challenging would be devices in a particular CIDR range that end in a specific octet.

These three examples are covered in the rsyslog.conf snippet below:

#Drop Cisco ASA Teardown packets :msg, contains, ": Teardown " ~ & stop #Route Cisco Meraki hosts to specific directory if ($msg contains ' events type=') then ?ciscoMerakiFile & stop #ICS Devices 10.160.0.0/11 (last octet being .150) :fromhost-ip, regex, "10\.\\(1[6-8][0-9]\\|19[0-1]\\)\..*\.150" -?icsDevices & stop

Routing logs via Port

I know we just provided you the ability to route packets via regex, however sometimes that can be inefficient--especially at high events per second. If you are really fortunate, the source sending the data has the ability to send to a different port. Then it may be worth looking into routing data to different files based on port.  The example file below provides port 6517 and 6518 as an example.

#Dynamic template names template(name="file6517" type="string" string="/rsyslog/port6517/%FROMHOST%/%$YEAR%-%$MONTH%-%$DAY%.log") template(name="file6518" type="string" string="/rsyslog/port6518/%FROMHOST%/%$YEAR%-%$MONTH%-%$DAY%.log") #Rulesets  ruleset(name="port6517"){     action(type="omfile" dynafile="file6517") } ruleset(name="port6518"){     action(type="omfile" dynafile="file6518") } input(type="imtcp" port="6517" ruleset="port6517") input(type="imtcp" port="6518" ruleset="port6518")

Synchronizing Multiple Rsyslog Servers

Since our architecture in part I outlined using a load balancer and multiple rsyslog servers, we will eventually need a way to synchronize the configuration files across the multiple rsyslog servers.  The example below provides two bash shell scripts to perform just that task. The first one will synchronize the rsyslog configuration and the second will synchronize the Splunk configuration--both scripts restart the respective service. Note: This is not the only method available for synchronization, but it is one possible method. Remember to replace <other_server> with the actual IP or FQDN of that server.

On the rsyslog server that you make the changes on, create these two bash scripts and modify the <other_server> section. Once you make a change to the rsyslog or Splunk UF configuration, run the necessary script.

sync-rsyslog.sh

scp /etc/rsyslog.conf <other_server>:/etc/rsyslog.conf ssh <other_server> service rsyslog restart

sync-splunk.sh

scp /opt/splunkforwarder/etc/apps/SplunkUniversalForwarder/local/inputs.conf <other_server>:/opt/splunkforwarder/etc/apps/SplunkUniversalForwarder/local/inputs.conf ssh <other_server> /opt/splunkforwarder/bin/splunk restart

Conclusion

In this article, we outlined key advanced features within rsyslog that may not be immediately evident. Hopefully this article will save you some Googling time when trying to operationalize log collection and forwarding using rsyslog in your environment. After all, eventually you will probably need to deal with file permissions, routing logs via regex and/or port, and configuration synchronization. We hope you enjoyed the article and found it useful.  Feel free to post your favorite tips and tricks in the comments section below. Happy Splunking!

Basic Troubleshooting

So you did everything above and you are still not seeing data...  Walk through some of these steps:

1. Ensure all network firewalls permit the traffic
2. Ensure iptables allows traffic to the rsyslog server
3. Run tcpdump on the source to ensure it is sending data
4. Run tcpdump on the rsyslog server to ensure it is receiving data
5. Verify permissions when writing the files to disk
6. Attempt to telnet (or even web browse) to the rsyslog server port to see if anything is written to the directories

rsyslog fun - Basic Splunk Log Collection and Forwarding - Part I

By Tony Lee

We found it a bit surprising that there are so few articles on how to use an rsyslog server to forward logs to Splunk. This provided the motivation to write this article and hopefully save others some Googling. Choosing between rsyslog, syslog-ng, or other software is entirely up to the reader and may depend on their environment and approved/available software. We realize that this is not the only option for log architecture or collection, but it may help those faced with this task—especially if rsyslog is the standard in their environment.

Warnings

Before we jump in, we wanted to remind you of three potential gotchas that may thwart your success and give you a troubleshooting migraine.

1. Network firewalls – You may not own this, but make sure the network path is clear
2. iptables – Complex rule sets can throw you for a loop
3. SE Linux – Believe it or not, SE Linux when locked down can prevent the writing of the log files

If something is not working the way you expect it to work, it is most likely due to one of the three items mentioned above. It could be worth temporarily disabling them until you get everything working. Just don’t forget to go back and lock it down.

Note:  We will also be using Splunk Universal Forwarders (UF) in this article.  Universal Forwarders have very little pre-processing or filtering capabilities when compared to Heavy Forwarders.  If significant filtering is necessary, consider using a Splunk Heavy Forwarder in the same fashion as we are using the UFs below.

Architecture

Whether your Splunk instance is on-prem or it is in the cloud, you will most likely need syslog collectors and forwarders at some point. The architecture diagram below shows one potential configuration. The number of each component is configurable and dependent upon the volume of traffic.

Figure 1:  Architecture diagram illustrating traffic flow from data sources to the Index Cluster

Rsyslog configuration

Rsyslog is a flexible service, but in this case rsyslog’s primary role will be to:

• Open the sockets to accept data from the sources
• Properly route traffic to local temporary files that Splunk will forward on to the indexers

If you are fortunate enough to be able to route network traffic to different ports, you may be able to reduce the if-then logic shown below for routing the events to separate files. In this case, we were not able to open separate ports from the load balancer, thus we needed to do the routing on our end. In the next article we will cover more advanced routing to include regex and traffic coming in on different ports.

Note:  Modern rsyslog is designed to run extra config files that exist in the /etc/rsyslog.d/ directory. If that directory exists, place the following 15-splunk-rsyslog.conf file in that directory. Otherwise, the /etc/rsyslog.conf file is interpreted from top to bottom, so make a copy of your current config file (cp /etc/rsyslog.conf /etc/rsyslog.bak) and selectively add the following at the top of the new active rsyslog.conf file. This addition to the rsyslog configuration will do the following (assuming the day is 2018-06-01:

• Open TCP and UDP 514
• Write all data from 192.168.1.1 to:  /rsyslog/cisco/192.168.1.1/2018-06-01.log
• Write all data from 192.168.1.2 to:  /rsyslog/cisco/192.168.1.2/2018-06-01.log
• Write all data from 10.1.1.* to /rsyslog/pan/10.1.1.*/2018-06-01.log (where * is the last octet of the source IP
• Write all remaining data to /rsyslog/unclaimed/<host>/2018-06-01.log (where <host> is the source IP or hostname of the sender)

Note:  If the rsyslog server sees the hosts by their hostname instead of IP address, feel free to use $fromhost == '<hostname>' in the configuration file below.

/etc/rsyslog.d/15-splunk-rsyslog.conf

$ModLoad imtcp $ModLoad imudp $UDPServerRun 514 $InputTCPServerRun 514 # do this in FRONT of the local/regular rules $template ciscoFile,"/rsyslog/cisco/%fromhost%/%$YEAR%-%$MONTH%-%$DAY%.log" $template PANFile,"/rsyslog/pan/%fromhost%/%$YEAR%-%$MONTH%-%$DAY%.log" $template unclaimedFile,"/rsyslog/unclaimed/%fromhost%/%$YEAR%-%$MONTH%-%$DAY%.log" if ($fromhost-ip == '192.168.1.1' or $fromhost-ip == '192.168.1.2') then ?ciscoFile & stop if $fromhost-ip startswith '10.1.1' then ?PANFile & stop else ?unclaimedFile & stop # local/regular rules, like *.* /var/log/syslog.log

Note:  Rsyslog should create directories that don't already exist, but just in case it doesn't, you need to create the directories and make them writable.  For example:

mkdir -p /rsyslog/cisco/ mkdir -p /rsyslog/pan/ mkdir -p /rsyslog/unclaimed/

Pro tip:  After making changes to the rsyslog config file, you can verify that there are no syntax errors BEFORE you restart the rsyslog daemon.  For a simple rsyslog config validation.  Try using the following command: 

rsyslogd -N 1

If there are no errors, then you should be good to restart the rsyslog service so your changes take effect:

service rsyslog restart

Log cleanup

The rsyslog servers in our setup are not intended to store the data permanently. They are intended to act as a caching server for temporary storage before shipping the logs off to the Splunk Indexers for proper long-term storage. Since disk space is not unlimited on these caching servers we will need to implement log rotation and deletion so we do not fill up the hard disk. Our rsyslog config file already takes care of the log rotation with the template parameter specifying the name of the file as “%$YEAR%-%$MONTH%-%$DAY%.log", however, we still need to clean up the files, so they don’t sit there indefinitely. One possible solution is to use a daily cron job to clean up files in the /rsyslog/ directory that are more than x days old (where x is defined by the organization). Once you have some files in the /rsyslog/ directory, try the following command to see what would potentially be deleted. The command below lists files in the rsyslog directory that are older than two days.

find /rsyslog/ -type f -mtime +1 -exec ls -l "{}" \;

If you are happy with a two-day cache period, add it to a daily cron job (as shown below).  Otherwise feel free to play with the +1 until you are comfortable with what it will delete and use that for your cron job.

/etc/cron.daily/logdelete.sh

find /rsyslog/ -type f -mtime +1 -delete

Splunk Universal Forwarder (UF) Configuration

Splunk Forwarders are very flexible in terms of data ingest. For example, they can create listening ports, monitor directories, run scripts, etc. In this case, since rsyslog is writing the information to a directory, we will use a Splunk UF to monitor those directories and send them to the appropriate indexes and label them with the appropriate sourcetypes.  See our example configuration below.

Note:  Make sure the indexes mentioned below exist prior to trying to send data there. These will need to be created within Splunk.  Also ensure that the UF is configured to forward data to indexers (out of the scope of this write up).

/opt/splunkforwarder/etc/apps/SplunkForwarder/local/inputs.conf 

[monitor:///rsyslog/cisco/] whitelist = \.log$ host_segment=3 sourcetype = cisco:ios index = cisco [monitor:///rsyslog/pan/] whitelist = \.log$ host_segment=3 sourcetype = pan:traffic index = pan_logs [monitor:///rsyslog/unclaimed/] whitelist = \.log$ host_segment=3 sourcetype = syslog index = lastchanceindex

Pro tip:  Remember to restart the Splunk UF after modifying files.  

/opt/splunkforwarder/bin/splunk restart

Conclusion

A simple Splunk search of index=cisco, index=pan_logs, or index=lastchanceindex should be able to confirm that you are now receiving data in Splunk. Keep monitoring the lastchanceindex to move hosts to where they need to go as they come on-line. Moving the hosts is accomplished by editing the rsyslog.conf file and possibly adding another monitor stanza within the Splunk UF config. This process can be challenging to create, but once it is going, it just needs a little care from time to time to make sure that all is well.  We hope you found this article helpful.  Happy Splunking!

Basic Troubleshooting

So you did everything above and you are still not seeing data...  Walk through some of these steps:

1. Ensure all network firewalls permit the traffic
2. Ensure iptables allows traffic to the rsyslog server
3. Run tcpdump on the source to ensure it is sending data
4. Run tcpdump on the rsyslog server to ensure it is receiving data
5. Verify permissions when writing the files to disk
6. Attempt to telnet (or even web browse) to the rsyslog server port to see if anything is written to the directories

References

• https://www.rsyslog.com/storing-messages-from-a-remote-system-into-a-specific-file/ 
• https://serverfault.com/questions/400293/syslog-ip-ranges-to-specific-files-using-rsyslog  
• https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/7/html/system_administrators_guide/s1-debugging_rsyslog