MaxMind Writer创建自定义数据库，以与Geoip一起在Elk Stack中使用

Question

我正在尝试使用MaxMind的Perl Writer创建一个内部网络的自定义数据库，MaxMind到目前为止运作良好。然后，我尝试使用LogStash中的GeoIP滤波器插件将这些IP范围分配给这些IP范围。我可以验证创建的数据库使用使用MaxMind Reader的Perl脚本正确包含我的数据。

#!/usr/bin/env perl
use strict;
use warnings;
use feature qw( say );
use local::lib 'local';
use MaxMind::DB::Writer::Tree;
use Net::Works::Network;
my $filename = 'test.mmdb';
# Your top level data structure will always be a map (hash).  The MMDB format
# is strongly typed.  Describe your data types here.
# See https://metacpan.org/pod/MaxMind::DB::Writer::Tree#DATA-TYPES
my %types = (
    geoname_id => 'uint32',
    locale_code => 'utf8_string',
    continent_code => 'utf8_string',
    continent_name => 'utf8_string',
    country_iso_code => 'utf8_string',
    country_name => 'utf8_string',
    subdivision_1_iso_code => 'utf8_string',
    subdivision_1_name => 'utf8_string',
    city_name => 'utf8_string',
    metro_code => 'uint32',
    time_zone => 'utf8_string',
    postal_code => 'utf8_string',
    latitude => 'double',
    longitude => 'double',
);
my $tree = MaxMind::DB::Writer::Tree->new(
    database_type => 'GeoIP2-City',
    description => { en => 'IP-Ranges', de => q{IP-Bereiche}, },
    ip_version => 4,
    map_key_type_callback => sub { $types{ $_[0] } },
    record_size => 24,
    remove_reserved_networks => 0,
);
my %address_for_employee = (
    'xx.xx.xx.xx/24' => {
        geoname_id => 3164699,
        locale_code => 'de',
        continent_code => 'EU',
        continent_name => 'Europa',
        country_iso_code => 'IT',
        country_name => 'Italien',
        subdivision_1_iso_code => '09',
        city_name => 'Varese',
        time_zone => 'Europe/Rome',
        postal_code => '21100',
        latitude => 45.7908,
        longitude => 8.8515,
    },    
    'xx.xx.xx.xx/24' => {
        geoname_id => 3164699,
        locale_code => 'de',
        continent_code => 'EU',
        continent_name => 'Europa',
        country_iso_code => 'IT',
        country_name => 'Italien',
        subdivision_1_iso_code => '09',
        city_name => 'Varese',
        time_zone => 'Europe/Rome',
        postal_code => '21100',
        latitude => 45.7908,
        longitude => 8.8515,
    },   
);
for my $address ( keys %address_for_employee ) {
    # Create one network and insert it into our database
    my $network = Net::Works::Network->new_from_string( string => $address );
    $tree->insert_network( $network, $address_for_employee{$address} );
        say "Created Subnet from ___ $network ===> $address ___";
}
# Write the database to disk.
open my $fh, '>:raw', $filename;
$tree->write_tree( $fh );
close $fh;
say "$filename has now been created";

logstash滤镜配置：

filter {
geoip {
  source => "src_ip"
  target => "src_ip_geo"
  database => "test.mmdb"
  add_field => [ "[src_ip_geo][coordinates]", "%{[src_ip_geo][longitude]}" ]
  add_field => [ "[src_ip_geo][coordinates]", "%{[src_ip_geo][latitude]}"  ]
}
mutate {
  convert => [ "[geoip][coordinates]", "float"]
}
}

将其与旧的MaxMind数据库GeoliteCity进行比较。DAT我尝试根据我在数据库中定义的IP地址将一些字段添加到输出中。但是，日志被正确推向elasticsearch，但没有其他字段，但添加了标签，上面添加了： _geoip_lookup_failure

有一种方法可以以某种方式调试此过滤器，因为我的LogStash日志没有提供任何提示。我还使用

输出

