nfs: Implement frames

Feature #4872

Frames:
  - RPC Frames: Generic over TCP/UDP
     - rpc.pdu
     - rpc.hdr
     - rpc.data
     - rpc.creds -- for rpc calls

  - NFSv2, NFSv3
     - nfs.pdu
     - nfs.status -- for nfs responses

  - NFSv4 Only Frames
     - nfs4.pdu
     - nfs4.hdr
     - nfs4.ops -- for compound request/response operations
     - nfs4.status -- for nfs4 responses

RPC tcp/udp frames created with separate registeration functions e.g:
add_rpc_tcp_tc_frames()
add_rpc_udp_tc_frames()

rust/src/nfs/nfs.rs

@@ -26,6 +26,7 @@ use nom7::{Err, Needed};
 
 use crate::applayer;
 use crate::applayer::*;
+use crate::frames::*;
 use crate::core::*;
 use crate::conf::*;
 use crate::filetracker::*;
@@ -43,6 +44,9 @@ pub const NFS_MIN_FRAME_LEN: u16 = 32;
 
 static mut NFS_MAX_TX: usize = 1024;
 
+pub const RPC_TCP_PRE_CREDS: usize = 28;
+pub const RPC_UDP_PRE_CREDS: usize = 24;
+
 static mut ALPROTO_NFS: AppProto = ALPROTO_UNKNOWN;
 /*
  * Record parsing.
@@ -87,6 +91,22 @@ static mut ALPROTO_NFS: AppProto = ALPROTO_UNKNOWN;
  * Transaction lookup.
  */
 
+#[derive(AppLayerFrameType)]
+pub enum NFSFrameType {
+    RPCPdu,
+    RPCHdr,
+    RPCData,
+    RPCCreds, // for rpc calls | rpc.creds [creds_flavor + creds_len + creds]
+
+    NFSPdu,
+    NFSStatus,
+
+    NFS4Pdu,
+    NFS4Hdr,
+    NFS4Ops,
+    NFS4Status,
+}
+
 #[derive(FromPrimitive, Debug, AppLayerEvent)]
 pub enum NFSEvent {
     MalformedData = 0,
@@ -437,6 +457,98 @@ impl NFSState {
         }
     }
 
