jjnicola · jjnicola · Jul 4, 2023 · Jul 5, 2023
diff --git a/rust/Cargo.lock b/rust/Cargo.lock
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
@@ -22,4 +22,5 @@ members = [
   "models",
   "osp",
   "openvasd",
+  "streamer",
 ]
diff --git a/rust/examples/simple_scan.json b/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"
+    }
+  ]
+}
diff --git a/rust/openvasd/Cargo.toml b/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]
diff --git a/rust/openvasd/src/controller/entry.rs b/rust/openvasd/src/controller/entry.rs
@@ -6,10 +6,15 @@
 //!
 //! All known paths must be handled in the entrypoint function.
 
+use std::convert::Infallible;
+use std::marker::PhantomData;
 use std::{fmt::Display, sync::Arc};
-
+use futures::future::Map;
+use futures::stream::futures_unordered::IntoIter;
+use futures::stream::{self, StreamExt};
 use super::{context::Context, quit_on_poison};
-use hyper::{Body, Method, Request, Response};
+use futures_util::Stream;
+use hyper::{Body, Method, Request, Response, body::Bytes};
 
 use crate::scan::{Error, ScanDeleter, ScanStarter, ScanStopper};
 /// The supported paths of scannerd
@@ -105,11 +110,14 @@ where
 ///
 /// First it will be checked if a known path is requested and if the method is supported.
 /// Than corresponding functions will be called to handle the request.
-pub async fn entrypoint<'a, S>(
+pub async fn entrypoint<'a, S, J, O, E>(
     req: Request<Body>,
     ctx: Arc<Context<S>>,
 ) -> Result<Response<Body>, Error>
 where
+    J: Stream<Item = Result<O, E>> + Send + 'static,
+    O: Into<Bytes> + 'static,
+    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
     S: ScanStarter
         + ScanStopper
         + ScanDeleter
@@ -294,8 +302,149 @@ where
         }
         (&Method::GET, Vts) => {
             let (_, oids) = ctx.oids.read()?.clone();
-            Ok(ctx.response.ok(&oids))
+
+            let mut v: Vec<Foo<J, O, E>> = Vec::new();
+            let j1 = Foo::new(vec!['['.to_string()], Box::new(|s: String| -> Result<String, Infallible>
+                             {
+                                 Ok(s)
+                             }));
+            v.push(j1);
+
+
+            let j2 = Foo::new(oids, Box::new(|s: String| -> Result<String, Infallible>
+                             {
+                                 Ok(format!("{},",serde_json::to_string(&s).unwrap()).to_string())
+                             }));
+            v.push(j2);
+
+
+            let j3 = Foo::new(vec![']'.to_string()], Box::new(|s: String| -> Result<String, Infallible>
+                             {
+                                 Ok(s)
+                             }));
+            v.push(j3);
+
+            let stream = futures::stream::iter(v.into_iter().map(
+                |v| v.exec() 
+            ));
+            Ok(ctx.response.ok_stream(stream).await)
+
         }
         _ => Ok(ctx.response.not_found("path", req.uri().path())),
     }
 }
