Backport ccm main

.github/workflows/ccm_tests.yml

@@ -0,0 +1,29 @@
+name: CCM tests
+
+on:
+  push:
+    branches:
+    - main
+    - 'branch-*'
+  pull_request:
+    branches:
+    - '**'
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BACKTRACE: full
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+      - name: Install scylla-ccm
+        run: pip3 install https://github.com/scylladb/scylla-ccm/archive/master.zip
+      - name: Run CCM command to create the ~/.ccm dir
+        continue-on-error: true
+        run: ccm status
+      - name: Run CCM tests
+        run: RUST_LOG=trace make ccm-test

Makefile

@@ -52,6 +52,10 @@ test: up
 	 SCYLLA_URI3=172.42.0.4:9042 \
 	 cargo test
 
+.PHONY: ccm-test
+ccm-test:
+	RUSTFLAGS="${RUSTFLAGS} --cfg ccm_tests" cargo test --test integration ccm
+
 .PHONY: dockerized-test
 dockerized-test: up
 	test/dockerized/run.sh

scylla/Cargo.toml

@@ -84,11 +84,14 @@ bigdecimal-04 = { package = "bigdecimal", version = "0.4" }
 scylla-proxy = { version = "0.0.4", path = "../scylla-proxy" }
 ntest = "0.9.3"
 criterion = "0.5"
-tokio = { version = "1.34", features = ["test-util"] }
+tokio = { version = "1.34", features = ["test-util", "process", "fs"] }
 tracing-subscriber = { version = "0.3.14", features = ["env-filter"] }
 assert_matches = "1.5.0"
 rand_chacha = "0.9.0"
 time = "0.3"
+anyhow = "1.0.98"
+tempfile = "3.19"
+
 
 [[bench]]
 name = "benchmark"
@@ -102,4 +105,5 @@ unexpected_cfgs = { level = "warn", check-cfg = [
     'cfg(scylla_cloud_tests)',
     'cfg(cassandra_tests)',
     'cfg(cpp_rust_unstable)',
+    'cfg(ccm_tests)',
 ] }

scylla/tests/integration/ccm/authenticate.rs

@@ -0,0 +1,134 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::{BufMut, BytesMut};
+use scylla::authentication::{AuthenticatorProvider, AuthenticatorSession};
+use scylla::errors::AuthError;
+use tokio::sync::Mutex;
+
+use crate::ccm::lib::cluster::{Cluster, ClusterOptions};
+use crate::ccm::lib::{run_ccm_test_with_configuration, CLUSTER_VERSION};
+use crate::utils::{setup_tracing, unique_keyspace_name, PerformDDL};
+
+fn cluster_1_node() -> ClusterOptions {
+    ClusterOptions {
+        name: "cluster_auth_1_node".to_string(),
+        version: CLUSTER_VERSION.clone(),
+        nodes_per_dc: vec![1],
+        ..ClusterOptions::default()
+    }
+}
+
+async fn run_ccm_auth_test_cluster_one_node<T, TFut>(test: T)
+where
+    T: FnOnce(Arc<Mutex<Cluster>>) -> TFut,
+    TFut: std::future::Future<Output = ()>,
+{
+    run_ccm_test_with_configuration(
+        cluster_1_node,
+        |mut cluster| async move {
+            cluster
+                .enable_password_authentication()
+                .await
+                .expect("Failed to enable password authenticator");
+            cluster
+        },
+        test,
+    )
+    .await
+}
+
+#[tokio::test]
+#[cfg_attr(not(ccm_tests), ignore)]
+async fn authenticate_superuser_cluster_one_node() {
+    setup_tracing();
+    async fn test(cluster: Arc<Mutex<Cluster>>) {
+        let cluster = cluster.lock().await;
+
+        tracing::info!(
+            "Connecting to {:?} with cassandra superuser...",
+            cluster.nodes().get_contact_endpoints().await
+        );
+
+        let session = cluster
+            .make_session_builder()
+            .await
+            .user("cassandra", "cassandra")
+            .build()
+            .await
+            .unwrap();
+        let ks = unique_keyspace_name();
+
+        session.ddl(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks)).await.unwrap();
+        session.use_keyspace(ks, false).await.unwrap();
+        session.ddl("DROP TABLE IF EXISTS t;").await.unwrap();
+
+        tracing::info!("Ok.");
+    }
+
+    run_ccm_auth_test_cluster_one_node(test).await
+}
+
+struct CustomAuthenticator;
+
+#[async_trait]
+impl AuthenticatorSession for CustomAuthenticator {
+    async fn evaluate_challenge(
+        &mut self,
+        _token: Option<&[u8]>,
+    ) -> Result<Option<Vec<u8>>, AuthError> {
+        Err("Challenges are not expected".to_string())
+    }
+
+    async fn success(&mut self, _token: Option<&[u8]>) -> Result<(), AuthError> {
+        Ok(())
+    }
+}
+
+struct CustomAuthenticatorProvider;
+
+#[async_trait]
+impl AuthenticatorProvider for CustomAuthenticatorProvider {
+    async fn start_authentication_session(
+        &self,
+        _authenticator_name: &str,
+    ) -> Result<(Option<Vec<u8>>, Box<dyn AuthenticatorSession>), AuthError> {
+        let mut response = BytesMut::new();
+        let cred = "\0cassandra\0cassandra";
+
+        response.put_slice(cred.as_bytes());
+
+        Ok((Some(response.to_vec()), Box::new(CustomAuthenticator)))
+    }
+}
+
+#[tokio::test]
+#[cfg_attr(not(ccm_tests), ignore)]
+async fn custom_authentication_cluster_one_node() {
+    setup_tracing();
+    async fn test(cluster: Arc<Mutex<Cluster>>) {
+        let cluster = cluster.lock().await;
+
+        tracing::info!(
+            "Connecting to {:?} with custom authenticator as cassandra superuser...",
+            cluster.nodes().get_contact_endpoints().await
+        );
+
+        let session = cluster
+            .make_session_builder()
+            .await
+            .authenticator_provider(Arc::new(CustomAuthenticatorProvider))
+            .build()
+            .await
+            .unwrap();
+        let ks = unique_keyspace_name();
+
+        session.ddl(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks)).await.unwrap();
+        session.use_keyspace(ks, false).await.unwrap();
+        session.ddl("DROP TABLE IF EXISTS t;").await.unwrap();
+
+        tracing::info!("Ok.");
+    }
+
+    run_ccm_auth_test_cluster_one_node(test).await
+}