fix/feat: Improve io and fix serde features (#50)

* fix/feat: Improve io and fix serde features
* fix: test with no default features

Signed-off-by: Jeremy HERGAULT <jeremy.hergault@worldline.com>

Author: reneca

Cargo.toml

@@ -28,6 +28,7 @@ thiserror = "2"
 aquamarine = "0.6"
 bytes = "1"
 chrono = "0.4"
+serde = { version = "1", features = ["derive"] }
 url = { version = "2", features = ["serde"] }
 tokio = { version = "1", features = ["fs", "macros", "net", "parking_lot", "rt", "rt-multi-thread", "signal", "sync", "time"] }
 config = { version = "0.15", default-features = false, features = ["toml", "json", "yaml", "json5", "convert-case", "async"] }

cargo-prosa/Cargo.toml

@@ -18,7 +18,7 @@ thiserror.workspace = true
 aquamarine.workspace = true
 clap = "4"
 clap_complete = "4"
-serde = "1"
+serde.workspace = true
 toml.workspace = true
 toml_edit = { version = "0.22", features = ["serde"] }
 serde_json = "1"

prosa/Cargo.toml

@@ -43,6 +43,7 @@ bytes.workspace = true
 tracing = "0.1"
 tracing-subscriber = {version = "0.3", features = ["std", "env-filter"]}
 thiserror.workspace = true
+base64 = "0.22"
 url = { version = "2", features = ["serde"] }
 rlimit = "0.10"
 
@@ -53,7 +54,7 @@ tokio.workspace = true
 tokio-openssl = { version = "0.6", optional = true }
 async-http-proxy = { version = "1", optional = true, features = ["runtime-tokio","basic-auth"] }
 
-serde = { version = "1", features = ["derive"] }
+serde.workspace = true
 config.workspace = true
 glob = { version = "0.3" }
 toml.workspace = true

prosa/src/io.rs

@@ -1,6 +1,7 @@
 //! Module that define IO that could be use by a ProSA processor
 use std::{
     fmt,
+    hash::{Hash, Hasher},
     net::{SocketAddrV4, SocketAddrV6},
     path::Path,
 };
@@ -50,7 +51,7 @@ pub fn url_is_ssl(url: &Url) -> bool {
 }
 
 /// Internal Socket adress enum to define IPv4, IPv6 and unix socket.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum SocketAddr {
     #[cfg(target_family = "unix")]
     /// UNIX socket address
@@ -73,6 +74,50 @@ impl SocketAddr {
         }
     }
 
+    /// Returns `true` if the adress is a UNIX address, and `false` otherwise.
+    pub fn is_unix(&self) -> bool {
+        #[cfg(target_family = "unix")]
+        {
+            matches!(self, SocketAddr::Unix(_))
+        }
+        #[cfg(not(target_family = "unix"))]
+        {
+            false
+        }
+    }
+
+    /// Returns `true` if the adress is an IPV4 address, and `false` otherwise.
+    pub fn is_ipv4(&self) -> bool {
+        matches!(self, SocketAddr::V4(_))
+    }
+
+    /// Returns `true` if the adress is an IPV6 address, and `false` otherwise.
+    pub fn is_ipv6(&self) -> bool {
+        matches!(self, SocketAddr::V6(_))
+    }
+
+    /// Returns the IP address associated with this socket address.
+    pub fn ip(&self) -> std::net::IpAddr {
+        match self {
+            #[cfg(target_family = "unix")]
+            SocketAddr::Unix(_) => std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
+            SocketAddr::V4(ipv4) => std::net::IpAddr::V4(*ipv4.ip()),
+            SocketAddr::V6(ipv6) => std::net::IpAddr::V6(*ipv6.ip()),
+        }
+    }
+
+    /// Changes the IP address associated with this socket address.
+    pub const fn set_ip(&mut self, new_ip: std::net::IpAddr) {
+        match new_ip {
+            std::net::IpAddr::V4(ipv4_addr) => {
+                *self = SocketAddr::V4(SocketAddrV4::new(ipv4_addr, self.port()))
+            }
+            std::net::IpAddr::V6(ipv6_addr) => {
+                *self = SocketAddr::V6(SocketAddrV6::new(ipv6_addr, self.port(), 0, 0))
+            }
+        }
+    }
+
     /// Returns the port number associated with this socket address.
     pub const fn port(&self) -> u16 {
         match self {
@@ -106,6 +151,17 @@ impl PartialEq for SocketAddr {
     }
 }
 
+impl Hash for SocketAddr {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        match self {
+            #[cfg(target_family = "unix")]
+            SocketAddr::Unix(unix) => unix.as_pathname().hash(state),
+            SocketAddr::V4(ipv4) => ipv4.hash(state),
+            SocketAddr::V6(ipv6) => ipv6.hash(state),
+        }
+    }
+}
+
 impl fmt::Display for SocketAddr {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
@@ -123,6 +179,27 @@ impl fmt::Display for SocketAddr {
     }
 }
 
