Fix handling of non-integer port values in community_id processor (#70148)

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/CommunityIdProcessor.java

@@ -170,17 +170,16 @@ public final class CommunityIdProcessor extends AbstractProcessor {
             case Tcp:
             case Udp:
             case Sctp:
-                flow.sourcePort = parseIntFromObjectOrString(d.getFieldValue(sourcePortField, Object.class, ignoreMissing), "source port");
-                if (flow.sourcePort == 0) {
-                    throw new IllegalArgumentException("invalid source port [0]");
+                Object sourcePortValue = d.getFieldValue(sourcePortField, Object.class, ignoreMissing);
+                flow.sourcePort = parseIntFromObjectOrString(sourcePortValue, "source port");
+                if (flow.sourcePort < 1 || flow.sourcePort > 65535) {
+                    throw new IllegalArgumentException("invalid source port [" + sourcePortValue + "]");
                 }
 
-                flow.destinationPort = parseIntFromObjectOrString(
-                    d.getFieldValue(destinationPortField, Object.class, ignoreMissing),
-                    "destination port"
-                );
-                if (flow.destinationPort == 0) {
-                    throw new IllegalArgumentException("invalid destination port [0]");
+                Object destinationPortValue = d.getFieldValue(destinationPortField, Object.class, ignoreMissing);
+                flow.destinationPort = parseIntFromObjectOrString(destinationPortValue, "destination port");
+                if (flow.destinationPort < 1 || flow.sourcePort > 65535) {
+                    throw new IllegalArgumentException("invalid destination port [" + destinationPortValue + "]");
                 }
                 break;
             case Icmp:
@@ -216,7 +215,7 @@ public final class CommunityIdProcessor extends AbstractProcessor {
         if (o == null) {
             return 0;
         } else if (o instanceof Number) {
-            return (int) o;
+            return ((Number) o).intValue();
         } else if (o instanceof String) {
             try {
                 return Integer.parseInt((String) o);

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/CommunityIdProcessorTests.java

@@ -106,7 +106,8 @@ public class CommunityIdProcessorTests extends ESTestCase {
         var destination = (Map<String, Object>) event.get("destination");
         destination.put("port", null);
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> testCommunityIdProcessor(event, null));
-        assertThat(e.getMessage(), containsString("invalid destination port [0]"));
+        // slightly modified from the beats test in that this one reports the actual invalid value rather than '0'
+        assertThat(e.getMessage(), containsString("invalid destination port [null]"));
     }
 
     public void testBeatsUnknownProtocol() throws Exception {
@@ -269,6 +270,38 @@ public class CommunityIdProcessorTests extends ESTestCase {
         testCommunityIdProcessor(event, "1:KF3iG9XD24nhlSy4r1TcYIr5mfE=");
     }
 
+    public void testLongsForNumericValues() throws Exception {
+        event = buildEvent();
+        @SuppressWarnings("unchecked")
+        var source2 = (Map<String, Object>) event.get("source");
+        source2.put("port", 34855L);
+        testCommunityIdProcessor(event, "1:LQU9qZlK+B5F3KDmev6m5PMibrg=");
+    }
+
+    public void testFloatsForNumericValues() throws Exception {
+        event = buildEvent();
+        @SuppressWarnings("unchecked")
+        var source2 = (Map<String, Object>) event.get("source");
+        source2.put("port", 34855.0);
+        testCommunityIdProcessor(event, "1:LQU9qZlK+B5F3KDmev6m5PMibrg=");
+    }
+
+    public void testInvalidPort() throws Exception {
+        event = buildEvent();
+        @SuppressWarnings("unchecked")
+        var source = (Map<String, Object>) event.get("source");
+        source.put("port", 0);
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> testCommunityIdProcessor(event, null));
+        assertThat(e.getMessage(), containsString("invalid source port [0]"));
+
+        event = buildEvent();
+        @SuppressWarnings("unchecked")
+        var source2 = (Map<String, Object>) event.get("source");
+        source2.put("port", 65536);
+        e = expectThrows(IllegalArgumentException.class, () -> testCommunityIdProcessor(event, null));
+        assertThat(e.getMessage(), containsString("invalid source port [65536]"));
+    }
+
     public void testIgnoreMissing() throws Exception {
         @SuppressWarnings("unchecked")
         var network = (Map<String, Object>) event.get("network");