Logstash：运用 Logstash 对 Service API 数据进行分析

Posted 2021-10-22 中国社区官方博客

我记得在之前的文章 “Logstash：使用 ELK 堆栈进行 API 分析” 运用 Logstash 对一些指标的 API 进行分析。在今天的练习中，我将展示如何使用 Logstash 来对一些日志类的 Service API 进行分析。我们知道在很多的时候，我们可以很快速地运用一些脚本对一些 Service API 进行数据分析。这对于我们快速地分析一些数据进行分析是有帮助的。在数据导入中，我们可以使用 Logstash 所提供的丰富的 filter 来对数据进行清洗，丰富，转换等。

在今天的练习中，我将使用 shodan.io 网站为例：

 Shodan 是一个万物互联的搜索引擎。在上面，我们搜索 china 这个单词的时候，它会显示那些主机中有这个单词的显示。关于 Shodan，我这里就不做更多的介绍。Shodan 实际上也提供一种 Service API 的接口并供我们的客户端进行调用。我们可以在地址来申请一个开发者账号，并得到开发者秘钥。

在下面，我使用一个 Python 的应用程序来获取查询的结果：

shodan_scanner.py

import os
from shodan import Shodan
import time
import requests
import pprint
import sys
import logging

logging.basicConfig(level=logging.INFO,
                    filename='shodan.log',
                    format='{%(message)s}')

# api_key = os.environ[SHODAN_API_KEY]
api_key = "YOUR DEVELOPER KEY"
api = Shodan(api_key)

search = sys.argv[1]

# Search Shodan
results = api.search(search)

# print(f"Total count: {len(results['matches'])}")

# Show the results
for result in results['matches']:
    ip_address = result['ip_str']
    domains = result['domains']
    logging.info('"ip_addr":"{}","domains":"{}"'.format(ip_address, domains))
    time.sleep(1)

为了运行上面的应用，我们必须安装 shodan：

pip3 install shodan

我们可以通过如下的方式来运行：

python3 shodan_scanner.py "china" > shodan.log

在上面，我们搜索 china 这个单词，并把搜索的结果保存到 shodan.log  里。等我们运行完后，我们可以在 shodan.log 文件里查看到如下的这样的日志：

shodan.log

{"ip_addr":"46.171.98.231","domains":"['tpnet.pl']"}
{"ip_addr":"50.212.192.99","domains":"[]"}
{"ip_addr":"189.68.169.82","domains":"['telesp.net.br']"}
{"ip_addr":"157.25.97.179","domains":"[]"}
{"ip_addr":"116.106.51.100","domains":"['viettel.vn']"}
{"ip_addr":"147.161.68.134","domains":"['adamo.es']"}
{"ip_addr":"37.252.122.48","domains":"['tilaa.com']"}
{"ip_addr":"134.73.16.32","domains":"['smbccojp14.com']"}
{"ip_addr":"177.182.214.4","domains":"['virtua.com.br']"}
{"ip_addr":"115.72.1.76","domains":"['viettel.vn']"}
{"ip_addr":"118.70.44.0","domains":"[]"}
{"ip_addr":"113.165.201.60","domains":"['vnpt.vn']"}
{"ip_addr":"45.201.208.251","domains":"[]"}
{"ip_addr":"42.119.67.22","domains":"[]"}
{"ip_addr":"14.166.150.83","domains":"['vnpt.vn']"}
{"ip_addr":"181.46.231.207","domains":"['telecentro-reversos.com.ar']"}
{"ip_addr":"109.50.128.189","domains":"['netcabo.pt']"}
{"ip_addr":"81.193.147.140","domains":"['telepac.pt']"}
{"ip_addr":"41.38.136.142","domains":"['tedata.net']"}
{"ip_addr":"85.187.243.54","domains":"['villagenet-bg.com']"}
{"ip_addr":"103.217.236.75","domains":"[]"}
{"ip_addr":"115.79.68.162","domains":"['viettel.vn']"}

从上面，我们可以看到，它有一个叫做 ip_addr 这样的 IP 地址。我们可以使用 Logstash 所提供的 geoip 过滤器来丰富这个数据。

logstash.conf

input {
  file {
    path => [ "/Users/liuxg/python/shodan/shodan.log" ]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec   => "json"
  }
}

filter {
    geoip {
        source => "ip_addr"
        target => "geo"
    }

    if [geo][latitude] and [geo][logitude] {
        mutate {
            add_field => {
                "location" => ["%{[geo][latitude]},%{[geo][logitude]}"]
            }
        }
        mutate {
            convert => {
                "location" => "float"
            }
        }
    }
}


output { 
  stdout {
    codec => rubydebug
  }  

  elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "shodan"
  }
}

在上面，我们通过 file input 把数据导入。在使用时，你需要使用自己的文件路径来代替上面的 path。在上面，我使用了 geoip 这个过滤器来对数据进行丰富。

在运行上面的 Logstash pipeline 之前，我们在 Kibana 中打入如下的命令：

PUT shodan
{
  "mappings": {
    "properties": {
      "geo": {
        "properties": {
          "location": {
            "type": "geo_point"
          }
        }
      }
    }
  }
}

上面的命令定义了 geo.location 的数据类型。它是一个 geo_point 的数据类型。

我们可以使用如下的命令来运行 Logstash：

sudo ./bin/logstash -f logstash.conf

我们可以在 Logstash 运行的 terminal 中看到如下的输出：

从上面，我们可以看出来，通过 geoip 的使用，我们得到了更多的关于位置的字段。

我们需要为 shaodan 索引创建一个索引模式。我们在 Discover 中可以对这些数据进行查找：

 

由于它也有位置字段，我们也可以使用 Maps 应用来对文档的位置来进行显示：