+impl<I: Into<std::net::IpAddr>> From<(I, u16)> for SocketAddr {
+    fn from(pieces: (I, u16)) -> Self {
+        match pieces.0.into() {
+            std::net::IpAddr::V4(ipv4) => SocketAddr::V4(SocketAddrV4::new(ipv4, pieces.1)),
+            std::net::IpAddr::V6(ipv6) => SocketAddr::V6(SocketAddrV6::new(ipv6, pieces.1, 0, 0)),
+        }
+    }
+}
+
+impl From<std::net::SocketAddrV4> for SocketAddr {
+    fn from(ipv4: std::net::SocketAddrV4) -> Self {
+        SocketAddr::V4(ipv4)
+    }
+}
+
+impl From<std::net::SocketAddrV6> for SocketAddr {
+    fn from(ipv6: std::net::SocketAddrV6) -> Self {
+        SocketAddr::V6(ipv6)
+    }
+}
+
 impl From<std::net::SocketAddr> for SocketAddr {
     fn from(addr: std::net::SocketAddr) -> Self {
         match addr {

prosa/src/io/listener.rs

@@ -176,6 +176,7 @@ impl StreamListener {
     /// use prosa_utils::config::ssl::{SslConfig, SslConfigContext};
     /// use prosa::io::listener::StreamListener;
     ///
+    /// # #[cfg(feature="openssl")]
     /// async fn accepting() -> Result<(), io::Error> {
     ///     let ssl_acceptor = SslConfig::default().init_tls_server_context(None).unwrap().build();
     ///     let stream_listener: StreamListener = StreamListener::bind("0.0.0.0:10000").await?.ssl_acceptor(ssl_acceptor, None);
@@ -232,6 +233,7 @@ impl StreamListener {
     /// use prosa_utils::config::ssl::{SslConfig, SslConfigContext};
     /// use prosa::io::listener::StreamListener;
     ///
+    /// # #[cfg(feature="openssl")]
     /// async fn accepting() -> Result<(), io::Error> {
     ///     let ssl_acceptor = SslConfig::default().init_tls_server_context(None).unwrap().build();
     ///     let stream_listener: StreamListener = StreamListener::bind("0.0.0.0:10000").await?.ssl_acceptor(ssl_acceptor, None);

prosa/src/io/stream.rs

@@ -12,6 +12,7 @@ use prosa_utils::config::ssl::SslConfig;
 #[cfg(feature = "openssl")]
 use prosa_utils::config::ssl::SslConfigContext;
 
+use base64::{Engine as _, engine::general_purpose::STANDARD};
 use serde::{Deserialize, Serialize};
 use tokio::{
     io::{AsyncRead, AsyncWrite, ReadBuf},
@@ -41,7 +42,7 @@ pub enum Stream {
 }
 
 impl Stream {
-    /// Returns the local address that this stream is bound to.
+    /// Returns the socket address of the remote peer of this TCP connection.
     ///
     /// ```
     /// use tokio::io;
@@ -51,10 +52,41 @@ impl Stream {
     /// use std::net::{Ipv4Addr, SocketAddrV4};
     ///
     /// async fn accepting() -> Result<(), io::Error> {
-    ///     let stream: Stream = Stream::connect_tcp("127.0.0.1:80").await?;
+    ///     let stream: Stream = Stream::connect_tcp("127.0.0.1:8080").await?;
+    ///
+    ///     assert_eq!(stream.peer_addr()?,
+    ///                SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 8080)));
+    ///
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn peer_addr(&self) -> Result<SocketAddr, io::Error> {
+        match self {
+            #[cfg(target_family = "unix")]
+            Stream::Unix(s) => s.peer_addr().map(|addr| addr.into()),
+            Stream::Tcp(s) => s.peer_addr().map(|addr| addr.into()),
+            #[cfg(feature = "openssl")]
+            Stream::OpenSsl(s) => s.get_ref().peer_addr().map(|addr| addr.into()),
+            #[cfg(feature = "http-proxy")]
+            Stream::TcpHttpProxy(s) => s.peer_addr().map(|addr| addr.into()),
+            #[cfg(all(feature = "openssl", feature = "http-proxy"))]
+            Stream::OpenSslHttpProxy(s) => s.get_ref().peer_addr().map(|addr| addr.into()),
+        }
+    }
+
+    /// Returns the local address that this stream is bound to.
+    ///
+    /// ```
+    /// use tokio::io;
+    /// use url::Url;
+    /// use prosa::io::stream::Stream;
+    /// use prosa::io::SocketAddr;
+    /// use std::net::{IpAddr, Ipv4Addr, SocketAddrV4};
+    ///
+    /// async fn accepting() -> Result<(), io::Error> {
+    ///     let stream: Stream = Stream::connect_tcp("127.0.0.1:8080").await?;
     ///
-    ///     assert_eq!(stream.local_addr()?,
-    ///                SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 80)));
+    ///     assert_eq!(stream.local_addr()?.ip(), IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
     ///
     ///     Ok(())
     /// }
@@ -752,6 +784,36 @@ impl TargetSetting {
         self.ssl.is_some() || url_is_ssl(&self.url)
     }
 
+    /// Method to get authentication value out of URL username/password
+    ///
+    /// - If user password is provided, it return *Basic* authentication with base64 encoded username:password
+    /// - If only password is provided, it return *Bearer* authentication with the password as token
+    ///
+    /// ```
+    /// use url::Url;
+    /// use prosa::io::stream::TargetSetting;
+    ///
+    /// let basic_auth_target = TargetSetting::from(Url::parse("http://user:pass@localhost:8080").unwrap());
+    /// assert_eq!(Some(String::from("Basic dXNlcjpwYXNz")), basic_auth_target.get_authentication());
+    ///
+    /// let bearer_auth_target = TargetSetting::from(Url::parse("http://:token@localhost:8080").unwrap());
+    /// assert_eq!(Some(String::from("Bearer token")), bearer_auth_target.get_authentication());
+    /// ```
+    pub fn get_authentication(&self) -> Option<String> {
+        if let Some(password) = self.url.password() {
+            if self.url.username().is_empty() {
+                Some(format!("Bearer {password}"))
+            } else {
+                Some(format!(
+                    "Basic {}",
+                    STANDARD.encode(format!("{}:{}", self.url.username(), password))
+                ))
+            }
+        } else {
+            None
+        }
+    }
+
     /// Method to init the ssl context out of the ssl target configuration.
     /// Must be call when the configuration is retrieved
     pub fn init_ssl_context(&mut self) {
@@ -876,6 +938,14 @@ impl fmt::Display for TargetSetting {
             }
         }
 
+        // Mask username and password to avoid data leak in logs
+        if !url.username().is_empty() {
+            let _ = url.set_username("***");
+        }
+        if url.password().is_some() {
+            let _ = url.set_password(Some("***"));
+        }
+
         if let Some(proxy_url) = &self.proxy {
             write!(f, "{url} -proxy {proxy_url}")
         } else {

prosa_macros/src/io.rs

@@ -52,7 +52,7 @@ fn generate_struct(mut item_struct: syn::ItemStruct) -> syn::parse::Result<syn::
         // Add the Address field that is the remote address
         fields.named.push(
             syn::Field::parse_named
-                .parse2(quote! { addr: std::option::Option<std::net::SocketAddr> })
+                .parse2(quote! { addr: std::option::Option<prosa::io::SocketAddr> })
                 .unwrap(),
         );
         // Add the buffer object to read from the net object
@@ -115,11 +115,11 @@ fn generate_struct_impl(
                 }
             }
         }
-        impl #item_generics std::convert::From<(IO, std::net::SocketAddr)> for #item_ident #item_generics
+        impl #item_generics std::convert::From<(IO, prosa::io::SocketAddr)> for #item_ident #item_generics
         where
             IO: 'static + tokio::io::AsyncReadExt + tokio::io::AsyncWriteExt + std::os::fd::AsRawFd + std::marker::Unpin + std::marker::Send
         {
-            fn from(socket: (IO, std::net::SocketAddr)) -> Self {
+            fn from(socket: (IO, prosa::io::SocketAddr)) -> Self {
                 let (stream, addr) = socket;
                 let socket_id = stream.as_raw_fd() as u32;
                 #item_ident {

prosa_utils/Cargo.toml

@@ -35,7 +35,7 @@ hex = "0.4"
 
 # Config
 glob = { version = "0.3", optional = true }
-serde = { version = "1", optional = true, features = ["derive"] }
+serde = { workspace = true, optional = true }
 serde_yaml = { version = "0.9", optional = true }
 
 # Config OpenSSL

prosa_utils/src/config/ssl.rs

@@ -19,6 +19,7 @@ pub trait SslStore<C, S> {
     /// use prosa_utils::config::ssl::{Store, SslStore};
     ///
     /// let store = Store::File { path: "./target".into() };
+    /// # #[cfg(feature="config-openssl")]
     /// let ssl_store = store.get_store().unwrap();
     /// ```
     fn get_store(&self) -> Result<S, ConfigError>;
@@ -29,9 +30,11 @@ pub trait SslStore<C, S> {
     /// use prosa_utils::config::ssl::{Store, SslStore};
     ///
     /// let store = Store::File { path: "./target".into() };
+    /// # #[cfg(feature="config-openssl")]
     /// let certs_map = store.get_certs().unwrap();
     ///
     /// // No cert in target
+    /// # #[cfg(feature="config-openssl")]
     /// assert!(certs_map.is_empty());
     /// ```
     fn get_certs(&self) -> Result<HashMap<String, C>, ConfigError>;
@@ -91,6 +94,7 @@ pub trait SslConfigContext<C, S> {
     ///
     /// let mut client_config = SslConfig::default();
     /// client_config.set_store(Store::File { path: "./target".into() });
+    /// # #[cfg(feature="config-openssl")]
     /// if let Ok(mut ssl_context_builder) = client_config.init_tls_client_context() {
     ///     let ssl_context = ssl_context_builder.build();
     /// }
@@ -103,6 +107,7 @@ pub trait SslConfigContext<C, S> {
     /// use prosa_utils::config::ssl::{SslConfig, SslConfigContext as _};
     ///
     /// let server_config = SslConfig::new_pkcs12("server.pkcs12".into());
+    /// # #[cfg(feature="config-openssl")]
     /// if let Ok(mut ssl_context_builder) = server_config.init_tls_server_context(None) {
     ///     let ssl_context = ssl_context_builder.build();
     /// }
@@ -118,9 +123,11 @@ pub trait SslConfigContext<C, S> {
 /// use std::pin::Pin;
 /// use tokio::net::TcpStream;
 /// use tokio_openssl::SslStream;
+/// # #[cfg(feature="config-openssl")]
 /// use openssl::ssl::{ErrorCode, Ssl, SslMethod, SslVerifyMode};
 /// use prosa_utils::config::ssl::{SslConfig, SslConfigContext};
 ///
+/// # #[cfg(feature="config-openssl")]
 /// async fn client() -> Result<(), io::Error> {
 ///     let mut stream = TcpStream::connect("localhost:4443").await?;
 ///
@@ -147,9 +154,11 @@ pub trait SslConfigContext<C, S> {
 /// use std::pin::Pin;
 /// use tokio::net::TcpListener;
 /// use tokio_openssl::SslStream;
+/// # #[cfg(feature="config-openssl")]
 /// use openssl::ssl::{ErrorCode, Ssl, SslMethod, SslVerifyMode};
 /// use prosa_utils::config::ssl::{SslConfig, SslConfigContext};
 ///
+/// # #[cfg(feature="config-openssl")]
 /// async fn server() -> Result<(), io::Error> {
 ///     let listener = TcpListener::bind("0.0.0.0:4443").await?;
 ///