Merge pull request #2 from maksimryndin/debug-ssh-monitoring

Fix ssh logs parsing

.github/ISSUE_TEMPLATE/BUG_REPORT.yaml

@@ -82,6 +82,7 @@ body:
       label: Which version of Goral do you run (`goral --version`)?
       multiple: false
       options:
+        - 0.1.8
         - 0.1.7
         - 0.1.6
         - 0.1.5

.github/site/src/install.sh

@@ -21,7 +21,7 @@ main() {
 
     get_architecture || return 1
     local _arch="$RETVAL"
-    local _version=${1:-'0.1.7'}
+    local _version=${1:-'0.1.8'}
     assert_nz "$_arch" "arch"
 
     local _file="goral-${_version}-${_arch}"

.github/site/src/installation.md

@@ -11,9 +11,9 @@ curl --proto '=https' --tlsv1.2 -sSf https://maksimryndin.github.io/goral/instal
 </summary>
 
 ```sh
-wget https://github.com/maksimryndin/goral/releases/download/0.1.7/goral-0.1.7-x86_64-unknown-linux-gnu.tar.gz
-tar -xzf goral-0.1.7-x86_64-unknown-linux-gnu.tar.gz
-cd goral-0.1.7-x86_64-unknown-linux-gnu/
+wget https://github.com/maksimryndin/goral/releases/download/0.1.8/goral-0.1.8-x86_64-unknown-linux-gnu.tar.gz
+tar -xzf goral-0.1.8-x86_64-unknown-linux-gnu.tar.gz
+cd goral-0.1.8-x86_64-unknown-linux-gnu/
 shasum -a 256 -c sha256_checksum.txt 
 ```
 </details>
@@ -23,7 +23,7 @@ shasum -a 256 -c sha256_checksum.txt
 
 ```sh
 curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
-git clone --depth 1 --branch 0.1.7 https://github.com/maksimryndin/goral
+git clone --depth 1 --branch 0.1.8 https://github.com/maksimryndin/goral
 cd goral
 RUSTFLAGS='-C target-feature=+crt-static' cargo build --release --target <target triple>
 ```

CHANGELOG.md

@@ -1,3 +1,6 @@
+* 0.1.8
+    * fix ssh logs parsing
+
 * 0.1.7
     * ssh log monitoring
     * rules for text now support "is" and "is not" conditions

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "goral"
-version = "0.1.7"
+version = "0.1.8"
 edition = "2021"
 author = "Maksim Ryndin"
 license = "Apache-2.0"
@@ -50,7 +50,7 @@ tracing-subscriber = { version = "0.3", features = ["fmt", "json", "env-filter"]
 url = { version = "2", features = ["serde"] }
 
 [target.'cfg(target_os = "linux")'.dependencies]
-logwatcher2 = { git = "https://github.com/maksimryndin/logwatcher2.git" }
+logwatcher2 = { git = "https://github.com/maksimryndin/logwatcher2.git", rev="9124084dedf7cca548a7be01f0195b876683749a" }
 psutil = { version = "3.2.2", default-features = false, features = ["disk"]}
 
 [dev-dependencies]

src/services/system/configuration.rs

@@ -23,10 +23,17 @@ pub(super) fn scrape_push_rule(
         ));
     }
 
+    #[cfg(target_os = "linux")]
+    const AVERAGE_DATAROWS_PER_SCRAPE: u16 = 15; // see collector.rs, ssh.rs
+    #[cfg(target_os = "linux")]
+    const LIMIT: u16 = 45;
+    #[cfg(not(target_os = "linux"))]
     const AVERAGE_DATAROWS_PER_SCRAPE: u16 = 10; // see collector.rs
+    #[cfg(not(target_os = "linux"))]
+    const LIMIT: u16 = 20;
+
     let number_of_rows_in_batch =
         ceiled_division(*push_interval_secs, *scrape_interval_secs) * AVERAGE_DATAROWS_PER_SCRAPE;
