【告警自动化处置脚本】

发布时间:2023年12月17日

【告警自动化处置脚本0】

以下是一个Python脚本,用于定时查询历史告警,找到攻击结果为成功的告警,并根据告警结果查询威胁情报。如果是恶意IP,则调用防火墙进行封禁,并记录封禁动作的日志。

首先,确保已经安装了pymysqlrequests库。如果没有,请使用以下命令安装:

pip install pymysql requests

接下来,创建一个名为block_malicious_ips.py的脚本,并将以下代码复制到其中:

# 导入所需库
import pymysql  # 用于与MySQL数据库进行交互
import requests  # 用于发送HTTP请求
import json  # 用于处理JSON数据
import subprocess  # 用于执行系统命令
import time  # 用于获取当前时间

# 数据库配置
db_config = {
    'host': 'localhost',  # 数据库主机名
    'user': 'root',  # 数据库用户名
    'password': 'your_password',  # 数据库密码
    'db': 'your_database',  # 数据库名称
    'charset': 'utf8mb4',  # 数据库字符集
}

# 微步威胁情报API配置
threat_intelligence_api_key = 'your_api_key'  # 微步威胁情报API密钥
threat_intelligence_url = 'https://api.example.com/v1/ip/query'  # 微步威胁情报API的URL

# 防火墙配置
firewall_cmd = 'iptables'  # 防火墙命令,这里使用iptables作为示例

# 日志文件
log_file = 'block_malicious_ips.log'  # 日志文件名

# 定义查询成功攻击的函数
def query_successful_attacks():
    connection = pymysql.connect(**db_config)  # 连接到MySQL数据库
    try:
        with connection.cursor() as cursor:  # 创建一个数据库游标
            sql = "SELECT * FROM `alerts` WHERE `attack_result` = 'success'"  # 查询成功攻击的SQL语句
            cursor.execute(sql)  # 执行SQL查询
            return cursor.fetchall()  # 返回查询结果
    finally:
        connection.close()  # 关闭数据库连接

# 定义查询威胁情报的函数
def query_threat_intelligence(ip):
    headers = {
        'Authorization': f'Bearer {threat_intelligence_api_key}',  # 设置API密钥
    }
    response = requests.get(threat_intelligence_url, headers=headers, params={'ip': ip})  # 发送GET请求查询威胁情报
    return response.json()  # 返回JSON格式的响应数据

# 定义封禁IP的函数
def block_ip(ip):
    cmd = f'{firewall_cmd} -A INPUT -s {ip} -j DROP'  # 构造防火墙命令
    subprocess.run(cmd, shell=True, check=True)  # 执行防火墙命令

# 定义记录封禁IP日志的函数
def log_blocked_ip(ip):
    with open(log_file, 'a') as f:  # 以追加模式打开日志文件
        f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} Blocked IP: {ip}\n")  # 写入封禁IP的日志

# 主函数
def main():
    successful_attacks = query_successful_attacks()  # 查询成功攻击
    for attack in successful_attacks:  # 遍历成功攻击
        ip = attack['source_ip']  # 获取攻击源IP
        threat_info = query_threat_intelligence(ip)  # 查询威胁情报
        if threat_info.get('malicious'):  # 判断是否为恶意IP
            block_ip(ip)  # 封禁恶意IP
            log_blocked_ip(ip)  # 记录封禁日志

# 当脚本作为主程序运行时,执行主函数
if __name__ == '__main__':
    main()

在上述脚本中,需要替换以下几个变量:

  • db_config:数据库配置,包括主机名、用户名、密码和数据库名称。
  • threat_intelligence_api_key:微步威胁情报API密钥。
  • threat_intelligence_url:微步威胁情报API的URL。
  • firewall_cmd:防火墙命令,这里使用iptables作为示例。

接下来,设置一个Linux计划任务,定期运行此脚本。首先,打开终端并运行以下命令:

crontab -e

在打开的编辑器中,添加以下行以每小时运行一次脚本:

0 * * * * /usr/bin/python3 /path/to/block_malicious_ips.py

保存并退出编辑器。现在,脚本将每小时运行一次,自动封禁恶意IP并记录封禁日志。

关于Linux计划任务(Cron)的详细解释:

Cron是Linux系统中用于定期执行任务的工具。它允许用户在特定的日期和时间自动执行命令或者脚本。

在这个例子中,我们使用crontab -e命令来编辑当前用户的Cron配置。crontab命令用于管理用户的Cron表,-e选项表示编辑模式。

编辑器中的每一行都代表一个计划任务。每行的格式如下:

