Working SOCKS4(a) TCP session tagger/parser #1048

tap/nzyme-tap.conf.example

@@ -29,7 +29,7 @@ ethernet_broker_buffer_capacity = 65535
 [protocols.tcp]
 pipeline_size = 16384
 reassembly_buffer_size = 1048576
-session_timeout_seconds = 60
+session_timeout_seconds = 86400
 
 [protocols.udp]
 pipeline_size = 16384

tap/src/data/tcp_table.rs

@@ -26,6 +26,7 @@ pub struct TcpTable {
 
 #[derive(Debug)]
 pub struct TcpSession {
+    pub session_key: TcpSessionKey,
     pub state: TcpSessionState,
     pub source_mac: String,
     pub destination_mac: String,
@@ -123,6 +124,7 @@ impl TcpTable {
                         // We only record new sessions, not mid-session.
                         if session_state == SynSent {
                             let new_session = TcpSession {
+                                session_key: segment.session_key.clone(),
                                 state: session_state.clone(),
                                 start_time: segment.timestamp,
                                 end_time: None,
@@ -329,6 +331,7 @@ fn timeout_sweep(sessions: &mut MutexGuard<HashMap<TcpSessionKey, TcpSession>>,
         if Utc::now() - session.most_recent_segment_time
             > Duration::try_seconds(timeout).unwrap() {
             session.state = ClosedTimeout;
+            session.end_time = Some(Utc::now());
         }
     }
 }

tap/src/ethernet/detection/l7_tagger.rs

@@ -40,23 +40,21 @@ pub fn tag_tcp_sessions(sessions: &mut MutexGuard<HashMap<TcpSessionKey, TcpSess
 }
 
 fn tag_all(client_to_server: Vec<u8>, server_to_client: Vec<u8>, session: &TcpSession) -> Vec<L7SessionTag> {
-    // Remove all control characters. Case insensitivity. Ignore redirects.
-    let cts = String::from_utf8_lossy(&client_to_server).to_string();
-    let stc = String::from_utf8_lossy(&server_to_client).to_string();
+    let client_to_server_string = String::from_utf8_lossy(&client_to_server).to_string();
+    let server_to_client_string = String::from_utf8_lossy(&server_to_client).to_string();
 
     let mut tags = Vec::new();
 
-    if http_tagger::tag(&cts, &stc) {
+    if http_tagger::tag(&client_to_server_string, &server_to_client_string).is_some() {
         tags.extend([Unencrypted, Http]);
     }
 
-    if socks_tagger::tag(&client_to_server, &server_to_client) {
+    // TODO match, send to socks channel/processor
+    if let Some(socks) = socks_tagger::tag(&client_to_server, &server_to_client, session) {
+        info!("SOCKS: {:?}", socks);
+
         tags.extend([Socks]);
     }
 
-    if tags.contains(&Socks) {
-        info!("Detected new SOCKS4 ({:?}) TCP session: {} {}:{} -> {}:{} ({} byte)", tags, session.state, session.source_address, session.source_port, session.destination_address, session.destination_port, session.bytes_count);
-    }
-
     tags
 }

tap/src/ethernet/detection/taggers/http_tagger.rs

@@ -1,15 +1,19 @@
 use crate::ethernet::detection::taggers::tagger_utils;
 
-pub fn tag(cts: &String, stc: &String) -> bool {
+pub fn tag(cts: &String, stc: &String) -> Option<()> {
     let methods = &vec!["get", "post", "put", "delete", "head", "options", "patch"];
     let versions = &vec!["HTTP/1.0", "HTTP/1.1", "HTTP/2", "HTTP/3"];
 
     let lowercase_cts = &cts.to_lowercase();
     let lowercase_stc = &stc.to_lowercase();
 
-    tagger_utils::scan_body_substrings_or(lowercase_cts, methods)
+    if tagger_utils::scan_body_substrings_or(lowercase_cts, methods)
         && tagger_utils::scan_body_substrings_or(lowercase_cts, versions)
         && tagger_utils::scan_body_substring(lowercase_cts, "\r\n")
         && tagger_utils::scan_body_substrings_or(lowercase_stc, versions)
-        && tagger_utils::scan_body_substring(lowercase_stc, "\r\n")
+        && tagger_utils::scan_body_substring(lowercase_stc, "\r\n") {
+        return Some(())
+    } else {
+        None
+    }
 }

tap/src/ethernet/detection/taggers/socks_tagger.rs

@@ -1,10 +1,116 @@
-pub fn tag(cts: &Vec<u8>, stc: &Vec<u8>) -> bool {
+use byteorder::{BigEndian, ByteOrder};
+use log::info;
+use crate::data::tcp_table::{TcpSession, TcpSessionState};
+use crate::ethernet::packets::{SocksConnectionHandshakeStatus, SocksConnectionStatus, SocksTunnel, SocksType};
+use crate::ethernet::packets::SocksType::{Socks4, Socks4A};
+use crate::helpers::network::{string_up_to_null_byte, to_ipv4_address};
 
-    if cts.len() < 3 { // TODO set properly
-        return false;
+pub fn tag(cts: &[u8], stc: &[u8], session: &TcpSession) -> Option<SocksTunnel> {
+
+    if cts.len() < 9 || stc.len() < 2 {
+        return None;
+    }
+
+    if *cts.first().unwrap() == 0x04
+        && (*cts.get(1).unwrap() == 0x01 || *cts.get(1).unwrap() == 0x02) {
+        // Potentially SOCKS4, check if the response was SOCKS as well.
+        if *stc.first().unwrap() != 0x04
+            || *stc.get(1).unwrap() < 0x5A
+            || *stc.get(1).unwrap() > 0x5D {
+            return None
+        }
+
+        let tunneled_destination_port = BigEndian::read_u16(&cts[2..4]);
+
+        let tunneled_destination_address = if !is_socks_4a_address(&cts[4..8]) {
+            Some(to_ipv4_address(&cts[4..8]))
+        } else {
+            None
+        };
+
+        let username = string_up_to_null_byte(&cts[8..]);
+
+        let cursor = 8 + match &username {
+            Some(username) => username.len() + 1,
+            None => 1
+        };
+
+        let tunneled_destination_host = if tunneled_destination_address.is_none() {
+           string_up_to_null_byte(&cts[cursor..])
+        } else {
+            None
+        };
+
+        let handshake_status = match *stc.get(1).unwrap() {
+            0x5A => SocksConnectionHandshakeStatus::Granted,
+            0x5B => SocksConnectionHandshakeStatus::Rejected,
+            0x5C => SocksConnectionHandshakeStatus::FailedIdentdUnreachable,
+            0x5D => SocksConnectionHandshakeStatus::FailedIdentdAuth,
+            _    => SocksConnectionHandshakeStatus::Invalid,
+        };
+
+        // Overwrite connection status in case of closed TCP connection.
+        let (connection_status, terminated_at) = match session.state {
+            TcpSessionState::SynSent
+            | TcpSessionState::SynReceived
+            | TcpSessionState::Established
+            | TcpSessionState::FinWait1
+            | TcpSessionState::FinWait2 => (SocksConnectionStatus::Active, None),
+            TcpSessionState::ClosedFin
+            | TcpSessionState::ClosedRst
+            | TcpSessionState::Refused => (SocksConnectionStatus::Inactive, session.end_time),
+            TcpSessionState::ClosedTimeout =>
+                (SocksConnectionStatus::InactiveTimeout, session.end_time)
+        };
+
+        let socks_type = if tunneled_destination_address.is_some() {
+            Socks4
+        } else {
+            Socks4A
+        };
+
+        return Some(SocksTunnel {
+            socks_type,
+            handshake_status,
+            connection_status,
+            username,
+            tunneled_bytes: session.bytes_count,
+            tcp_session_key: session.session_key.clone(),
+            tunneled_destination_address,
+            tunneled_destination_host,
+            tunneled_destination_port,
+            source_mac: session.source_mac.clone(),
+            destination_mac: session.destination_mac.clone(),
+            source_address: session.source_address,
+            destination_address: session.destination_address,
+            source_port: session.source_port,
+            destination_port: session.destination_port,
+            established_at: session.start_time,
+            terminated_at
+        })
+    }
+
+    if *cts.first().unwrap() == 0x05 && *cts.get(1).unwrap() <= 9 
+        && *cts.get(2).unwrap() <= 9 {
+        // Potentially SOCKS5, check if the response was SOCKS as well.
+        if *stc.first().unwrap() != 0x05 || *stc.get(1).unwrap() > 9 {
+            return None;
+        }
+
+        info!("SOCKS5");
+
+        // TODO update bounds check for cts on top of method
+
+        return None
     }
 
-    return *cts.first().unwrap() == 0x04
-        && (*cts.get(1).unwrap() == 0x01 || *cts.get(1).unwrap() == 0x02)
+    None
+}
 
+fn is_socks_4a_address(address: &[u8]) -> bool {
+    address.len() == 4
+        && address[0] == 0x00
+        && address[1] == 0x00
+        && address[2] == 0x00
+        && address[3] != 0x00
 }

tap/src/ethernet/packets.rs

@@ -135,3 +135,45 @@ pub struct DNSData {
 
 #[derive(Debug)]
 pub struct IPv6Packet { }
+
+#[derive(Debug)]
+pub enum SocksType {
+    Socks4,
+    Socks4A,
+    Socks5
+}
+
+#[derive(Debug)]
+pub enum SocksConnectionHandshakeStatus {
+    Granted,
+    Rejected,
+    FailedIdentdUnreachable,
+    FailedIdentdAuth,
+    Invalid
+}
+
+#[derive(Debug)]
+pub enum SocksConnectionStatus {
+    Active, Inactive, InactiveTimeout
+}
+
+#[derive(Debug)]
+pub struct SocksTunnel {
+    pub socks_type: SocksType,
+    pub handshake_status: SocksConnectionHandshakeStatus,
+    pub connection_status: SocksConnectionStatus,
+    pub username: Option<String>,
+    pub tunneled_bytes: u64,
+    pub tunneled_destination_address: Option<IpAddr>,
+    pub tunneled_destination_host: Option<String>,
+    pub tunneled_destination_port: u16,
+    pub tcp_session_key: TcpSessionKey,
+    pub source_mac: String,
+    pub destination_mac: String,
+    pub source_address: IpAddr,
+    pub destination_address: IpAddr,
+    pub source_port: u16,
+    pub destination_port: u16,
+    pub established_at: DateTime<Utc>,
+    pub terminated_at: Option<DateTime<Utc>>,
+}

tap/src/helpers/network.rs

@@ -177,4 +177,18 @@ pub fn is_ipv6_address(s: &str) -> bool {
 
 pub fn is_ip_address(s: &str) -> bool {
     is_ipv4_address(s) || is_ipv6_address(s)
+}
+
+pub fn string_up_to_null_byte(b: &[u8]) -> Option<String> {
+    b.iter()
+        .position(|&x| x == 0x00)
+        .and_then(|index| {
+            if index == 0 { 
+                None
+            } else {
+                std::str::from_utf8(&b[..index])
+                    .map(|s| s.to_string())
+                    .ok()
+            }
+        })
 }