import json import subprocess with open('/var/log/suricata/fast.log', 'r') as f: for line in f: try: event = json.loads(line) except json.JSONDecodeError: # 如果解析失败，则跳过这一行 continue if event['alert']['severity'] <= 2: ip = event['src_ip'] # 调用iptables命令封禁IP result = subprocess.call(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP']) if result == 0: print(f'封禁IP {ip} 成功！') else: print(f'封禁IP {ip} 失败！')修改优化一下，fast.log中内容并不是json格式

好的，那我们可以使用正则表达式来解析fast.log中的内容，而不是使用json。代码如下：

import re
import subprocess

with open('/var/log/suricata/fast.log', 'r') as f:
    for line in f:
        # 使用正则表达式匹配IP地址
        match = re.search(r'\d+\.\d+\.\d+\.\d+', line)
        if not match:
            continue
        ip = match.group()
        # 调用iptables命令封禁IP
        result = subprocess.call(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP'])
        if result == 0:
            print(f'封禁IP {ip} 成功！')
        else:
            print(f'封禁IP {ip} 失败！')

这段代码使用了正则表达式来匹配IP地址，在匹配成功后调用iptables命令封禁该IP地址。

相关问题

import json import subprocess with open('/var/log/suricata/fast.log', 'r') as f: events = json.load(f) for event in events: if event['alert']['severity'] <= 2: ip = event['src_ip'] # 调用iptables命令封禁IP result = subprocess.call(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP']) if result == 0: print(f'封禁IP {ip} 成功！') else: print(f'封禁IP {ip} 失败！')修改优化一下这段代码

可以考虑以下几点进行优化：

1. 将代码封装成函数，方便复用。

2. 使用 try-except 语句处理 subprocess.call() 函数返回的异常，可以更好地处理命令执行失败的情况。

3. 使用 f-string 格式化字符串，使代码更加简洁明了。

4. 将常量提取出来，避免硬编码。

下面是优化后的代码：

import json
import subprocess

LOG_FILE = '/var/log/suricata/fast.log'
IPTABLES_CMD = ['iptables', '-A', 'INPUT', '-s']

def ban_ip(ip):
    try:
        result = subprocess.call(IPTABLES_CMD + [ip, '-j', 'DROP'])
        if result == 0:
            print(f'封禁IP {ip} 成功！')
        else:
            print(f'封禁IP {ip} 失败！')
    except subprocess.CalledProcessError as e:
        print(f'命令执行失败：{e}')

with open(LOG_FILE, 'r') as f:
    events = json.load(f)
    for event in events:
        if event['alert']['severity'] <= 2:
            ip = event['src_ip']
            ban_ip(ip)

这样优化后的代码更具可读性和可维护性，也更加健壮。

import re import subprocess import time while True: with open('/var/log/suricata/fast.log', 'r') as f: for line in f: # 使用正则表达式匹配IP地址 match = re.search(r'\d+\.\d+\.\d+\.\d+', line) if not match: continue ip = match.group() # 调用iptables命令封禁IP result = subprocess.call(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP']) if result == 0: print(f'封禁IP {ip} 成功！') else: print(f'封禁IP {ip} 失败！') time.sleep(5) # 每隔5秒读取一次日志 将这两行代码加入到上面代码中，应该如何修改优化file = open('/var/log/suricata/eve.json') file.seek(0, 2)

可以将文件操作部分放入一个函数中，每次读取完文件后关闭文件，避免文件句柄泄漏。同时，可以使用 `time.sleep()` 函数控制读取日志的时间间隔。修改后代码如下：

import re 
import subprocess 
import time 

def process_log_file():
    # 打开日志文件
    with open('/var/log/suricata/fast.log', 'r') as f:
        for line in f:
            # 使用正则表达式匹配IP地址
            match = re.search(r'\d+\.\d+\.\d+\.\d+', line)
            if not match:
                continue
            ip = match.group()
            # 调用iptables命令封禁IP
            result = subprocess.call(['iptables', '-A', 'INPUT', '-s', ip, '-j', 'DROP'])
            if result == 0:
                print(f'封禁IP {ip} 成功！')
            else:
                print(f'封禁IP {ip} 失败！')

while True:
    process_log_file()
    time.sleep(5) # 每隔5秒读取一次日志

# 打开日志文件
with open('/var/log/suricata/eve.json', 'r') as file:
    file.seek(0, 2)
    # 在文件结尾处添加新内容
    # ...