MaxMind编写器创建用于ELK堆栈中geoip的自定义数据库

Question

我正在尝试使用来自MaxMind的perl为内部网络创建一个自定义数据库，到目前为止，该数据库运行良好。然后，我尝试使用logstash中的geoip过滤器插件来确定这些IP范围。我可以使用利用MaxMind阅读器的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" } }

但是没有错误或警告，只有简单的json，然后被推送到elasticsearch。数据库是否以正确的方式创建？

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"
          }
        }
      }
      }
    }
    }
}