feat: add auth and tls for mirroring connections

crates/fluvio-sc/src/services/auth/basic.rs → crates/fluvio-auth/src/basic.rs

@@ -4,9 +4,9 @@ use tracing::instrument;
 use async_trait::async_trait;
 pub use policy::BasicRbacPolicy;
 
-use fluvio_auth::{AuthContext, Authorization, TypeAction, InstanceAction, AuthError};
+use crate::{AuthContext, Authorization, TypeAction, InstanceAction, AuthError};
 use fluvio_controlplane_metadata::extended::ObjectType;
-use fluvio_auth::x509::X509Identity;
+use crate::x509::X509Identity;
 
 #[derive(Debug, Clone)]
 pub struct BasicAuthorization {
@@ -84,8 +84,8 @@ mod policy {
     use tracing::debug;
     use serde::{Serialize, Deserialize};
 
-    use fluvio_auth::{AuthError, TypeAction, InstanceAction};
-    use fluvio_auth::x509::X509Identity;
+    use crate::{AuthError, TypeAction, InstanceAction};
+    use crate::x509::X509Identity;
 
     use super::ObjectType;
 
@@ -105,6 +105,7 @@ mod policy {
             match action {
                 TypeAction::Create => Action::Create,
                 TypeAction::Read => Action::Read,
+                TypeAction::Update => Action::Update,
             }
         }
     }
@@ -147,6 +148,7 @@ mod policy {
             //   let (action,object,_instance) = request;
             // For each scope provided in the identity,
             // check if there is a match;
+            //
             let is_allowed = identity.scopes().iter().any(|scope| {
                 self.0
                     .get(scope)
@@ -179,6 +181,7 @@ mod policy {
             root_policy.insert(ObjectType::Partition, vec![Action::All]);
             root_policy.insert(ObjectType::TableFormat, vec![Action::All]);
             root_policy.insert(ObjectType::Mirror, vec![Action::All]);
+            root_policy.insert(ObjectType::RemoteToHomeConnection, vec![Action::All]);
 
             let mut policy = HashMap::new();
 
@@ -197,7 +200,7 @@ mod test {
     use std::convert::TryFrom;
     use std::collections::HashMap;
 
-    use fluvio_auth::x509::X509Identity;
+    use crate::x509::X509Identity;
 
     use super::policy::*;
     use super::ObjectType;

crates/fluvio-auth/src/lib.rs

@@ -1,6 +1,9 @@
 mod policy;
 mod error;
 
+pub mod basic;
+pub mod root;
+
 pub mod x509;
 
 pub use policy::*;

crates/fluvio-auth/src/policy.rs

@@ -12,14 +12,15 @@ use super::AuthError;
 pub enum TypeAction {
     Create,
     Read,
+    Update,
 }
 
 pub enum InstanceAction {
     Delete,
 }
 
 #[async_trait]
-pub trait AuthContext: Debug {
+pub trait AuthContext: Debug + Send + Sync + 'static {
     /// check if any allow type specific action can be allowed
     async fn allow_type_action(
         &self,

crates/fluvio-auth/src/root.rs

@@ -0,0 +1,154 @@
+use std::fmt::Debug;
+
+use async_trait::async_trait;
+
+use crate::{AuthContext, Authorization, TypeAction, InstanceAction, AuthError};
+use fluvio_socket::FluvioSocket;
+use fluvio_controlplane_metadata::extended::ObjectType;
+
+/// Authorization that allows anything
+/// Used for personal development
+#[derive(Debug, Clone, Default)]
+pub struct RootAuthorization {}
+
+#[async_trait]
+impl Authorization for RootAuthorization {
+    type Context = RootAuthContext;
+
+    async fn create_auth_context(
+        &self,
+        _socket: &mut FluvioSocket,
+    ) -> Result<Self::Context, AuthError> {
+        Ok(RootAuthContext {})
+    }
+}
+
+impl RootAuthorization {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+#[derive(Debug)]
+pub struct RootAuthContext {}
+
+#[async_trait]
+impl AuthContext for RootAuthContext {
+    async fn allow_type_action(
+        &self,
+        _ty: ObjectType,
+        _action: TypeAction,
+    ) -> Result<bool, AuthError> {
+        Ok(true)
+    }
+
+    /// check if specific instance of spec can be deleted
+    async fn allow_instance_action(
+        &self,
+        _ty: ObjectType,
+        _action: InstanceAction,
+        _key: &str,
+    ) -> Result<bool, AuthError> {
+        Ok(true)
+    }
+}
+
+/// Authorization that allows only read only ops
+#[derive(Debug, Clone, Default)]
+pub struct ReadOnlyAuthorization {}
+
+#[async_trait]
+impl Authorization for ReadOnlyAuthorization {
+    type Context = ReadOnlyAuthContext;
+
+    async fn create_auth_context(
+        &self,
+        _socket: &mut FluvioSocket,
+    ) -> Result<Self::Context, AuthError> {
+        Ok(ReadOnlyAuthContext {})
+    }
+}
+
+impl ReadOnlyAuthorization {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+#[derive(Debug)]
+pub struct ReadOnlyAuthContext {}
+
+#[async_trait]
+impl AuthContext for ReadOnlyAuthContext {
+    async fn allow_type_action(
+        &self,
+        ty: ObjectType,
+        action: TypeAction,
+    ) -> Result<bool, AuthError> {
+        Ok(matches!(action, TypeAction::Read) || matches!(ty, ObjectType::CustomSpu))
+    }
+
+    /// check if specific instance of spec can be deleted
+    async fn allow_instance_action(
+        &self,
+        _ty: ObjectType,
+        _action: InstanceAction,
+        _key: &str,
+    ) -> Result<bool, AuthError> {
+        Ok(true)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::AuthContext;
+
+    use super::{ObjectType, ReadOnlyAuthContext, RootAuthContext, TypeAction};
+
+    /// test read only context
+    /// read only context allows read on everything
+    /// and create on spu
+    #[fluvio_future::test]
+    async fn test_read_only_context() {
+        let auth_context = ReadOnlyAuthContext {};
+        assert!(auth_context
+            .allow_type_action(ObjectType::Spu, TypeAction::Read)
+            .await
+            .unwrap());
+        assert!(!auth_context
+            .allow_type_action(ObjectType::Spu, TypeAction::Create)
+            .await
+            .unwrap());
+        assert!(auth_context
+            .allow_type_action(ObjectType::Topic, TypeAction::Read)
+            .await
+            .unwrap());
+        assert!(!auth_context
+            .allow_type_action(ObjectType::Topic, TypeAction::Create)
+            .await
+            .unwrap());
+    }
+
+    /// test root context
+    /// root context allows everything
+    #[fluvio_future::test]
+    async fn test_root_context() {
+        let auth_context = RootAuthContext {};
+        assert!(auth_context
+            .allow_type_action(ObjectType::Spu, TypeAction::Read)
+            .await
+            .unwrap());
+        assert!(auth_context
+            .allow_type_action(ObjectType::Spu, TypeAction::Create)
+            .await
+            .unwrap());
+        assert!(auth_context
+            .allow_type_action(ObjectType::Topic, TypeAction::Read)
+            .await
+            .unwrap());
+        assert!(auth_context
+            .allow_type_action(ObjectType::Topic, TypeAction::Create)
+            .await
+            .unwrap());
+    }
+}

crates/fluvio-auth/src/x509/authenticator.rs

@@ -36,15 +36,15 @@ impl ScopeBindings {
 
 #[derive(Debug)]
 pub struct X509Authenticator {
-    scope_bindings: ScopeBindings,
+    scope_bindings: Option<ScopeBindings>,
 }
 
 impl X509Authenticator {
-    pub fn new(scope_binding_file_path: &Path) -> Self {
-        Self {
-            scope_bindings: ScopeBindings::load(scope_binding_file_path)
-                .expect("unable to create ScopeBindings"),
-        }
+    pub fn new(scope_binding_file_path: Option<&Path>) -> Self {
+        let scope_bindings = scope_binding_file_path.map(|scope_binding_file_path| {
+            ScopeBindings::load(scope_binding_file_path).expect("unable to create ScopeBindings")
+        });
+        Self { scope_bindings }
     }
 
     async fn send_authorization_request(
@@ -79,13 +79,9 @@ impl X509Authenticator {
     fn principal_from_tls_stream(tls_stream: &DefaultServerTlsStream) -> Result<String, IoError> {
         trace!("tls_stream {:?}", tls_stream);
 
-        let peer_certificate = tls_stream.peer_certificate();
-
-        trace!("peer_certificate {:?}", peer_certificate);
-
         let client_certificate = tls_stream.peer_certificate().ok_or(IoErrorKind::NotFound)?;
 
-        trace!("client_certificate {:?}", tls_stream);
+        trace!("client_certificate {:?}", client_certificate);
 
         let principal = Self::principal_from_raw_certificate(
             &client_certificate
@@ -96,7 +92,7 @@ impl X509Authenticator {
         Ok(principal)
     }
 
-    fn principal_from_raw_certificate(certificate_bytes: &[u8]) -> Result<String, IoError> {
+    pub fn principal_from_raw_certificate(certificate_bytes: &[u8]) -> Result<String, IoError> {
         parse_x509_certificate(certificate_bytes)
             .map_err(|err| IoError::new(IoErrorKind::InvalidData, err))
             .and_then(|(_, parsed_cert)| Self::common_name_from_parsed_certificate(parsed_cert))
@@ -131,7 +127,13 @@ impl Authenticator for X509Authenticator {
         target_tcp_stream: &TcpStream,
     ) -> Result<bool, IoError> {
         let principal = Self::principal_from_tls_stream(incoming_tls_stream)?;
-        let scopes = self.scope_bindings.get_scopes(&principal);
+
+        let scopes = if let Some(scope_bindings) = &self.scope_bindings {
+            scope_bindings.get_scopes(&principal)
+        } else {
+            Vec::new()
+        };
+
         let authorization_request = AuthRequest::new(principal, scopes);
         let success =
             Self::send_authorization_request(target_tcp_stream, authorization_request).await?;

crates/fluvio-auth/src/x509/identity.rs

@@ -37,11 +37,12 @@ impl X509Identity {
                             principal: req_msg.request.principal,
                         },
                     },
-                    Err(_e) => {
+                    Err(err) => {
+                        tracing::error!(%err, "error decoding auth request");
                         return Err(std::io::Error::new(
                             std::io::ErrorKind::Interrupted,
                             "connection closed",
-                        ))
+                        ));
                     }
                 }
             } else {

crates/fluvio-cli/Cargo.toml

@@ -84,6 +84,7 @@ k8-types = { workspace = true , features = ["core"] }
 fluvio-cluster = { path = "../fluvio-cluster", default-features = false, features = ["cli"], optional = true }
 
 fluvio = { workspace = true }
+fluvio-auth = { workspace = true }
 fluvio-socket = { workspace = true }
 fluvio-command = { workspace = true  }
 fluvio-package-index = { workspace = true }