This guide explains the basic steps for creating a simple Graylog Pipeline to consume logs sent from Sophos XG . It is not meant to be a comprehensive exploration of this topic nor the ultimate authority (or any authority for that matter) on how to accomplish this. What follows is what works for us, your mileage may vary.
A Syslog UDP input on Graylog configured as a Syslog Server within XG. Please see the documentation for Graylog and/or XG for help with these configurations.
Note: No extractors were configured on the input used for this guide.
Our Graylog server consumes Syslog messages from various sources, including networking appliances and Linux servers. All messages entering our UDP Syslog input are split into streams dedicated to processing pipelines specific to the message source type. We also have a generic Syslog pipeline which is shared by (almost) all of these streams. There are many ways to setup processing flows in Graylog, and this model just seemed to work for us.
In this guide, we will assume you have a similar configuration with all XG messages entering a stream dedicated to XG. If this is not the case, you will most likely want to add a stage at the beginning of the XG Pipeline which filters our any non-XG message from further processing.
There are a few rules we like to apply to all syslog messages. Rather than add these rules to every pipeline, we created one pipeline that can be shared by different streams. Several of our rules are specific to our network, but here is an example of what you can (and probably should) do for your XG logs as well as any other Syslog source
Since XG is sending Syslog format, go ahead and give yourself the text value of the log level as follows:
rule "level name"
when
true
then
set_field("severity",syslog_level($message.level));
end
This pipeline contains only rules for XG messages. We accomplish the basic message parsing in two separate stages which, in our environment are configured as stage 1 and 2. As noted above, if you have attached your XG Pipeline to a stream containing non-XG messages, you should include a stage at the beginning of the pipeline to exclude non-XG messages from further processing.
The first stage of pipeline processing is used to parse the XG log types into distinct message fields. The following rule creates log_type , log_component , and log_sub_type which are used in the next stage for individual message type parsing.
rule "XG Log Type"
when
to_string($message.source) == "xg.lan.paravantage.com"
then
set_fields(
grok(
pattern: "log_type=%{QUOTEDSTRING:log_type} log_component=%{QUOTEDSTRING:log_component} log_subtype=%{QUOTEDSTRING:log_sub_type}",
value: to_string($message.message),
only_named_captures: true
)
);
end
This stage performs the bulk of extraction work, with rules dedicated to specific log type and sub-types. To implement the optional Stage 3 you should configure the pipeline to continue as long as one of the rules matches.
rule "XG Firewall Type"
when
to_string($message.log_type) == "Firewall"
then
set_fields(
grok(
pattern: "status=%{QUOTEDSTRING:action}%{SPACE}priority=%{WORD:priority}%{SPACE}duration=%{INT:duration}%{SPACE}fw_rule_id=%{INT:fw_rule_id}%{SPACE}policy_type=%{INT:policy_type}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{SPACE}user_gp=%{QUOTEDSTRING:user_group}%{SPACE}iap=%{INT:iap}%{SPACE}ips_policy_id=%{INT:ips_policy_id}%{SPACE}appfilter_policy_id=%{INT:app_filter_policy_id}%{SPACE}application=%{QUOTEDSTRING:application}%{SPACE}application_risk=%{INT:application_risk}%{SPACE}application_technology=%{QUOTEDSTRING:application_technology}%{SPACE}application_category=%{QUOTEDSTRING:application_category}%{SPACE}in_interface=%{QUOTEDSTRING:in_interface}%{SPACE}out_interface=%{QUOTEDSTRING:out_interface}%{SPACE}src_mac=%{DATA:src_mac}%{SPACE}src_ip=%{DATA:src_ip}%{SPACE}src_country_code=%{DATA:src_country_code}%{SPACE}dst_ip=%{DATA:dst_ip}%{SPACE}dst_country_code=%{DATA:dst_country_code}%{SPACE}protocol=%{QUOTEDSTRING:protocol}%{SPACE}(src_port=%{INT:src_port}%{SPACE}dst_port=%{INT:dst_port})?(icmp_type=%{INT:icmp_type}%{SPACE}icmp_code=%{INT:icmp_code})?%{SPACE}sent_pkts=%{INT:sent_pkts;int}%{SPACE}recv_pkts=%{INT:recv_pkts;int}%{SPACE}sent_bytes=%{INT:sent_bytes;int}%{SPACE}recv_bytes=%{INT:recv_bytes;int}%{SPACE}tran_src_ip=%{DATA:tran_src_ip}%{SPACE}tran_src_port=%{INT:tran_src_port}%{SPACE}tran_dst_ip=%{DATA:tran_dst_ip}%{SPACE}tran_dst_port=%{INT:tran_dst_port}%{SPACE}srczonetype=%{QUOTEDSTRING:src_zone_type}%{SPACE}srczone=%{QUOTEDSTRING:src_zone}%{SPACE}dstzonetype=%{QUOTEDSTRING:dst_zone_type}%{SPACE}dstzone=%{QUOTEDSTRING:dst_zone}%{SPACE}dir_disp=%{QUOTEDSTRING:dir_disp}%{SPACE}(connevent=%{QUOTEDSTRING:conn_event})?%{SPACE}connid=%{QUOTEDSTRING:conn_id}%{SPACE}vconnid=%{QUOTEDSTRING:v_conn_id}%{SPACE}hb_health=%{QUOTEDSTRING:hb_health}%{SPACE}message=%{QUOTEDSTRING:fw_message}%{SPACE}appresolvedby=%{QUOTEDSTRING:app_resolved_by}%{SPACE}app_is_cloud=%{INT:app_is_cloud;boolean}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG Content Filter Type"
when
to_string($message.log_type) == "Content Filtering"
then
set_fields(
grok(
pattern: "status=%{QUOTEDSTRING:action}%{SPACE}priority=%{GREEDYDATA:priority}%{SPACE}fw_rule_id=%{INT:fw_rule_id}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{SPACE}user_gp=%{QUOTEDSTRING:user_group}%{SPACE}iap=%{INT:iap}%{SPACE}category=%{QUOTEDSTRING:category}%{SPACE}category_type=%{QUOTEDSTRING:category_type}%{SPACE}url=%{QUOTEDSTRING:url}%{SPACE}contenttype=%{QUOTEDSTRING:content_type}%{SPACE}override_token=%{QUOTEDSTRING:override_token}%{SPACE}httpresponsecode=%{QUOTEDSTRING:http_response_code}%{SPACE}src_ip=%{IP:src_ip}%{SPACE}dst_ip=%{IP:dst_ip}%{SPACE}protocol=%{QUOTEDSTRING:protocol}%{SPACE}src_port=%{INT:src_port}%{SPACE}dst_port=%{INT:dst_port}%{SPACE}sent_bytes=%{INT:sent_bytes;int}%{SPACE}recv_bytes=%{INT:recv_bytes;int}%{SPACE}domain=%{URIHOST:domain}%{SPACE}exceptions=%{DATA:exceptions}%{SPACE}activityname=%{QUOTEDSTRING:activity_name}%{SPACE}reason=%{QUOTEDSTRING:reason}%{SPACE}user_agent=%{QUOTEDSTRING:user_agent}%{SPACE}status_code=%{QUOTEDSTRING:status_code}%{SPACE}transactionid=%{DATA:transaction_id}%{SPACE}referer=%{QUOTEDSTRING:referer}%{SPACE}download_file_name=%{QUOTEDSTRING:downloaded_file_name}%{SPACE}download_file_type=%{QUOTEDSTRING:downloaded_file_type}%{SPACE}upload_file_name=%{QUOTEDSTRING}%{SPACE}upload_file_type=%{QUOTEDSTRING}%{SPACE}con_id=%{INT:con_id}%{SPACE}application=%{QUOTEDSTRING:application}%{SPACE}app_is_cloud=%{INT:app_is_cloud;boolean}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG IDP Type"
when
to_string($message.log_type) == "IDP"
then
set_fields(
grok(
pattern: "priority=%{WORD:priority}%{SPACE}idp_policy_id=%{INT:idp_policy_id}%{SPACE}fw_rule_id=%{INT:fw_rule_id}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{SPACE}signature_id=%{INT:signature_id}%{SPACE}signature_msg=%{QUOTEDSTRING:signature_msg}%{SPACE}classification=%{QUOTEDSTRING:classification}%{SPACE}rule_priority=%{INT:rule_priority}%{SPACE}src_ip=%{DATA:src_ip}%{SPACE}src_country_code=%{DATA:src_country_code}%{SPACE}dst_ip=%{DATA:dst_ip}%{SPACE}dst_country_code=%{DATA:dst_country_code}%{SPACE}protocol=%{QUOTEDSTRING:protocol}%{SPACE}src_port=%{INT:src_port}%{SPACE}dst_port=%{INT:dst_port}%{SPACE}platform=%{QUOTEDSTRING:platform}%{SPACE}category=%{QUOTEDSTRING:category}%{SPACE}target=%{QUOTEDSTRING:target}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG Event Authentication Type"
when
(to_string($message.log_type) == "Event") && (to_string($message.log_sub_type) == "Authentication")
then
set_fields(
grok(
pattern: "status=%{QUOTEDSTRING:status}%{SPACE}priority=%{WORD:priority}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{SPACE}usergroupname=%{QUOTEDSTRING:user_group}%{SPACE}auth_client=%{QUOTEDSTRING:auth_client}%{SPACE}auth_mechanism=%{QUOTEDSTRING:auth_mechanism}%{SPACE}reason=%{QUOTEDSTRING:reason}%{SPACE}src_ip=%{DATA:src_ip}%{SPACE}message=%{QUOTEDSTRING:auth_message}%{SPACE}name=%{QUOTEDSTRING:name}%{SPACE}src_mac=%{DATA:src_mac}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG System Type"
when
to_string($message.log_sub_type) == "System"
then
set_fields(
grok(
pattern: "(status=%{QUOTEDSTRING:status})?%{SPACE}priority=%{WORD:priority}%{SPACE}(status=%{QUOTEDSTRING:status})?%{GREEDYDATA}message=%{QUOTEDSTRING:system_message}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG SMTP Type"
when
to_string($message.log_component) == "SMTP"
then
set_fields(
grok(
pattern: "priority=%{WORD:priority}%{SPACE}fw_rule_id=%{INT:fw_rule_id}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{GREEDYDATA}from_email_address=%{QUOTEDSTRING:from_email_address}%{SPACE}to_email_address=%{QUOTEDSTRING:to_email_address}%{SPACE}email_subject=%{QUOTEDSTRING:email_subject}%{GREEDYDATA}src_domainname=%{QUOTEDSTRING:src_domain}",
value: to_string($message.message),
only_named_captures: true
)
);
end
rule "XG GUI Event Type"
when
to_string($message.log_type) == "Event" && to_string($message.log_component) == "GUI"
then
set_fields(
grok(
pattern: "status=%{DATA:QUOTEDSTRING}%{SPACE}priority=%{WORD:priority}%{SPACE}user_name=%{QUOTEDSTRING:user_name}%{SPACE}src_ip=%{DATA:src_ip}%{SPACE}ZONE=%{QUOTEDSTRING:zone}%{SPACE}message=%{QUOTEDSTRING:event_message}",
value: to_string($message.message),
only_named_captures: true
)
);
end
This is optional, and reflects our method of alerting for messages that we somehow missed in our processing rules. The following sets a flag to indicate the XG message was captured and parsed by one of the rules in Stage 2 . If any of our XG log entries exit the pipeline without this flag, a notification is sent so we can investigate and update or add rules. We haven't had a anything hit this in a while, but our XG configuration and usage may not match yours, so it is recommended that you perform some form of capture to identify rule deficiencies.
rule "Pipeline Processed Flag"
when
true
then
set_field("pipeline_processed",true);
end
GaryParr
free!
Hi there - I tried the above but for some reason when I created a generic pipeline for all syslog and then created a second pipeline for sophos XG - the messages would not filter to the second pipeline so I had to have all the steps in one pipeline for the messages to be processed. More importantly though can you explain how you distinguish between multiple Sophos XG firewalls as when I undertake the steps you did - the "message.source" value as "device="SFW". This is the same on all Sophos XG firewalls we have. Any ideas how to change this so the source name can be changed to identify the relevant Sophos firewall?