+
+struct Foo<J, O, E>
+where
+    J: Stream<Item = Result<O, E>> + Send + 'static,
+    O: Into<Bytes> + 'static,
+    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+{
+    pub v: Vec<String>,
+    pub c: Box<dyn Fn(String) -> Result<String, Infallible> + 'static>,
+    p: PhantomData<J>
+}
+
+impl<J, O, E> Foo<J, O, E>
+where
+    J: Stream<Item = Result<O, E>> + Send + 'static,
+    O: Into<Bytes> + 'static,
+    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+{
+    fn new(v: Vec<String>, c: impl Fn(String) -> Result<String, Infallible> + 'static) -> Self {
+        Self {
+            v,
+            c: Box::new(c),
+            p: PhantomData
+        }
+    }
+
+    pub fn exec(self) ->  Result<String, Infallible> {
+        let e = self.v.iter().map(
+            |s| (self.c)(s.to_string())
+        ).collect();
+        e
+    }
+}
+
+// let stream = futures::stream::iter(
+//                oids
+//                    .into_iter()
+//                    .map(|s: String| -> Result<String, Infallible>
+//                         {
+//                             Ok(format!("{},",serde_json::to_string(&s).unwrap()).to_string())
+//                         }
+//                    )
+//            );
+//fn async_read() -> impl Stream<Item = Result<u8, std::io::Error>> {
+//    let f = File::open("/dev/random").expect("Could not open file");
+//    let reader = BufReader::new(f);
+//    stream::iter(reader.bytes())
+//}
+// 
+//async fn async_main(oids: Vec<String>) {
+//    while let Some(b) = async_read().next().await {
+//        println!("{:?}", b);
+//    }
+//}
+
+
+
+//use futures::stream::futures_unordered::FuturesUnordered;
+//async fn read_oids(oids: Vec<String>) -> Vec<String> {
+//    oids.iter()
+//        .map(|oid| oid)
+//        .collect::<FuturesUnordered<_>>()
+//        .collect::<Vec<_>>()
+//        .await
+//}
+// 
+//async fn read_oid(oid: &String) -> String {
+//    let oidstr = serde_json::to_string(&oid).unwrap();
+//    println!("{:?}", oid);
+//    oidstr
+//}
+
+//struct Oids(Vec<String>);
+// 
+//impl Oids {
+//    fn new(oids: Vec<String>) -> Self {
+//        Self(oids)
+//    }
+//}
+// 
+//impl AsyncRead for Oids {
+//    fn poll_read(
+//        mut self: Pin<&mut Self>,
+//        _cx: &mut Context<'_>,
+//        buf: &mut ReadBuf<'_>,
+//    ) -> Poll<io::Result<()>> {
+//        let amt = std::cmp::min(self.len(), buf.remaining());
+//        let (a, b) = self.split_at(amt);
+//        buf.put_slice(a);
+//        *self = b;
+//        Poll::Ready(Ok(()))
+//    }
+//}
+
+//use futures_util::stream::iter;
+//async fn vec_to_stream<S, O, E, I>(oids: Vec<String>) -> Iter<std::vec::IntoIter<std::string::String>>
+//where
+//    S: Stream<Item = Result<O, E>> + Send + 'static,
+//    O: Into<Bytes> + 'static,
+//    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+//    I: IntoIterator,
+//{
+//    //let v: Vec<Result<_, std::io::Error>> = oids.into_iter().map(|x| Ok(x)).collect();
+//    //let v: Vec<Result<_, std::io::Error>> = oids.into_iter().map(|x| Ok(x)).collect();
+// 
+//    
+//    futures_util::stream::iter(oids)
+//    
+// 
+//    //hyper::body::HttpBody::Data(stream)
+//}
+
+//Iter<std::vec::IntoIter<std::result::Result<std::string::String, std::io::Error>>>
+
+
diff --git a/rust/openvasd/src/response.rs b/rust/openvasd/src/response.rs
@@ -2,10 +2,14 @@
 //
 // SPDX-License-Identifier: GPL-2.0-or-later
 
-use std::error::Error;
+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 +19,40 @@ pub struct Response {
 }
 
 impl Response {
-    #[tracing::instrument]
+    async fn create_stream<J, O, E>(&self, code: hyper::StatusCode, value: J) -> Result
+    where
+        J: 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<J, O, E>(&self, value: J) -> Result
+    where
+        J: 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,
@@ -49,7 +86,7 @@ impl Response {
             }
         }
     }
-
+    
     pub fn ok<T>(&self, value: &T) -> Result
     where
         T: ?Sized + Serialize + std::fmt::Debug,

diff --git a/rust/streamer/Cargo.toml b/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"