How to match multi ip and port

hi,

I try to separate ip and port:

src="10.20.20.131:0 125.167.15.161:0 147.75.111.32:0 223.202.110.11:0 36.74.149.118:0 36.86.63.182:0 203.190.242.244:0 "

in this way (?%{IP}( %{IP})*) but the results are only one ip that appears 10.20.20.131

I want to separate between ip and port, which later I ip mutate split,

What should I do

regadrs,
hambali

What do you want the result to look like?

the result :

{
      ip : 10.20.20.131 125.167.15.161 147.75.111.32 223.202.110.11 36.74.149.118 203.190.242.244
      port : 0 0 0 0 0 0
}
    mutate { add_field => { "ip" => "%{src}" "port" => "%{src}" } }
    mutate { gsub => [ "ip", ":[0-9]+ ", " ", "port", "[0-9\.]+:", "" ] }
1 Like

i'm not succes sparate ip and port

log:

<164> USG6300 %%01SEC/4/ATCKDF(l): AttackType="ICMP unreachable attack", slot="0", receive interface="TEST GE1/0/0(TEST) ", proto="ICMP", src="10.20.20.131:0 125.167.15.161:0 147.75.111.32:0 223.202.110.11:0 36.74.149.118:0 36.86.63.182:0 203.190.242.244:0 ", dst="192.168.137.2:0 192.168.137.9:0 192.168.137.34:0 192.168.137.1:0 192.168.137.2:0 ", begin time="2019-01-28 17:06:33", end time="2019-01-28 17:07:02", total packets="92", max speed="0", User="", Action="discard".

this is configuration logstash.conf

input {
   udp {
    port => 514
    type => syslog 
   }
}
filter {
 if [type] == "syslog" {
    grok {
	match => { "message" => "%{HUAWEI_ATCKDEF}"}
	}
	   mutate { add_field => { "ip" => "%{attack.src}" "port" => "%{attack.src}" } }
    mutate { gsub => [ "attack.src_IP", ":[0-9]+ ", " ", "attack.src_port", "[0-9\.]+:", "" ] }
#	mutate {
#      split => {"attack.src" => " "}
#	  split => {"attack.dst" => " "}
#	  split => {"ids.src_ip" => " "}
#	  split => {"ids.dst_ip" => " "}
#   }
       geoip  {
		source => "ids.src_ip"
		
  }
     geoip  {
		source => "ids.dst_ip"
		
  }

	mutate{
	 remove_tag => ["_geoip_lookup_failure"]
}
      translate {
	  field => "syslog.program"
      destination => "program"
      override => "false"
      dictionary_path => "/etc/logstash/translate.yml"
        }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }

	if "_grokparsefailure" in [tags] {
		drop { }
	}
 }

this is grok configuration :

HUAWEI %{TEST3}%{NUMBER}%{TEST3} %{SYSLOGHOST:hostname} %{DATA:syslog.program}(?:[%{POSINT:syslog.pid}])?:

HUAWEI_ATCKDEF %{HUAWEI} AttackType="%{DATA:attack.type}", slot="%{DATA:attack.slot}", receive interface="%{DATA:receive_interface}", proto="%{DATA:attack.protocol}", src="%{TEST7}", dst="%{TEST7}", begin time="%{TEST8:time.begin}", end time="%{TEST8:time.end}", total packets="%{DATA:attack.total_packets}", max speed="%{NUMBER:attack.max_speed}", User="%{DATA:attack.user}", Action="%{DATA:action}".

TEST3 ([<>]+)
TEST7 ([(?%{TEST4}( %{TEST4})*)]+)
TEST8 %{TEST6} %{NOTSPACE}
TEST6 %{YEAR}-%{MONTHNUM}-%{MONTHDAY}

any idea my problem ?

The syslog input will parse some of the fields from the message and remove them. Add

output { stdout { codec => rubydebug } }

and see what is in the message field.

Thank you so much for your help,
now I can separate ip and port.

this is my configure logstash.conf

input {
   udp {
    port => 514
    type => syslog 
   }
}
filter {
 if [type] == "syslog" {
    grok {
	match => { "message" => "%{HUAWEI_ATCKDEF}"}
	}
     mutate { add_field => { "attack.src" => "%{src}" "attack.src_port" => "%{src}" }}
	 mutate { gsub => [ "attack.src", ":[0-9]+ ", " ", "attack.src_port", "[0-9\.]+:", "" ] }
	 mutate { add_field => { "attack.dst" => "%{dst}" "attack.dst_port" => "%{dst}" }}
	 mutate { gsub => [ "attack.dst", ":[0-9]+ ", " ", "attack.dst_port", "[0-9\.]+:", "" ] }
#	mutate {
#      split => {"attack.src" => " "}
#	  split => {"attack.dst" => " "}
#	  split => {"ids.src_ip" => " "}
#	  split => {"ids.dst_ip" => " "}
#   }
       geoip  {
		source => "ids.src_ip"
		
  }
     geoip  {
		source => "ids.dst_ip"
		
  }

	mutate{
	 remove_tag => ["_geoip_lookup_failure"]
}
      translate {
	  field => "syslog.program"
      destination => "program"
      override => "false"
      dictionary_path => "/etc/logstash/translate.yml"
        }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }

	if "_grokparsefailure" in [tags] {
		drop { }
	}
 }
 output {
  stdout { codec => rubydebug }
  	if "%%01SEC/4/ATCKDF(l)" in [message] {
		elasticsearch {
			hosts => ["localhost:9200"]
			index => "attack-%{+YYYY.MM.dd}"
			document_type => "huawei"
		}
	}
	    else {
		elasticsearch {
			hosts => ["localhost:9200"]
			index => "huawei-%{+YYYY.MM.dd}"
			document_type => "huawei"
		}
	}
} 

thank you very much :smile:

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.