feat: support icmp close type

yuanchaoa · yuanchaoa · commit c014f2533a0a · 2024-09-11T16:26:32.000+08:00
diff --git a/agent/src/collector/quadruple_generator.rs b/agent/src/collector/quadruple_generator.rs
@@ -986,6 +986,20 @@ impl QuadrupleGenerator {
                     | CloseType::TcpFin
                     | CloseType::Unknown
                     | CloseType::TcpFinClientRst
+                    | CloseType::IcmpOtherTimeout
+                    | CloseType::IcmpAddressMaskTimeout
+                    | CloseType::IcmpDestinationUnreachable
+                    | CloseType::IcmpEchoTimeout
+                    | CloseType::IcmpInformationTimeout
+                    | CloseType::IcmpNeighborTimeout
+                    | CloseType::IcmpNormal
+                    | CloseType::IcmpPacketTooBig
+                    | CloseType::IcmpParameterProblem
+                    | CloseType::IcmpRouterTimeout
+                    | CloseType::IcmpSourceQuench
+                    | CloseType::IcmpTimeExceeded
+                    | CloseType::IcmpTimestampTimeout
+                    | CloseType::IcmpTraceroute
                     | CloseType::Max => (),
                 }
             }
diff --git a/agent/src/common/flow.rs b/agent/src/common/flow.rs
@@ -77,7 +77,20 @@ pub enum CloseType {
     ClientEstablishReset = 18,  // 18: 建连-客户端其他重置
     ServerEstablishReset = 19,  // 19: 建连-服务端其他重置
     TcpFinClientRst = 20,       // 20: 正常结束-客户端重置
-    Max = 21,
+    IcmpNormal = 21,                 // 21: ICMP-正常结束
+    IcmpAddressMaskTimeout = 22,     // 22: ICMP-地址掩码超时
+    IcmpDestinationUnreachable = 23, // 23: ICMP-目标不可达
+    IcmpEchoTimeout = 24,            // 24: ICMP-PING超时
+    IcmpInformationTimeout = 25,     // 25: ICMP-信息超时
+    IcmpParameterProblem = 26,       // 26: ICMP-参数问题
+    IcmpRouterTimeout = 27,          // 27: ICMP-路由器超时
+    IcmpSourceQuench = 28,           // 28: ICMP-源站抑制
+    IcmpTimeExceeded = 29,           // 29: ICMP-TTL溢出
+    IcmpTimestampTimeout = 30,       // 30: ICMP-时间戳超时
+    IcmpNeighborTimeout = 31,        // 31: ICMP-邻居发现超时
+    IcmpPacketTooBig = 32,           // 32: ICMP-Packet过大
+    IcmpOtherTimeout = 33,           // 33: ICMP-未知超时
+    Max = 34,
 }
 
 impl CloseType {
@@ -97,6 +110,18 @@ impl CloseType {
             || self == CloseType::ServerReset
             || self == CloseType::ServerQueueLack
             || self == CloseType::ServerEstablishReset
+            || self == CloseType::IcmpAddressMaskTimeout
+            || self == CloseType::IcmpEchoTimeout
+            || self == CloseType::IcmpInformationTimeout
+            || self == CloseType::IcmpRouterTimeout
+            || self == CloseType::IcmpTimestampTimeout
+            || self == CloseType::IcmpNeighborTimeout
+            || self == CloseType::IcmpPacketTooBig
+            || self == CloseType::IcmpTimeExceeded
+            || self == CloseType::IcmpSourceQuench
+            || self == CloseType::IcmpDestinationUnreachable
+            || self == CloseType::IcmpParameterProblem
+            || self == CloseType::IcmpOtherTimeout
     }
 }
 
