Fix community_id processor so that ports greater than 65535 aren't …

…valid (#25409)

* initial commit

* update function

* update function per comments

CHANGELOG.next.asciidoc

@@ -249,6 +249,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix discovery of short-living and failing pods in Kubernetes autodiscover {issue}22718[22718] {pull}24742[24742]
 - Fix panic when overwriting metadata {pull}24741[24741]
 - Fix role_arn to work with access keys for AWS. {pull}25446[25446]
+- Fix `community_id` processor so that ports greater than 65535 aren't valid. {pull}25409[25409]
 
 *Auditbeat*
 

libbeat/processors/communityid/communityid.go

@@ -154,20 +154,22 @@ func (p *processor) buildFlow(event *beat.Event) *flowhash.Flow {
 		if err != nil {
 			return nil
 		}
-		flow.SourcePort, ok = tryToUint16(v)
-		if !ok || flow.SourcePort == 0 {
+		sp, ok := tryToUint(v)
+		if !ok || sp < 1 || sp > 65535 {
 			return nil
 		}
+		flow.SourcePort = uint16(sp)
 
 		// destination port
 		v, err = event.GetValue(p.Fields.DestinationPort)
 		if err != nil {
 			return nil
 		}
-		flow.DestinationPort, ok = tryToUint16(v)
-		if !ok || flow.DestinationPort == 0 {
+		dp, ok := tryToUint(v)
+		if !ok || dp < 1 || dp > 65535 {
 			return nil
 		}
+		flow.DestinationPort = uint16(dp)
 	case icmpProtocol, icmpIPv6Protocol:
 		// Return a flow even if the ICMP type/code is unavailable.
 		if t, c, ok := getICMPTypeCode(event, p.Fields.ICMPType, p.Fields.ICMPCode); ok {
@@ -211,43 +213,43 @@ func tryToIP(from interface{}) (net.IP, bool) {
 	}
 }
 
-// tryToUint16 tries to coerce the given interface to an uint16. On success it
+// tryToUint tries to coerce the given interface to an uint16. On success it
 // returns the int value and true.
-func tryToUint16(from interface{}) (uint16, bool) {
+func tryToUint(from interface{}) (uint, bool) {
 	switch v := from.(type) {
 	case int:
-		return uint16(v), true
+		return uint(v), true
 	case int8:
-		return uint16(v), true
+		return uint(v), true
 	case int16:
-		return uint16(v), true
+		return uint(v), true
 	case int32:
-		return uint16(v), true
+		return uint(v), true
 	case int64:
-		return uint16(v), true
+		return uint(v), true
 	case uint:
-		return uint16(v), true
+		return uint(v), true
 	case uint8:
-		return uint16(v), true
+		return uint(v), true
 	case uint16:
-		return v, true
+		return uint(v), true
 	case uint32:
-		return uint16(v), true
+		return uint(v), true
 	case uint64:
-		return uint16(v), true
+		return uint(v), true
 	case string:
-		num, err := strconv.ParseUint(v, 0, 16)
+		num, err := strconv.ParseUint(v, 0, 64)
 		if err != nil {
 			return 0, false
 		}
-		return uint16(num), true
+		return uint(num), true
 	default:
 		return 0, false
 	}
 }
 
 func tryToUint8(from interface{}) (uint8, bool) {
-	to, ok := tryToUint16(from)
+	to, ok := tryToUint(from)
 	return uint8(to), ok
 }
 

libbeat/processors/communityid/communityid_test.go

@@ -72,6 +72,12 @@ func TestRun(t *testing.T) {
 		testProcessor(t, 0, e, nil)
 	})
 
+	t.Run("invalid source port1", func(t *testing.T) {
+		e := evt()
+		e.Put("source.port", 123456)
+		testProcessor(t, 0, e, nil)
+	})
+
 	t.Run("invalid destination IP", func(t *testing.T) {
 		e := evt()
 		e.Put("destination.ip", "308.111.1.2.3")
@@ -80,7 +86,13 @@ func TestRun(t *testing.T) {
 
 	t.Run("invalid destination port", func(t *testing.T) {
 		e := evt()
-		e.Put("source.port", nil)
+		e.Put("destination.port", 0)
+		testProcessor(t, 0, e, nil)
+	})
+
+	t.Run("invalid destination port1", func(t *testing.T) {
+		e := evt()
+		e.Put("destination.port", 123456)
 		testProcessor(t, 0, e, nil)
 	})
 

x-pack/filebeat/module/zeek/connection/test/connection-json.log

@@ -1,4 +1,4 @@
 {"ts":1547188415.857497,"uid":"CAcJw21BbVedgFnYH3","id.orig_h":"192.168.86.167","id.orig_p":38339,"id.resp_h":"192.168.86.1","id.resp_p":53,"proto":"udp","service":"dns","duration":0.076967,"orig_bytes":75,"resp_bytes":178,"conn_state":"SF","local_orig":true,"local_resp":true,"missed_bytes":0,"history":"Dd","orig_pkts":1,"orig_ip_bytes":103,"resp_pkts":1,"resp_ip_bytes":206,"tunnel_parents":[]}
 {"ts":1547188416.857497,"uid":"CAcJw21BbVedgFnYH4","id.orig_h":"192.168.86.167","id.orig_p":38340,"id.resp_h":"8.8.8.8","id.resp_p":53,"proto":"udp","service":"dns","duration":0.076967,"orig_bytes":75,"resp_bytes":178,"conn_state":"SF","local_orig":true,"local_resp":false,"missed_bytes":0,"history":"Dd","orig_pkts":1,"orig_ip_bytes":103,"resp_pkts":1,"resp_ip_bytes":206,"tunnel_parents":[]}
-{"ts":1547188417.857497,"uid":"CAcJw21BbVedgFnYH5","id.orig_h":"4.4.2.2","id.orig_p":383341,"id.resp_h":"8.8.8.8","id.resp_p":53,"proto":"udp","service":"dns","duration":0.076967,"orig_bytes":75,"resp_bytes":178,"conn_state":"SF","local_orig":false,"local_resp":false,"missed_bytes":0,"history":"Dd","orig_pkts":1,"orig_ip_bytes":103,"resp_pkts":1,"resp_ip_bytes":206,"tunnel_parents":[]}
+{"ts":1547188417.857497,"uid":"CAcJw21BbVedgFnYH5","id.orig_h":"4.4.2.2","id.orig_p":38341,"id.resp_h":"8.8.8.8","id.resp_p":53,"proto":"udp","service":"dns","duration":0.076967,"orig_bytes":75,"resp_bytes":178,"conn_state":"SF","local_orig":false,"local_resp":false,"missed_bytes":0,"history":"Dd","orig_pkts":1,"orig_ip_bytes":103,"resp_pkts":1,"resp_ip_bytes":206,"tunnel_parents":[]}
 {"ts":1551399000.57855,"uid":"Cc6NJ3GRlfjE44I3h","id.orig_h":"192.0.2.205","id.orig_p":3,"id.resp_h":"198.51.100.249","id.resp_p":3,"proto":"icmp","conn_state":"OTH","local_orig":false,"local_resp":false,"missed_bytes":0,"orig_pkts":1,"orig_ip_bytes":107,"resp_pkts":0,"resp_ip_bytes":0,"tunnel_parents":[]}

x-pack/filebeat/module/zeek/connection/test/connection-json.log-expected.json

@@ -140,7 +140,7 @@
         "input.type": "log",
         "log.offset": 792,
         "network.bytes": 309,
-        "network.community_id": "1:9xAq+MIBct9Is73ErTrU/RZ+Nq0=",
+        "network.community_id": "1:7pTO7SRt6R5Ms7DZet2wPuZnXSs=",
         "network.direction": "external",
         "network.packets": 2,
         "network.protocol": "dns",
@@ -161,7 +161,7 @@
         "source.geo.location.lon": -97.822,
         "source.ip": "4.4.2.2",
         "source.packets": 1,
-        "source.port": 383341,
+        "source.port": 38341,
         "tags": [
             "zeek.connection"
         ],
@@ -192,7 +192,7 @@
         ],
         "fileset.name": "connection",
         "input.type": "log",
-        "log.offset": 1181,
+        "log.offset": 1180,
         "network.bytes": 107,
         "network.community_id": "1:gzTID87+KHoT4RFDSqb5aInTPeg=",
         "network.direction": "external",