Add: integrate openvasctl into openvasd and make it configurable

rust/models/src/scan.rs

@@ -12,7 +12,7 @@ use super::{scanner_preference::ScannerPreference, target::Target, vt::VT};
     serde(deny_unknown_fields)
 )]
 pub struct Scan {
-    #[cfg_attr(feature = "serde_support", serde(skip_deserializing, default = "uuid"))]
+    #[cfg_attr(feature = "serde_support", serde(default = "uuid"))]
     /// Unique ID of a scan
     pub scan_id: String,
     /// Information about the target to scan

rust/openvasctl/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "openvas"
+name = "openvasctl"
 version = "0.1.0"
 edition = "2021"
 
@@ -15,3 +15,8 @@ storage = { version = "0.1.0", path = "../storage" }
 sysinfo = "0.30.5"
 tokio = { version = "1.36.0", features = ["full"] }
 tracing = "0.1.40"
+serde = { version = "1", features = ["derive"], optional = true }
+
+[features]
+default = ["serde_support"]
+serde_support = ["serde"]

rust/openvasctl/src/config.rs

@@ -1,9 +1,25 @@
 use std::time::Duration;
 
-#[derive(Default)]
+#[derive(Debug, Clone)]
+#[cfg_attr(
+    feature = "serde_support",
+    derive(serde::Serialize, serde::Deserialize),
+    serde(deny_unknown_fields)
+)]
 pub struct Config {
     pub max_queued_scans: Option<usize>,
     pub max_running_scans: Option<usize>,
     pub min_free_mem: Option<u64>,
     pub check_interval: Duration,
 }
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            max_queued_scans: Default::default(),
+            max_running_scans: Default::default(),
+            min_free_mem: Default::default(),
+            check_interval: Duration::from_secs(1),
+        }
+    }
+}

rust/openvasctl/src/ctl.rs

@@ -71,7 +71,7 @@ impl OpenvasController {
 
         let mut scan = match self.remove_running(&scan_id) {
             Some(scan) => scan,
-            None => return Err(OpenvasError::ScanNotFound),
+            None => return Err(OpenvasError::ScanNotFound(scan_id)),
         };
 
         cmd::stop(&scan_id, self.sudo)

rust/openvasctl/src/error.rs

@@ -1,13 +1,23 @@
-use std::io;
+use std::{fmt::Display, io};
 
 pub enum OpenvasError {
-    MissingID,
-    DuplicateScanID,
+    DuplicateScanID(String),
     MissingExec,
-    ScanNotFound,
-    ScanAlreadyExists,
+    ScanNotFound(String),
     CmdError(io::Error),
-    BrokenChannel,
     MaxQueuedScans,
     UnableToRunExec,
 }
+
+impl Display for OpenvasError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            OpenvasError::DuplicateScanID(id) => write!(f, "a scan with ID {id} already exists"),
+            OpenvasError::MissingExec => write!(f, "unable to launch openvas executable"),
+            OpenvasError::ScanNotFound(id) => write!(f, "a scan with ID {id} not found"),
+            OpenvasError::CmdError(e) => write!(f, "unable to run command: {e}"),
+            OpenvasError::MaxQueuedScans => write!(f, "maximum number of queued scan reached"),
+            OpenvasError::UnableToRunExec => write!(f, "unable to run openvas"),
+        }
+    }
+}

rust/openvasctl/src/openvas.rs

