Added new three detection events

CHANGELOG.md

@@ -5,6 +5,18 @@ file is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and
 this project adheres to [Semantic
 Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Add new `WindowsThreat` event message for Windows sysmon events.
+- Add new `NetworkThreat` event message for network events.
+- Add new `ExtraThreat` event message for misc log events.
+
+### Changed
+
+- Updated review-database to 0.23.0.
+
 ## [0.16.0] - 2024-01-15
 
 ### Added
@@ -359,6 +371,7 @@ across our system.
 
 - An initial version.
 
+[Unreleased]: https://github.com/aicers/review-web/compare/0.16.0...main
 [0.16.0]: https://github.com/aicers/review-web/compare/0.15.0...0.16.0
 [0.15.0]: https://github.com/aicers/review-web/compare/0.14.5...0.15.0
 [0.14.5]: https://github.com/aicers/review-web/compare/0.14.4...0.14.5

Cargo.toml

@@ -25,7 +25,7 @@ oinq = { git = "https://github.com/petabi/oinq.git", tag = "0.9.1" }
 reqwest = { version = "0.11", default-features = false, features = [
   "rustls-tls-native-roots",
 ] }
-review-database = { git = "https://github.com/petabi/review-database.git", tag = "0.22.1" }
+review-database = { git = "https://github.com/petabi/review-database.git", tag = "0.23.0" }
 roxy = { git = "https://github.com/aicers/roxy.git", tag = "0.2.1" }
 rustls = "0.21"
 rustls-native-certs = "0.6"

src/graphql/category.rs

@@ -49,7 +49,7 @@ impl CategoryMutation {
     async fn add_category(&self, ctx: &Context<'_>, name: String) -> Result<ID> {
         let db = ctx.data::<Arc<RwLock<Store>>>()?.read().await;
         let map = db.category_map();
-        let id = map.add(&name)?;
+        let id = map.insert(&name)?;
         Ok(ID(id.to_string()))
     }
 

src/graphql/event.rs

@@ -6,13 +6,16 @@ mod group;
 mod http;
 mod kerberos;
 mod ldap;
+mod log;
 mod mqtt;
+mod network;
 mod nfs;
 mod ntlm;
 mod rdp;
 mod smb;
 mod smtp;
 mod ssh;
+mod sysmon;
 mod tls;
 
 pub(super) use self::group::EventGroupQuery;
@@ -22,9 +25,10 @@ use self::{
     dns::DnsCovertChannel, ftp::BlockListFtp, ftp::FtpBruteForce, ftp::FtpPlainText,
     http::BlockListHttp, http::DomainGenerationAlgorithm, http::HttpThreat, http::NonBrowser,
     http::RepeatedHttpSessions, http::TorConnection, kerberos::BlockListKerberos,
-    ldap::BlockListLdap, ldap::LdapBruteForce, ldap::LdapPlainText, mqtt::BlockListMqtt,
-    nfs::BlockListNfs, ntlm::BlockListNtlm, rdp::BlockListRdp, rdp::RdpBruteForce,
-    smb::BlockListSmb, smtp::BlockListSmtp, ssh::BlockListSsh, tls::BlockListTls,
+    ldap::BlockListLdap, ldap::LdapBruteForce, ldap::LdapPlainText, log::ExtraThreat,
+    mqtt::BlockListMqtt, network::NetworkThreat, nfs::BlockListNfs, ntlm::BlockListNtlm,
+    rdp::BlockListRdp, rdp::RdpBruteForce, smb::BlockListSmb, smtp::BlockListSmtp,
+    ssh::BlockListSsh, sysmon::WindowsThreat, tls::BlockListTls,
 };
 use super::{
     customer::{Customer, HostNetworkGroupInput},
@@ -145,6 +149,9 @@ async fn fetch_events(
     let mut block_list_smtp_time = start_time;
     let mut block_list_ssh_time = start_time;
     let mut block_list_tls_time = start_time;
+    let mut windows_threat_time = start_time;
+    let mut network_threat_time = start_time;
+    let mut extra_threat_time = start_time;
 
     loop {
         itv.tick().await;
@@ -179,7 +186,10 @@ async fn fetch_events(
             .min(block_list_smb_time)
             .min(block_list_smtp_time)
             .min(block_list_ssh_time)
-            .min(block_list_tls_time);
+            .min(block_list_tls_time)
+            .min(windows_threat_time)
+            .min(network_threat_time)
+            .min(extra_threat_time);
 
         // Fetch event iterator based on time
         let start = i128::from(start) << 64;
@@ -376,7 +386,24 @@ async fn fetch_events(
                         block_list_tls_time = event_time + ADD_TIME_FOR_NEXT_COMPARE;
                     }
                 }
-                EventKind::Log => continue,
+                EventKind::WindowsThreat => {
+                    if event_time >= windows_threat_time {
+                        tx.unbounded_send(value.into())?;
+                        windows_threat_time = event_time + ADD_TIME_FOR_NEXT_COMPARE;
+                    }
+                }
+                EventKind::NetworkThreat => {
+                    if event_time >= network_threat_time {
+                        tx.unbounded_send(value.into())?;
+                        network_threat_time = event_time + ADD_TIME_FOR_NEXT_COMPARE;
+                    }
+                }
+                EventKind::ExtraThreat => {
+                    if event_time >= extra_threat_time {
+                        tx.unbounded_send(value.into())?;
+                        extra_threat_time = event_time + ADD_TIME_FOR_NEXT_COMPARE;
+                    }
+                }
             }
         }
     }
@@ -500,6 +527,12 @@ enum Event {
     BlockListSsh(BlockListSsh),
 
     BlockListTls(BlockListTls),
+
+    WindowsThreat(WindowsThreat),
+
+    NetworkThreat(NetworkThreat),
+
+    ExtraThreat(ExtraThreat),
 }
 
 impl From<database::Event> for Event {
@@ -543,6 +576,9 @@ impl From<database::Event> for Event {
                 RecordType::Ssh(event) => Event::BlockListSsh(event.into()),
                 RecordType::Tls(event) => Event::BlockListTls(event.into()),
             },
+            database::Event::WindowsThreat(event) => Event::WindowsThreat(event.into()),
+            database::Event::NetworkThreat(event) => Event::NetworkThreat(event.into()),
+            database::Event::ExtraThreat(event) => Event::ExtraThreat(event.into()),
         }
     }
 }

