批量查询威胁情报

• 批量查询威胁情报
  • 1 编写目的
  • 2 实现目标
  • 3 脚本实例

1 编写目的

在提升蓝队进行对 IP 研判的效率工作时，通常有以下场景：

1. 需要对多个攻击 IP 进行查询研判；
2. 为了提高得到 IP 信息的准确程度，通常会到威胁情报共享平台查询关于IP的威胁情报。

2 实现目标

1. 将整理成文档的可疑 IP 地址列表进行批量查询；
2. 输出微步与奇安信TI情报库的信息对 IP 进行研判。

3 脚本实例

1. 在脚本的目录下创建iplist.txt文档，并写入可疑 IP ，每个 IP 为一行；
2. 执行脚本后会自动生成 ipinfo.xlsx 文档

import os
import sys
import time

from openpyxl import Workbook
import chardet
import requests


def check_code(f_path):
    """
    检查文件所属编码
    :param f_path:文件名称
    :return:文件所属编码
    """
    file_text = open(f_path, 'rb').readline()
    file_code = chardet.detect(file_text)['encoding']
    # 由于windows系统的编码有可能是Windows-1254,打印出来后还是乱码,所以不直接用adchar['encoding']编码
    if file_code == 'gbk' or file_code == 'GBK' or file_code == 'GB2312':
        file_code = 'GB2312'
    else:
        file_code = 'utf-8'
    return file_code


def existsfile(filename):
    if not os.path.exists(filename):  # 判断目标文件是否存在
        print('%s 不存在,请创建!' % filename)
        sys.exit(2)
    else:
        return filename


def save_result(x_result_parse: dict, f_path):
    # 将查询结果记录在csv中
    wb = Workbook()
    # 获取默认的工作表
    ws = wb.active
    ws.title = "ipinfo"
    i = 2
    ws["A1"].value = "序号"
    ws["B1"].value = "攻击IP"
    ws["C1"].value = "IP区域"
    ws["D1"].value = "微步情报"
    ws["E1"].value = "微步情报时间"
    ws["F1"].value = "奇安信情报"
    ws["G1"].value = "奇安信情报时间"
    for ip, ip_info in x_result_parse.items():
        ws["A{}".format(i)].value = str(i - 1)
        ws["B{}".format(i)].value = ip
        ws["C{}".format(i)].value = ip_info['location']
        ws["D{}".format(i)].value = ip_info['x_judgments']
        ws["E{}".format(i)].value = ip_info['x_update_time']
        ws["F{}".format(i)].value = ip_info['ti_judgments']
        ws["G{}".format(i)].value = ip_info['ti_update_time']
        i += 1
    wb.save(filename=f_path)
    print("报告收集完成，输出位置: {}".format(f_path))


def get_result(f_path):
    ip_info_dict = {}
    ip_file_code = check_code(f_path)
    with open(f_path, 'r', encoding=ip_file_code) as ip_file_content:
        iplist = [ip.strip() for ip in ip_file_content.readlines()]
    if len(iplist) > 60:
        print("超出微步数量限制，请缩小数量在60以下")
        sys.exit(1)
    """
        asn：asn信息。
        ports：端口信息。
        cas：SSL证书等信息。
        rdns_list：rdns记录信息。
        intelligences：威胁情报。
        judgments：从威胁情报中分析，综合判定的威胁类型。
        tags_classes：相关攻击团伙或安全事件信息标签等。
        samples：相关样本。
        update_time：情报最近更新时间。
        sum_cur_domains：当前指向域名的数量。
        scene：应用场景。
    """

    for query_ip in iplist:
        print("正在分析{} 中...".format(query_ip))
        x_url = "https://api.threatbook.cn/v3/ip/query"
        x_query = {
            "apikey": "threatbook_Api_key",
            "resource": "{}".format(query_ip),
            "exclude": "ports,cas,rdns_list,samples,sum_cur_domains,scene",
            "lang": "zh"
        }
        x_response = requests.request("GET", url=x_url, params=x_query)
        x_result = x_response.json()

        # Ti
        ti_url = "https://webapi.ti.qianxin.com/ip/v3/reputation?param=" + query_ip
        ti_headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
                          "Chrome/76.0.3809.132 Safari/537.36",
            'Api-Key': 'ti_Api_key',
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6"
        }
        ti_response = requests.get(ti_url, headers=ti_headers)
        ti_result = ti_response.json()
        if x_response.status_code == 200 and ti_response.status_code == 200:
            # X情报社区奖励计划来源情报
            # 从威胁情报中分析，综合判定的威胁类型
            x_judgments = ",".join(x_result['data']['{}'.format(query_ip)]['judgments'])
            x_update_time = x_result['data']['{}'.format(query_ip)]['update_time']

            # 奇安信情报
            ti_geo_country = ti_result['data']['{}'.format(query_ip)]['geo']['country']
            ti_geo_city = ti_result['data']['{}'.format(query_ip)]['geo']['city']
            ti_geo = ti_geo_country + '-' + ti_geo_city
            ti_update_time = ti_result['data']['{}'.format(query_ip)]['summary_info']['latest_reputation_time']
            ti_judgments_reputation = ti_result['data']['{}'.format(query_ip)]['summary_info']['reputation']
            ti_judgments_malicious_label = ",".join(
                ti_result['data']['{}'.format(query_ip)]['summary_info']['malicious_label'])
            ti_judgments = ti_judgments_reputation + ":" + ti_judgments_malicious_label
            # 调用写excel函数将查询结果写到excel表格中
            ip_info_dict[query_ip] = {
                "location": ti_geo,
                "x_judgments": x_judgments,
                "x_update_time": x_update_time,
                "ti_judgments": ti_judgments,
                "ti_update_time": ti_update_time
            }
        else:
            print("x_result: ", x_result)
            print("ti_result: ", ti_result)
    return ip_info_dict


if __name__ == '__main__':
    script_path = os.path.dirname(__file__)
    # script_path = os.path.dirname(os.path.realpath(sys.executable))

    # 生成 iplist.txt 路径
    ip_file = os.path.join(script_path, 'iplist.txt')
    # 校验 iplist.txt 文件是否存在
    existsfile(ip_file)

    # 生成时间
    date_time = time.strftime('%Y%m%d%H%M%S', time.localtime())
    # 生成带有时间的 ipinfo.xlsx 文件
    info_file = os.path.join(script_path, 'ipinfo-{}.xlsx'.format(date_time))

    # 获取查询结果，需要是字典类型
    ip_result = get_result(ip_file)

    # 保存结果
    save_result(ip_result, info_file)