* * * * * /path/to/command arg1 arg2
| | | | |
| | | | ----- Day of week (0 - 7) (Sunday = 0 or 7)
| | | ------- Month (1 - 12)
| | --------- Day of month (1 - 31)
| ----------- Hour (0 - 23)
------------- Minute (0 - 59)

在这个例子中,我们将添加以下行以每小时运行一次脚本:

0 * * * * /usr/bin/python3 /path/to/block_malicious_ips.py

这里的0 * * * *表示每小时的第0分钟执行任务,也就是每小时运行一次。/usr/bin/python3是Python 3解释器的路径,/path/to/block_malicious_ips.py是脚本的路径。请确保将其替换为实际的路径。

保存并退出编辑器后,Cron将自动加载新的配置,并按照计划执行任务。这样,脚本将每小时运行一次,自动封禁恶意IP并记录封禁日志。

【告警自动化处置脚本1】

这是一个Python脚本,使用了schedule库来实现定时任务,requests库来发送HTTP请求。请确保已安装这两个库:

pip install schedule requests
import schedule
import time
import requests

# 定义查询历史告警的函数
def get_historical_alerts():
    # 这里替换为实际的查询历史告警的API地址
    url = "https://your-alert-api.example.com/alerts"
    response = requests.get(url)
    alerts = response.json()
    return alerts

# 定义根据告警结果查询威胁情报的函数
def get_threat_intelligence(alert):
    # 这里替换为实际的查询威胁情报的API地址
    url = "https://your-threat-intelligence-api.example.com/intelligence"
    response = requests.get(url, params={"ip": alert["source_ip"]})
    threat_info = response.json()
    return threat_info

# 定义调用防火墙进行封禁的函数
def block_ip(ip):
    # 这里替换为实际的调用防火墙封禁IP的API地址
    url = "https://your-firewall-api.example.com/block"
    response = requests.post(url, json={"ip": ip})
    return response.ok

# 定义定时任务
def check_alerts():
    alerts = get_historical_alerts()
    for alert in alerts:
        if alert["attack_result"] == "success":
            threat_info = get_threat_intelligence(alert)
            if threat_info["malicious"]:
                blocked = block_ip(alert["source_ip"])
                if blocked:
                    print(f"IP {alert['source_ip']} 已被封禁")
                else:
                    print(f"无法封禁 IP {alert['source_ip']}")

# 每小时执行一次定时任务
schedule.every(1).hours.do(check_alerts)

# 无限循环执行定时任务
while True:
    schedule.run_pending()
    time.sleep(1)

请将上述代码中的API地址替换为实际的查询历史告警、查询威胁情报和调用防火墙封禁IP的API地址。运行此脚本后,它将每小时执行一次定时任务,检查历史告警中攻击结果为成功的告警,并根据告警结果查询威胁情报。如果是恶意IP,则调用防火墙进行封禁。

【告警自动化处置脚本2】

这是一个使用 Python 编写的示例脚本,它定时查询历史告警,找到攻击结果为成功的告警,根据告警结果查询威胁情报,如果是恶意IP则调用防火墙进行封禁。

请注意,这个示例脚本仅用于说明目的,并没有实际的 API 调用。你需要根据你的实际环境和 API 进行修改。

import time
import requests

# 配置参数
INTERVAL = 60  # 查询间隔(秒)
ALARM_API_URL = "http://example.com/api/alarm"  # 历史告警 API URL
THREAT_INTEL_API_URL = "http://example.com/api/threat-intel"  # 威胁情报 API URL
FIREWALL_API_URL = "http://example.com/api/firewall"  # 防火墙 API URL

def get_successful_alarms():
    response = requests.get(ALARM_API_URL)
    alarms = response.json()
    successful_alarms = [alarm for alarm in alarms if alarm["attack_result"] == "success"]
    return successful_alarms

def get_threat_intel(ip):
    response = requests.get(THREAT_INTEL_API_URL, params={"ip": ip})
    threat_intel = response.json()
    return threat_intel

def block_ip(ip):
    response = requests.post(FIREWALL_API_URL, json={"action": "block", "ip": ip})
    return response.status_code == 200

def main():
    while True:
        print("查询历史告警...")
        successful_alarms = get_successful_alarms()
        for alarm in successful_alarms:
            ip = alarm["source_ip"]
            print(f"查询威胁情报: {ip}")
            threat_intel = get_threat_intel(ip)
            if threat_intel["malicious"]:
                print(f"封禁恶意 IP: {ip}")
                if block_ip(ip):
                    print(f"封禁成功: {ip}")
                else:
                    print(f"封禁失败: {ip}")
        time.sleep(INTERVAL)