@@ -59,7 +59,7 @@ impl ScanController for OpenvasControl {
 
         let mut scan = match self.remove_running(&scan_id) {
             Some(scan) => scan,
-            None => return Err(OpenvasError::ScanNotFound),
+            None => return Err(OpenvasError::ScanNotFound(id.to_string())),
         };
 
         cmd::stop(&scan_id, self.sudo)

rust/openvasctl/src/scheduler.rs

@@ -1,16 +1,11 @@
-use std::{
-    collections::VecDeque,
-    sync::{Arc, Mutex},
-};
+use std::{collections::VecDeque, sync::Arc};
 
 use sysinfo::System;
-use tokio::task;
+use tokio::{sync::Mutex, task};
 
 use crate::{config::Config, ctl::ScanController, error::OpenvasError};
 
-/// Estimated memory a scan will use
-const SCAN_MEM: u64 = 1024 * 1024 * 512;
-
+#[derive(Debug, Clone)]
 pub struct Scheduler<S> {
     queue: Arc<Mutex<VecDeque<models::Scan>>>,
     config: Config,
@@ -25,32 +20,32 @@ where
     S: ScanController + std::marker::Send + std::marker::Sync + 'static,
 {
     /// Create a new OpenvasController
-    pub fn new(config: Option<Config>, controller: S) -> Self {
+    pub fn new(config: Config, controller: S) -> Self {
         Self {
             queue: Default::default(),
-            config: config.unwrap_or_default(),
+            config,
             controller: Arc::new(controller),
         }
     }
 
     /// Add a new scan to the queue. An error is returned, if the queue is either full or the scan
     /// ID is already registered.
     pub async fn add(&mut self, scan: models::Scan) -> Result<(), OpenvasError> {
-        let queue = self.queue.lock().unwrap();
+        let queue = self.queue.lock().await;
         if let Some(max_queued_scans) = self.config.max_queued_scans {
             if queue.len() == max_queued_scans {
                 return Err(OpenvasError::MaxQueuedScans);
             }
         }
         if queue.iter().any(|x| x.scan_id == scan.scan_id) {
-            return Err(OpenvasError::DuplicateScanID);
+            return Err(OpenvasError::DuplicateScanID(scan.scan_id));
         }
-        self.queue.lock().unwrap().push_back(scan);
+        self.queue.lock().await.push_back(scan);
         Ok(())
     }
 
-    fn remove_queue(&mut self, id: &str) -> bool {
-        let mut queue = self.queue.lock().unwrap();
+    async fn remove_queue(&mut self, id: &str) -> bool {
+        let mut queue = self.queue.lock().await;
         if let Some(index) = queue.iter().position(|x| x.scan_id == id) {
             queue.remove(index);
             return true;
@@ -63,7 +58,7 @@ where
     /// as a scan should not actually start, when it is removed. An error is returned, if
     /// the scan in not registered in the scheduler.
     pub async fn stop(&mut self, id: &str) -> Result<(), OpenvasError> {
-        if self.remove_queue(id) {
+        if self.remove_queue(id).await {
             return Ok(());
         }
 
@@ -85,7 +80,7 @@ where
             // TODO: Remove forgotten finished scans
 
             // Are scans in the queue?
-            if self.queue.lock().unwrap().is_empty() {
+            if self.queue.lock().await.is_empty() {
                 continue;
             }
 
@@ -99,15 +94,15 @@ where
             // Check available resources
             sys.refresh_memory();
             if let Some(min_free_mem) = self.config.min_free_mem {
-                if sys.available_memory() - self.controller.num_init() as u64 * SCAN_MEM
+                if sys.available_memory() - self.controller.num_init() as u64 * min_free_mem
                     < min_free_mem
                 {
                     continue;
                 }
             }
 
             // Start next scan in queue
-            let mut queue = self.queue.lock().unwrap();
+            let mut queue = self.queue.lock().await;
             if let Some(scan) = queue.pop_front() {
                 let controller = self.controller.clone();
                 controller.set_init(&scan.scan_id);

rust/openvasd/Cargo.toml

@@ -14,6 +14,7 @@ storage = { path = "../storage" }
 redis-storage = { path = "../redis-storage" }
 infisto = { path = "../infisto" }
 notus = { path = "../notus" }
+openvasctl = { path = "../openvasctl" }
 hyper = { version = "1", features = ["full"] }
 hyper-rustls = "0"
 tokio = { version = "1.28.1", features = ["full"] }

rust/openvasd/src/config.rs

@@ -30,6 +30,54 @@ pub struct Redis {
     pub url: String,
 }
 
+#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+pub struct Wrapper {
+    #[serde(default)]
+    pub wrapper_type: WrapperType,
+    #[serde(default)]
+    pub ospd: OspdWrapper,
+    #[serde(default)]
+    pub openvasctl: openvasctl::config::Config,
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+pub enum WrapperType {
+    #[serde(rename = "ospd")]
+    OSPD,
+    #[serde(rename = "openvasctl")]
+    Openvasctl,
+}
+
+impl Default for WrapperType {
+    fn default() -> Self {
+        Self::OSPD
+    }
+}
+
+impl TypedValueParser for WrapperType {
+    type Value = WrapperType;
+
+    fn parse_ref(
+        &self,
+        cmd: &clap::Command,
+        _: Option<&clap::Arg>,
+        value: &std::ffi::OsStr,
+    ) -> Result<Self::Value, clap::Error> {
+        Ok(match value.to_str().unwrap_or_default() {
+            "ospd" => WrapperType::OSPD,
+            "openvasctl" => WrapperType::Openvasctl,
+            _ => {
+                let mut cmd = cmd.clone();
+                let err = cmd.error(
+                    clap::error::ErrorKind::InvalidValue,
+                    "`{}` is not a wrapper type.",
+                );
+                return Err(err);
+            }
+        })
+    }
+}
+
 #[derive(Deserialize, Serialize, Debug, Clone)]
 pub struct OspdWrapper {
     pub result_check_interval: Duration,
@@ -142,7 +190,7 @@ impl TypedValueParser for StorageType {
                 let mut cmd = cmd.clone();
                 let err = cmd.error(
                     clap::error::ErrorKind::InvalidValue,
-                    "`{}` is not an storage type.",
+                    "`{}` is not a storage type.",
                 );
                 return Err(err);
             }
@@ -186,7 +234,7 @@ pub struct Config {
     #[serde(default)]
     pub tls: Tls,
     #[serde(default)]
-    pub ospd: OspdWrapper,
+    pub wrapper: Wrapper,
     #[serde(default)]
     pub listener: Listener,
     #[serde(default)]
@@ -318,6 +366,43 @@ impl Config {
                     .action(ArgAction::Set)
                     .help("API key that must be set as X-API-KEY header to gain access"),
             )
+            .arg(
+                clap::Arg::new("wrapper-type")
+                    .env("WRAPPER_TYPE")
+                    .long("wrapper-type")
+                    .value_name("ospd,openvasctl")
+                    .value_parser(WrapperType::OSPD)
+                    .help("Type of wrapper used to manage scans")
+            )
+            .arg(
+                clap::Arg::new("max-queued-scans")
+                    .env("MAX_QUEUED_SCANS")
+                    .long("max-queued-scans")
+                    .action(ArgAction::Set)
+                    .help("Maximum number of queued scans")
+            )
+            .arg(
+                clap::Arg::new("max-running-scans")
+                    .env("MAX_RUNNING_SCANS")
+                    .long("max-running-scans")
+                    .action(ArgAction::Set)
+                    .help("Maximum number of active running scans, omit for no limits")
+            )
+            .arg(
+                clap::Arg::new("min-free-mem")
+                    .env("MIN_FREE_MEMORY")
+                    .long("min-free-mem")
+                    .action(ArgAction::Set)
+                    .help("Minimum memory available to start a new scan")
+            )
+            .arg(
+                clap::Arg::new("check-interval")
+                    .env("SCHEDULER_CHECK_INTERVAL")
+                    .long("check_interval")
+                    .value_parser(clap::value_parser!(u64))
+                    .value_name("SECONDS")
+                    .help("Check interval of the Scheduler if a new scan can be started")
+            )
             .arg(
                 clap::Arg::new("ospd-socket")
                     .env("OSPD_SOCKET")
@@ -394,14 +479,29 @@ impl Config {
         if let Some(interval) = cmds.get_one::<u64>("feed-check-interval") {
             config.feed.check_interval = Duration::from_secs(*interval);
         }
+        if let Some(wrapper_type) = cmds.get_one::<WrapperType>("wrapper-type") {
+            config.wrapper.wrapper_type = wrapper_type.clone()
+        }
+        if let Some(max_queued_scans) = cmds.get_one::<usize>("max-queued-scans") {
+            config.wrapper.openvasctl.max_queued_scans = Some(*max_queued_scans)
+        }
+        if let Some(max_running_scans) = cmds.get_one::<usize>("max_running_scans") {
+            config.wrapper.openvasctl.max_running_scans = Some(*max_running_scans)
+        }
+        if let Some(min_free_mem) = cmds.get_one::<u64>("min-free-mem") {
+            config.wrapper.openvasctl.min_free_mem = Some(*min_free_mem)
+        }
+        if let Some(check_interval) = cmds.get_one::<u64>("check-interval") {
+            config.wrapper.openvasctl.check_interval = Duration::from_secs(*check_interval)
+        }
         if let Some(interval) = cmds.get_one::<u64>("result-check-interval") {
-            config.ospd.result_check_interval = Duration::from_secs(*interval);
+            config.wrapper.ospd.result_check_interval = Duration::from_secs(*interval);
         }
         if let Some(path) = cmds.get_one::<PathBuf>("ospd-socket") {
-            config.ospd.socket = path.clone();
+            config.wrapper.ospd.socket = path.clone();
         }
         if let Some(interval) = cmds.get_one::<u64>("read-timeout") {
-            config.ospd.read_timeout = Some(Duration::from_secs(*interval));
+            config.wrapper.ospd.read_timeout = Some(Duration::from_secs(*interval));
         }
 
         if let Some(path) = cmds.get_one::<PathBuf>("feed-path") {
@@ -475,9 +575,15 @@ mod tests {
         assert!(config.tls.key.is_none());
         assert!(config.tls.client_certs.is_none());
 
-        assert_eq!(config.ospd.result_check_interval, Duration::from_secs(1));
-        assert_eq!(config.ospd.socket, PathBuf::from("/var/run/ospd/ospd.sock"));
-        assert!(config.ospd.read_timeout.is_none());
+        assert_eq!(
+            config.wrapper.ospd.result_check_interval,
+            Duration::from_secs(1)
+        );
+        assert_eq!(
+            config.wrapper.ospd.socket,
+            PathBuf::from("/var/run/ospd/ospd.sock")
+        );
+        assert!(config.wrapper.ospd.read_timeout.is_none());
 
         assert_eq!(config.listener.address, ([127, 0, 0, 1], 3000).into());