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

2021年8月9日   |   by mebius

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

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

%title插图%num

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

在下面,我使tgcode用一个 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 pipeltgcodeine 之前,我们在 Kibana 中打入如下的命令:

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

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

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

sudo ./bin/logstash -f logstash.conf

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

%title插图%num

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

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

%title插图%num

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

%title插图%num

文章来源于互联网:Logstash:运用 Logstash 对 Service API 数据进行分析

相关推荐: Logstash:导入 RSS feed 数据

在实际的使用过程中,我们有时希望正对 RSS 的数据来进行导入并进行搜索。在现实的许多微服务中,有很多的数据是根据 RSS feed 的形式来提供的,比如我们常见的点评网站等。那么我们有没有办法把这些数据导入到 Elasticsearch 并进行搜索呢?答案是…

Tags: , ,