output { stdout{ codec => "json" } }

，但是在那里没有错误或警告，只是将其推向Elasticsearch的普通JSON。数据库是否以正确的方式创建？

Answer 1

您需要参考官方的geolite2-city.mmdb结构（例如，使用读取器）

perl代码样本：

my %types = (
 city => 'map',
 geoname_id => 'uint32',
 names => 'map',
 en => 'utf8_string',
 iso_code => 'utf8_string',
 country => 'map',
 registered_country => 'map',
 location => 'map',
 latitude => 'double',
 accuracy_radius => 'uint32',
 time_zone => 'utf8_string',
 longitude => 'double',
 continent => 'map',
);
my $tree = MaxMind::DB::Writer::Tree->new(
ip_version            => 4,
record_size           => 28,
database_type         => 'GeoIP2-City',
languages             => ['en'],
description           => { en => 'GeoLite2 City database' },
map_key_type_callback => sub { $types{ $_[0] } },
);
$tree->insert_network(
 '183.136.190.0/24' => {
     city => {
                 geoname_id => 1808926,
                 names => {
                              'en' => 'test1',
                          },
             },
     country => {
                     geoname_id => 1814991,
                     iso_code => 'CN',
                     names => {
                                  'en' => 'test2',
                              },
                },
     registered_country => {
                               geoname_id => 1814991,
                               iso_code => 'CN',
                               names => {
                                            'en' => 'test3',
                                        },
                           },
#         subdivisions => (
#                           {
#                             geoname_id => 1784764,
#                             iso_code => 'ZJ',
#                             names => {
#                                          en => 'Zhejiang',
#                                      },
#                           },
#                         ),
     location => {
                     latitude => 30.2936,
                     accuracy_radius => 50,
                     time_zone => 'Asia/Shanghai',
                     longitude => 120.1614,
                 },
     continent => {
                      geoname_id => 6255147,
                      iso_code => 'AS',
                      names => {
                                   en => 'Asia',
                               },
                  },
      },
);

logstash过滤器配置：

    geoip {
      source => "remote_addr"
      database => "/temp/csv2mmdb/IPB.mmdb"
      fields => ["city_name", "country_name"]
      target => "geoip"
    }

输出结果：

"remote_addr" => "183.136.190.40",
"geoip" => {
       "city_name" => "test1",
       "country_name" => "test2"
},
"X-Forwarded-Proto" => "-",

Answer 2

您不需要使用MaxMind GeoIP格式。

如果您构建自己的MMDB。您只需创建一个Python脚本即可使用MaxMind MaxMindDB模块查找IP并输出JSON对象。您需要为GEO_POINT数据更新弹性索引模板。

为了使Ruby脚本变得容易，我使用Pyinstaller使单个Python可以拒绝。

logstash配置

    if [dest_ip] =~ /.+/ {
       ruby {
        code => 'require "open3"
                if(File.file?("/path/to/mmdb/file/data_ip4.mmdb"))
                dest_ip = event.get("dest_ip")
                cmd =  "ip_search /path/to/mmdb/file/data_ip4.mmdb #{dest_ip}" dst
                stdin, stdout, stderr = Open3.popen3(cmd)
                event.set("process_result1", stdout.read)
                end'
       }
    }
    if [process_result1] =~ /.+/ {
         json {
            source => "process_result1"
            skip_on_invalid_json => true
         }
    }

示例脚本输出

{
  "dst": {
    "country": {
      "iso_code": "US"
    }, 
    "domain": "akamaitechnologies.com", 
    "company": "Akamai Technologies, Inc.", 
    "asn": 32787, 
    "location": {
      "lat": 39.043800354003906, 
      "lon": -77.48789978027344
    }
  }
}

更新地理数据的索引模板

PUT _template/logstash/
{
    "index_patterns": "some_index*",
    "mappings": {
    "_default_": {
    "properties": {
      "dst":{
        "properties": {
          "location": {
            "type": "geo_point"
          }
        }
      }
      }
    }
    }
}