+    fn add_rpc_udp_ts_pdu(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64) -> Option<Frame> {
+        let rpc_udp_ts_pdu = Frame::new_ts(flow, stream_slice, input, rpc_len, NFSFrameType::RPCPdu as u8);
+        SCLogDebug!("rpc_udp_pdu ts frame {:?}", rpc_udp_ts_pdu);
+        rpc_udp_ts_pdu
+    }
+
+    fn add_rpc_udp_ts_creds(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], creds_len: i64) {
+        let _rpc_udp_ts_creds = Frame::new_ts(flow, stream_slice, input, creds_len, NFSFrameType::RPCCreds as u8);
+        SCLogDebug!("rpc_creds ts frame {:?}", _rpc_udp_ts_creds);
+    }
+
+    fn add_rpc_tcp_ts_pdu(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64) -> Option<Frame> {
+        let rpc_tcp_ts_pdu = Frame::new_ts(flow, stream_slice, input, rpc_len, NFSFrameType::RPCPdu as u8);
+        SCLogDebug!("rpc_tcp_pdu ts frame {:?}", rpc_tcp_ts_pdu);
+        rpc_tcp_ts_pdu
+    }
+
+    fn add_rpc_tcp_ts_creds(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], creds_len: i64) {
+        let _rpc_tcp_ts_creds = Frame::new_ts(flow, stream_slice, input, creds_len, NFSFrameType::RPCCreds as u8);
+        SCLogDebug!("rpc_tcp_ts_creds {:?}", _rpc_tcp_ts_creds);
+    }
+
+    fn add_nfs_ts_frame(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], nfs_len: i64) {
+        let _nfs_req_pdu = Frame::new_ts(flow, stream_slice, input, nfs_len, NFSFrameType::NFSPdu as u8);
+        SCLogDebug!("nfs_ts_pdu Frame {:?}", _nfs_req_pdu);
+    }
+
+    fn add_nfs4_ts_frames(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], nfs4_len: i64) {
+        let _nfs4_ts_pdu = Frame::new_ts(flow, stream_slice, input, nfs4_len, NFSFrameType::NFS4Pdu as u8);
+        SCLogDebug!("nfs4_ts_pdu Frame: {:?}", _nfs4_ts_pdu);
+        if nfs4_len > 8 {
+            let _nfs4_ts_hdr = Frame::new_ts(flow, stream_slice, input, 8, NFSFrameType::NFS4Hdr as u8);
+            SCLogDebug!("nfs4_ts_hdr Frame {:?}", _nfs4_ts_hdr);
+            let _nfs4_ts_ops = Frame::new_ts(flow, stream_slice, &input[8..], nfs4_len - 8, NFSFrameType::NFS4Ops as u8);
+            SCLogDebug!("nfs4_ts_ops Frame {:?}", _nfs4_ts_ops);
+        }
+    }
+
+    fn add_rpc_udp_tc_pdu(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64) -> Option<Frame> {
+        let rpc_udp_tc_pdu = Frame::new_ts(flow, stream_slice, input, rpc_len, NFSFrameType::RPCPdu as u8);
+        SCLogDebug!("rpc_tc_pdu frame {:?}", rpc_udp_tc_pdu);
+        rpc_udp_tc_pdu
+    }
+
+    fn add_rpc_udp_tc_frames(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64) {
+        if rpc_len > 8 {
+            let _rpc_udp_tc_hdr = Frame::new_ts(flow, stream_slice, input, 8, NFSFrameType::RPCHdr as u8);
+            let _rpc_udp_tc_data = Frame::new_ts(flow, stream_slice, &input[8..], rpc_len - 8, NFSFrameType::RPCData as u8);
+            SCLogDebug!("rpc_udp_tc_hdr frame {:?}", _rpc_udp_tc_hdr);
+            SCLogDebug!("rpc_udp_tc_data frame {:?}", _rpc_udp_tc_data);
+        }
+    }
+
+    fn add_rpc_tcp_tc_pdu(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_tcp_len: i64) -> Option<Frame> {
+        let rpc_tcp_tc_pdu = Frame::new_tc(flow, stream_slice, input, rpc_tcp_len, NFSFrameType::RPCPdu as u8);
+        SCLogDebug!("rpc_tcp_pdu tc frame {:?}", rpc_tcp_tc_pdu);
+        rpc_tcp_tc_pdu
+    }
+
+    fn add_rpc_tcp_tc_frames(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], rpc_tcp_len: i64) {
+        if rpc_tcp_len > 12 {
+            let _rpc_tcp_tc_hdr = Frame::new_tc(flow, stream_slice, input, 12, NFSFrameType::RPCHdr as u8);
+            let _rpc_tcp_tc_data = Frame::new_tc(flow, stream_slice, &input[12..], rpc_tcp_len - 12, NFSFrameType::RPCData as u8);
+            SCLogDebug!("rpc_tcp_tc_hdr frame {:?}", _rpc_tcp_tc_hdr);
+            SCLogDebug!("rpc_tcp_tc_data frame {:?}", _rpc_tcp_tc_data);
+        }
+    }
+
+    fn add_nfs_tc_frames(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], nfs_len: i64) {
+        if nfs_len > 0 {
+            let _nfs_tc_pdu = Frame::new_tc(flow, stream_slice, input, nfs_len, NFSFrameType::NFSPdu as u8);
+            SCLogDebug!("nfs_tc_pdu frame {:?}", _nfs_tc_pdu);
+            let _nfs_res_status = Frame::new_tc(flow, stream_slice, input, 4, NFSFrameType::NFSStatus as u8);
+            SCLogDebug!("nfs_tc_status frame {:?}", _nfs_res_status);
+        }
+    }
+
+    fn add_nfs4_tc_frames(&mut self, flow: *const Flow, stream_slice: &StreamSlice, input: &[u8], nfs4_len: i64) {
+        if nfs4_len > 0 {
+            let _nfs4_tc_pdu = Frame::new_tc(flow, stream_slice, input, nfs4_len, NFSFrameType::NFS4Pdu as u8);
+            SCLogDebug!("nfs4_tc_pdu frame {:?}", _nfs4_tc_pdu);
+            let _nfs4_tc_status = Frame::new_tc(flow, stream_slice, input, 4, NFSFrameType::NFS4Status as u8);
+            SCLogDebug!("nfs4_tc_status frame {:?}", _nfs4_tc_status);
+        }
+        if nfs4_len > 8 {
+            let _nfs4_tc_hdr = Frame::new_tc(flow, stream_slice, input, 8, NFSFrameType::NFS4Hdr as u8);
+            SCLogDebug!("nfs4_tc_hdr frame {:?}", _nfs4_tc_hdr);
+            let _nfs4_tc_ops = Frame::new_tc(flow, stream_slice, &input[8..], nfs4_len - 8, NFSFrameType::NFS4Ops as u8);
+            SCLogDebug!("nfs4_tc_ops frame {:?}", _nfs4_tc_ops);
+        }
+    }
+
     fn post_gap_housekeeping_for_files(&mut self)
     {
         let mut post_gap_txs = false;
@@ -530,18 +642,21 @@ impl NFSState {
     }
 
     /// complete request record
-    fn process_request_record<'b>(&mut self, r: &RpcPacket<'b>) {
+    fn process_request_record<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice, r: &RpcPacket<'b>) {
         SCLogDebug!("REQUEST {} procedure {} ({}) blob size {}",
                 r.hdr.xid, r.procedure, self.requestmap.len(), r.prog_data.len());
 
         match r.progver {
             4 => {
+                self.add_nfs4_ts_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_request_record_v4(r)
             },
             3 => {
+                self.add_nfs_ts_frame(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_request_record_v3(r)
             },
             2 => {
+                self.add_nfs_ts_frame(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_request_record_v2(r)
             },
             _ => { },
@@ -669,7 +784,7 @@ impl NFSState {
         return self.process_write_record(r, w);
     }
 
-    fn process_reply_record<'b>(&mut self, r: &RpcReplyPacket<'b>) -> u32 {
+    fn process_reply_record<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice, r: &RpcReplyPacket<'b>) -> u32 {
         let mut xidmap;
         match self.requestmap.remove(&r.hdr.xid) {
             Some(p) => { xidmap = p; },
@@ -692,16 +807,19 @@ impl NFSState {
         match xidmap.progver {
             2 => {
                 SCLogDebug!("NFSv2 reply record");
+                self.add_nfs_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_reply_record_v2(r, &xidmap);
                 return 0;
             },
             3 => {
                 SCLogDebug!("NFSv3 reply record");
+                self.add_nfs_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_reply_record_v3(r, &mut xidmap);
                 return 0;
             },
             4 => {
                 SCLogDebug!("NFSv4 reply record");
+                self.add_nfs4_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                 self.process_reply_record_v4(r, &mut xidmap);
                 return 0;
             },
@@ -1051,8 +1169,8 @@ impl NFSState {
     }
 
     /// Parsing function, handling TCP chunks fragmentation
-    pub fn parse_tcp_data_ts<'b>(&mut self, i: &'b[u8]) -> AppLayerResult {
+    pub fn parse_tcp_data_ts<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice) -> AppLayerResult {
-        let mut cur_i = i;
+        let mut cur_i = stream_slice.as_slice();
         // take care of in progress file chunk transfers
         // and skip buffer beyond it
         let consumed = self.filetracker_update(Direction::ToServer, cur_i, 0);
@@ -1080,7 +1198,7 @@ impl NFSState {
                     0 => {
                         SCLogDebug!("incomplete, queue and retry with the next block (input {}). Looped {} times.",
                                 cur_i.len(), _cnt);
-                        return AppLayerResult::incomplete((i.len() - cur_i.len()) as u32, (cur_i.len() + 1) as u32);
+                        return AppLayerResult::incomplete(stream_slice.len() - cur_i.len() as u32, (cur_i.len() + 1) as u32);
                     },
                     -1 => {
                         cur_i = &cur_i[1..];
@@ -1097,13 +1215,14 @@ impl NFSState {
         }
 
         while cur_i.len() > 0 { // min record size
+            self.add_rpc_tcp_ts_pdu(flow, stream_slice, cur_i, cur_i.len() as i64);
             match parse_rpc_request_partial(cur_i) {
                 Ok((_, ref rpc_phdr)) => {
                     let rec_size = (rpc_phdr.hdr.frag_len + 4) as usize;
 
                     // Handle partial records
                     if rec_size > cur_i.len() {
-                        return self.parse_tcp_partial_data_ts(i, cur_i, rpc_phdr, rec_size);
+                        return self.parse_tcp_partial_data_ts(stream_slice.as_slice(), cur_i, rpc_phdr, rec_size);
                     }
 
                     // we have the full records size worth of data,
@@ -1112,7 +1231,8 @@ impl NFSState {
                     // go to the next record.
                     match parse_rpc(cur_i, true) {
                         Ok((_, ref rpc_record)) => {
-                            self.process_request_record(rpc_record);
+                            self.add_rpc_tcp_ts_creds(flow, stream_slice, &cur_i[RPC_TCP_PRE_CREDS..], (rpc_record.creds_len + 8) as i64);
+                            self.process_request_record(flow, stream_slice, rpc_record);
                         }
                         Err(Err::Incomplete(_)) => {
                             self.set_event(NFSEvent::MalformedData);
@@ -1132,7 +1252,7 @@ impl NFSState {
                         // looks for.
                         let n = usize::from(n);
                         let need = if n > 28 { n } else { 28 };
-                        return AppLayerResult::incomplete((i.len() - cur_i.len()) as u32, need as u32);
+                        return AppLayerResult::incomplete(stream_slice.len() - cur_i.len() as u32, need as u32);
                     }
                     return AppLayerResult::err();
                 }
@@ -1213,8 +1333,8 @@ impl NFSState {
     }
 
     /// Parsing function, handling TCP chunks fragmentation
-    pub fn parse_tcp_data_tc<'b>(&mut self, i: &'b[u8]) -> AppLayerResult {
+    pub fn parse_tcp_data_tc<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice) -> AppLayerResult {
-        let mut cur_i = i;
+        let mut cur_i = stream_slice.as_slice();
         // take care of in progress file chunk transfers
         // and skip buffer beyond it
         let consumed = self.filetracker_update(Direction::ToClient, cur_i, 0);
@@ -1242,7 +1362,7 @@ impl NFSState {
                     0 => {
                         SCLogDebug!("incomplete, queue and retry with the next block (input {}). Looped {} times.",
                                 cur_i.len(), _cnt);
-                        return AppLayerResult::incomplete((i.len() - cur_i.len()) as u32, (cur_i.len() + 1) as u32);
+                        return AppLayerResult::incomplete(stream_slice.len() - cur_i.len() as u32, (cur_i.len() + 1) as u32);
                     },
                     -1 => {
                         cur_i = &cur_i[1..];
@@ -1259,18 +1379,20 @@ impl NFSState {
         }
 
         while cur_i.len() > 0 {
+            self.add_rpc_tcp_tc_pdu(flow, stream_slice, cur_i, cur_i.len() as i64);
             match parse_rpc_packet_header(cur_i) {
                 Ok((_, ref rpc_phdr)) => {
                     let rec_size = (rpc_phdr.frag_len + 4) as usize;
                     // see if we have all data available
                     if rec_size > cur_i.len() {
-                        return self.parse_tcp_partial_data_tc(i, cur_i, rpc_phdr, rec_size);
+                        return self.parse_tcp_partial_data_tc(stream_slice.as_slice(), cur_i, rpc_phdr, rec_size);
                     }
 
                     // we have the full data of the record, lets parse
                     match parse_rpc_reply(cur_i, true) {
                         Ok((_, ref rpc_record)) => {
-                            self.process_reply_record(rpc_record);
+                            self.add_rpc_tcp_tc_frames(flow, stream_slice, cur_i, cur_i.len() as i64);
+                            self.process_reply_record(flow, stream_slice, rpc_record);
                         }
                         Err(Err::Incomplete(_)) => {
                             // we shouldn't get incomplete as we have the full data
@@ -1293,7 +1415,7 @@ impl NFSState {
                         // looks for.
                         let n = usize::from(n);
                         let need = if n > 12 { n } else { 12 };
-                        return AppLayerResult::incomplete((i.len() - cur_i.len()) as u32, need as u32);
+                        return AppLayerResult::incomplete(stream_slice.len() - cur_i.len() as u32, need as u32);
                     }
                     return AppLayerResult::err();
                 }
@@ -1315,17 +1437,21 @@ impl NFSState {
         AppLayerResult::ok()
     }
     /// Parsing function
-    pub fn parse_udp_ts<'b>(&mut self, input: &'b[u8]) -> AppLayerResult {
+    pub fn parse_udp_ts<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice) -> AppLayerResult {
+        let input = stream_slice.as_slice();
         SCLogDebug!("parse_udp_ts ({})", input.len());
+        self.add_rpc_udp_ts_pdu(flow, stream_slice, input, input.len() as i64);
         if input.len() > 0 {
             match parse_rpc_udp_request(input) {
                 Ok((_, ref rpc_record)) => {
                     self.is_udp = true;
+                    self.add_rpc_udp_ts_creds(flow, stream_slice, &input[RPC_UDP_PRE_CREDS..], (rpc_record.creds_len + 8) as i64);
                     match rpc_record.progver {
                         3 => {
-                            self.process_request_record(rpc_record);
+                            self.process_request_record(flow, stream_slice, rpc_record);
                         },
                         2 => {
+                            self.add_nfs_ts_frame(flow, stream_slice, rpc_record.prog_data, rpc_record.prog_data_size as i64);
                             self.process_request_record_v2(rpc_record);
                         },
                         _ => { },
@@ -1343,13 +1469,16 @@ impl NFSState {
     }
 
     /// Parsing function
-    pub fn parse_udp_tc<'b>(&mut self, input: &'b[u8]) -> AppLayerResult {
+    pub fn parse_udp_tc<'b>(&mut self, flow: *const Flow, stream_slice: &StreamSlice) -> AppLayerResult {
+        let input = stream_slice.as_slice();
         SCLogDebug!("parse_udp_tc ({})", input.len());
+        self.add_rpc_udp_tc_pdu(flow, stream_slice, input, input.len() as i64);
         if input.len() > 0 {
             match parse_rpc_udp_reply(input) {
                 Ok((_, ref rpc_record)) => {
                     self.is_udp = true;
-                    self.process_reply_record(rpc_record);
+                    self.add_rpc_udp_tc_frames(flow, stream_slice, input, input.len() as i64);
+                    self.process_reply_record(flow, stream_slice, rpc_record);
                 },
                 Err(Err::Incomplete(_)) => {
                 },
@@ -1418,7 +1547,7 @@ pub unsafe extern "C" fn rs_nfs_parse_request(flow: *const Flow,
     SCLogDebug!("parsing {} bytes of request data", stream_slice.len());
 
     state.update_ts(flow.get_last_time().as_secs());
-    state.parse_tcp_data_ts(stream_slice.as_slice())
+    state.parse_tcp_data_ts(flow, &stream_slice)
 }
 
 #[no_mangle]
@@ -1449,7 +1578,7 @@ pub unsafe extern "C" fn rs_nfs_parse_response(flow: *const Flow,
     SCLogDebug!("parsing {} bytes of response data", stream_slice.len());
 
     state.update_ts(flow.get_last_time().as_secs());
-    state.parse_tcp_data_tc(stream_slice.as_slice())
+    state.parse_tcp_data_tc(flow, &stream_slice)
 }
 
 #[no_mangle]
@@ -1475,7 +1604,7 @@ pub unsafe extern "C" fn rs_nfs_parse_request_udp(f: *const Flow,
     rs_nfs_setfileflags(Direction::ToServer.into(), state, file_flags);
 
     SCLogDebug!("parsing {} bytes of request data", stream_slice.len());
-    state.parse_udp_ts(stream_slice.as_slice())
+    state.parse_udp_ts(f, &stream_slice)
 }
 
 #[no_mangle]
@@ -1490,7 +1619,7 @@ pub unsafe extern "C" fn rs_nfs_parse_response_udp(f: *const Flow,
     let file_flags = FileFlowToFlags(f, Direction::ToClient.into());
     rs_nfs_setfileflags(Direction::ToClient.into(), state, file_flags);
     SCLogDebug!("parsing {} bytes of response data", stream_slice.len());
-    state.parse_udp_tc(stream_slice.as_slice())
+    state.parse_udp_tc(f, &stream_slice)
 }
 
 #[no_mangle]
@@ -1844,8 +1973,8 @@ pub unsafe extern "C" fn rs_nfs_register_parser() {
         apply_tx_config: None,
         flags: APP_LAYER_PARSER_OPT_ACCEPT_GAPS,
         truncate: None,
-        get_frame_id_by_name: None,
+        get_frame_id_by_name: Some(NFSFrameType::ffi_id_from_name),
-        get_frame_name_by_id: None,
+        get_frame_name_by_id: Some(NFSFrameType::ffi_name_from_id),
     };
 
     let ip_proto_str = CString::new("tcp").unwrap();
@@ -1922,8 +2051,8 @@ pub unsafe extern "C" fn rs_nfs_udp_register_parser() {
         apply_tx_config: None,
         flags: APP_LAYER_PARSER_OPT_UNIDIR_TXS,
         truncate: None,
-        get_frame_id_by_name: None,
+        get_frame_id_by_name: Some(NFSFrameType::ffi_id_from_name),
-        get_frame_name_by_id: None,
+        get_frame_name_by_id: Some(NFSFrameType::ffi_name_from_id),
     };
 
     let ip_proto_str = CString::new("udp").unwrap();