Minimal protocol analysis beginnings.

src/rust/lqos_bus/src/ip_stats.rs

@@ -202,4 +202,6 @@ pub struct FlowbeeData {
   pub remote_asn_name: String,
   /// Remote ASN Country
   pub remote_asn_country: String,
+  /// Analysis
+  pub analysis: String,
 }

src/rust/lqos_node_manager/static/main.html

@@ -318,7 +318,6 @@
                 html += "<th>Protocol</th>";
                 html += "<th>Local IP</th>";
                 html += "<th>Remote IP</th>";
-                html += "<th>Ports</th>";
                 html += "<th>UL ⬆️</th>";
                 html += "<th>DL ⬇️</th>";
                 html += "<th>UL RTT</th>";
@@ -329,10 +328,9 @@
                 for (var i = 0; i<data.length; i++) {
                     //console.log(data[i]);
                     html += "<tr>";
-                    html += "<td>" + data[i].ip_protocol + "</td>";
+                    html += "<td>" + data[i].analysis + "</td>";
                     html += "<td>" + data[i].local_ip + "</td>";
                     html += "<td>" + data[i].remote_ip + "</td>";
-                    html += "<td>" + data[i].src_port + " / " + data[i].dst_port + "</td>";
                     // TODO: Check scaling
                     html += "<td>" + scaleNumber(data[i].rate_estimate_bps[0] * 8) + "</td>";
                     html += "<td>" + scaleNumber(data[i].rate_estimate_bps[1] * 8) + "</td>";

src/rust/lqosd/src/throughput_tracker/flow_data/flow_analysis/mod.rs

@@ -1,16 +1,20 @@
 use std::{net::IpAddr, sync::Mutex};
+use lqos_sys::flowbee_data::FlowbeeKey;
 use once_cell::sync::Lazy;
-
 use self::asn::AsnTable;
 mod asn;
+mod protocol;
+pub use protocol::FlowProtocol;
 
-static ANALYSIS: Lazy<FlowAnalysis> = Lazy::new(|| FlowAnalysis::new());
+use super::AsnId;
 
-pub struct FlowAnalysis {
+static ANALYSIS: Lazy<FlowAnalysisSystem> = Lazy::new(|| FlowAnalysisSystem::new());
+
+pub struct FlowAnalysisSystem {
     asn_table: Mutex<Option<asn::AsnTable>>,
 }
 
-impl FlowAnalysis {
+impl FlowAnalysisSystem {
     pub fn new() -> Self {
         // Periodically update the ASN table
         std::thread::spawn(|| {
@@ -42,6 +46,24 @@ pub fn setup_flow_analysis() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct FlowAnalysis {
+    pub asn_id: AsnId,
+    pub protocol_analysis: FlowProtocol,
+}
+
+impl FlowAnalysis {
+    pub fn new(key: &FlowbeeKey) -> Self {
+        let asn_id = lookup_asn_id(key.remote_ip.as_ip());
+        let protocol_analysis = FlowProtocol::new(key);
+        Self {
+            asn_id: AsnId(asn_id.unwrap_or(0)),
+            protocol_analysis,
+        }
+    }
+}
+
+
 pub fn lookup_asn_id(ip: IpAddr) -> Option<u32> {
     let table_lock = ANALYSIS.asn_table.lock();
     if table_lock.is_err() {

src/rust/lqosd/src/throughput_tracker/flow_data/flow_analysis/protocol.rs

@@ -0,0 +1,93 @@
+use std::fmt::Display;
+
+use lqos_sys::flowbee_data::FlowbeeKey;
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum FlowProtocol {
+    Smtp,
+    Ftp,
+    Http,
+    Https,
+    Ssh,
+    Telnet,
+    Imap,
+    Rdp,
+    Dns,
+    Pop3,
+    Quic,
+    Other { proto: u8, src_port: u16, dst_port: u16 }
+}
+
+impl FlowProtocol {
+    pub fn new(key: &FlowbeeKey) -> Self {
+        match key.ip_protocol {
+            6 => Self::tcp(key),
+            17 => Self::udp(key),
+            _ => Self::Other { 
+                proto: key.ip_protocol, 
+                src_port: key.src_port, 
+                dst_port: key.dst_port, 
+            }
+        }
+    }
+
+    fn tcp(key: &FlowbeeKey) -> Self {
+        match key.src_port {
+            25 => Self::Smtp,
+            80 => Self::Http,
+            443 => Self::Https,
+            21 | 20 => Self::Ftp,
+            22 => Self::Ssh,
+            23 => Self::Telnet,
+            3389 => Self::Rdp,
+            143 => Self::Imap,
+            53 => Self::Dns,
+            110 => Self::Pop3,
+            _ => Self::Other { 
+                proto: key.ip_protocol, 
+                src_port: key.src_port, 
+                dst_port: key.dst_port, 
+            }
+        }
+    }
+
+    fn udp(key: &FlowbeeKey) -> Self {
+        match key.src_port {
+            53 => Self::Dns,
+            80 | 443 => Self::Quic,            
+            _ => Self::Other { 
+                proto: key.ip_protocol, 
+                src_port: key.src_port, 
+                dst_port: key.dst_port, 
+            }
+        }
+    }
+}
+
+impl Display for FlowProtocol {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Smtp => write!(f, "SMTP"),
+            Self::Ftp => write!(f, "FTP"),
+            Self::Http => write!(f, "HTTP"),
+            Self::Https => write!(f, "HTTPS"),
+            Self::Ssh => write!(f, "SSH"),
+            Self::Telnet => write!(f, "Telnet"),
+            Self::Imap => write!(f, "IMAP"),
+            Self::Rdp => write!(f, "RDP"),
+            Self::Dns => write!(f, "DNS"),
+            Self::Pop3 => write!(f, "POP3"),
+            Self::Quic => write!(f, "QUIC"),
+            Self::Other { proto, src_port, dst_port } => write!(f, "{} {}/{}", proto_name(proto), src_port, dst_port),
+        }
+    }
+}
+
+fn proto_name(proto: &u8) -> &'static str {
+    match proto {
+        6 => "TCP",
+        17 => "UDP",
+        1 => "ICMP",
+        _ => "Other",
+    }
+}

src/rust/lqosd/src/throughput_tracker/flow_data/flow_tracker.rs

@@ -4,9 +4,10 @@
 use dashmap::DashMap;
 use lqos_sys::flowbee_data::{FlowbeeData, FlowbeeKey};
 use once_cell::sync::Lazy;
+use super::flow_analysis::FlowAnalysis;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct AsnId(pub u32);
 
-pub static ALL_FLOWS: Lazy<DashMap<FlowbeeKey, (FlowbeeData, AsnId)>> = Lazy::new(|| DashMap::new());
+pub static ALL_FLOWS: Lazy<DashMap<FlowbeeKey, (FlowbeeData, FlowAnalysis)>> = Lazy::new(|| DashMap::new());
 

src/rust/lqosd/src/throughput_tracker/flow_data/mod.rs

@@ -13,7 +13,7 @@ use std::sync::{
     mpsc::{channel, Sender},
     Arc,
 };
-pub(crate) use flow_analysis::{setup_flow_analysis, lookup_asn_id, get_asn_name_and_country};
+pub(crate) use flow_analysis::{setup_flow_analysis, lookup_asn_id, get_asn_name_and_country, FlowAnalysis};
 
 
 trait FlowbeeRecipient {

src/rust/lqosd/src/throughput_tracker/mod.rs

@@ -18,7 +18,7 @@ use tokio::{
     time::{Duration, Instant},
 };
 
-use self::flow_data::{get_asn_name_and_country, AsnId, ALL_FLOWS};
+use self::flow_data::{get_asn_name_and_country, AsnId, FlowAnalysis, ALL_FLOWS};
 
 const RETIRE_AFTER_SECONDS: u64 = 30;
 
@@ -486,7 +486,7 @@ pub fn all_unknown_ips() -> BusResponse {
   /// For debugging: dump all active flows!
   pub fn dump_active_flows() -> BusResponse {
     let result: Vec<lqos_bus::FlowbeeData> = ALL_FLOWS.iter().map(|row| {
-      let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(row.value().1.0);
+      let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(row.value().1.asn_id.0);
 
       lqos_bus::FlowbeeData {
         remote_ip: row.key().remote_ip.as_ip().to_string(),
@@ -502,9 +502,10 @@ pub fn all_unknown_ips() -> BusResponse {
         end_status: row.value().0.end_status,
         tos: row.value().0.tos,
         flags: row.value().0.flags,
-        remote_asn: row.value().1.0,
+        remote_asn: row.value().1.asn_id.0,
         remote_asn_name,
         remote_asn_country,
+        analysis: row.value().1.protocol_analysis.to_string(),
       }
     }).collect();
 
@@ -518,7 +519,7 @@ pub fn all_unknown_ips() -> BusResponse {
 
   /// Top Flows Report
   pub fn top_flows(n: u32, flow_type: TopFlowType) -> BusResponse {
-    let mut table: Vec<(FlowbeeKey, (FlowbeeData, AsnId))> = ALL_FLOWS
+    let mut table: Vec<(FlowbeeKey, (FlowbeeData, FlowAnalysis))> = ALL_FLOWS
       .iter()
       .map(|row| (
         row.key().clone(),
@@ -568,7 +569,7 @@ pub fn all_unknown_ips() -> BusResponse {
       .iter()
       .take(n as usize)
       .map(|(ip, flow)| {
-        let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(flow.1.0);
+        let (remote_asn_name, remote_asn_country) = get_asn_name_and_country(flow.1.asn_id.0);
         lqos_bus::FlowbeeData {
           remote_ip: ip.remote_ip.as_ip().to_string(),
           local_ip: ip.local_ip.as_ip().to_string(),
@@ -583,9 +584,10 @@ pub fn all_unknown_ips() -> BusResponse {
           end_status: flow.0.end_status,
           tos: flow.0.tos,
           flags: flow.0.flags,
-          remote_asn: flow.1.0,
+          remote_asn: flow.1.asn_id.0,
           remote_asn_name,
           remote_asn_country,
+          analysis: flow.1.protocol_analysis.to_string(),
         }
       })
       .collect();

src/rust/lqosd/src/throughput_tracker/tracking_data.rs

@@ -1,6 +1,6 @@
 use std::{sync::atomic::AtomicU64, time::Duration};
 use crate::{shaped_devices_tracker::{SHAPED_DEVICES, NETWORK_JSON}, stats::{HIGH_WATERMARK_DOWN, HIGH_WATERMARK_UP}};
-use super::{flow_data::{lookup_asn_id, AsnId, ALL_FLOWS}, throughput_entry::ThroughputEntry, RETIRE_AFTER_SECONDS};
+use super::{flow_data::{lookup_asn_id, AsnId, FlowAnalysis, ALL_FLOWS}, throughput_entry::ThroughputEntry, RETIRE_AFTER_SECONDS};
 use dashmap::DashMap;
 use lqos_bus::TcHandle;
 use lqos_sys::{flowbee_data::{FlowbeeData, FlowbeeKey}, iterate_flows, throughput_for_each};
@@ -214,9 +214,9 @@ impl ThroughputTracker {
             this_flow.0.flags = data.flags;  
           } else {
             // Insert it into the map
-            let asn = lookup_asn_id(key.remote_ip.as_ip()).unwrap_or(0);
+            let flow_analysis = FlowAnalysis::new(&key);
 
-            ALL_FLOWS.insert(key.clone(), (data.clone(), AsnId(asn)));
+            ALL_FLOWS.insert(key.clone(), (data.clone(), flow_analysis));
             // TODO: Submit it for analysis
           }