-    const LIMIT: u16 = 20;
     if number_of_rows_in_batch > LIMIT {
         return Err(serde_valid::validation::Error::Custom(
             format!("push interval ({push_interval_secs}) is too big or scrape interval ({scrape_interval_secs}) is too small - too much data ({number_of_rows_in_batch} rows vs limit of {LIMIT}) would be accumulated before saving to a spreadsheet")

src/services/system/ssh.rs

@@ -46,24 +46,29 @@ pub(super) fn process_sshd_log(
     log_watcher.watch(&mut move |result| {
         let result = match result {
             Ok(event) => match event {
-                LogWatcherEvent::Line(line) => match parse(&line) {
-                    Some(mut datarow) => {
-                        lookup_connection(&mut datarow, &mut connections);
-                        let Datavalue::Text(ref status) = datarow.data[4].1 else {
-                            panic!("assert: ssh status is parsed")
-                        };
-                        if status == SSH_LOG_STATUS_CONNECTED && connections.len() > 100 {
-                            let message =
-                                format!("there are {} active ssh connections", connections.len());
-                            tracing::warn!("{}", message);
-                            messenger.send_nonblock(Notification::new(message, Level::WARN));
+                LogWatcherEvent::Line(line) => {
+                    tracing::debug!("new auth log line: {line}");
+                    match parse(&line) {
+                        Some(mut datarow) => {
+                            lookup_connection(&mut datarow, &mut connections);
+                            let Datavalue::Text(ref status) = datarow.data[4].1 else {
+                                panic!("assert: ssh status is parsed")
+                            };
+                            if status == SSH_LOG_STATUS_CONNECTED && connections.len() > 100 {
+                                let message = format!(
+                                    "there are {} active ssh connections",
+                                    connections.len()
+                                );
+                                tracing::warn!("{}", message);
+                                messenger.send_nonblock(Notification::new(message, Level::WARN));
+                            }
+                            Ok(Data::Single(datarow))
+                        }
+                        None => {
+                            return LogWatcherAction::None;
                         }
-                        Ok(Data::Single(datarow))
-                    }
-                    None => {
-                        return LogWatcherAction::None;
                     }
-                },
+                }
                 LogWatcherEvent::LogRotation => {
                     tracing::info!("auth log file rotation");
                     return LogWatcherAction::None;
@@ -74,6 +79,7 @@ pub(super) fn process_sshd_log(
                 Err(Data::Message(message))
             }
         };
+        tracing::debug!("sending ssh result: {result:?}");
         if sender.blocking_send(TaskResult { id: 0, result }).is_err() {
             if is_shutdown.load(Ordering::Relaxed) {
                 return LogWatcherAction::Finish;
@@ -82,6 +88,7 @@ pub(super) fn process_sshd_log(
                 "assert: ssh monitoring messages queue shouldn't be closed before shutdown signal"
             );
         }
+        tracing::debug!("sent ssh result");
 
         LogWatcherAction::None
     });
@@ -150,7 +157,7 @@ fn parse(line: &str) -> Option<Datarow> {
         static ref RE: Regex = Regex::new(
             r"(?x)
             (?P<datetime>
-                [A-Za-z]{3,9}\s\d{1,2}\s\d{2}:\d{2}:\d{2}
+                [A-Z][a-z]{2}(\s\d{2}|\s{2}\d)\s\d{2}:\d{2}:\d{2}
             )
             \s\S+\s
             sshd\[(?P<id>\d+)\]:\s
@@ -175,7 +182,7 @@ fn parse(line: &str) -> Option<Datarow> {
             .map(|datetime| {
                 let captured = datetime.as_str();
                 let captured = format!("{} {captured}", Utc::now().format("%Y"));
-                NaiveDateTime::parse_from_str(&captured, "%Y %b %d %H:%M:%S")
+                NaiveDateTime::parse_from_str(&captured, "%Y %b %e %H:%M:%S")
                     .expect("assert: can parse auth log datetime")
             })
             .expect("assert: can get auth log datetime");
@@ -426,6 +433,21 @@ mod tests {
             Datavalue::Text("wrong_params".to_string())
         );
         assert_eq!(parsed.data[5].1, Datavalue::NotAvailable);
+
+        let line = "Jun  3 18:21:14 household sshd[219219]: Connection closed by invalid user dell 141.98.10.125 port 60878 [preauth]";
+        let parsed = parse(line).unwrap();
+        assert_eq!(parsed.data[0].1, Datavalue::IntegerID(219219));
+        assert_eq!(parsed.data[1].1, Datavalue::Text("dell".to_string()));
+        assert_eq!(
+            parsed.data[2].1,
+            Datavalue::Text("141.98.10.125".to_string())
+        );
+        assert_eq!(parsed.data[3].1, Datavalue::IntegerID(60878));
+        assert_eq!(
+            parsed.data[4].1,
+            Datavalue::Text("invalid_user".to_string())
+        );
+        assert_eq!(parsed.data[5].1, Datavalue::NotAvailable);
     }
 
     #[test]