Add sysmon event to publish stream data (#510)

- `FileCreate`, `FileDelete`

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "giganto"
-version = "0.12.4-alpha.1"
+version = "0.13.0-alpha.1"
 edition = "2021"
 
 [dependencies]
@@ -15,7 +15,7 @@ ctrlc = { version = "3", features = ["termination"] }
 data-encoding = "2.4"
 directories = "5.0"
 futures-util = "0.3"
-giganto-client = { git = "https://github.com/aicers/giganto-client.git", tag = "0.12.1" }
+giganto-client = { git = "https://github.com/aicers/giganto-client.git", tag = "0.12.2" }
 humantime = "2.1"
 humantime-serde = "1"
 libc = "0.2"

src/graphql/network.rs

@@ -1604,7 +1604,6 @@ fn network_connection(
                         NetworkRawEvents::ConnRawEvent(ConnRawEvent::from_key_value(&key, value)?),
                     ));
                     conn_data = conn_iter.next();
-                } else {
                 };
             }
             _ if selected == dns_ts => {
@@ -1614,7 +1613,6 @@ fn network_connection(
                         NetworkRawEvents::DnsRawEvent(DnsRawEvent::from_key_value(&key, value)?),
                     ));
                     dns_data = dns_iter.next();
-                } else {
                 };
             }
             _ if selected == http_ts => {
@@ -1624,7 +1622,6 @@ fn network_connection(
                         NetworkRawEvents::HttpRawEvent(HttpRawEvent::from_key_value(&key, value)?),
                     ));
                     http_data = http_iter.next();
-                } else {
                 };
             }
             _ if selected == rdp_ts => {
@@ -1634,7 +1631,6 @@ fn network_connection(
                         NetworkRawEvents::RdpRawEvent(RdpRawEvent::from_key_value(&key, value)?),
                     ));
                     rdp_data = rdp_iter.next();
-                } else {
                 };
             }
             _ if selected == ntlm_ts => {
@@ -1644,7 +1640,6 @@ fn network_connection(
                         NetworkRawEvents::NtlmRawEvent(NtlmRawEvent::from_key_value(&key, value)?),
                     ));
                     ntlm_data = ntlm_iter.next();
-                } else {
                 };
             }
             _ if selected == kerberos_ts => {
@@ -1656,7 +1651,6 @@ fn network_connection(
                         )?),
                     ));
                     kerberos_data = kerberos_iter.next();
-                } else {
                 };
             }
             _ if selected == ssh_ts => {
@@ -1666,7 +1660,6 @@ fn network_connection(
                         NetworkRawEvents::SshRawEvent(SshRawEvent::from_key_value(&key, value)?),
                     ));
                     ssh_data = ssh_iter.next();
-                } else {
                 };
             }
             _ if selected == dce_rpc_ts => {
@@ -1678,7 +1671,6 @@ fn network_connection(
                         )?),
                     ));
                     dce_rpc_data = dce_rpc_iter.next();
-                } else {
                 };
             }
             _ if selected == ftp_ts => {
@@ -1688,7 +1680,6 @@ fn network_connection(
                         NetworkRawEvents::FtpRawEvent(FtpRawEvent::from_key_value(&key, value)?),
                     ));
                     ftp_data = ftp_iter.next();
-                } else {
                 };
             }
             _ if selected == mqtt_ts => {
@@ -1698,7 +1689,6 @@ fn network_connection(
                         NetworkRawEvents::MqttRawEvent(MqttRawEvent::from_key_value(&key, value)?),
                     ));
                     mqtt_data = mqtt_iter.next();
-                } else {
                 };
             }
             _ if selected == ldap_ts => {
@@ -1708,7 +1698,6 @@ fn network_connection(
                         NetworkRawEvents::LdapRawEvent(LdapRawEvent::from_key_value(&key, value)?),
                     ));
                     ldap_data = ldap_iter.next();
-                } else {
                 };
             }
             _ if selected == tls_ts => {
@@ -1718,7 +1707,6 @@ fn network_connection(
                         NetworkRawEvents::TlsRawEvent(TlsRawEvent::from_key_value(&key, value)?),
                     ));
                     tls_data = tls_iter.next();
