add multithreaded logic, add package format checking script

src/nat_format.cpp

@@ -21,7 +21,7 @@
 #define HW_EVENT_ADD "SESSION_BUILT"
 #define HW_EVENT_DEL "SESSION_TEARDOWN"
 
-#define HW_EVENT_NEW 0x01
+#define HW_EVENT_NEW 0x00
 #define HW_EVENT_AGED 0x02
 #define HW_EVENT_PERIOD 0x03
 
@@ -42,10 +42,13 @@
 
 struct nat_format_global_info g_nat_format_info;
 
-char *multicast_payload;
+// 组播报文缓冲区和socket都是每个线程一个
-int cur_pkt = 0;
+int platform_thread_num = get_thread_count();
 
-int udp_socket;
+char **multicast_payloads;
+int *cur_pkts;
+
+int *udp_sockets;
 struct sockaddr_in dst_addr = {0};
 
 // 函数：子串匹配，返回子串结尾+1的位置，即值的起始位置
@@ -138,22 +141,22 @@ int extract_action_hs(char *data, int data_len, const char *key, char *dst) {
 }
 
 // 组播报文发送
-int send_multicast() {
+int send_multicast(int thread_seq) {
     // 发之前统一填充发送时间
     time_t now;
     time(&now);
     unsigned int now_timestamp = (int)now;
     for (int i = 0; i < g_nat_format_info.batch_size; i++) {
-        memcpy(multicast_payload + i*PAYLOAD_LEN+SEND_TIME_OFFSET, &now_timestamp, 4);
+        memcpy(multicast_payloads[thread_seq] + i*PAYLOAD_LEN+SEND_TIME_OFFSET, &now_timestamp, 4);
     }
     // 进行发送
-    if (sendto(udp_socket, multicast_payload, PAYLOAD_LEN*g_nat_format_info.batch_size, 0, (struct sockaddr *)&dst_addr, sizeof(dst_addr)) < 0) {
+    if (sendto(udp_sockets[thread_seq], multicast_payloads[thread_seq], PAYLOAD_LEN*g_nat_format_info.batch_size, 0, (struct sockaddr *)&dst_addr, sizeof(dst_addr)) < 0) {
         MESA_handle_runtime_log(g_nat_format_info.log, RLOG_LV_INFO, "nat_format", "Send multicast failed: %s", strerror(errno));
-        cur_pkt--;
+        cur_pkts[thread_seq]--;
         return -1;
     }
 
-    cur_pkt = 0;
+    cur_pkts[thread_seq] = 0;
     return 0;
 }
 
@@ -185,31 +188,54 @@ int nat_format_init(void) {
     }
 
     // 分配并初始化组播报文存储空间，长度为 batch_size*46Bytes
-    multicast_payload = (char *)malloc(PAYLOAD_LEN * g_nat_format_info.batch_size);
+    multicast_payloads = (char **)malloc(platform_thread_num * sizeof(char *));
-    memset(multicast_payload, 0, PAYLOAD_LEN * g_nat_format_info.batch_size);
+    if (multicast_payloads == NULL) {
-
+        printf("Memory of multicast_payloads allocation failure.\n");
-    // 创建socket用于发包
-    udp_socket = socket(AF_INET, SOCK_DGRAM, 0);
-    if (udp_socket == -1) {
-        printf("UDP multicast socket creation failed:%d, %s\n", errno, strerror(errno));
         return -1;
     }
-    // 绑定组播地址
+    for (int i = 0; i < platform_thread_num; i++) {
+        multicast_payloads[i] = (char *)malloc(PAYLOAD_LEN * g_nat_format_info.batch_size);
+        if (multicast_payloads[i] == NULL) {
+            for (int j = 0; j < i; j++) {
+                free(multicast_payloads[j]);
+            }
+            free(multicast_payloads);
+            printf("Memory of multicast_payload for thread %d allocation failure.\n", i);
+            return -1;
+        }
+        memset(multicast_payloads[i], 0, PAYLOAD_LEN * g_nat_format_info.batch_size);
+    }
+    cur_pkts = (int *)malloc(platform_thread_num * sizeof(int));
+    memset(cur_pkts, 0, platform_thread_num * sizeof(int));
+
+    // 组播地址
     dst_addr.sin_family = AF_INET;
     dst_addr.sin_addr.s_addr = inet_addr(g_nat_format_info.multicast_ip);
     dst_addr.sin_port = htons(g_nat_format_info.multicast_port);
     struct in_addr multicast_addr;
     inet_pton(AF_INET, g_nat_format_info.multicast_ip, &multicast_addr.s_addr);
-    setsockopt(udp_socket, IPPROTO_IP, IP_MULTICAST_IF, &multicast_addr, sizeof(multicast_addr));
+    // 发包端口
-    // 绑定发包端口
     struct sockaddr_in src_addr;
     src_addr.sin_family = AF_INET;
     src_addr.sin_port = htons(g_nat_format_info.host_port);
     src_addr.sin_addr.s_addr = inet_addr(g_nat_format_info.host_ip);
-    if (bind(udp_socket, (struct sockaddr*)&src_addr, sizeof(src_addr)) < 0) {
+
-        printf("socket bind failed\n");
+    // 创建socket用于发包
-        close(udp_socket);
+    udp_sockets = (int *)malloc(platform_thread_num * sizeof(int));
-        return -1;
+    for (int i = 0; i < platform_thread_num; i++) {
+        udp_sockets[i] = socket(AF_INET, SOCK_DGRAM, 0);
+        if (udp_sockets[i] == -1) {
+            free(udp_sockets);
+            printf("UDP multicast socket for thread %d creation failed:%d, %s\n", i, errno, strerror(errno));
+            return -1;
+        }
+        setsockopt(udp_sockets[i], IPPROTO_IP, IP_MULTICAST_IF, &multicast_addr, sizeof(multicast_addr));
+        if (bind(udp_sockets[i], (struct sockaddr*)&src_addr, sizeof(src_addr)) < 0) {
+            free(udp_sockets);
+            printf("socket bind for thread %d failed:%d, %s\n", i, errno, strerror(errno));
+            close(udp_sockets[i]);
+            return -1;
+        }
     }
 
     return 0;
@@ -217,7 +243,14 @@ int nat_format_init(void) {
 
 // 卸载函数
 void nat_format_destroy(void) {
-    free(multicast_payload);
+    for (int j = 0; j < platform_thread_num; j++) {
+        free(multicast_payloads[j]);
+    }
+    free(multicast_payloads);
+
+    free(cur_pkts);
+
+    free(udp_sockets);
 }
 
 // 入口函数
@@ -328,12 +361,12 @@ char nat_format_entry(struct streaminfo *a_udp, void **pme, int thread_seq, void
         }
 
         // 将提取出来的信息写进组播载荷
-        memcpy(multicast_payload + cur_pkt*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
+        memcpy(multicast_payloads[thread_seq] + cur_pkts[thread_seq]*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
-        cur_pkt++;
+        cur_pkts[thread_seq]++;
         
         // 攒够20个进行发送
-        if (cur_pkt == g_nat_format_info.batch_size) {
+        if (cur_pkts[thread_seq] == g_nat_format_info.batch_size) {
-            send_multicast();
+            send_multicast(thread_seq);
         }
     } 
     // 华三syslog
@@ -409,12 +442,12 @@ char nat_format_entry(struct streaminfo *a_udp, void **pme, int thread_seq, void
         }        
 
         // 将提取出来的信息写进组播载荷
-        memcpy(multicast_payload + cur_pkt*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
+        memcpy(multicast_payloads[thread_seq] + cur_pkts[thread_seq]*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
-        cur_pkt++;
+        cur_pkts[thread_seq]++;
         
         // 攒够20个进行发送
-        if (cur_pkt == g_nat_format_info.batch_size) {
+        if (cur_pkts[thread_seq] == g_nat_format_info.batch_size) {
-            send_multicast();
+            send_multicast(thread_seq);
         }
     }
     // 迪普syslog TODO
@@ -447,7 +480,7 @@ char nat_format_entry(struct streaminfo *a_udp, void **pme, int thread_seq, void
             MESA_handle_runtime_log(g_nat_format_info.log, RLOG_LV_INFO, "nat_format Huawei Binary", "Unknown Version %d", hb_head->Version);
             return APP_STATE_GIVEME;
         }
-        u_int16_t log_num = hb_head->Count;
+        u_int16_t log_num = ntohs(hb_head->Count);
         // 提取防火墙日志生成时间
         nat_payload.fw_log_timestamp = hb_head->Second;
         // 分别处理每一个body
@@ -486,12 +519,12 @@ char nat_format_entry(struct streaminfo *a_udp, void **pme, int thread_seq, void
             }
 
             // 将提取出来的信息写进组播载荷
-            memcpy(multicast_payload + cur_pkt*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
+            memcpy(multicast_payloads[thread_seq] + cur_pkts[thread_seq]*PAYLOAD_LEN, &nat_payload, PAYLOAD_LEN);
-            cur_pkt++;
+            cur_pkts[thread_seq]++;
 
             // 攒够20个进行发送
-            if (cur_pkt == g_nat_format_info.batch_size) {
+            if (cur_pkts[thread_seq] == g_nat_format_info.batch_size) {
-                send_multicast();
+                send_multicast(thread_seq);
             }
 
             // 定位下一个body的offset -- 03版本28,08版本的IPV4、NOPAT为44，USER为209，URL、TLV每个不同，由附加长度确定
@@ -500,7 +533,7 @@ char nat_format_entry(struct streaminfo *a_udp, void **pme, int thread_seq, void
                     offset += body_len;
                 } else {
                     char *tmp = (char *)hb_body;
-                    u_int16_t AppendLength = *(u_int16_t *)(tmp + sizeof(struct hw_binary_log_body) + HW_BINARY_BODY_V8_TLV_OFFSET);
+                    u_int16_t AppendLength = ntohs(*(u_int16_t *)(tmp + sizeof(struct hw_binary_log_body) + HW_BINARY_BODY_V8_TLV_OFFSET));
                     offset += (HW_BINARY_BODY_LENGTH_V8_IPV4 + AppendLength);
                 }
             }

tools/binary_filed_extraction.py

@@ -0,0 +1,102 @@
+from scapy.all import rdpcap, UDP
+
+def parser_hw_binary_detail(s):
+    len_list = [i * 2 for i in [1, 1, 2, 4, 4, 2, 1, 1]]
+    head_s = s[0:sum(len_list)]
+    head = []
+    start = 0
+    for l in len_list:
+        head.append(head_s[start:start + l])
+        start += l
+    [Version, LogType, Count, Second, FlowSequence, DeviceId, Slot, Reserved] = head
+    print(head)
+
+    body_start = sum(len_list)
+    len_list = [i * 2 for i in [1, 1, 1, 1, 4, 4, 4, 4, 2, 2, 2, 2, 4, 4, 4, 4, 4, 4, 2, 2, 1, 1, 2, 4]]
+    bodys = []
+    for i in range(int(Count, 16)):
+        body_s = s[body_start:body_start+sum(len_list)]
+        body = []
+        start = 0
+        for l in len_list:
+            body.append(body_s[start:start + l])
+            start += l
+        [Prot, Operator, IpVersion, TosIPv4, SourceIP, SrcNatIP, DestIP, DestNatIP, SrcPort, SrcNatPort, DestPort, DestNatPort, StartTime, EndTime, InTotalPkg, InTotalByte, OutTotalPkg, OutTotalByte, SourVpnIndex, DestVpnIndex, Reserved1, EventTrend, Reserved2, Reserved3] = body
+        print(body)
+
+def parser_hw_binary(s):
+    sum_ = 0
+    counts = [0, 0, 0, 0, 0]
+
+    len_list = [i * 2 for i in [1, 1, 2, 4, 4, 2, 1, 1]]
+    head_s = s[0:sum(len_list)]
+    Count = head_s[4:8]
+    sum_ += int(Count, 16)
+
+    body_start = sum(len_list)
+    len_list = [i * 2 for i in [1, 1, 1, 1, 4, 4, 4, 4, 2, 2, 2, 2, 4, 4, 4, 4, 4, 4, 2, 2, 1, 1, 2, 4]]
+    for i in range(int(Count, 16)):
+        body_s = s[body_start:body_start+sum(len_list)]
+        EventTrend = min(int(body_s[114:116]), 4)
+        counts[EventTrend] += 1
+        body_start += sum(len_list)
+    return (sum_, counts)
+
+def parser_nat(s):
+    counts = [0, 0, 0]
+    for i in range(20):
+        body_s = s[i*46*2:(i+1)*46*2]
+        EventTrend = body_s[36:38]
+        EventTrend = min(int(body_s[36:38]), 2)
+        counts[EventTrend] += 1
+    return counts
+
+def read_and_filter_pcap_hw(pcap_file, protocol="UDP"):
+    nat_num = 0
+    count_nums = [0, 0, 0, 0, 0]
+    
+    packets = rdpcap(pcap_file)
+    for pkt in packets:
+        if protocol in pkt:
+            # print(packet.show())  # 显示每个符合条件的数据包详情
+            udp_payload = bytes(pkt[UDP].payload)
+            hex_payload = ''.join(f'{byte:02x}' for byte in udp_payload)
+            (sum_, counts) = parser_hw_binary(hex_payload)
+            nat_num += sum_
+            for i in range(5):
+                count_nums[i] += counts[i]
+        
+    print(f"Total {len(packets)} {protocol} packets in hw_binary found.")
+    print(f"unknown(00) log:\t{count_nums[0]}/{nat_num}({count_nums[0]/nat_num:.2f}%)")
+    print(f"build(01) log:\t{count_nums[1]}/{nat_num}({count_nums[1]/nat_num:.2f}%)")
+    print(f"aged(02) log:\t{count_nums[2]}/{nat_num}({count_nums[2]/nat_num:.2f}%)")
+    print(f"period(03) log:\t{count_nums[3]}/{nat_num}({count_nums[3]/nat_num:.2f}%)")
+    print(f"unknown(>03) log:\t{count_nums[4]}/{nat_num}({count_nums[4]/nat_num:.2f}%)")
+
+def read_and_filter_pcap_nat(pcap_file, protocol="UDP"):
+    nat_num = 0
+    count_nums = [0, 0, 0]
+    
+    packets = rdpcap(pcap_file)
+    for pkt in packets:
+        if protocol in pkt:
+            # print(packet.show())  # 显示每个符合条件的数据包详情
+            udp_payload = bytes(pkt[UDP].payload)
+            hex_payload = ''.join(f'{byte:02x}' for byte in udp_payload)
+            counts = parser_nat(hex_payload)
+            for i in range(3):
+                count_nums[i] += counts[i]
+        
+    print(f"Total {len(packets)} {protocol} packets in nat_format found.")
+    nat_num = len(packets)*20
+    print(f"add(00) log:\t{count_nums[0]}/{nat_num}({count_nums[0]/nat_num:.2f}%)")
+    print(f"del(01) log:\t{count_nums[1]}/{nat_num}({count_nums[1]/nat_num:.2f}%)")
+    print(f"unknown(>01) log:\t{count_nums[2]}/{nat_num}({count_nums[2]/nat_num:.2f}%)")
+    
+ 
+if __name__ == "__main__":
+    pcap_file_path = "D://MESA//搞点项目//NAT//nat-pcap//hw_binary_1021.pcap"
+    read_and_filter_pcap_hw(pcap_file_path)
+    pcap_file_path = "D://MESA//搞点项目//NAT//nat-pcap//nat_1021.pcap"
+    read_and_filter_pcap_nat(pcap_file_path)
+    # print('ok!')