小程序可以做企业网站销售的技巧与口才

使用 python 检测泛洪攻击的案例

本案例只使用python标准库通过执行命令来监控异常请求, 并封锁IP, 不涉及其他第三方库工具.

import os
import time
from collections import Counter# 1、update 命令, 采集CPU的平均负载
def get_cpu_load():"""uptime 命令获取系统开机时间和cpu负载, 比如:04:02:00 up 1 day,  3:49,  1 user,  load average: 0.71, 0.48, 0.46利用awk命令来筛选出CPU负载:-F 设置拆分数据的分隔符 ': ', 如果不设置默认以空格作为分隔符'{print $2}' 表示打印第二列数据"""cpu_load = os.popen("uptime | awk -F ': ' '{print $2}' | awk -F ',' '{print $1}'").read()cpu_load = float(cpu_load)return cpu_load# 2、netstat 命令, 采集 tcp 的连接数量
def get_conn_count():"""统计所有tcp连接数量-a 列出所有, -n 以数字形式显示端口, -t tcpwc -l 统计行数"""netstat = os.popen('netstat -ant | wc -l').read()return int(netstat)# 3、ss命令, 采集socket统计信息,
"""
ss命令比netstat命令更详细, 速度更快
端口队列默认大小是128, 当 Recv-Q 大于等于 Send-Q 时表示队列满了, 端口存在大量请求
"""
def get_queue_size():"""统计当前的队列情况-l 监听状态, -n 以数字显示, -t tcp"""# ss -lnt | grep :80 | awk '{print $2}' # 第二列是 Recv-Q# ss -lnt | grep :80 | awk '{print $3}' # 第三列是 Send-Qsslnt = os.popen("ss -lnt | grep :80").read()recv_q = int(sslnt.split()[1])send_q = int(sslnt.split()[2])return recv_q, send_q# 4、netstat 命令, 采集连接数量最多的IP地址
def get_most_ip():"""获取当前连接数量最多的IP地址"""# 获取所有访问 80 端口的ip添加到列表中result = os.popen('netstat -ant | grep :80').read()line_list = result.split('\n')ip_list = []for line in line_list:try:temp_list = line.split()ip = temp_list[4].split(':')[0]ip_list.append(ip)except Exception as e:pass# 统计列表中出现最多的ipcnt = Counter(ip_list)# most_common(n) 返回出现次数最多的前n项元组组成的列表t = cnt.most_common(1)return t[0][0]# 5、firewall-cmd 命令, 防火墙封锁IP地址
def firewall_ip(ip):"""使用防火墙封锁ip:param ip: 封锁目标ip"""result = os.popen(f"firewall-cmd --add-rich-rule='rule family=ipv4 source address={ip} port port=80 protocol=tcp reject'").read()if 'success' in result:print(f"成功封锁 {ip}.")else:print(f"封锁失败.")if __name__ == '__main__':while 1:cpu = get_cpu_load()conn = get_conn_count()recvq, sendq = get_queue_size()print(f"CPU-Load: {cpu}, TCP Conn: {conn}, TCP Queue: {recvq, sendq}")# 对采集到的数据进行判断，并进行预警提醒, 封锁ipif cpu > 55 and conn > 500 and recvq > sendq - 10:ip = get_most_ip()print(f"预警，可疑IP：{ip}.")firewall_ip(ip)time.sleep(5)