Add fuzzing test to flowgger

Cargo.toml

@@ -56,6 +56,7 @@ time-tz = "0.3"
 
 [dev-dependencies]
 tempdir = "0.3"
+quickcheck = "1"
 
 [profile.release]
 opt-level = 3

flowgger.toml

@@ -160,3 +160,11 @@ framing = "line"
 format = "rfc3164"
 # Format of the optional timestamp to be prepended to each event
 syslog_prepend_timestamp="[[[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:6]Z]"
+
+
+[test]
+
+###################
+#  Fuzzing Test   #
+###################
+fuzzed_message_count = 500

src/flowgger/config.rs

@@ -11,7 +11,7 @@ use toml::Value;
 /// [`Configuration`]: https://github.com/jedisct1/flowgger/wiki/Configuration
 #[derive(Clone)]
 pub struct Config {
-    config: Value,
+    pub config: Value,
 }
 
 impl Config {

src/flowgger/encoder/ltsv_encoder.rs

@@ -134,9 +134,9 @@ use time::Month;
 #[test]
 fn test_ltsv_full_encode_no_sd() {
     let full_msg = "<23>Aug  6 11:15:24 testhostname appname[69]: 42 - some test message";
-    let expected_msg = "host:testhostname\ttime:1659784524\tmessage:some test message\tfull_message:<23>Aug  6 11:15:24 testhostname appname[69]: 42 - some test message\tlevel:7\tfacility:2\tappname:appname\tprocid:69\tmsgid:42";
-    let cfg = Config::from_string("[input]\n[input.ltsv_schema]\nformat = \"ltsv\"\n").unwrap();
     let ts = ts_from_partial_date_time(Month::August, 6, 11, 15, 24);
+    let expected_msg = format!("host:testhostname\ttime:{}\tmessage:some test message\tfull_message:<23>Aug  6 11:15:24 testhostname appname[69]: 42 - some test message\tlevel:7\tfacility:2\tappname:appname\tprocid:69\tmsgid:42", ts);
+    let cfg = Config::from_string("[input]\n[input.ltsv_schema]\nformat = \"ltsv\"\n").unwrap();
 
     let record = Record {
         ts,
@@ -159,9 +159,10 @@ fn test_ltsv_full_encode_no_sd() {
 #[test]
 fn test_ltsv_full_encode_multiple_sd() {
     let full_msg = "<23>Aug  6 11:15:24 testhostname appname[69]: 42 [someid a=\"b\" c=\"123456\"][someid2 a2=\"b2\" c2=\"123456\"] some test message";
-    let expected_msg = "a:b\tc:123456\ta2:b2\tc2:123456\thost:testhostname\ttime:1659784524\tmessage:some test message\tfull_message:<23>Aug  6 11:15:24 testhostname appname[69]: 42 [someid a=\"b\" c=\"123456\"][someid2 a2=\"b2\" c2=\"123456\"] some test message\tlevel:7\tfacility:2\tappname:appname\tprocid:69\tmsgid:42";
-    let cfg = Config::from_string("[input]\n[input.ltsv_schema]\nformat = \"ltsv\"\n").unwrap();
     let ts = ts_from_partial_date_time(Month::August, 6, 11, 15, 24);
+    let expected_msg = format!("a:b\tc:123456\ta2:b2\tc2:123456\thost:testhostname\ttime:{}\tmessage:some test message\tfull_message:<23>Aug  6 11:15:24 testhostname appname[69]: 42 [someid a=\"b\" c=\"123456\"][someid2 a2=\"b2\" c2=\"123456\"] some test message\tlevel:7\tfacility:2\tappname:appname\tprocid:69\tmsgid:42", ts);
+    let cfg = Config::from_string("[input]\n[input.ltsv_schema]\nformat = \"ltsv\"\n").unwrap();
+
 
     let record = Record {
         ts,

src/flowgger/input/mod.rs

@@ -7,7 +7,7 @@ mod tcp;
 #[cfg(feature = "tls")]
 mod tls;
 #[cfg(feature = "syslog")]
-mod udp_input;
+pub mod udp_input;
 
 #[cfg(feature = "file")]
 pub use self::file::FileInput;

src/flowgger/input/udp_input.rs

@@ -97,7 +97,7 @@ impl Input for UdpInput {
 /// but could not be handled
 /// `Invalid UTF-8 input`: Bubble up from handle_record, the record is not in a valid utf-8 format, it could be a non
 /// supported compression format
-fn handle_record_maybe_compressed(
+pub fn handle_record_maybe_compressed(
     line: &[u8],
     tx: &SyncSender<Vec<u8>>,
     decoder: &Box<dyn Decoder>,

src/flowgger/mod.rs

@@ -1,9 +1,9 @@
-mod config;
-mod decoder;
-mod encoder;
-mod input;
-mod merger;
-mod output;
+pub mod config;
+pub mod decoder;
+pub mod encoder;
+pub mod input;
+pub mod merger;
+pub mod output;
 mod record;
 mod splitter;
 mod utils;
@@ -275,12 +275,12 @@ fn get_encoder_passthrough(config: &Config) -> Box<dyn Encoder + Send> {
 }
 
 #[cfg(feature = "rfc3164")]
-fn get_decoder_rfc3164(config: &Config) -> Box<dyn Decoder + Send> {
+pub fn get_decoder_rfc3164(config: &Config) -> Box<dyn Decoder + Send> {
     Box::new(RFC3164Decoder::new(config)) as Box<dyn Decoder + Send>
 }
 
 #[cfg(feature = "rfc3164")]
-fn get_encoder_rfc3164(config: &Config) -> Box<dyn Encoder + Send> {
+pub fn get_encoder_rfc3164(config: &Config) -> Box<dyn Encoder + Send> {
     Box::new(RFC3164Encoder::new(config)) as Box<dyn Encoder + Send>
 }
 

tests/fuzzer.rs

@@ -0,0 +1,156 @@
+use flowgger;
+use quickcheck;
+
+use quickcheck::QuickCheck;
+
+use flowgger::flowgger::config::Config;
+use flowgger::flowgger::encoder::Encoder;
+use flowgger::flowgger::decoder::Decoder;
+use flowgger::flowgger::merger;
+use flowgger::flowgger::output;
+
+use std::sync::mpsc::{Receiver, sync_channel, SyncSender};
+
+use flowgger::flowgger::get_decoder_rfc3164;
+use flowgger::flowgger::get_encoder_rfc3164;
+use flowgger::flowgger::input::udp_input::handle_record_maybe_compressed;
+
+use self::merger::{LineMerger, Merger};
+use self::output::FileOutput;
+use self::output::Output;
+
+use std::sync::{Arc, Mutex};
+use toml::Value;
+use std::fs;
+use std::{thread, time};
+
+const DEFAULT_CONFIG_FILE: &str = "flowgger.toml";
+const DEFAULT_OUTPUT_FILEPATH: &str = "output.log";
+const DEFAULT_QUEUE_SIZE: usize = 10_000_000;
+
+const DEFAULT_OUTPUT_FORMAT: &str = "gelf";
+const DEFAULT_OUTPUT_FRAMING: &str = "noop";
+const DEFAULT_OUTPUT_TYPE: &str = "file";
+
+const DEFAULT_FUZZED_MESSAGE_COUNT: u64 = 500;
+
+fn get_file_output(config: &Config) -> Box<dyn Output> {
+    Box::new(FileOutput::new(config)) as Box<dyn Output>
+}
+
+pub fn start_file_output(config: &Config, rx: Receiver<Vec<u8>>){
+
+    let output_format = config
+        .lookup("output.format")
+        .map_or(DEFAULT_OUTPUT_FORMAT, |x| {
+            x.as_str().expect("output.format must be a string")
+        });
+
+    let output = get_file_output(&config);
+    let output_type = config
+        .lookup("output.type")
+        .map_or(DEFAULT_OUTPUT_TYPE, |x| {
+            x.as_str().expect("output.type must be a string")
+        });
+
+    let _output_framing = match config.lookup("output.framing") {
+        Some(framing) => framing.as_str().expect("output.framing must be a string"),
+        None => match (output_format, output_type) {
+            ("capnp", _) | (_, "kafka") => "noop",
+            (_, "debug") | ("ltsv", _) => "line",
+            ("gelf", _) => "nul",
+            _ => DEFAULT_OUTPUT_FRAMING,
+        },
+    };
+    let merger: Option<Box<dyn Merger>> = Some(Box::new(LineMerger::new(&config)) as Box<dyn Merger>);
+
+    let arx = Arc::new(Mutex::new(rx));
+    output.start(arx, merger);
+
+}
+
+pub fn get_config() -> Config {
+    let mut config = match Config::from_path(DEFAULT_CONFIG_FILE) {
+        Ok(config) => config,
+        Err(e) => panic!(
+            "Unable to read the config file [{}]: {}",
+            "flowgger.toml",
+            e.to_string()
+        ),
+    };
+
+    if let Some(entry) = config.config.get_mut("output").unwrap().get_mut("file_rotation_time"){
+        *entry = Value::Integer(0);
+    }else{
+        panic!("Failed to find config entry");
+    }
+
+    return config;
+}
+
+pub fn remove_output_file(file_output_path: &str){
+    fs::remove_file(file_output_path);
+}
+
+pub fn fuzz_target_rfc3164(data: &[u8]) {
+    let config = get_config();
+    let file_output_path = config.lookup("output.file_path").map_or(DEFAULT_OUTPUT_FILEPATH, |x| {
+        x.as_str().expect("File output path missing in config")
+    });
+    remove_output_file(&file_output_path);
+
+    if let Ok(s) = std::str::from_utf8(data) {
+        let (tx, rx): (SyncSender<Vec<u8>>, Receiver<Vec<u8>>) = sync_channel(DEFAULT_QUEUE_SIZE);
+        start_file_output(&config, rx);
+
+        let encoder = get_encoder_rfc3164(&config);
+        let decoder = get_decoder_rfc3164(&config);
+        let (decoder, encoder): (Box<dyn Decoder>, Box<dyn Encoder>) =
+            (decoder.clone_boxed(), encoder.clone_boxed());
+        let result = handle_record_maybe_compressed(s.as_bytes(), &tx, &decoder, &encoder);
+
+        match result {
+            Ok(_) => {
+                drop(tx);
+                thread::sleep(time::Duration::from_millis(100));
+
+                let file_contents = match fs::read_to_string(file_output_path){
+                    Ok(contents) => contents,
+                    Err(_) => {
+                        println!("Failed to read file");
+                        "".to_string()
+                    }
+                };
+
+                let split_file_content: Vec<&str> = file_contents.split(" ").filter(|s| !s.is_empty()).collect();
+                let split_input: Vec<&str> = s.split(" ").filter(|s| !s.is_empty()).collect();
+
+                let hostnames_match = split_file_content[3].trim() == split_input[3].trim();
+                let appnames_match = split_file_content[4].trim() == split_input[4].trim();
+
+                if !(hostnames_match && appnames_match){
+                    panic!("Log output invalid");
+                }
+            }
+            Err(_) => {
+            }
+        }
+
+
+    }
+}
+
+
+#[test]
+fn test_fuzzer(){
+    let config = get_config();
+    let fuzzed_message_count = match config.lookup("test.fuzzed_message_count"){
+        Some(count) => count.as_integer().unwrap() as u64,
+        None => DEFAULT_FUZZED_MESSAGE_COUNT,
+    };
+
+    fn fuzz(data: String){
+        fuzz_target_rfc3164(data.as_bytes());
+    }
+    QuickCheck::new().max_tests(fuzzed_message_count).quickcheck(fuzz as fn(String));
+}