rust: intercept module review

Author: rizsotto

rust/bear/src/bin/wrapper.rs

@@ -18,8 +18,10 @@
 extern crate core;
 
 use anyhow::{Context, Result};
-use bear::intercept::reporter::{Reporter, TcpReporter};
-use bear::intercept::{Event, Execution, ProcessId, KEY_DESTINATION};
+use bear::ipc::tcp::ReporterOnTcp;
+use bear::ipc::Reporter;
+use bear::ipc::{Event, Execution, ProcessId};
+use bear::modes::intercept::KEY_DESTINATION;
 use std::path::{Path, PathBuf};
 
 /// Implementation of the wrapper process.
@@ -108,7 +110,7 @@ fn report(execution: Execution) -> Result<()> {
     std::env::var(KEY_DESTINATION)
         .with_context(|| format!("${} is missing from the environment", KEY_DESTINATION))
         // Create a new reporter
-        .and_then(TcpReporter::new)
+        .and_then(ReporterOnTcp::new)
         .with_context(|| "Cannot create TCP execution reporter")
         // Report the execution
         .and_then(|reporter| reporter.report(event))

rust/bear/src/fixtures.rs

@@ -7,6 +7,13 @@ pub mod fixtures {
         ($($x:expr),*) => (vec![$($x.to_string()),*]);
     }
 
+    #[macro_export]
+    macro_rules! map_of_strings {
+        ($($k:expr => $v:expr),* $(,)?) => {{
+            core::convert::From::from([$(($k.to_string(), $v.to_string()),)*])
+        }};
+    }
+
     #[macro_export]
     macro_rules! vec_of_pathbuf {
         ($($x:expr),*) => (vec![$(PathBuf::from($x)),*]);

rust/bear/src/input.rs

@@ -8,7 +8,7 @@ use std::io::BufReader;
 use std::path::PathBuf;
 
 use super::args;
-use super::intercept::Execution;
+use super::ipc::Execution;
 
 /// Responsible for reading the build events from the intercept mode.
 ///

rust/bear/src/intercept/collector.rs

@@ -1,46 +1,60 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
+//! The module contains the intercept reporting and collecting functionality.
+//!
+//! When a command execution is intercepted, the interceptor sends the event to the collector.
+//! This happens in two different processes, requiring a communication channel between these
+//! processes.
+//!
+//! The module provides abstractions for the reporter and the collector. And it also defines
+//! the data structures that are used to represent the events.
+
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::io::{Read, Write};
 use std::path::PathBuf;
+use std::sync::mpsc::Sender;
 
-use chrono::Utc;
-use rand::random;
-use serde::{Deserialize, Serialize};
+pub mod tcp;
 
-pub mod collector;
-pub mod reporter;
+/// Represents the remote sink of supervised process events.
+///
+/// This allows the reporters to send events to a remote collector.
+pub trait Reporter {
+    fn report(&self, event: Event) -> Result<(), anyhow::Error>;
+}
 
-/// Reporter id is a unique identifier for a reporter.
+/// Represents the local sink of supervised process events.
 ///
-/// It is used to identify the process that sends the execution report.
-/// Because the OS PID is not unique across a single build (PIDs are
-/// recycled), we need to use a new unique identifier to identify the process.
-#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
-pub struct ReporterId(pub u64);
+/// The collector is responsible for collecting the events from the reporters.
+///
+/// To share the collector between threads, we use the `Arc` type to wrap the
+/// collector. This way we can clone the collector and send it to other threads.
+pub trait Collector {
+    /// Returns the address of the collector.
+    ///
+    /// The address is in the format of `ip:port`.
+    fn address(&self) -> String;
 
-impl ReporterId {
-    pub fn new() -> Self {
-        let id = random::<u64>();
-        ReporterId(id)
-    }
-}
+    /// Collects the events from the reporters.
+    ///
+    /// The events are sent to the given destination channel.
+    ///
+    /// The function returns when the collector is stopped. The collector is stopped
+    /// when the `stop` method invoked (from another thread).
+    fn collect(&self, destination: Sender<Envelope>) -> Result<(), anyhow::Error>;
 
-/// Process id is a OS identifier for a process.
-#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
-pub struct ProcessId(pub u32);
+    /// Stops the collector.
+    fn stop(&self) -> Result<(), anyhow::Error>;
+}
 
-/// Execution is a representation of a process execution.
+/// Envelope is a wrapper around the event.
 ///
-/// It does not contain information about the outcome of the execution,
-/// like the exit code or the duration of the execution. It only contains
-/// the information that is necessary to reproduce the execution.
+/// It contains the reporter id, the timestamp of the event and the event itself.
 #[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
-pub struct Execution {
-    pub executable: PathBuf,
-    pub arguments: Vec<String>,
-    pub working_dir: PathBuf,
-    pub environment: HashMap<String, String>,
+pub struct Envelope {
+    pub rid: ReporterId,
+    pub timestamp: u64,
+    pub event: Event,
 }
 
 /// Represent a relevant life cycle event of a process.
@@ -54,58 +68,27 @@ pub struct Event {
     pub execution: Execution,
 }
 
-/// Envelope is a wrapper around the event.
+/// Execution is a representation of a process execution.
 ///
-/// It contains the reporter id, the timestamp of the event and the event itself.
+/// It does not contain information about the outcome of the execution,
+/// like the exit code or the duration of the execution. It only contains
+/// the information that is necessary to reproduce the execution.
 #[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
-pub struct Envelope {
-    pub rid: ReporterId,
-    pub timestamp: u64,
-    pub event: Event,
+pub struct Execution {
+    pub executable: PathBuf,
+    pub arguments: Vec<String>,
+    pub working_dir: PathBuf,
+    pub environment: HashMap<String, String>,
 }
 
-impl Envelope {
-    pub fn new(rid: &ReporterId, event: Event) -> Self {
-        let timestamp = Utc::now().timestamp_millis() as u64;
-        Envelope {
-            rid: rid.clone(),
-            timestamp,
-            event,
-        }
-    }
-
-    /// Read an envelope from a reader using TLV format.
-    ///
-    /// The envelope is serialized using JSON and the length of the JSON
-    /// is written as a 4 byte big-endian integer before the JSON.
-    pub fn read_from(reader: &mut impl Read) -> Result<Self, anyhow::Error> {
-        let mut length_bytes = [0; 4];
-        reader.read_exact(&mut length_bytes)?;
-        let length = u32::from_be_bytes(length_bytes) as usize;
-
-        let mut buffer = vec![0; length];
-        reader.read_exact(&mut buffer)?;
-        let envelope = serde_json::from_slice(buffer.as_ref())?;
-
-        Ok(envelope)
-    }
-
-    /// Write an envelope to a writer using TLV format.
-    ///
-    /// The envelope is serialized using JSON and the length of the JSON
-    /// is written as a 4 byte big-endian integer before the JSON.
-    pub fn write_into(&self, writer: &mut impl Write) -> Result<u32, anyhow::Error> {
-        let serialized_envelope = serde_json::to_string(&self)?;
-        let bytes = serialized_envelope.into_bytes();
-        let length = bytes.len() as u32;
-
-        writer.write_all(&length.to_be_bytes())?;
-        writer.write_all(&bytes)?;
-
-        Ok(length)
-    }
-}
+/// Reporter id is a unique identifier for a reporter.
+///
+/// It is used to identify the process that sends the execution report.
+/// Because the OS PID is not unique across a single build (PIDs are
+/// recycled), we need to use a new unique identifier to identify the process.
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
+pub struct ReporterId(pub u64);
 
-/// Declare the environment variable name for the reporter address.
-pub const KEY_DESTINATION: &str = "INTERCEPT_REPORTER_ADDRESS";
-pub const KEY_PRELOAD_PATH: &str = "LD_PRELOAD";
+/// Process id is a OS identifier for a process.
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
+pub struct ProcessId(pub u32);