feat: parse web_sphere_mq TSH message

Author: lzf575

agent/crates/enterprise-utils/src/lib.rs

@@ -334,14 +334,21 @@ pub mod l7 {
             pub struct WebSphereMqParser {
                 pub base: L7LogBase,
                 pub orig_send_time: String,
+                pub skip_frame: bool,
             }
 
             impl WebSphereMqParser {
-                pub fn check_payload(&mut self, _: &[u8]) -> Option<LogMessageType> {
+                pub fn check_payload(&mut self, _: &[u8], _: bool) -> Option<LogMessageType> {
                     unimplemented!()
                 }
 
-                pub fn parse_payload(&mut self, _: &[u8], _: PacketDirection, _: bool) -> usize {
+                pub fn parse_payload(
+                    &mut self,
+                    _: &[u8],
+                    _: PacketDirection,
+                    _: bool,
+                    _: bool,
+                ) -> usize {
                     unimplemented!()
                 }
             }

agent/src/config/config.rs

@@ -1834,12 +1834,14 @@ impl Default for Iso8583Config {
 #[serde(default)]
 pub struct WebSphereMqConfig {
     pub parse_xml_enabled: bool,
+    pub decompress_enabled: bool,
 }
 
 impl Default for WebSphereMqConfig {
     fn default() -> Self {
         Self {
             parse_xml_enabled: true,
+            decompress_enabled: true,
         }
     }
 }
@@ -3641,12 +3643,14 @@ impl Default for Iso8583ParseConfig {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct WebSphereMqParseConfig {
     pub parse_xml_enabled: bool,
+    pub decompress_enabled: bool,
 }
 
 impl Default for WebSphereMqParseConfig {
     fn default() -> Self {
         Self {
             parse_xml_enabled: true,
+            decompress_enabled: true,
         }
     }
 }

agent/src/config/handler.rs

@@ -768,6 +768,13 @@ impl From<&UserConfig> for FlowConfig {
                     .protocol_special_config
                     .web_sphere_mq
                     .parse_xml_enabled,
+                decompress_enabled: conf
+                    .processors
+                    .request_log
+                    .application_protocol_inference
+                    .protocol_special_config
+                    .web_sphere_mq
+                    .decompress_enabled,
             },
             obfuscate_enabled_protocols: L7ProtocolBitmap::from(
                 conf.processors

agent/src/flow_generator/protocol_logs/mq/web_sphere_mq.rs

@@ -291,8 +291,9 @@ pub struct WebSphereMqLog {
 }
 
 impl L7ProtocolParserInterface for WebSphereMqLog {
-    fn check_payload(&mut self, payload: &[u8], _param: &ParseParam) -> Option<LogMessageType> {
-        self.parser.check_payload(payload)
+    fn check_payload(&mut self, payload: &[u8], param: &ParseParam) -> Option<LogMessageType> {
+        self.parser
+            .check_payload(payload, param.web_sphere_mq_parse_conf.decompress_enabled)
     }
 
     fn parse_payload(&mut self, payload: &[u8], param: &ParseParam) -> Result<L7ParseResult> {
@@ -303,6 +304,8 @@ impl L7ProtocolParserInterface for WebSphereMqLog {
 
         let custom_policies =
             config.get_custom_field_policies(L7Protocol::WebSphereMq.into(), param);
+        let parse_xml_enabled = param.web_sphere_mq_parse_conf.parse_xml_enabled;
+        let decompress_enabled = param.web_sphere_mq_parse_conf.decompress_enabled;
         let mut pos = 0;
         let mut loop_count = 0;
         let mut results: Vec<L7ProtocolInfo> = Vec::with_capacity(Self::INIT_L7_RESULT_CAPACITY);
@@ -311,11 +314,16 @@ impl L7ProtocolParserInterface for WebSphereMqLog {
             let parsed_size = self.parser.parse_payload(
                 &payload[pos..],
                 param.direction,
-                param.web_sphere_mq_parse_conf.parse_xml_enabled,
+                parse_xml_enabled,
+                decompress_enabled,
             );
             if parsed_size == 0 {
                 break;
             }
+            if self.parser.skip_frame {
+                pos += parsed_size;
+                continue;
+            }
             let mut info = WebSphereMqInfo::default();
             info.is_tls = param.is_tls();
             info.base.is_async = true;

server/agent_config/README-CH.md

@@ -8114,6 +8114,36 @@ processors:
 
 是否解析 XML 数据内容。
 
+###### 解压数据包 {#processors.request_log.application_protocol_inference.protocol_special_config.web_sphere_mq.decompress_enabled}
+
+**标签**:
+
+<mark>agent_restart</mark>
+<mark>ee_feature</mark>
+
+**FQCN**:
+
+`processors.request_log.application_protocol_inference.protocol_special_config.web_sphere_mq.decompress_enabled`
+
+**默认值**:
+```yaml
+processors:
+  request_log:
+    application_protocol_inference:
+      protocol_special_config:
+        web_sphere_mq:
+          decompress_enabled: true
+```
+
+**模式**:
+| Key  | Value                        |
+| ---- | ---------------------------- |
+| Type | bool |
+
+**详细描述**:
+
+部分 web_sphere_mq 的消息中使用 zlib 压缩，开启此选项后，agent 在解析时会对数据包进行解压。
+
 ##### MySQL {#processors.request_log.application_protocol_inference.protocol_special_config.mysql}
 
 ###### 解压 MySQL 数据包 {#processors.request_log.application_protocol_inference.protocol_special_config.mysql.decompress_payload}

server/agent_config/README.md

@@ -8297,6 +8297,37 @@ processors:
 
 Whether to parse XML.
 
+###### Decompress Payload {#processors.request_log.application_protocol_inference.protocol_special_config.web_sphere_mq.decompress_enabled}
+
+**Tags**:
+
+<mark>agent_restart</mark>
+<mark>ee_feature</mark>
+
+**FQCN**:
+
+`processors.request_log.application_protocol_inference.protocol_special_config.web_sphere_mq.decompress_enabled`
+
+**Default value**:
+```yaml
+processors:
+  request_log:
+    application_protocol_inference:
+      protocol_special_config:
+        web_sphere_mq:
+          decompress_enabled: true
+```
+
+**Schema**:
+| Key  | Value                        |
+| ---- | ---------------------------- |
+| Type | bool |
+
+**Description**:
+
+Some web_sphere_mq messages use zlib compression. When this option is enabled,
+the agent will decompress the data packets during parsing.
+
 ##### MySQL {#processors.request_log.application_protocol_inference.protocol_special_config.mysql}
 
 ###### Decompress MySQL Payload {#processors.request_log.application_protocol_inference.protocol_special_config.mysql.decompress_payload}

server/agent_config/template.yaml

@@ -5898,6 +5898,22 @@ processors:
           #   ch: |-
           #     是否解析 XML 数据内容。
           parse_xml_enabled: true
+          # type: bool
+          # name:
+          #   en: Decompress Payload
+          #   ch: 解压数据包
+          # unit:
+          # range: []
+          # enum_options: []
+          # modification: agent_restart
+          # ee_feature: true
+          # description:
+          #   en: |-
+          #     Some web_sphere_mq messages use zlib compression. When this option is enabled,
+          #     the agent will decompress the data packets during parsing.
+          #   ch: |-
+          #     部分 web_sphere_mq 的消息中使用 zlib 压缩，开启此选项后，agent 在解析时会对数据包进行解压。
+          decompress_enabled: true
         # type: section
         # name: MySQL
         # description: