Merge pull request #1 from stefanDeveloper/add-auth

Add authentication

Author: stefanDeveloper

.github/workflows/ci_e2e.yml

@@ -0,0 +1,57 @@
+name: CI E2E Auth
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  auth_e2e_matrix:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Install Rust
+      uses: actions-rs/toolchain@v1
+      with:
+        toolchain: stable
+        override: true
+
+    - name: Install Core Dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y cmake g++ libssl-dev pkg-config openssl
+
+    - name: Install Fluvio Local Cluster
+      run: |
+        curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | FVM_VERSION=dev bash
+        echo "$HOME/.fluvio/bin" >> $GITHUB_PATH
+
+    - name: Generate mTLS Evaluation Certificates
+      run: |
+        mkdir -p /tmp/certs && cd /tmp/certs
+        openssl req -x509 -new -nodes -newkey rsa:2048 -keyout ca.key -out ca.crt -days 3650 -subj '/CN=fluvio-ca' -extensions v3_ca -config <(printf "[req]\ndistinguished_name=dn\n[dn]\n[v3_ca]\nbasicConstraints=CA:TRUE\nkeyUsage=keyCertSign,cRLSign")
+        openssl req -new -newkey rsa:2048 -nodes -keyout server.key -out server.csr -subj '/CN=localhost'
+        openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365 -extfile <(printf "subjectAltName=DNS:localhost,DNS:custom-spu-5001.localhost,IP:127.0.0.1\nbasicConstraints=CA:FALSE\nextendedKeyUsage=serverAuth")
+        openssl req -new -newkey rsa:2048 -nodes -keyout client.key -out client.csr -subj '/CN=fluvio-client'
+        openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt -days 365 -extfile <(printf "subjectAltName=DNS:localhost,IP:127.0.0.1\nbasicConstraints=CA:FALSE\nextendedKeyUsage=clientAuth")
+
+    - name: Start Authenticated Local TLS Cluster
+      run: |
+        fluvio cluster start --local --tls --server-cert /tmp/certs/server.crt --server-key /tmp/certs/server.key --client-cert /tmp/certs/client.crt --client-key /tmp/certs/client.key --ca-cert /tmp/certs/ca.crt --domain localhost
+
+    - name: Dynamically Build C++ Drivers representing E2E Target
+      run: |
+        cmake -B build
+        cmake --build build
+
+    - name: Execute Strict mTLS Validation Suite
+      run: |
+        export FLUVIO_E2E_TLS_DOMAIN="localhost"
+        export FLUVIO_E2E_TLS_KEY="/tmp/certs/client.key"
+        export FLUVIO_E2E_TLS_CERT="/tmp/certs/client.crt"
+        export FLUVIO_E2E_TLS_CA="/tmp/certs/ca.crt"
+        fluvio topic create test-auth-topic || true
+        cd build
+        ctest --output-on-failure -R fluvio_auth_test

CMakeLists.txt

@@ -15,9 +15,12 @@ execute_process(
 
 set(RUST_TARGET_DIR "target/debug")
 file(GLOB_RECURSE CXXBRIDGE_HEADERS "${RUST_TARGET_DIR}/build/*lib.rs.h")
-file(GLOB_RECURSE CXXBRIDGE_SOURCES "${RUST_TARGET_DIR}/build/*lib.rs.cc")
+file(GLOB_RECURSE CXXBRIDGE_SOURCES_ALL "${RUST_TARGET_DIR}/build/*lib.rs.cc")
 file(GLOB_RECURSE CXX_CORE_HEADER "${RUST_TARGET_DIR}/build/*/cxx.h")
+file(GLOB RUST_STATIC_LIB "${RUST_TARGET_DIR}/libfluvio*.a")
+list(GET RUST_STATIC_LIB 0 FLUVIO_STATIC_LIB)
 
+list(GET CXXBRIDGE_SOURCES_ALL 0 CXXBRIDGE_SOURCES)
 list(GET CXXBRIDGE_HEADERS 0 CXXBRIDGE_HEADER)
 get_filename_component(CXXBRIDGE_HDR_DIR ${CXXBRIDGE_HEADER} DIRECTORY)
 list(GET CXX_CORE_HEADER 0 CXX_CORE_HDR)
@@ -31,7 +34,7 @@ target_include_directories(test_producer PRIVATE
     ${CXX_ROOT_DIR}
 )
 target_link_libraries(test_producer PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/${RUST_TARGET_DIR}/libfluvio_client_cpp.a
+    ${FLUVIO_STATIC_LIB}
     pthread dl m
 )
 add_test(NAME fluvio_producer_test COMMAND test_producer)
@@ -43,7 +46,7 @@ target_include_directories(test_consumer PRIVATE
     ${CXX_ROOT_DIR}
 )
 target_link_libraries(test_consumer PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/${RUST_TARGET_DIR}/libfluvio_client_cpp.a
+    ${FLUVIO_STATIC_LIB}
     pthread dl m
 )
 add_test(NAME fluvio_consumer_test COMMAND test_consumer)
@@ -52,7 +55,7 @@ add_test(NAME fluvio_consumer_test COMMAND test_consumer)
 add_executable(test_admin tests/test_admin.cpp ${CXXBRIDGE_SOURCES})
 target_include_directories(test_admin PRIVATE ${CXXBRIDGE_HDR_DIR}/../.. ${CXX_ROOT_DIR})
 target_link_libraries(test_admin PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/${RUST_TARGET_DIR}/libfluvio_client_cpp.a
+    ${FLUVIO_STATIC_LIB}
     pthread dl m
 )
 add_test(NAME fluvio_admin_test COMMAND test_admin)
@@ -61,7 +64,7 @@ add_test(NAME fluvio_admin_test COMMAND test_admin)
 add_executable(test_config tests/test_config.cpp ${CXXBRIDGE_SOURCES})
 target_include_directories(test_config PRIVATE ${CXXBRIDGE_HDR_DIR}/../.. ${CXX_ROOT_DIR})
 target_link_libraries(test_config PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/${RUST_TARGET_DIR}/libfluvio_client_cpp.a
+    ${FLUVIO_STATIC_LIB}
     pthread dl m
 )
 add_test(NAME fluvio_config_test COMMAND test_config)
@@ -70,7 +73,25 @@ add_test(NAME fluvio_config_test COMMAND test_config)
 add_executable(test_c tests/test_c.c)
 target_include_directories(test_c PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
 target_link_libraries(test_c PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}/${RUST_TARGET_DIR}/libfluvio_client_cpp.a
+    ${FLUVIO_STATIC_LIB}
     stdc++ pthread dl m
 )
 add_test(NAME fluvio_c_native_test COMMAND test_c)
+
+# Auth E2E CXX Test
+add_executable(test_auth tests/test_auth.cpp ${CXXBRIDGE_SOURCES})
+target_include_directories(test_auth PRIVATE ${CXXBRIDGE_HDR_DIR}/../.. ${CXX_ROOT_DIR})
+target_link_libraries(test_auth PRIVATE
+    ${FLUVIO_STATIC_LIB}
+    pthread dl m
+)
+add_test(NAME fluvio_auth_test COMMAND test_auth)
+
+# Auth E2E C Native Test
+add_executable(test_auth_c tests/test_auth_c.c)
+target_include_directories(test_auth_c PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include)
+target_link_libraries(test_auth_c PRIVATE
+    ${FLUVIO_STATIC_LIB}
+    stdc++ pthread dl m
+)
+add_test(NAME fluvio_auth_c_test COMMAND test_auth_c)

include/fluvio.h

@@ -11,13 +11,21 @@ extern "C" {
 // Opaque types
 typedef struct fluvio_client_opaque fluvio_client_t;
 typedef struct fluvio_producer_opaque fluvio_topic_producer_t;
-typedef struct fluvio_consumer_opaque fluvio_partition_consumer_t;
 typedef struct fluvio_produce_output_opaque fluvio_produce_output_t;
 typedef struct fluvio_stream_opaque fluvio_stream_t;
 typedef struct fluvio_record_opaque fluvio_record_t;
+typedef struct fluvio_config_opaque fluvio_config_t;
 
 // Client
+int fluvio_c_config_load(fluvio_config_t** out_config);
+void fluvio_c_config_set_endpoint(fluvio_config_t* config, const char* endpoint);
+void fluvio_c_config_set_client_id(fluvio_config_t* config, const char* client_id);
+void fluvio_c_config_disable_tls(fluvio_config_t* config);
+void fluvio_c_config_set_anonymous_tls(fluvio_config_t* config);
+void fluvio_c_config_set_inline_tls(fluvio_config_t* config, const char* domain, const char* key, const char* cert, const char* ca_cert);
+void fluvio_c_config_set_tls_file_paths(fluvio_config_t* config, const char* domain, const char* key_path, const char* cert_path, const char* ca_cert_path);
 int fluvio_c_connect(fluvio_client_t** out_client);
+int fluvio_c_connect_with_config(fluvio_config_t* config, fluvio_client_t** out_client);
 void fluvio_c_client_free(fluvio_client_t* client);
 
 // Producer
@@ -29,13 +37,11 @@ void fluvio_c_producer_free(fluvio_topic_producer_t* producer);
 void fluvio_c_produce_output_free(fluvio_produce_output_t* out);
 
 // Consumer
-int fluvio_c_partition_consumer(fluvio_client_t* client, const char* topic, uint32_t partition, fluvio_partition_consumer_t** out_consumer);
-int fluvio_c_consumer_stream(fluvio_partition_consumer_t* consumer, int64_t offset_index, fluvio_stream_t** out_stream);
+int fluvio_c_consumer_stream(fluvio_client_t* client, const char* topic, uint32_t partition, int64_t offset_index, fluvio_stream_t** out_stream);
 int fluvio_c_stream_next(fluvio_stream_t* stream, fluvio_record_t** out_record);
 int fluvio_c_record_value(fluvio_record_t* record, const uint8_t** out_buf, size_t* out_len);
 void fluvio_c_record_free(fluvio_record_t* record);
 void fluvio_c_stream_free(fluvio_stream_t* stream);
-void fluvio_c_consumer_free(fluvio_partition_consumer_t* consumer);
 
 #ifdef __cplusplus
 }

src/c_api.rs

@@ -1,10 +1,16 @@
 use crate::client::{FluvioClient, fluvio_connect};
 use crate::producer::{FluvioProducer, create_producer, producer_send, producer_flush};
-use crate::consumer::{FluvioConsumer, FluvioStream, FluvioRecord, partition_consumer, consumer_stream, stream_next};
+use crate::consumer::{FluvioStream, FluvioRecord, consumer_stream, stream_next};
 use crate::produce_output::{FluvioProduceOutput, produce_output_wait};
+use crate::config::{FluvioConfigWrapper, fluvio_config_load};
 use std::os::raw::c_char;
 use std::ffi::CStr;
 
+#[repr(C)]
+pub struct fluvio_config_t {
+    _private: [u8; 0],
+}
+
 #[unsafe(no_mangle)]
 pub extern "C" fn fluvio_c_connect(out_client: *mut *mut FluvioClient) -> i32 {
     match fluvio_connect() {
@@ -16,6 +22,19 @@ pub extern "C" fn fluvio_c_connect(out_client: *mut *mut FluvioClient) -> i32 {
     }
 }
 
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_connect_with_config(config: *mut fluvio_config_t, out_client: *mut *mut FluvioClient) -> i32 {
+    if config.is_null() || out_client.is_null() { return -1; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    match crate::client::fluvio_connect_with_config(config_wrapper) {
+        Ok(client) => {
+            unsafe { *out_client = Box::into_raw(client); }
+            0
+        }
+        Err(_) => -1,
+    }
+}
+
 #[unsafe(no_mangle)]
 pub extern "C" fn fluvio_c_client_free(client: *mut FluvioClient) {
     if !client.is_null() { unsafe { let _ = Box::from_raw(client); } }
@@ -65,19 +84,73 @@ pub extern "C" fn fluvio_c_produce_output_free(out: *mut FluvioProduceOutput) {
 }
 
 #[unsafe(no_mangle)]
-pub extern "C" fn fluvio_c_partition_consumer(client: *mut FluvioClient, topic: *const c_char, partition: u32, out_consumer: *mut *mut FluvioConsumer) -> i32 {
-    if client.is_null() || topic.is_null() || out_consumer.is_null() { return -1; }
-    let topic_str = unsafe { CStr::from_ptr(topic).to_str() }.unwrap_or("");
-    match partition_consumer(unsafe { &*client }, topic_str, partition) {
-        Ok(consumer) => { unsafe { *out_consumer = Box::into_raw(consumer); } 0 }
+pub extern "C" fn fluvio_c_config_load(out_config: *mut *mut fluvio_config_t) -> i32 {
+    if out_config.is_null() { return -1; }
+    match fluvio_config_load() {
+        Ok(config) => { unsafe { *out_config = Box::into_raw(config) as *mut fluvio_config_t; } 0 }
         Err(_) => -1,
     }
 }
 
 #[unsafe(no_mangle)]
-pub extern "C" fn fluvio_c_consumer_stream(consumer: *mut FluvioConsumer, offset_index: i64, out_stream: *mut *mut FluvioStream) -> i32 {
-    if consumer.is_null() || out_stream.is_null() { return -1; }
-    match consumer_stream(unsafe { &*consumer }, offset_index) {
+pub unsafe extern "C" fn fluvio_c_config_set_endpoint(config: *mut fluvio_config_t, endpoint: *const std::ffi::c_char) {
+    if config.is_null() || endpoint.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    let ep_str = std::ffi::CStr::from_ptr(endpoint).to_string_lossy();
+    crate::config::fluvio_config_set_endpoint(config_wrapper, &ep_str);
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_config_set_client_id(config: *mut fluvio_config_t, client_id: *const std::ffi::c_char) {
+    if config.is_null() || client_id.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    let client_id_str = std::ffi::CStr::from_ptr(client_id).to_string_lossy();
+    crate::config::fluvio_config_set_client_id(config_wrapper, &client_id_str);
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_config_disable_tls(config: *mut fluvio_config_t) {
+    if config.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    crate::config::fluvio_config_disable_tls(config_wrapper);
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_config_set_anonymous_tls(config: *mut fluvio_config_t) {
+    if config.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    crate::config::fluvio_config_set_anonymous_tls(config_wrapper);
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_config_set_inline_tls(config: *mut fluvio_config_t, domain: *const std::ffi::c_char, key: *const std::ffi::c_char, cert: *const std::ffi::c_char, ca_cert: *const std::ffi::c_char) {
+    if config.is_null() || domain.is_null() || key.is_null() || cert.is_null() || ca_cert.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    crate::config::fluvio_config_set_inline_tls(config_wrapper, 
+        &std::ffi::CStr::from_ptr(domain).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(key).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(cert).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(ca_cert).to_string_lossy(),
+    );
+}
+
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn fluvio_c_config_set_tls_file_paths(config: *mut fluvio_config_t, domain: *const std::ffi::c_char, key_path: *const std::ffi::c_char, cert_path: *const std::ffi::c_char, ca_cert_path: *const std::ffi::c_char) {
+    if config.is_null() || domain.is_null() || key_path.is_null() || cert_path.is_null() || ca_cert_path.is_null() { return; }
+    let config_wrapper = &mut *(config as *mut FluvioConfigWrapper);
+    crate::config::fluvio_config_set_tls_file_paths(config_wrapper, 
+        &std::ffi::CStr::from_ptr(domain).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(key_path).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(cert_path).to_string_lossy(),
+        &std::ffi::CStr::from_ptr(ca_cert_path).to_string_lossy(),
+    );
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn fluvio_c_consumer_stream(client: *mut FluvioClient, topic: *const c_char, partition: u32, offset_index: i64, out_stream: *mut *mut FluvioStream) -> i32 {
+    if client.is_null() || topic.is_null() || out_stream.is_null() { return -1; }
+    let topic_str = unsafe { CStr::from_ptr(topic).to_str() }.unwrap_or("");
+    match consumer_stream(unsafe { &*client }, topic_str, partition, offset_index) {
         Ok(stream) => { unsafe { *out_stream = Box::into_raw(stream); } 0 }
         Err(_) => -1,
     }
@@ -110,7 +183,3 @@ pub extern "C" fn fluvio_c_stream_free(stream: *mut FluvioStream) {
     if !stream.is_null() { unsafe { let _ = Box::from_raw(stream); } }
 }
 
-#[unsafe(no_mangle)]
-pub extern "C" fn fluvio_c_consumer_free(consumer: *mut FluvioConsumer) {
-    if !consumer.is_null() { unsafe { let _ = Box::from_raw(consumer); } }
-}

src/config.rs

@@ -45,3 +45,29 @@ pub fn fluvio_config_set_endpoint(c: &mut FluvioConfigWrapper, endpoint: &str) {
 pub fn fluvio_config_set_client_id(c: &mut FluvioConfigWrapper, client_id: &str) {
     c.inner.client_id = Some(client_id.to_string());
 }
+
+pub fn fluvio_config_disable_tls(c: &mut FluvioConfigWrapper) {
+    c.inner.tls = fluvio::config::TlsPolicy::Disabled;
+}
+
+pub fn fluvio_config_set_anonymous_tls(c: &mut FluvioConfigWrapper) {
+    c.inner.tls = fluvio::config::TlsPolicy::Anonymous;
+}
+
+pub fn fluvio_config_set_inline_tls(c: &mut FluvioConfigWrapper, domain: &str, key: &str, cert: &str, ca_cert: &str) {
+    c.inner.tls = fluvio::config::TlsPolicy::Verified(fluvio::config::TlsConfig::Inline(fluvio::config::TlsCerts {
+        domain: domain.to_string(),
+        key: key.to_string(),
+        cert: cert.to_string(),
+        ca_cert: ca_cert.to_string(),
+    }));
+}
+
+pub fn fluvio_config_set_tls_file_paths(c: &mut FluvioConfigWrapper, domain: &str, key_path: &str, cert_path: &str, ca_cert_path: &str) {
+    c.inner.tls = fluvio::config::TlsPolicy::Verified(fluvio::config::TlsConfig::Files(fluvio::config::TlsPaths {
+        domain: domain.to_string(),
+        key: std::path::PathBuf::from(key_path),
+        cert: std::path::PathBuf::from(cert_path),
+        ca_cert: std::path::PathBuf::from(ca_cert_path),
+    }));
+}

src/consumer.rs

@@ -1,35 +1,27 @@
-use fluvio::{PartitionConsumer, Offset, consumer::Record as NativeRecord};
+use fluvio::{Offset, consumer::Record as NativeRecord, consumer::ConsumerConfigExtBuilder};
 use fluvio_future::task::run_block_on;
 use futures_util::stream::StreamExt;
 use futures_util::stream::Stream;
 use std::pin::Pin;
 use fluvio::dataplane::link::ErrorCode;
-use crate::config::ConsumerConfigWrapper;
 use crate::client::FluvioClient;
 
-pub struct FluvioConsumer { pub inner: PartitionConsumer }
 pub struct FluvioRecord { pub inner: NativeRecord }
 
-type PartitionConsumerStreamInner = Pin<Box<dyn Stream<Item = Result<NativeRecord, ErrorCode>> + Send>>;
-pub struct FluvioStream { pub inner: PartitionConsumerStreamInner }
+type ConsumerStreamInner = Pin<Box<dyn Stream<Item = Result<NativeRecord, ErrorCode>> + Send>>;
+pub struct FluvioStream { pub inner: ConsumerStreamInner }
 
-pub fn partition_consumer(client: &FluvioClient, topic: &str, partition: u32) -> Result<Box<FluvioConsumer>, String> {
-    run_block_on(client.inner.partition_consumer(topic, partition)).map(|c| Box::new(FluvioConsumer { inner: c })).map_err(|e| e.to_string())
-}
-
-#[allow(deprecated)]
-pub fn consumer_with_config(client: &FluvioClient, topic: &str, partition: u32, config: &ConsumerConfigWrapper) -> Result<Box<FluvioStream>, String> {
-    let consumer = run_block_on(client.inner.partition_consumer(topic, partition)).map_err(|e| e.to_string())?;
-    let built_config = config.inner.build().map_err(|e| e.to_string())?;
-    run_block_on(consumer.stream_with_config(Offset::beginning(), built_config))
-        .map(|s| Box::new(FluvioStream { inner: Box::pin(s) }))
-        .map_err(|e| e.to_string())
-}
-
-#[allow(deprecated)]
-pub fn consumer_stream(consumer: &FluvioConsumer, offset_index: i64) -> Result<Box<FluvioStream>, String> {
+pub fn consumer_stream(client: &FluvioClient, topic: &str, partition: u32, offset_index: i64) -> Result<Box<FluvioStream>, String> {
     let offset = if offset_index == -1 { Offset::end() } else if offset_index == 0 { Offset::beginning() } else { Offset::absolute(offset_index).unwrap() };
-    run_block_on(consumer.inner.stream(offset)).map(|s| Box::new(FluvioStream { inner: Box::pin(s) })).map_err(|e| e.to_string())
+    let config = ConsumerConfigExtBuilder::default()
+        .topic(topic.to_string())
+        .partition(partition)
+        .offset_start(offset)
+        .build()
+        .map_err(|e| e.to_string())?;
+
+    let consumer_stream = run_block_on(client.inner.consumer_with_config(config)).map_err(|e| e.to_string())?;
+    Ok(Box::new(FluvioStream { inner: Box::pin(consumer_stream) }))
 }
 
 pub fn stream_next(stream: &mut FluvioStream) -> Result<Box<FluvioRecord>, String> {