Test stream response

rust/Cargo.toml

@@ -22,4 +22,5 @@ members = [
   "models",
   "osp",
   "openvasd",
+  "streamer",
 ]

rust/examples/simple_scan.json

@@ -0,0 +1,45 @@
+{
+  "scan_id": "6c591f83-8f7b-452a-8c78-ba35779e682f",
+  "target": {
+    "hosts": [
+      "127.0.0.1",
+      "::1",
+      "10.0.0.1"
+    ],
+    "ports": [
+      {
+        "protocol": "udp",
+        "range": [
+          {
+            "start": 1,
+            "end": 3000
+          }
+        ]
+      },
+      {
+        "protocol": "tcp",
+        "range": [
+          {
+            "start": 1,
+            "end": 3000
+          }
+        ]
+      }
+    ],
+    "credentials": [
+      {
+        "service": "ssh",
+        "port": 22,
+        "up": {
+          "username": "user",
+          "password": "pw"
+        }
+      }
+    ]
+  },
+  "vts": [
+    {
+      "oid": "1.3.6.1.4.1.25623.1.0.50282"
+    }
+  ]
+}

rust/openvasd/Cargo.toml

@@ -26,5 +26,10 @@ rustls-pemfile = "1.0.2"
 async-trait = "0.1.68"
 clap = { version = "4.3.0", features = ["derive", "env"] }
 toml = "0.7.4"
+tokio-util = { version = "0.7.8", features = ["full"] }
+tokio-serde = "0.8.0"
+tokio-stream = "0.1.14"
+file-format = "0.17.3"
+futures = "0.3.28"
 
 [dev-dependencies]

rust/openvasd/src/controller/entry.rs

@@ -6,6 +6,8 @@
 //!
 //! All known paths must be handled in the entrypoint function.
 
+use std::convert::Infallible;
+
 use std::{fmt::Display, sync::Arc};
 
 use super::{context::Context, quit_on_poison};
@@ -294,7 +296,14 @@ where
         }
         (&Method::GET, Vts) => {
             let (_, oids) = ctx.oids.read()?.clone();
-            Ok(ctx.response.ok(&oids))
+
+            let stream = futures::stream::iter(oids.into_iter().map(
+                |s: String| -> Result<String, Infallible> {
+                    Ok(format!("{},", serde_json::to_string(&s).unwrap()))
+                },
+            ));
+
+            Ok(ctx.response.ok_stream(stream).await)
         }
         _ => Ok(ctx.response.not_found("path", req.uri().path())),
     }

rust/openvasd/src/response.rs

@@ -4,8 +4,9 @@
 
 use std::error::Error;
 
+use futures::Stream;
+use hyper::body::Bytes;
 use serde::Serialize;
-
 type Result = hyper::Response<hyper::Body>;
 
 #[derive(Debug, Default)]
@@ -15,7 +16,39 @@ pub struct Response {
 }
 
 impl Response {
-    #[tracing::instrument]
+    async fn create_stream<S, O, E>(&self, code: hyper::StatusCode, value: S) -> Result
+    where
+        S: Stream<Item = std::result::Result<O, E>> + Send + 'static,
+        O: Into<Bytes> + 'static,
+        E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+    {
+        match hyper::Response::builder()
+            .status(code)
+            .header("Content-Type", "application/json")
+            .header("authentication", &self.authentication)
+            .header("version", &self.version)
+            .body(hyper::Body::wrap_stream(value))
+        {
+            Ok(resp) => resp,
+            Err(e) => {
+                tracing::error!("Error creating response: {}", e);
+                hyper::Response::builder()
+                    .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
+                    .body(hyper::Body::empty())
+                    .unwrap()
+            }
+        }
+    }
+
+    pub async fn ok_stream<S, O, E>(&self, value: S) -> Result
+    where
+        S: Stream<Item = std::result::Result<O, E>> + Send + 'static,
+        O: Into<Bytes> + 'static,
+        E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+    {
+        self.create_stream(hyper::StatusCode::OK, value).await
+    }
+
     fn create<T>(&self, code: hyper::StatusCode, value: &T) -> Result
     where
         T: ?Sized + Serialize + std::fmt::Debug,

rust/streamer/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "streamer"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+models = {path = "../models"}
+osp = {path = "../osp"}
+nasl-interpreter = { path = "../nasl-interpreter" }
+feed = {path = "../feed"}
+storage = { path = "../storage" }
+hyper = { version = "0.14.26", features = ["full", "stream"] }
+tokio = { version = "1.28.1", features = ["full"] }
+tracing = "0.1.37"
+tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
+serde_json = "1.0.96"
+serde = { version = "1.0.163", features = ["derive"] }
+uuid = {version = "1", features = ["v4", "fast-rng", "serde"]}
+hyper-rustls = "0.24.0"
+rustls = "0.21.1"
+tokio-rustls = "0.24.0"
+futures-util = "0.3.28"
+rustls-pemfile = "1.0.2"
+async-trait = "0.1.68"
+clap = { version = "4.3.0", features = ["derive", "env"] }
+toml = "0.7.4"
+http-body-util = "0.1.0-rc.2" 
+tokio-utils = "0.1.2"
+tokio-util = "0.7.8"
+pretty_env_logger = "0.5.0"

rust/streamer/src/main.rs

@@ -0,0 +1,68 @@
+#![deny(warnings)]
+
+use tokio::fs::File;
+
+use tokio_util::codec::{BytesCodec, FramedRead};
+
+use hyper::service::{make_service_fn, service_fn};
+use hyper::{Body, Method, Request, Response, Result, Server, StatusCode};
+
+static INDEX: &str = "/tmp/feed.json";
+static NOTFOUND: &[u8] = b"Not Found";
+
+#[tokio::main]
+async fn main() {
+    pretty_env_logger::init();
+
+    let addr = "127.0.0.1:1337".parse().unwrap();
+
+    let make_service =
+        make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(response_examples)) });
+
+    let server = Server::bind(&addr).serve(make_service);
+
+    println!("Listening on http://{}", addr);
+
+    if let Err(e) = server.await {
+        eprintln!("server error: {}", e);
+    }
+}
+
+async fn response_examples(req: Request<Body>) -> Result<Response<Body>> {
+    match (req.method(), req.uri().path()) {
+        (&Method::GET, "/") | (&Method::GET, "/index.html") => simple_file_send(INDEX).await,
+        (&Method::GET, "/no_file.html") => {
+            // Test what happens when file cannot be be found
+            simple_file_send("this_file_should_not_exist.html").await
+        }
+        _ => Ok(not_found()),
+    }
+}
+
+/// HTTP status code 404
+fn not_found() -> Response<Body> {
+    Response::builder()
+        .status(StatusCode::NOT_FOUND)
+        .body(NOTFOUND.into())
+        .unwrap()
+}
+
+async fn simple_file_send(filename: &str) -> Result<Response<Body>> {
+    // Serve a file by asynchronously reading it by chunks using tokio-util crate.
+
+    if let Ok(file) = File::open(filename).await {
+        let stream = FramedRead::new(file, BytesCodec::new());
+        let body = Body::wrap_stream(stream);
+        //return Ok(Response::new(body));
+        match hyper::Response::builder()
+            .status(hyper::StatusCode::OK)
+            .header(hyper::header::CONTENT_TYPE, "application/json")
+            .header(hyper::header::CONTENT_ENCODING, "chuncked")
+            .body(body) {
+                Ok(resp) => return Ok(resp),
+                _ => return Ok(not_found())
+            }
+    };
+    Ok(not_found())
+}
+