src/graphql/event/log.rs

@@ -0,0 +1,65 @@
+use super::TriageScore;
+use async_graphql::Object;
+use chrono::{DateTime, Utc};
+use review_database as database;
+
+#[allow(clippy::module_name_repetitions)]
+pub(super) struct ExtraThreat {
+    inner: database::ExtraThreat,
+}
+
+#[Object]
+impl ExtraThreat {
+    async fn time(&self) -> DateTime<Utc> {
+        self.inner.time
+    }
+
+    async fn source(&self) -> &str {
+        &self.inner.source
+    }
+
+    async fn service(&self) -> &str {
+        &self.inner.service
+    }
+
+    async fn content(&self) -> &str {
+        &self.inner.content
+    }
+
+    async fn db_name(&self) -> &str {
+        &self.inner.db_name
+    }
+
+    async fn rule_id(&self) -> u32 {
+        self.inner.rule_id
+    }
+
+    async fn matched_to(&self) -> &str {
+        &self.inner.matched_to
+    }
+
+    async fn cluster_id(&self) -> usize {
+        self.inner.cluster_id
+    }
+
+    async fn attack_kind(&self) -> &str {
+        &self.inner.attack_kind
+    }
+
+    async fn confidence(&self) -> f32 {
+        self.inner.confidence
+    }
+
+    async fn triage_scores(&self) -> Option<Vec<TriageScore>> {
+        self.inner
+            .triage_scores
+            .as_ref()
+            .map(|scores| scores.iter().map(Into::into).collect::<Vec<TriageScore>>())
+    }
+}
+
+impl From<database::ExtraThreat> for ExtraThreat {
+    fn from(inner: database::ExtraThreat) -> Self {
+        Self { inner }
+    }
+}

src/graphql/event/network.rs