if __name__ == "__main__":
    main()

这个脚本首先定义了一些配置参数,如查询间隔、API URL 等。接下来,定义了三个函数:get_successful_alarms 用于查询历史告警并筛选出攻击成功的告警;get_threat_intel 用于查询指定 IP 的威胁情报;block_ip 用于调用防火墙 API 封禁指定 IP。

main 函数是脚本的主要逻辑,它会循环执行以下操作:查询历史告警,找到攻击成功的告警;对于每个成功的告警,查询其源 IP 的威胁情报;如果威胁情报显示该 IP 是恶意的,则调用防火墙 API 进行封禁。

为了运行此脚本,你需要安装 requests 库。你可以使用以下命令安装:

pip install requests

请注意,这个脚本仅作为示例,你需要根据你的实际环境和 API 进行修改。在实际使用中,你还需要考虑异常处理、日志记录等方面的问题。

【告警自动化处置脚本3】

以下是一个更详细的 Python 脚本示例,使用 pymysql 库查询数据库中的历史告警,并使用微步在线的威胁情报 API 查询威胁情报。该脚本包含详细的注释以解释每个步骤。

请注意,此示例脚本仅用于说明目的,你需要根据你的实际环境和数据库配置进行修改。

import time
import pymysql
import requests

# 配置参数
INTERVAL = 60  # 查询间隔(秒)
DB_CONFIG = {  # 数据库配置
    'host': 'localhost',
    'user': 'username',
    'password': 'password',
    'db': 'database_name',
    'charset': 'utf8mb4'
}
THREAT_INTEL_API_URL = "https://api.example.com/v4/ip"  # 微步威胁情报 API URL
THREAT_INTEL_API_KEY = "your_api_key"  # 微步 API 密钥
FIREWALL_API_URL = "http://example.com/api/firewall"  # 防火墙 API URL

# 连接数据库
def connect_db():
    connection = pymysql.connect(**DB_CONFIG)
    return connection

# 查询成功攻击的告警
def get_successful_alarms(connection):
    with connection.cursor() as cursor:
        sql = "SELECT * FROM alarms WHERE attack_result = 'success'"
        cursor.execute(sql)
        successful_alarms = cursor.fetchall()
    return successful_alarms

# 查询微步威胁情报
def get_threat_intel(ip):
    headers = {"Authorization": f"ApiKey {THREAT_INTEL_API_KEY}"}
    response = requests.get(f"{THREAT_INTEL_API_URL}/{ip}", headers=headers)
    threat_intel = response.json()
    return threat_intel

# 调用防火墙 API 进行封禁
def block_ip(ip):
    response = requests.post(FIREWALL_API_URL, json={"action": "block", "ip": ip})
    return response.status_code == 200

# 主函数
def main():
    connection = connect_db()
    while True:
        print("查询成功攻击的告警...")
        successful_alarms = get_successful_alarms(connection)
        for alarm in successful_alarms:
            ip = alarm["source_ip"]
            print(f"查询微步威胁情报: {ip}")
            threat_intel = get_threat_intel(ip)
            if threat_intel["malicious"]:
                print(f"封禁恶意 IP: {ip}")
                if block_ip(ip):
                    print(f"封禁成功: {ip}")
                else:
                    print(f"封禁失败: {ip}")
        time.sleep(INTERVAL)

if __name__ == "__main__":
    main()

在这个脚本中,我们首先定义了一些配置参数,如查询间隔、数据库配置、API URL 等。接下来,定义了四个函数:

  1. connect_db:连接到数据库并返回连接对象。
  2. get_successful_alarms:查询数据库中成功攻击的告警。
  3. get_threat_intel:查询指定 IP 的微步威胁情报。
  4. block_ip:调用防火墙 API 封禁指定 IP。

main 函数是脚本的主要逻辑,它会循环执行以下操作:连接到数据库,查询成功攻击的告警;对于每个成功的告警,查询其源 IP 的微步威胁情报;如果威胁情报显示该 IP 是恶意的,则调用防火墙 API 进行封禁。

为了运行此脚本,你需要安装 pymysqlrequests 库。你可以使用以下命令安装:

pip install pymysql requests

请注意,这个脚本仅作为示例,你需要根据你的实际环境和数据库配置进行修改。在实际使用中,你还需要考虑异常处理、日志记录等方面的问题。

文章来源:https://blog.csdn.net/weixin_46356409/article/details/134955642
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。