@@ -1070,7 +1095,40 @@ impl Flow {
             FlowState::Exception => CloseType::Unknown,
             FlowState::Opening1 => CloseType::ClientSynRepeat,
             FlowState::Opening2 => CloseType::ServerSynAckRepeat,
-            FlowState::Established => CloseType::Timeout,
+            FlowState::IcmpAddressMaskReply => CloseType::IcmpNormal,
+            FlowState::IcmpAddressMaskRequest => CloseType::IcmpAddressMaskTimeout,
+            FlowState::IcmpEchoReply => CloseType::IcmpNormal,
+            FlowState::IcmpEchoRequest => CloseType::IcmpEchoTimeout,
+            FlowState::IcmpDestinationUnreachable => CloseType::IcmpDestinationUnreachable,
+            FlowState::IcmpInformationReply => CloseType::IcmpNormal,
+            FlowState::IcmpInformationRequest => CloseType::IcmpInformationTimeout,
+            FlowState::IcmpNeighborAdvert => CloseType::IcmpNormal,
+            FlowState::IcmpNeighborSolicit => CloseType::IcmpNeighborTimeout,
+            FlowState::IcmpPacketTooBig => CloseType::IcmpPacketTooBig,
+            FlowState::IcmpParameterProblem => CloseType::IcmpParameterProblem,
+            FlowState::IcmpRedirectMessage => CloseType::IcmpNormal,
+            FlowState::IcmpRouterAdvertisement => CloseType::IcmpNormal,
+            FlowState::IcmpRouterSolicitation => CloseType::IcmpRouterTimeout,
+            FlowState::IcmpSourceQuench => CloseType::IcmpSourceQuench,
+            FlowState::IcmpTimeExceeded => CloseType::IcmpTimeExceeded,
+            FlowState::IcmpTimestamp => CloseType::IcmpTimestampTimeout,
+            FlowState::IcmpTimestampReply => CloseType::IcmpNormal,
+            FlowState::IcmpTraceroute => CloseType::IcmpNormal,
+            FlowState::Established => {
+                if self.flow_key.proto == IpProtocol::ICMPV4
+                    || self.flow_key.proto == IpProtocol::ICMPV6
+                {
+                    if self.flow_metrics_peers[0].total_packet_count
+                        != self.flow_metrics_peers[1].total_packet_count
+                    {
+                        CloseType::IcmpOtherTimeout
+                    } else {
+                        CloseType::IcmpNormal
+                    }
+                } else {
+                    CloseType::Timeout
+                }
+            }
             FlowState::ClosingTx1 => CloseType::ServerHalfClose,
             FlowState::ClosingRx1 => CloseType::ClientHalfClose,
             FlowState::ClosingTx2 | FlowState::ClosingRx2 | FlowState::Closed => CloseType::TcpFin,
diff --git a/agent/src/flow_generator/flow_config.rs b/agent/src/flow_generator/flow_config.rs
@@ -26,6 +26,7 @@ pub const TIMEOUT_OTHERS: Timestamp = Timestamp::from_secs(5);
 pub const TIMEOUT_ESTABLISHED: Timestamp = Timestamp::from_secs(300);
 pub const TIMEOUT_CLOSING: Timestamp = Timestamp::from_secs(35);
 pub const TIMEOUT_OPENING_RST: Timestamp = Timestamp::from_secs(1);
+pub const TIMEOUT_ICMP: Timestamp = Timestamp::from_secs(5);
 
 pub struct TcpTimeout {
     pub established: Timestamp,
@@ -55,6 +56,7 @@ pub struct FlowTimeout {
     pub closed_fin: Timestamp,
     pub single_direction: Timestamp,
     pub opening_rst: Timestamp,
+    pub icmp_timeout: Timestamp,
 
     pub min: Timestamp,
     pub max: Timestamp, // time window
@@ -71,6 +73,8 @@ impl From<TcpTimeout> for FlowTimeout {
             closed_fin: Timestamp::from_secs(2),
             single_direction: t.others,
             opening_rst: t.opening_rst,
+            icmp_timeout: TIMEOUT_ICMP,
+
             min: Timestamp::from_secs(0),
             max: Timestamp::from_secs(0),
         };
diff --git a/agent/src/flow_generator/flow_map.rs b/agent/src/flow_generator/flow_map.rs
@@ -915,7 +915,12 @@ impl FlowMap {
     ) -> bool {
         self.update_flow(config, node, meta_packet);
         let peers = &node.tagged_flow.flow.flow_metrics_peers;
-        if peers[FLOW_METRICS_PEER_SRC].packet_count > 0
+        if node.tagged_flow.flow.flow_key.proto == IpProtocol::ICMPV4
+            || node.tagged_flow.flow.flow_key.proto == IpProtocol::ICMPV6
+        {
+            node.timeout = config.flow.flow_timeout.icmp_timeout;
+            node.set_icmp_flow_state(meta_packet);
+        } else if peers[FLOW_METRICS_PEER_SRC].packet_count > 0
             && peers[FLOW_METRICS_PEER_DST].packet_count > 0
         {
             node.timeout = config.flow.flow_timeout.established_rst;
@@ -1732,9 +1737,15 @@ impl FlowMap {
         meta_packet.flow_id = node.tagged_flow.flow.flow_id;
         meta_packet.second_in_minute =
             (node.tagged_flow.flow.start_time.as_secs() % SECONDS_IN_MINUTE) as u8;
-        node.flow_state = FlowState::Established;
-        // opening timeout
-        node.timeout = config.flow.flow_timeout.opening;
+        if meta_packet.lookup_key.proto == IpProtocol::ICMPV4
+            || meta_packet.lookup_key.proto == IpProtocol::ICMPV6
+        {
+            node.set_icmp_flow_state(meta_packet);
+            node.timeout = config.flow.flow_timeout.icmp_timeout;
+        } else {
+            node.flow_state = FlowState::Established;
+            node.timeout = config.flow.flow_timeout.opening;
+        }
         if let Some(meta_flow_log) = node.meta_flow_log.as_mut() {
             let _ = meta_flow_log.parse_l3(meta_packet);
         }
diff --git a/agent/src/flow_generator/flow_node.rs b/agent/src/flow_generator/flow_node.rs
@@ -16,18 +16,23 @@
 
 use std::{net::IpAddr, sync::Arc};
 
+use pnet::packet::{
+    icmp::{IcmpType, IcmpTypes},
+    icmpv6::{Icmpv6Type, Icmpv6Types},
+};
+
 use super::{perf::FlowLog, FlowState, FLOW_METRICS_PEER_DST, FLOW_METRICS_PEER_SRC};
 use crate::common::{
     decapsulate::TunnelType,
     endpoint::EndpointDataPov,
     enums::{EthernetType, TapType, TcpFlags},
     flow::{FlowMetricsPeer, L7PerfStats, PacketDirection, SignalSource, TcpPerfStats},
     lookup_key::LookupKey,
-    meta_packet::MetaPacket,
+    meta_packet::{MetaPacket, ProtocolData},
     tagged_flow::TaggedFlow,
     TapPort, Timestamp,
 };
-use public::{proto::common::TridentType, utils::net::MacAddr};
+use public::{enums::IpProtocol, proto::common::TridentType, utils::net::MacAddr};
 
 use npb_pcap_policy::PolicyData;
 use packet_sequence_block::PacketSequenceBlock;
@@ -145,6 +150,65 @@ pub struct FlowNode {
 }
 
 impl FlowNode {
+    pub(super) fn set_icmpv4_flow_state(&mut self, meta_packet: &MetaPacket) {
+        let ProtocolData::IcmpData(icmp_data) = &meta_packet.protocol_data else {
+            return;
+        };
+        let icmp_type = IcmpType::new(icmp_data.icmp_type);
+        match icmp_type {
+            IcmpTypes::AddressMaskReply => self.flow_state = FlowState::IcmpAddressMaskReply,
+            IcmpTypes::AddressMaskRequest => self.flow_state = FlowState::IcmpAddressMaskRequest,
+            IcmpTypes::DestinationUnreachable => {
+                self.flow_state = FlowState::IcmpDestinationUnreachable
+            }
+            IcmpTypes::EchoReply => self.flow_state = FlowState::IcmpEchoReply,
+            IcmpTypes::EchoRequest => self.flow_state = FlowState::IcmpEchoRequest,
+            IcmpTypes::InformationReply => self.flow_state = FlowState::IcmpInformationReply,
+            IcmpTypes::InformationRequest => self.flow_state = FlowState::IcmpInformationRequest,
+            IcmpTypes::ParameterProblem => self.flow_state = FlowState::IcmpParameterProblem,
+            IcmpTypes::RedirectMessage => self.flow_state = FlowState::IcmpRedirectMessage,
+            IcmpTypes::RouterAdvertisement => self.flow_state = FlowState::IcmpRouterAdvertisement,
+            IcmpTypes::RouterSolicitation => self.flow_state = FlowState::IcmpRouterSolicitation,
+            IcmpTypes::SourceQuench => self.flow_state = FlowState::IcmpSourceQuench,
+            IcmpTypes::TimeExceeded => self.flow_state = FlowState::IcmpTimeExceeded,
+            IcmpTypes::Timestamp => self.flow_state = FlowState::IcmpTimestamp,
+            IcmpTypes::TimestampReply => self.flow_state = FlowState::IcmpTimestampReply,
+            IcmpTypes::Traceroute => self.flow_state = FlowState::IcmpTraceroute,
+            _ => self.flow_state = FlowState::Established,
+        }
+    }
+
+    pub(super) fn set_icmpv6_flow_state(&mut self, meta_packet: &MetaPacket) {
+        let ProtocolData::IcmpData(icmp_data) = &meta_packet.protocol_data else {
+            return;
+        };
+        let icmp_type = Icmpv6Type::new(icmp_data.icmp_type);
+        match icmp_type {
+            Icmpv6Types::DestinationUnreachable => {
+                self.flow_state = FlowState::IcmpDestinationUnreachable
+            }
+            Icmpv6Types::EchoReply => self.flow_state = FlowState::IcmpEchoReply,
+            Icmpv6Types::EchoRequest => self.flow_state = FlowState::IcmpEchoRequest,
+            Icmpv6Types::NeighborAdvert => self.flow_state = FlowState::IcmpNeighborAdvert,
+            Icmpv6Types::NeighborSolicit => self.flow_state = FlowState::IcmpNeighborSolicit,
+            Icmpv6Types::PacketTooBig => self.flow_state = FlowState::IcmpPacketTooBig,
+            Icmpv6Types::ParameterProblem => self.flow_state = FlowState::IcmpParameterProblem,
+            Icmpv6Types::Redirect => self.flow_state = FlowState::IcmpRedirectMessage,
+            Icmpv6Types::RouterAdvert => self.flow_state = FlowState::IcmpRouterAdvertisement,
+            Icmpv6Types::RouterSolicit => self.flow_state = FlowState::IcmpRouterSolicitation,
+            Icmpv6Types::TimeExceeded => self.flow_state = FlowState::IcmpTimeExceeded,
+            _ => self.flow_state = FlowState::Established,
+        }
+    }
+
+    pub(super) fn set_icmp_flow_state(&mut self, meta_packet: &MetaPacket) {
+        match meta_packet.lookup_key.proto {
+            IpProtocol::ICMPV4 => self.set_icmpv4_flow_state(meta_packet),
+            IpProtocol::ICMPV6 => self.set_icmpv6_flow_state(meta_packet),
+            _ => unreachable!(),
+        }
+    }
+
     pub(super) fn reset_flow_stat_info(&mut self) {
         self.policy_in_tick = [false; 2];
         self.packet_in_tick = false;
diff --git a/agent/src/flow_generator/flow_state.rs b/agent/src/flow_generator/flow_state.rs
@@ -44,6 +44,26 @@ pub enum FlowState {
     EstablishReset,
     OpeningRst,
 
+    IcmpAddressMaskReply,
+    IcmpAddressMaskRequest,
+    IcmpDestinationUnreachable,
+    IcmpEchoReply,
+    IcmpEchoRequest,
+    IcmpInformationReply,
+    IcmpInformationRequest,
+    IcmpParameterProblem,
+    IcmpRedirectMessage,
+    IcmpRouterAdvertisement,
+    IcmpRouterSolicitation,
+    IcmpSourceQuench,
+    IcmpTimeExceeded,
+    IcmpTimestamp,
+    IcmpTimestampReply,
+    IcmpTraceroute,
+    IcmpNeighborAdvert,
+    IcmpNeighborSolicit,
+    IcmpPacketTooBig,
+
     Max,
 }
 
diff --git a/server/ingester/flow_log/log_data/l4_flow_log.go b/server/ingester/flow_log/log_data/l4_flow_log.go
@@ -858,11 +858,11 @@ func (k *KnowledgeGraph) FillL4(f *pb.Flow, isIPv6 bool, platformData *grpc.Plat
 
 func getStatus(t datatype.CloseType, p layers.IPProtocol) datatype.LogMessageStatus {
 	if t == datatype.CloseTypeTCPFin || t == datatype.CloseTypeForcedReport || t == datatype.CloseTypeTCPFinClientRst ||
-		(p != layers.IPProtocolTCP && t == datatype.CloseTypeTimeout) {
+		(p != layers.IPProtocolTCP && t == datatype.CloseTypeTimeout) || t == atatype.CloseTypeIcmpNormal {
 		return datatype.STATUS_OK
 	} else if t.IsClientError() {
 		return datatype.STATUS_CLIENT_ERROR
-	} else if p == layers.IPProtocolTCP && t.IsServerError() {
+	} else if (p == layers.IPProtocolTCP || p == layers.IPProtocolICMPv4 || p == layers.IPProtocolICMPv6) && t.IsServerError() {
 		return datatype.STATUS_SERVER_ERROR
 	} else {
 		return datatype.STATUS_NOT_EXIST
diff --git a/server/libs/datatype/flow.go b/server/libs/datatype/flow.go
@@ -49,26 +49,40 @@ const (
 	//	ANOMALY_CLIENT_RST_FLOW: {"client_rst_flow", "传输-客户端重置"},
 	//	...
 
-	CloseTypeTCPFin                //  1: 正常结束
-	CloseTypeTCPServerRst          //  2: 传输-服务端重置
-	CloseTypeTimeout               //  3: 连接超时
-	_                              //  4: 【废弃】CloseTypeFlood
-	CloseTypeForcedReport          //  5: 周期性上报
-	_                              //  6: 【废弃】CloseTypeFoecedClose
-	CloseTypeClientSYNRepeat       //  7: 建连-客户端SYN结束
-	CloseTypeServerHalfClose       //  8: 断连-服务端半关
-	CloseTypeTCPClientRst          //  9: 传输-客户端重置
-	CloseTypeServerSYNACKRepeat    // 10: 建连-服务端SYN结束
-	CloseTypeClientHalfClose       // 11: 断连-客户端半关
-	_                              // 12: 【废弃】CloseTypeClientNoResponse
-	CloseTypeClientSourcePortReuse // 13: 建连-客户端端口复用
-	_                              // 14: 【废弃】CloseTypeClientSYNRetryLack
-	CloseTypeServerReset           // 15: 建连-服务端直接重置
-	_                              // 16: 【废弃】CloseTypeServerNoResponse
-	CloseTypeServerQueueLack       // 17: 传输-服务端队列溢出
-	CloseTypeClientEstablishReset  // 18: 建连-客户端其他重置
-	CloseTypeServerEstablishReset  // 19: 建连-服务端其他重置
-	CloseTypeTCPFinClientRst       // 20: 正常结束-客户端重置
+	CloseTypeTCPFin                     //  1: 正常结束
+	CloseTypeTCPServerRst               //  2: 传输-服务端重置
+	CloseTypeTimeout                    //  3: 连接超时
+	_                                   //  4: 【废弃】CloseTypeFlood
+	CloseTypeForcedReport               //  5: 周期性上报
+	_                                   //  6: 【废弃】CloseTypeFoecedClose
+	CloseTypeClientSYNRepeat            //  7: 建连-客户端SYN结束
+	CloseTypeServerHalfClose            //  8: 断连-服务端半关
+	CloseTypeTCPClientRst               //  9: 传输-客户端重置
+	CloseTypeServerSYNACKRepeat         // 10: 建连-服务端SYN结束
+	CloseTypeClientHalfClose            // 11: 断连-客户端半关
+	_                                   // 12: 【废弃】CloseTypeClientNoResponse
+	CloseTypeClientSourcePortReuse      // 13: 建连-客户端端口复用
+	_                                   // 14: 【废弃】CloseTypeClientSYNRetryLack
+	CloseTypeServerReset                // 15: 建连-服务端直接重置
+	_                                   // 16: 【废弃】CloseTypeServerNoResponse
+	CloseTypeServerQueueLack            // 17: 传输-服务端队列溢出
+	CloseTypeClientEstablishReset       // 18: 建连-客户端其他重置
+	CloseTypeServerEstablishReset       // 19: 建连-服务端其他重置
+	CloseTypeTCPFinClientRst            // 20: 正常结束-客户端重置
+	CloseTypeIcmpNormal                 // 21: ICMP-正常结束
+	CloseTypeIcmpAddressMaskTimeout     // 22: ICMP-地址掩码超时
+	CloseTypeIcmpDestinationUnreachable // 23: ICMP-目标不可达
+	CloseTypeIcmpEchoTimeout            // 24: ICMP-PING超时
+	CloseTypeIcmpInformationTimeout     // 25: ICMP-信息超时
+	CloseTypeIcmpParameterProblem       // 26: ICMP-参数问题
+	CloseTypeIcmpRouterTimeout          // 27: ICMP-路由器超时
+	CloseTypeIcmpSourceQuench           // 28: ICMP-源站抑制
+	CloseTypeIcmpTimeExceeded           // 29: ICMP-TTL溢出
+	CloseTypeIcmpTimestampTimeout       // 30: ICMP-时间戳超时
+	CloseTypeIcmpNeighborTimeout        // 31: ICMP-邻居发现超时
+	CloseTypeIcmpPacketTooBig           // 32: ICMP-Packet过大
+	CloseTypeIcmpOtherTimeout           // 33: ICMP-未知超时
+
 	MaxCloseType
 )
 
@@ -81,7 +95,13 @@ func (t CloseType) IsClientError() bool {
 func (t CloseType) IsServerError() bool {
 	return t == CloseTypeTCPServerRst || t == CloseTypeTimeout ||
 		t == CloseTypeServerHalfClose || t == CloseTypeServerSYNACKRepeat ||
-		t == CloseTypeServerReset || t == CloseTypeServerQueueLack || t == CloseTypeServerEstablishReset
+		t == CloseTypeServerReset || t == CloseTypeServerQueueLack || t == CloseTypeServerEstablishReset ||
+		t == CloseTypeIcmpAddressMaskTimeout || t == CloseTypeIcmpDestinationUnreachable ||
+		t == CloseTypeIcmpEchoTimeout || t == CloseTypeIcmpInformationTimeout ||
+		t == CloseTypeIcmpParameterProblem || t == CloseTypeIcmpRouterTimeout ||
+		t == CloseTypeIcmpSourceQuench || t == CloseTypeIcmpTimeExceeded ||
+		t == CloseTypeIcmpTimestampTimeout || t == CloseTypeIcmpNeighborTimeout ||
+		t == CloseTypeIcmpPacketTooBig || t == CloseTypeIcmpOtherTimeout
 }
 
 type DeviceType uint8
diff --git a/server/querier/db_descriptions/clickhouse/tag/enum/close_type.ch b/server/querier/db_descriptions/clickhouse/tag/enum/close_type.ch
@@ -15,3 +15,16 @@
 11      , 断连-客户端半关       ,
 8       , 断连-服务端半关       ,
 20      , 正常结束-客户端重置   ,
+21      , ICMP-正常结束         ,
+22      , ICMP-地址掩码超时     ,
+23      , ICMP-目标不可达       ,
+24      , ICMP-PING超时         ,
+25      , ICMP-信息超时         ,
+26      , ICMP-参数问题         ,
+27      , ICMP-路由器超时       ,
+28      , ICMP-源站抑制         ,
+29      , ICMP-TTL溢出          ,
+30      , ICMP-时间戳超时       ,
+31      , ICMP-邻居发现超时     ,
+32      , ICMP-Packet过大       ,
+33      , ICMP-未知超时         ,
diff --git a/server/querier/db_descriptions/clickhouse/tag/enum/close_type.en b/server/querier/db_descriptions/clickhouse/tag/enum/close_type.en