@@ -0,0 +1,128 @@
+use super::{country_code, find_ip_customer, find_ip_network, TriageScore};
+use crate::graphql::{customer::Customer, network::Network};
+use async_graphql::{Context, Object, Result};
+use chrono::{DateTime, Utc};
+use review_database as database;
+
+#[allow(clippy::module_name_repetitions)]
+pub(super) struct NetworkThreat {
+    inner: database::NetworkThreat,
+}
+
+#[Object]
+impl NetworkThreat {
+    async fn time(&self) -> DateTime<Utc> {
+        self.inner.time
+    }
+
+    async fn source(&self) -> &str {
+        &self.inner.source
+    }
+
+    async fn src_addr(&self) -> String {
+        self.inner.orig_addr.to_string()
+    }
+
+    async fn src_port(&self) -> u16 {
+        self.inner.orig_port
+    }
+
+    /// The two-letter country code of the source IP address. `"XX"` if the
+    /// location of the address is not known, and `"ZZ"` if the location
+    /// database is unavailable.
+    async fn src_country(&self, ctx: &Context<'_>) -> String {
+        country_code(ctx, self.inner.orig_addr)
+    }
+
+    async fn src_customer(&self, ctx: &Context<'_>) -> Result<Option<Customer>> {
+        let store = crate::graphql::get_store(ctx).await?;
+        let map = store.customer_map();
+        find_ip_customer(&map, self.inner.orig_addr)
+    }
+
+    async fn src_network(&self, ctx: &Context<'_>) -> Result<Option<Network>> {
+        let store = crate::graphql::get_store(ctx).await?;
+        let map = store.network_map();
+        find_ip_network(&map, self.inner.orig_addr)
+    }
+
+    async fn dst_addr(&self) -> String {
+        self.inner.resp_addr.to_string()
+    }
+
+    async fn dst_port(&self) -> u16 {
+        self.inner.resp_port
+    }
+
+    /// The two-letter country code of the destination IP address. `"XX"` if the
+    /// location of the address is not known, and `"ZZ"` if the location
+    /// database is unavailable.
+    async fn dst_country(&self, ctx: &Context<'_>) -> String {
+        country_code(ctx, self.inner.resp_addr)
+    }
+
+    async fn dst_customer(&self, ctx: &Context<'_>) -> Result<Option<Customer>> {
+        let store = crate::graphql::get_store(ctx).await?;
+        let map = store.customer_map();
+        find_ip_customer(&map, self.inner.resp_addr)
+    }
+
+    async fn dst_network(&self, ctx: &Context<'_>) -> Result<Option<Network>> {
+        let store = crate::graphql::get_store(ctx).await?;
+        let map = store.network_map();
+        find_ip_network(&map, self.inner.resp_addr)
+    }
+
+    async fn proto(&self) -> u8 {
+        self.inner.proto
+    }
+
+    async fn service(&self) -> &str {
+        &self.inner.service
+    }
+
+    async fn last_time(&self) -> i64 {
+        self.inner.last_time
+    }
+
+    async fn content(&self) -> &str {
+        &self.inner.content
+    }
+
+    async fn db_name(&self) -> &str {
+        &self.inner.db_name
+    }
+
+    async fn rule_id(&self) -> u32 {
+        self.inner.rule_id
+    }
+
+    async fn matched_to(&self) -> &str {
+        &self.inner.matched_to
+    }
+
+    async fn cluster_id(&self) -> usize {
+        self.inner.cluster_id
+    }
+
+    async fn attack_kind(&self) -> &str {
+        &self.inner.attack_kind
+    }
+
+    async fn confidence(&self) -> f32 {
+        self.inner.confidence
+    }
+
+    async fn triage_scores(&self) -> Option<Vec<TriageScore>> {
+        self.inner
+            .triage_scores
+            .as_ref()
+            .map(|scores| scores.iter().map(Into::into).collect::<Vec<TriageScore>>())
+    }
+}
+
+impl From<database::NetworkThreat> for NetworkThreat {
+    fn from(inner: database::NetworkThreat) -> Self {
+        Self { inner }
+    }
+}