python-3.x - 绘制 Snort 警报数量随时间变化的连续图

Question

我有 snort 将 DDOS 警报记录到文件；我使用 Syslog-ng 解析日志并以 json 格式输出到 redis(想将其设置为缓冲区，我使用 70 秒到期的“setex”命令)。

整个事情似乎不太顺利；欢迎任何让它变得更容易的想法。

我写了一个简单的 python 脚本来监听 redis KA 事件并计算每秒 snort 警报的数量。我尝试创建另外两个线程；一个用于从 snort 检索 json 格式的警报，第二个用于计算警报。第三个应该使用 matplotlib.pyplot 绘制图形

#import time 
from redis import StrictRedis as sr 
import os
import json
import matplotlib.pyplot as plt
import threading as th
import time


redis = sr(host='localhost', port = 6379, decode_responses = True)


#file = open('/home/lucidvis/vis_app_py/log.json','w+')

# This function is still being worked on
def  do_plot():
    print('do_plot loop running')
    while accumulated_data:

        x_values = [int(x['time_count']) for x in accumulated_data]
        y_values = [y['date'] for y in accumulated_data]

        plt.title('Attacks Alerts per time period')

        plt.xlabel('Time', fontsize=14)
        plt.ylabel('Snort Alerts/sec')

        plt.tick_params(axis='both', labelsize=14)

        plt.plot(y_values,x_values, linewidth=5)
        plt.show()
        time.sleep(0.01)




def accumulator():
    # first of, check the current json data and see if its 'sec' value is same 
    #that is the last in the accumulated data list
    #if it is the same, increase time_count by one else pop that value
    pointer_data = {}

    print('accumulator loop running')

    while True: 
        # pointer data is the current sec of json data used for comparison
        #new_data is the latest json formatted alert received
        # received_from_redis is a list declared in the main function
        if received_from_redis:
            new_data = received_from_redis.pop(0)
        if not pointer_data:
            pointer_data = new_data.copy()

        print(">>", type(pointer_data), " >> ", pointer_data)

        if pointer_data and pointer_data['sec']==new_data["sec"]
            pointer_data['time_count'] +=1


        elif pointer_data: 
            accumulated_data.append(pointer_data)
            pointer_data = new_data.copy()
            pointer_data.setdefault('time_count',1)

        else:
            time.sleep(0.01)




# main function creates the redis object and receives messages based on events
#this function calls two other functions and creates threads so they appear to run concurrently

def main():
    p = redis.pubsub()
    # 
    p.psubscribe('__keyspace@0__*')

    print('Starting message loop')

    while True:
        try:
            time.sleep(2)
            message = p.get_message()

    # Obtain the key from the redis emmitted event if the event is a set event
            if message and message['data']=='set':
         # the format emmited by redis is in a dict form
         # the key is the value to the key 'channel'
         # The key is in '__keyspace@0__*' form
         # obtain the last field of the list returned by split function
                key = message['channel'].split('__:')[-1]

                data_redis = json.loads(redis.get(str(key)))
                received_from_redis.append(data_redis)
        except Exception e:
            print(e)

            continue




if __name__ == "__main__":
    accumulated_data = []
    received_from_redis = []
# main function creates the redis object and receives messages based on events
#this function calls two other functions and creates threads so they appear to run concurrently
    thread_accumulator = th.Thread(target = accumulator, name ='accumulator')
    do_plot_thread = th.Thread(target = do_plot, name ='do_plot')

    while True:
        thread_accumulator.start()
        do_plot_thread.start()

        main()

    thread_accumulator.join()
    do_plot_thread.join()

我目前确实遇到了错误；我只是不知道线程是否已创建或是否运行良好。我需要让事情变得更好的想法。

以 json 格式格式化并从下面的 redis 中获取的警报示例

{"victim_port":"","victim":"192.168.204.130","protocol":"ICMP","msg":"Ping_Flood_Attack_Detected","key":"1000","date":"06/01-09:26:13","attacker_port":"","attacker":"192.168.30.129","sec":"13"}

Answer 1

我不确定我是否完全理解你的场景，但如果你想计算本质上是日志消息的事件，你可以在 syslog-ng 中完成。作为 Python destination (因为你已经在 python 中工作)，或者甚至可能不需要使用 grouping-by parser 进行额外的编程。 .