updated url enrichment to work more consistently

config/enrichments/06_url.conf

@@ -1,68 +1,104 @@
 # Copyright [2021] [Cargill, Incorporated.] 
 # SPDX-License-Identifier: Apache-2.0
-# 1. Copy all url.original to url.full if full does not exsist
-# 2. Removes all url.x fields, except for [url][full] and [url][original]
-# 2. Copies [url][full] to [url][original] if does not exsit
-# 4. Lowercase [url][full]
-# 5. If [url][full] then parses [url][full] into subfields
-# 6. if [url][domain] processes [url][domain] with Top Level Domain (TLD) filter
-# 7. Rename tld fields to ECS fields
-# 8. Remove the remaining TLD fields not ECS
+# 1. Check to see if url.original or url.full exsist and 
+# 2. Ensures url.original and url.full both exsist
+# 3. Lowercase [url][full]
+# 4. Parses [url][full] into subfields tmp fields
+# 5. Rename tmp fields to ECS fields if fields do not exsist
+# 6. Convert "[url][port]" to "integer" and remove tmp field
+# 7. Run Top Level Domain (TLD) against [url][domain] filter
+# 8. Rename tld fields to ECS fields if ECS field does not exsist
+# 9. Remove the remaining TLD fields
 filter {
   if "disable_url_enrichment" in [tags] or "disable_code_reduction" in [tags] or "disable_enrichments" in [tags] {
     mutate {
       remove_tag => ["disable_url_enrichment"]
     }
   } else {
-    ### URL enrichment
-    if [url][original] and [url][original] != "" and ![url][full] {
-      mutate {
-        copy => { "[url][original]" => "[url][full]" }
-      }
-    }
-    if [url][full] and [url][full] != "" {
-      mutate {
-        remove_field => [ "[url][domain]", "[url][extension]", "[url][fragment]", "[url][full][text]", "[url][original][text]", "[url][password]", "[url][path]", "[url][port]", "[url][query]", "[url][registered_domain]", "[url][scheme]", "[url][top_level_domain]", "[url][username]" ]
-      }
-      ### mutate order of operation has lowercase before copy
-      if ![url][original] {
+      if [url][full] or [url][original] {
+        if [url][full] and (![url][original] or [url][original] == "") {
+          mutate {
+            copy => { "[url][full]" => "[url][original]" }
+          }
+        }
+        if [url][original] and (![url][full] or [url][full] == "") {
+          mutate {
+            copy => { "[url][original]" => "[url][full]" }
+          }
+        }
+        # parse out url into temp sub fields
         mutate {
-          copy => { "[url][full]" => "[url][original]" }
+          lowercase => [ "[url][full]" ]
         }
-      }
-      mutate {
-        lowercase => [ "[url][full]" ]
-      }
-      grok {
-        match => {"[url][full]" => "^((?<[url][scheme]>.*?)://)?((?<[user][name]>.*?):(?<[user][password]>.*?)@)?(?<[url][domain]>\w+((\.\w+){1,})|\d+\.\d+\.\d+\.\d+)(:(?<[url][port]>\d+))?(/|$)((?<[url][path]>.*?))?(\?(?<[url][query]>.*?))?(\#(?<[url][fragment]>.*?))?$" }
-        timeout_millis => 500
-        tag_on_failure => "_groktimeout_url_en_1"
-      }
-      if [url][path] and [url][path] != "" and [url][path] =~ "^.*?\..*?$" {
         grok {
-          match => {"[url][path]" => "^(.*)?\.(?<[url][extension]>.*?)$"}
+          match => {"[tmp][url][full]" => "^((?<[tmp][url][scheme]>.*?)://)?((?<[tmp][user][name]>.*?):(?<[tmp][user][password]>.*?)@)?(?<[tmp][url][domain]>\w+((\.\w+){1,})|\d+\.\d+\.\d+\.\d+)(:(?<[tmp][url][port]>\d+))?(/|$)((?<[tmp][url][path]>.*?))?(\?(?<[tmp][url][query]>.*?))?(\#(?<[tmp][url][fragment]>.*?))?$" }
           timeout_millis => 500
-          tag_on_failure => "_groktimeout_url_en_2"
+          tag_on_failure => "_groktimeout_url_en_1"
+        }
+        if [tmp][url][scheme] and (![url][scheme] or [url][scheme] == "" ) {
+          mutate {
+            rename => { "[tmp][url][scheme]" => "[url][scheme]" }
+          }
+        }
+        if [tmp][user][name] and (![user][name] or [user][name] == "" ) {
+          mutate {
+            rename => { "[tmp][user][name]" => "[url][username]" }
+          }
+        }
+        if [tmp][user][password] and (![user][password] or [user][password] == "" ) {
+          mutate {
+            rename => { "[tmp][user][password]" => "[url][password]" }
+          }  
+        }
+        if [tmp][url][domain] and (![url][domain] or [url][domain] == "" ) {
+          mutate {
+            rename => { "[tmp][url][domain]" => "[url][domain]" }
+          }
+        }
+        if [tmp][url][port] and (![url][port] or [url][port] == "" ) {
+          mutate {
+            rename => { "[tmp][url][port]" => "[url][port]" }
+          }
         }
+        if [tmp][url][path] and (![url][path] or [url][path] == "" ) {
+          mutate {
+            rename => { "[tmp][url][path]" => "[url][path]" }
+          }
+        }
+        if [tmp][url][query] and (![url][query] or [url][query] == "" ) {
+          mutate {
+            rename => { "[tmp][url][query]" => "[url][query]" }
+          }
+        }
+        if [tmp][url][fragment] and (![url][fragment] or [url][fragment] == "" ) {
+          mutate {
+            rename => { "[tmp][url][fragment]" => "[url][fragment]" }
+          }  
+        }
+      mutate {
+        convert => { "[url][port]" => "integer" }
+        remove_field => [ "tmp" ]
       }
-      # add tld enrichment
-      # if [url][domain] exists parse it for top level domain
-      # tld filter produces object like
-      # {trd=clients4, domain=google.com, subdomain=clients4.google.com, sld=google, tld=com}
-      # mutate the tld and domain fields to [url][top_level_domain] and [url][registered_domain] respectively
-      ### Disabling tld enrichment for now as the logstash tld filter have issue
-      # https://github.com/logstash-plugins/logstash-filter-tld/issues/8
       tld {
         source => "[url][domain]"
         target => "tmp"
       }
-      mutate {
-        convert => { "[url][port]" => "integer" }
-        rename => {      
-          "[tmp][domain]" => "[url][registered_domain]"
-          "[tmp][sld]" => "[url][subdomain]"
-          "[tmp][tld]" => "[url][top_level_domain]"
+      if [tmp][domain] and (![url][registered_domain] or [url][registered_domain] == "" ) {
+        mutate {
+          rename => { "[tmp][domain]" => "[url][registered_domain]" }
+        }
+      }
+      if [tmp][sld] and (![url][subdomain] or [url][subdomain] == "" ) {
+        mutate {
+          rename => {  "[tmp][sld]" => "[url][subdomain]"  }
         }
+      }
+      if [tmp][tld] and (![url][top_level_domain] or [url][top_level_domain] == "" ) {
+          mutate {
+            rename => { "[tmp][tld]" => "[url][top_level_domain]" }
+          }
+      }
+      mutate {
         remove_field => [ "tmp" ]
       }
     }

config/processors/api_security_skyhigh.scp.conf

@@ -1,5 +1,6 @@
 # Copyright [2021] [Cargill, Incorporated.] 
 # SPDX-License-Identifier: Apache-2.0
+# exclude from url enrichment
 input {
   pipeline {
     address => VAR_PIPELINE_NAME
@@ -17,110 +18,138 @@ filter {
     drop {}
   }
   if [message] !~ "^test message for.*?$" {
-	  csv {
-		source => "message"
-		columns => ["num","usr","[source][nat][ip]","[http][request][method]","[destination][bytes]","[source][bytes]","[url][domain]","[url][path]","[event][action]","[rule][name]","request_timestamp_epoch","[event][time]","[url][scheme]","[rule][category]","[http][request][body][content]","[service][name]","[event][severity_name]", "[rule][uuid]", "[http][response][status_code]", "[source][ip]", "[rule][description]", "[rule][ruleset]", "[user_agent][name]", "[user_agent][version]", "[user_agent][original]", "[process][name]", "[destination][ip]", "[destination][port]", "[observer][geo][country_iso_code]", "[http][request][referrer]", "[ssl_scanned]", "[av_scanned_up]", "[av_scanned_down]", "[rbi]", "[dlp]", "[source][address]", "[file][name]", "[observer][egress][ip]", "[observer][ingress][ip]", "[source][nat][port]", "[event][risk_score]", "discarded_host", "[tls][client][x509][version_number]", "[tls][version]", "[tmp][filed]", "[service][id]", "[tmp][duration_1]", "[tmp][duration_2]",  "[tmp][duration_3]",  "[tmp][duration_4]" ]
-		convert => {
-		  "[destination][bytes]" => "integer"
-		  "[source][bytes]" => "integer"
-		}
-		skip_empty_columns => true
-		skip_empty_rows => true
-	  }
-	  # row empty
-	  if "_csvskippedemptyfield" in [tags] {
-		  drop {}
-	  }
+      csv {
+        source => "message"
+        columns => ["num","usr","[source][nat][ip]","[http][request][method]","[destination][bytes]","[source][bytes]","[url][domain]","[url][path]","[event][action]","[rule][name]","request_timestamp_epoch","[event][time]","[url][scheme]","[rule][category]","[http][request][body][content]","[service][name]","[event][severity_name]", "[rule][uuid]", "[http][response][status_code]", "[source][ip]", "[rule][description]", "[rule][ruleset]", "[user_agent][name]", "[user_agent][version]", "[user_agent][original]", "[process][name]", "[destination][ip]", "[destination][port]", "[observer][geo][country_iso_code]", "[http][request][referrer]", "[ssl_scanned]", "[av_scanned_up]", "[av_scanned_down]", "[rbi]", "[dlp]", "[source][address]", "[file][name]", "[observer][egress][ip]", "[observer][ingress][ip]", "[source][nat][port]", "[event][risk_score]", "discarded_host", "[tls][client][x509][version_number]", "[tls][version]", "[tmp][filed]", "[service][id]", "[tmp][duration_1]", "[tmp][duration_2]",  "[tmp][duration_3]",  "[tmp][duration_4]" ]
+        convert => {
+          "[destination][bytes]" => "integer"
+          "[source][bytes]" => "integer"
+        }
+        skip_empty_columns => true
+        skip_empty_rows => true
+      }
+      # row empty
+      if "_csvskippedemptyfield" in [tags] {
+          drop {}
+      }
 
-	  mutate {
-		split => { "[rule][category]" => ", " }
-	  }
-	  # Message is csv, this creates [event][original] to include fields names
-	  mutate {
-		add_field => { "[event][original]" => "request_timestamp_epoch: %{[event][time]}, num: %{num}, usr: %{usr}, source.nat.ip: %{[source][ip]}, http.request.method: %{[http][request][method]}, destination.bytes: %{[destination][bytes]}, source.bytes: %{[source][bytes]}, url.domain: %{[url][domain]}, url.path: %{[url][path]}, event.action: %{[event][action]}, rule.name: %{[rule][name]}, request_timestamp_epoch: %{request_timestamp_epoch}, url.scheme: %{[url][scheme]}, rule.category: %{[rule][category]}, http.request.body.content: %{[http][request][body][content]}, service.name: %{[network][application]}, event.severity_name: %{[event][severity_name]}, last_rule: %{[rule][uuid]}, http_status_code: %{[http][response][status_code]}, client_ip: %{[source][nat][ip]}, location: %{[rule][description]}, block_reason: %{[rule][ruleset]}, user_agent_comment: %{[user_agent][version]}, user_agent_product: %{[user_agent][name]}, user_agent_version: %{[user_agent][original]}, process_name: %{[process][name]}, destination_ip: %{[destination][ip]}, destination_port: %{[destination][port]}, event.risk_score: %{[event][risk_score]}, discarded_host: %{discarded_host}, tls.client.x509.version_number: %{[tls][client][x509][version_number]}, tls.version: %{[tls][version]}" }
-	  }
-	  mutate {
-		gsub => [ "[event][original]", "%\{.*?}(,)? ", "" ]
-		gsub => [ "[event][original]", "%\{.*?}", "" ]
-	  }
-	  
-	  if [usr] and [usr] =~ ".*?\\.*?" {
-		grok {
-		  match => { "usr" => "(?<[user][domain]>.*?)\\(?<[user][name]>.*?)$" }
-		  timeout_millis => 500
-		}
-	  } else {
-		mutate {
-		  rename => { "[usr]" => "[user][name]" }
-		}
-	  }
-	  if [source][ip] and [source][ip] =~ "source_ip" {
-		mutate {
-		  remove_field => ["[source][ip]"]
-		}
-	  }
-	  # Create [url][full] 
-	  mutate {
-		add_field => { "[url][full]" => "%{[[url][scheme]]}://%{[[url][domain]]}%{[[url][path]]}" }
-	  }
-	  date {
-		match => [ "request_timestamp_epoch", "UNIX" ]
-		timezone => "GMT"
-		locale => "en"
-		target => "[event][created]"
-	  }
-	  mutate {
-		add_field => { "[cloud][provider]" => "mcafee" }
-		#add_field => { "[event][module]" => "mcafee" }
-		#add_field => { "[event][dataset]" => "mcafee.mcp" }
-		add_field => { "[log][source][hostname]" => "api_mcp"}
-	  }
-	  mutate {
-		lowercase => [ "[event][action]" ]
-	  }
-	  if [event][action] =~ "observed" {
-		mutate {
-		  replace => { "[event][action]" => "allowed" }
-		}
-	  }
-	  if [event][action] =~ "denied" {
-		mutate {
-		  replace => { "[event][action]" => "denied" }
-		}
-	  }
-	  mutate {
-		add_field => { "[service][state]" => "ssl_scanned: %{ssl_scanned}" }
-	  }
-	  if [av_scanned_up] {
-		mutate {
-		  add_field => { "[service][state]" => "av_scanned_up: %{av_scanned_up}" }
-		}
-	  }
-	  if [av_scanned_down] {
-		mutate {
-		  add_field => { "[service][state]" => "av_scanned_down: %{av_scanned_down}" }
-		}
-	  }
-	  if [rbi] {
-		mutate {
-		  add_field => { "[service][state]" => "rbi: %{rbi}" }
-		}
-	  }
-	  if [dlp] {
-		mutate {
-		  add_field => { "[service][state]" => "dlp: %{dlp}" }
-		}
-	  }
-	  
+      mutate {
+        split => { "[rule][category]" => ", " }
+      }
+      # Message is csv, this creates [event][original] to include fields names
+      mutate {
+        add_field => { "[event][original]" => "request_timestamp_epoch: %{[event][time]}, num: %{num}, usr: %{usr}, source.nat.ip: %{[source][ip]}, http.request.method: %{[http][request][method]}, destination.bytes: %{[destination][bytes]}, source.bytes: %{[source][bytes]}, url.domain: %{[url][domain]}, url.path: %{[url][path]}, event.action: %{[event][action]}, rule.name: %{[rule][name]}, request_timestamp_epoch: %{request_timestamp_epoch}, url.scheme: %{[url][scheme]}, rule.category: %{[rule][category]}, http.request.body.content: %{[http][request][body][content]}, service.name: %{[network][application]}, event.severity_name: %{[event][severity_name]}, last_rule: %{[rule][uuid]}, http_status_code: %{[http][response][status_code]}, client_ip: %{[source][nat][ip]}, location: %{[rule][description]}, block_reason: %{[rule][ruleset]}, user_agent_comment: %{[user_agent][version]}, user_agent_product: %{[user_agent][name]}, user_agent_version: %{[user_agent][original]}, process_name: %{[process][name]}, destination_ip: %{[destination][ip]}, destination_port: %{[destination][port]}, event.risk_score: %{[event][risk_score]}, discarded_host: %{discarded_host}, tls.client.x509.version_number: %{[tls][client][x509][version_number]}, tls.version: %{[tls][version]}" }
+      }
+      mutate {
+        gsub => [ "[event][original]", "%\{.*?}(,)? ", "" ]
+        gsub => [ "[event][original]", "%\{.*?}", "" ]
+      }
+      
+      if [usr] and [usr] =~ ".*?\\.*?" {
+        grok {
+          match => { "usr" => "(?<[user][domain]>.*?)\\(?<[user][name]>.*?)$" }
+          timeout_millis => 500
+        }
+      } else {
+        mutate {
+          rename => { "[usr]" => "[user][name]" }
+        }
+      }
+      if [source][ip] and [source][ip] =~ "source_ip" {
+        mutate {
+          remove_field => ["[source][ip]"]
+        }
+      }
+      # Create [url][full] 
+      mutate {
+        add_field => { "[url][full]" => "%{[[url][scheme]]}://%{[[url][domain]]}%{[[url][path]]}" }
+      }
+      date {
+        match => [ "request_timestamp_epoch", "UNIX" ]
+        timezone => "GMT"
+        locale => "en"
+        target => "[event][created]"
+      }
+      mutate {
+        add_field => { "[cloud][provider]" => "mcafee" }
+        #add_field => { "[event][module]" => "mcafee" }
+        #add_field => { "[event][dataset]" => "mcafee.mcp" }
+        add_field => { "[log][source][hostname]" => "api_mcp"}
+      }
+      mutate {
+        lowercase => [ "[event][action]" ]
+      }
+      if [event][action] =~ "observed" {
+        mutate {
+          replace => { "[event][action]" => "allowed" }
+        }
+      }
+      if [event][action] =~ "denied" {
+        mutate {
+          replace => { "[event][action]" => "denied" }
+        }
+      }
+      mutate {
+        add_field => { "[service][state]" => "ssl_scanned: %{ssl_scanned}" }
+      }
+      if [av_scanned_up] {
+        mutate {
+          add_field => { "[service][state]" => "av_scanned_up: %{av_scanned_up}" }
+        }
+      }
+      if [av_scanned_down] {
+        mutate {
+          add_field => { "[service][state]" => "av_scanned_down: %{av_scanned_down}" }
+        }
+      }
+      if [rbi] {
+        mutate {
+          add_field => { "[service][state]" => "rbi: %{rbi}" }
+        }
+      }
+      if [dlp] {
+        mutate {
+          add_field => { "[service][state]" => "dlp: %{dlp}" }
+        }
+      }
+      
   }
   mutate {
     add_field => { "[event][code]" => "%{[tmp][duration_1]}, %{[tmp][duration_2]}, %{[tmp][duration_3]}, %{[tmp][duration_4]}" }
   }
+    json {
+      source => "tmp_j"
+      target => ""
+      tag_on_failure => "_jsonparsefailure_init"
+      remove_field => [ "tmp_j" ]
+    }  
   mutate {
     # host field is added by file input plugin
     remove_field => [ "tmp", "request_timestamp_epoch", "usr", "num", "[event][time]", "path", "ssl_scanned", "av_scanned_up", "av_scanned_down", "rbi", "dlp", "discarded_host", "[event][original]" ]
     split => { "[event][code]" => "," }
   }
+        tld {
+        source => "[url][domain]"
+        target => "tmp"
+      }
+      if [tmp][domain] and (![url][registered_domain] or [url][registered_domain] == "" ) {
+        mutate {
+          rename => { "[tmp][domain]" => "[url][registered_domain]" }
+        }
+      }
+      if [tmp][sld] and (![url][subdomain] or [url][subdomain] == "" ) {
+        mutate {
+          rename => {  "[tmp][sld]" => "[url][subdomain]"  }
+        }
+      }
+      if [tmp][tld] and (![url][top_level_domain] or [url][top_level_domain] == "" ) {
+          mutate {
+            rename => { "[tmp][tld]" => "[url][top_level_domain]" }
+          }
+      }
+      mutate {
+        remove_field => [ "tmp" ]
+      }
 }
 output {
   pipeline { send_to => [enrichments] }