Fix crash of pipeline due to NPE when lookup in DB with custom fields (#225)

Permit the user to handle an unexpected error (through event tag) instead of crashing the pipeline, when the data lookup encounter a customised field in the DB.

Protect the handleEvent method from NPE generated in the MAxMind GeoIP reader library.
That NPE happens when fetching data that contains customised attributes, this PR protects from NPE and return a lookup error so that the Event can be tagged with failure.

Author: andsel

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 7.3.1
+  - Avoid to crash pipelines when lookup a database with customised fields. [#225](https://github.com/logstash-plugins/logstash-filter-geoip/pull/225)
+
 ## 7.3.0
   - Added support for MaxMind GeoIP2 Enterprise and Anonymous-IP databases ([#223](https://github.com/logstash-plugins/logstash-filter-geoip/pull/223))
   - Updated MaxMind dependencies.

VERSION

@@ -1 +1 @@
-7.3.0
+7.3.1

build.gradle

@@ -40,7 +40,7 @@ plugins {
 group "org.logstash.filters"
 version "${new File("VERSION").text.trim()}"
 
-String junitVersion = '5.9.2'
+String junitVersion = '5.11.2' // at least 5.11 is needed to use @FieldSource with @ParameterizedTest
 String maxmindGeoip2Version = '2.17.0'
 String maxmindDbVersion = '2.1.0'
 String log4jVersion = '2.17.1'

lib/logstash-filter-geoip_jars.rb

@@ -3,4 +3,4 @@
 require 'jar_dependencies'
 require_jar('com.maxmind.geoip2', 'geoip2', '2.17.0')
 require_jar('com.maxmind.db', 'maxmind-db', '2.1.0')
-require_jar('org.logstash.filters', 'logstash-filter-geoip', '7.3.0')
+require_jar('org.logstash.filters', 'logstash-filter-geoip', '7.3.1')

spec/filters/geoip_ecs_spec.rb

@@ -22,7 +22,7 @@
         let(:common_options) { {"source" => "message", "database" => CITYDB, "target" => target} }
 
         before(:each) do
-          allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(ecs_compatibility)
+          allow(plugin).to receive(:ecs_compatibility).and_return(ecs_compatibility)
           plugin.register
         end
 
@@ -170,7 +170,7 @@
 
       context "ECS disabled" do
         before do
-          allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(:disabled)
+          allow(plugin).to receive(:ecs_compatibility).and_return(:disabled)
           plugin.register
           plugin.filter(event)
         end
@@ -193,7 +193,7 @@
 
       context "ECS mode" do
         before do
-          allow_any_instance_of(described_class).to receive(:ecs_compatibility).and_return(:v1)
+          allow(plugin).to receive(:ecs_compatibility).and_return(:v1)
         end
 
         context "`target` is unset" do

spec/filters/geoip_online_spec.rb

@@ -63,7 +63,7 @@ def setup_filter(db_path)
         end
 
         before(:each) do
-          allow_any_instance_of(described_class).to receive(:load_database_manager?).and_return(true)
+          allow(plugin).to receive(:load_database_manager?).and_return(true)
           stub_const("LogStash::Filters::Geoip::DatabaseManager", double("DatabaseManager.Class", :instance => mock_manager))
         end
 

src/main/java/org/logstash/filters/geoip/GeoIPFilter.java

@@ -48,6 +48,15 @@
 import java.util.stream.Collectors;
 
 public class GeoIPFilter implements Closeable {
+
+  // This exception could raise during the processing of datapoint with custom fields, check out
+  // for more details https://github.com/logstash-plugins/logstash-filter-geoip/issues/226
+  static class GeoIp2InvalidCustomFieldException extends GeoIp2Exception {
+    public GeoIp2InvalidCustomFieldException(Throwable cause) {
+      super("invalid custom field", cause);
+    }
+  }
+
   private static final Logger logger = LogManager.getLogger();
   private final String sourceField;
   private final String targetField;
@@ -152,7 +161,7 @@ public boolean handleEvent(RubyEvent rubyEvent) {
       throw new IllegalArgumentException("Expected input field value to be String or List type");
     }
 
-    if (ip.trim().isEmpty()){
+    if (ip.trim().isEmpty()) {
       return false;
     }
 
@@ -224,7 +233,12 @@ private boolean applyGeoData(Map<Field, Object> geoData, Event event) {
   }
 
   private Map<Field,Object> retrieveCityGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    CityResponse response = databaseReader.city(ipAddress);
+    CityResponse response;
+    try {
+      response = databaseReader.city(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
     Country country = response.getCountry();
     City city = response.getCity();
     Location location = response.getLocation();
@@ -337,7 +351,12 @@ private Map<Field,Object> retrieveCityGeoData(InetAddress ipAddress) throws GeoI
   }
 
   private Map<Field,Object> retrieveCountryGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    CountryResponse response = databaseReader.country(ipAddress);
+    CountryResponse response;
+    try {
+      response = databaseReader.country(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
     Country country = response.getCountry();
     Continent continent = response.getContinent();
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
@@ -372,7 +391,12 @@ private Map<Field,Object> retrieveCountryGeoData(InetAddress ipAddress) throws G
   }
 
   private Map<Field, Object> retrieveIspGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    IspResponse response = databaseReader.isp(ipAddress);
+    IspResponse response;
+    try {
+      response = databaseReader.isp(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
 
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
     for (Field desiredField : this.desiredFields) {
@@ -411,7 +435,12 @@ private Map<Field, Object> retrieveIspGeoData(InetAddress ipAddress) throws GeoI
   }
 
   private Map<Field, Object> retrieveAsnGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    AsnResponse response = databaseReader.asn(ipAddress);
+    AsnResponse response;
+    try {
+      response = databaseReader.asn(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
     Network network = response.getNetwork();
 
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
@@ -444,7 +473,12 @@ private Map<Field, Object> retrieveAsnGeoData(InetAddress ipAddress) throws GeoI
   }
 
   private Map<Field, Object> retrieveDomainGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    DomainResponse response = databaseReader.domain(ipAddress);
+    DomainResponse response;
+    try {
+      response = databaseReader.domain(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
     for (Field desiredField : this.desiredFields) {
       switch (desiredField) {
@@ -459,7 +493,12 @@ private Map<Field, Object> retrieveDomainGeoData(InetAddress ipAddress) throws G
   }
 
   private Map<Field, Object> retrieveEnterpriseGeoData(InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    EnterpriseResponse response = databaseReader.enterprise(ipAddress);
+    EnterpriseResponse response;
+    try {
+      response = databaseReader.enterprise(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
 
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
     Country country = response.getCountry();
@@ -567,7 +606,12 @@ private Map<Field, Object> retrieveEnterpriseGeoData(InetAddress ipAddress) thro
   }
 
   private Map<Field, Object> retrieveAnonymousIpGeoData(final InetAddress ipAddress) throws GeoIp2Exception, IOException {
-    AnonymousIpResponse response = databaseReader.anonymousIp(ipAddress);
+    AnonymousIpResponse response;
+    try {
+      response = databaseReader.anonymousIp(ipAddress);
+    } catch (NullPointerException e) {
+      throw new GeoIp2InvalidCustomFieldException(e);
+    }
 
     Map<Field, Object> geoData = new EnumMap<>(Field.class);
     boolean isHostingProvider = response.isHostingProvider();

src/test/java/org/logstash/filters/geoip/GeoIPFilterTest.java

@@ -2,6 +2,7 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.logstash.Event;
 
@@ -15,6 +16,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.logstash.RubyUtil.RUBY;
 import static org.logstash.ext.JrubyEventExtLibrary.RubyEvent;
 
@@ -24,6 +26,12 @@ class GeoIPFilterTest {
     private static final String SOURCE_FIELD = "ip";
     private static final String TARGET_FIELD = "data";
 
+    // used as parameters of givenDatabaseWithCustomizedFieldWhenItsAccessedTheCustomizedIPShouldntThrowAnyErrorAndReportTheLookupAsFailure
+    @SuppressWarnings("unused")
+    static Path[] GEO_DATABASES = {MaxMindDatabases.GEOIP2_COUNTRY, MaxMindDatabases.GEOIP2_ANONYMOUS_IP,
+            MaxMindDatabases.GEOIP2_ENTERPRISE, MaxMindDatabases.GEOIP2_ISP,
+            MaxMindDatabases.GEOLITE2_ASN};
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     void handleEventWithGeoIp2CityDatabaseShouldProperlyCreateEvent(boolean ecsEnabled) {
@@ -265,6 +273,22 @@ void handleEventWithNoCustomFieldsShouldUseDatabasesDefaultFields() {
         }
     }
 
+    @ParameterizedTest
+    @FieldSource("GEO_DATABASES")
+    void givenDatabaseWithCustomizedFieldWhenItsAccessedTheCustomizedIPShouldntThrowAnyErrorAndReportTheLookupAsFailure(Path geoDatabase) {
+        try (final GeoIPFilter filter = createFilter(geoDatabase, true, Collections.emptyList())) {
+            final RubyEvent rubyEvent = createRubyEvent("216.160.83.60");
+            assertFalse(filter.handleEvent(rubyEvent), "Lookup of data point with invalid custom fields should report as failed");
+        }
+    }
+    @Test
+    void givenDomainDatabaseWithCustomizedFieldWhenItsAccessedTheCustomizedIPShouldntThrowAnyErrorAndReportTheLookupAsFailure() {
+        try (final GeoIPFilter filter = createFilter(MaxMindDatabases.GEOIP2_DOMAIN, true, Collections.emptyList())) {
+            final RubyEvent rubyEvent = createRubyEvent("216.160.83.60");
+            assertTrue(filter.handleEvent(rubyEvent), "Lookup of data point with invalid custom fields should report as failed");
+        }
+    }
+
     @Test
     void handleEventWithListSourceFieldShouldParseFirstIp() {
         try (final GeoIPFilter filter = createFilter(MaxMindDatabases.GEOIP2_COUNTRY, true, Collections.emptyList())) {