-                } else {
                 };
             }
             _ if selected == smb_ts => {
@@ -1728,7 +1716,6 @@ fn network_connection(
                         NetworkRawEvents::SmbRawEvent(SmbRawEvent::from_key_value(&key, value)?),
                     ));
                     smb_data = smb_iter.next();
-                } else {
                 };
             }
             _ if selected == nfs_ts => {
@@ -1738,7 +1725,6 @@ fn network_connection(
                         NetworkRawEvents::NfsRawEvent(NfsRawEvent::from_key_value(&key, value)?),
                     ));
                     nfs_data = nfs_iter.next();
-                } else {
                 };
             }
             _ => {}
@@ -1787,7 +1773,7 @@ fn network_connection(
     }
     let mut connection: Connection<String, NetworkRawEvents> =
         Connection::new(has_previous_page, has_next_page);
-    connection.edges.extend(result_vec.into_iter());
+    connection.edges.extend(result_vec);
 
     Ok(connection)
 }

src/graphql/status.rs

@@ -144,14 +144,18 @@ impl GigantoStatusQuery {
             .get("peers")
             .context("peers not found")?
             .as_array()
-            .context("invaild peers format")?;
+            .context("invalid peers format")?;
         let mut peer_list = Vec::new();
-        for peer in peers_value.iter() {
+        for peer in peers_value {
             if let Some(peer_data) = peer.as_inline_table() {
-                let (Some(address_val),Some(host_name_val)) = (peer_data.get("address"),peer_data.get("host_name")) else{
+                let (Some(address_val), Some(host_name_val)) =
+                    (peer_data.get("address"), peer_data.get("host_name"))
+                else {
                     return Err(anyhow!("Invalid address/hostname Value format").into());
                 };
-                let (Some(address),Some(host_name)) = (address_val.as_str(),host_name_val.as_str()) else{
+                let (Some(address), Some(host_name)) =
+                    (address_val.as_str(), host_name_val.as_str())
+                else {
                     return Err(anyhow!("Invalid address/hostname String format").into());
                 };
                 peer_list.push(PeerList {
@@ -223,10 +227,10 @@ pub fn write_toml_file(doc: &Document, path: &str) -> Result<()> {
 
 fn parse_toml_element(key: &str, doc: &Document) -> Result<String> {
     let Some(item) = doc.get(key) else {
-        return Err(anyhow!("{} not found.",key).into());
+        return Err(anyhow!("{} not found.", key).into());
     };
     let Some(value) = item.as_str() else {
-        return Err(anyhow!("parse failed: {}'s format is not available.",key).into());
+        return Err(anyhow!("parse failed: {}'s format is not available.", key).into());
     };
     Ok(value.to_string())
 }

src/peer.rs

@@ -206,7 +206,7 @@ async fn client_run(
     local_host_name: String,
     wait_shutdown: Arc<Notify>,
 ) {
-    for peer in peer_conn_info.peer_list.read().await.iter() {
+    for peer in &*peer_conn_info.peer_list.read().await {
         tokio::spawn(client_connection(
             client_endpoint.clone(),
             peer.clone(),
@@ -300,7 +300,7 @@ async fn client_connection(
                 .await?;
 
                 // Share the received peer list with connected peers.
-                for (_, conn) in peer_conn_info.peer_conn.read().await.iter() {
+                for conn in (*peer_conn_info.peer_conn.read().await).values() {
                     tokio::spawn(update_peer_info::<HashSet<PeerInfo>>(
                         conn.clone(),
                         PeerCode::UpdatePeerList,
@@ -345,7 +345,7 @@ async fn client_connection(
                         },
                         _ = peer_conn_info.notify_source.notified() => {
                             let source_list: HashSet<String> = peer_conn_info.sources.read().await.keys().cloned().collect();
-                            for (_, conn) in peer_conn_info.peer_conn.write().await.iter() {
+                            for conn in (*peer_conn_info.peer_conn.write().await).values() {
                                 tokio::spawn(update_peer_info::<HashSet<String>>(
                                     conn.clone(),
                                     PeerCode::UpdateSourceList,
@@ -456,7 +456,7 @@ async fn server_connection(
     .await?;
 
     // Share the received peer list with your connected peers.
-    for (_, conn) in peer_conn_info.peer_conn.read().await.iter() {
+    for conn in (*peer_conn_info.peer_conn.read().await).values() {
         tokio::spawn(update_peer_info::<HashSet<PeerInfo>>(
             conn.clone(),
             PeerCode::UpdatePeerList,
@@ -501,7 +501,7 @@ async fn server_connection(
             },
             _ = peer_conn_info.notify_source.notified() => {
                 let source_list: HashSet<String> = peer_conn_info.sources.read().await.keys().cloned().collect();
-                for (_, conn) in peer_conn_info.peer_conn.read().await.iter() {
+                for conn in (*peer_conn_info.peer_conn.read().await).values() {
                     tokio::spawn(update_peer_info::<HashSet<String>>(
                         conn.clone(),
                         PeerCode::UpdateSourceList,

src/publish.rs

@@ -616,6 +616,46 @@ where
                 error!("Failed to open nfs store");
             }
         }
+        RequestStreamRecord::FileCreate => {
+            if let Ok(store) = db.file_create_store() {
+                if let Err(e) = send_stream(
+                    store,
+                    conn,
+                    record_type,
+                    request_msg,
+                    source,
+                    kind,
+                    node_type,
+                    stream_direct_channel,
+                )
+                .await
+                {
+                    error!("Failed to send sysmon stream : {}", e);
+                }
+            } else {
+                error!("Failed to open file_create store");
+            }
+        }
+        RequestStreamRecord::FileDelete => {
+            if let Ok(store) = db.file_delete_store() {
+                if let Err(e) = send_stream(
+                    store,
+                    conn,
+                    record_type,
+                    request_msg,
+                    source,
+                    kind,
+                    node_type,
+                    stream_direct_channel,
+                )
+                .await
+                {
+                    error!("Failed to send sysmon stream : {}", e);
+                }
+            } else {
+                error!("Failed to open file_delete store");
+            }
+        }
         RequestStreamRecord::Pcap => {}
     };
     Ok(())
@@ -628,7 +668,7 @@ pub async fn send_direct_stream(
     source: &str,
     stream_direct_channel: StreamDirectChannel,
 ) -> Result<()> {
-    for (req_key, sender) in stream_direct_channel.read().await.iter() {
+    for (req_key, sender) in &*stream_direct_channel.read().await {
         if req_key.contains(&network_key.source_key) || req_key.contains(&network_key.all_key) {
             let raw_len = u32::try_from(raw_event.len())?.to_le_bytes();
             let mut send_buf: Vec<u8> = Vec::new();

src/server.rs

@@ -49,7 +49,10 @@ pub fn extract_cert_from_conn(connection: &Connection) -> Result<Vec<Certificate
     let Some(conn_info) = connection.peer_identity() else {
         bail!("no peer identity");
     };
-    let Some(cert_info) = conn_info.downcast_ref::<Vec<rustls::Certificate>>().cloned() else {
+    let Some(cert_info) = conn_info
+        .downcast_ref::<Vec<rustls::Certificate>>()
+        .cloned()
+    else {
         bail!("non-certificate identity");
     };
     Ok(cert_info)

src/storage/migration.rs

@@ -11,7 +11,7 @@ use std::{
 };
 use tracing::info;
 
-const COMPATIBLE_VERSION_REQ: &str = ">0.12.4-alpha,<0.13.0-alpha";
+const COMPATIBLE_VERSION_REQ: &str = ">0.12.4-alpha,<=0.13.0-alpha.1";
 
 /// Migrates the data directory to the up-to-date format if necessary.
 ///

src/web.rs

@@ -9,6 +9,7 @@ use warp::{http::Response as HttpResponse, Filter};
 ///
 /// Note that `key` is not compatible with the DER-encoded key extracted by
 /// rustls-pemfile.
+#[allow(clippy::unused_async)]
 pub async fn serve(
     schema: Schema,
     addr: SocketAddr,