高端品牌网站建设哪家好,合肥仿站定制模板建站,建设银行手机官方网站下载,龙岗网站建设开发设计公司【告警自动化处置脚本0】
以下是一个Python脚本#xff0c;用于定时查询历史告警#xff0c;找到攻击结果为成功的告警#xff0c;并根据告警结果查询威胁情报。如果是恶意IP#xff0c;则调用防火墙进行封禁#xff0c;并记录封禁动作的日志。
首先#xff0c;确保已经…【告警自动化处置脚本0】
以下是一个Python脚本用于定时查询历史告警找到攻击结果为成功的告警并根据告警结果查询威胁情报。如果是恶意IP则调用防火墙进行封禁并记录封禁动作的日志。
首先确保已经安装了pymysql和requests库。如果没有请使用以下命令安装
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: fBearer {threat_intelligence_api_key}, # 设置API密钥}response requests.get(threat_intelligence_url, headersheaders, 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, shellTrue, checkTrue) # 执行防火墙命令# 定义记录封禁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] # 获取攻击源IPthreat_info query_threat_intelligence(ip) # 查询威胁情报if threat_info.get(malicious): # 判断是否为恶意IPblock_ip(ip) # 封禁恶意IPlog_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 requestsimport schedule
import time
import requests# 定义查询历史告警的函数
def get_historical_alerts():# 这里替换为实际的查询历史告警的API地址url https://your-alert-api.example.com/alertsresponse requests.get(url)alerts response.json()return alerts# 定义根据告警结果查询威胁情报的函数
def get_threat_intelligence(alert):# 这里替换为实际的查询威胁情报的API地址url https://your-threat-intelligence-api.example.com/intelligenceresponse 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/blockresponse 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(fIP {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 URLdef 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_alarmsdef get_threat_intel(ip):response requests.get(THREAT_INTEL_API_URL, params{ip: ip})threat_intel response.json()return threat_inteldef block_ip(ip):response requests.post(FIREWALL_API_URL, json{action: block, ip: ip})return response.status_code 200def 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 successcursor.execute(sql)successful_alarms cursor.fetchall()return successful_alarms# 查询微步威胁情报
def get_threat_intel(ip):headers {Authorization: fApiKey {THREAT_INTEL_API_KEY}}response requests.get(f{THREAT_INTEL_API_URL}/{ip}, headersheaders)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 等。接下来定义了四个函数
connect_db连接到数据库并返回连接对象。get_successful_alarms查询数据库中成功攻击的告警。get_threat_intel查询指定 IP 的微步威胁情报。block_ip调用防火墙 API 封禁指定 IP。
main 函数是脚本的主要逻辑它会循环执行以下操作连接到数据库查询成功攻击的告警对于每个成功的告警查询其源 IP 的微步威胁情报如果威胁情报显示该 IP 是恶意的则调用防火墙 API 进行封禁。
为了运行此脚本你需要安装 pymysql 和 requests 库。你可以使用以下命令安装
pip install pymysql requests请注意这个脚本仅作为示例你需要根据你的实际环境和数据库配置进行修改。在实际使用中你还需要考虑异常处理、日志记录等方面的问题。
