diff --git a/cassandra/build.gradle.kts b/cassandra/build.gradle.kts index da76cc01b8..5e89fef8ef 100644 --- a/cassandra/build.gradle.kts +++ b/cassandra/build.gradle.kts @@ -23,6 +23,7 @@ plugins { } dependencies { implementation(libs.pulsar.io.core) + implementation(libs.pulsar.io.common) implementation(libs.jackson.databind) implementation(libs.jackson.dataformat.yaml) implementation(libs.cassandra.driver) diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java index 4f96df280d..7653107c67 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraAbstractSink.java @@ -29,6 +29,7 @@ import com.google.common.util.concurrent.MoreExecutors; import java.util.Map; import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.common.IOConfigUtils; import org.apache.pulsar.io.core.KeyValue; import org.apache.pulsar.io.core.Sink; import org.apache.pulsar.io.core.SinkContext; @@ -47,7 +48,10 @@ public abstract class CassandraAbstractSink implements Sink { @Override public void open(Map config, SinkContext sinkContext) throws Exception { - cassandraSinkConfig = CassandraSinkConfig.load(config); + // loadWithSecrets rather than load(): it resolves @FieldDoc(sensitive = true) fields from the + // connector's secrets provider, so the password need not sit in plaintext config. Fields with + // no secret configured still come from the config map, as before. + cassandraSinkConfig = IOConfigUtils.loadWithSecrets(config, CassandraSinkConfig.class, sinkContext); if (cassandraSinkConfig.getRoots() == null || cassandraSinkConfig.getKeyspace() == null || cassandraSinkConfig.getKeyname() == null @@ -55,6 +59,7 @@ public void open(Map config, SinkContext sinkContext) throws Exc || cassandraSinkConfig.getColumnName() == null) { throw new IllegalArgumentException("Required property not set."); } + cassandraSinkConfig.validateCredentials(); createClient(cassandraSinkConfig.getRoots()); statement = session.prepare("INSERT INTO " + cassandraSinkConfig.getColumnFamily() + " (" + cassandraSinkConfig.getKeyname() + ", " + cassandraSinkConfig.getColumnName() + ") VALUES (?, ?)"); @@ -98,6 +103,11 @@ private void createClient(String roots) { b.withPort(Integer.parseInt(hostPort[1])); } } + // Authenticate only when credentials were supplied; an unset pair leaves the connection + // exactly as it was before these settings existed. + if (cassandraSinkConfig.hasCredentials()) { + b.withCredentials(cassandraSinkConfig.getUserName(), cassandraSinkConfig.getPassword()); + } cluster = b.withoutJMXReporting().build(); session = cluster.connect(); session.execute("USE " + cassandraSinkConfig.getKeyspace()); diff --git a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java index 1dfc69b4d1..b83af7a4c8 100644 --- a/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java +++ b/cassandra/src/main/java/org/apache/pulsar/io/cassandra/CassandraSinkConfig.java @@ -34,6 +34,22 @@ public class CassandraSinkConfig implements Serializable { private static final long serialVersionUID = 1L; + @FieldDoc( + required = false, + defaultValue = "", + sensitive = true, + help = "Username used to authenticate against the cluster specified by `roots`. " + + "Must be set together with the other half of the pair; leave both unset " + + "for a cluster that does not require authentication.") + private String userName; + @FieldDoc( + required = false, + defaultValue = "", + sensitive = true, + help = "Password used to authenticate against the cluster specified by `roots`. " + + "Must be set together with the other half of the pair; leave both unset " + + "for a cluster that does not require authentication.") + private String password; @FieldDoc( required = true, defaultValue = "", @@ -69,4 +85,31 @@ public static CassandraSinkConfig load(Map map) throws IOExcepti ObjectMapper mapper = new ObjectMapper(); return mapper.readValue(mapper.writeValueAsString(map), CassandraSinkConfig.class); } + + /** + * Rejects a credential pair with only one half set. Half a pair is never a meaningful + * configuration — whichever half is present, the intent was to authenticate — and connecting + * unauthenticated instead turns a typo into either a rejection from the server that names + * neither setting, or a silent success against a cluster that does not yet require + * authentication. Sinks call this before connecting, so the failure names what is wrong. + */ + public void validateCredentials() { + if (hasText(userName) != hasText(password)) { + throw new IllegalArgumentException("userName and password must be supplied together: " + + "set both to authenticate, or neither to connect to a cluster that does not " + + "require authentication."); + } + } + + /** + * Whether to authenticate at all. Only ever true for a complete pair, and + * {@link #validateCredentials()} has already rejected an incomplete one. + */ + public boolean hasCredentials() { + return hasText(userName) && hasText(password); + } + + private static boolean hasText(String value) { + return value != null && !value.trim().isEmpty(); + } } \ No newline at end of file diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java new file mode 100644 index 0000000000..a7bdfba1d9 --- /dev/null +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthEnforcementTest.java @@ -0,0 +1,248 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.cassandra; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.fail; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.exceptions.AuthenticationException; +import com.datastax.driver.core.exceptions.NoHostAvailableException; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SinkContext; +import org.testcontainers.containers.CassandraContainer; +import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.containers.wait.strategy.WaitAllStrategy; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Drives {@link CassandraStringSink} against a Cassandra that actually enforces authentication, and + * so asserts what {@link CassandraSinkAuthTest} cannot: that a connection without credentials — or + * with the wrong ones — is refused by the server, and that the credentials the sink is configured + * with are what gets it past that refusal. + * + *

The container runs {@code PasswordAuthenticator} instead of the image default + * {@code AllowAllAuthenticator}. It gets there by editing the shipped {@code cassandra.yaml} in + * place rather than by mounting a replacement: {@code withConfigurationOverride()} maps a whole + * directory over {@code /etc/cassandra}, which would mean carrying a complete, version-matched copy + * of every file Cassandra reads from there, while a one-line {@code sed} touches only the setting + * under test. The {@code grep} that follows it makes a no-op substitution fatal — if a future image + * tag expresses the authenticator differently, the container fails to start rather than quietly + * reverting to {@code AllowAllAuthenticator} and leaving every assertion here vacuous. + * {@link #serverRejectsUnauthenticatedConnection()} is the second half of that guard: it fails the + * suite if the server ever stops enforcing, whatever the reason. + */ +public class CassandraSinkAuthEnforcementTest { + + private static final String IMAGE = "cassandra:4.1"; + private static final String CONF = "/etc/cassandra/cassandra.yaml"; + private static final String SUPERUSER = "cassandra"; + private static final String SUPERUSER_PASSWORD = "cassandra"; + + private static final String ENABLE_PASSWORD_AUTHENTICATION = + "sed -i 's/^authenticator:.*/authenticator: PasswordAuthenticator/' " + CONF + + " && grep -q '^authenticator: PasswordAuthenticator' " + CONF + + " && exec docker-entrypoint.sh cassandra -f"; + + private static final String KEYSPACE = "auth_enforced_ks"; + private static final String TABLE = "auth_enforced_table"; + private static final String KEY_COLUMN = "key"; + private static final String VALUE_COLUMN = "value"; + + private CassandraContainer cassandraContainer; + + @BeforeClass + public void setUp() { + cassandraContainer = new CassandraContainer<>(IMAGE) + .withCommand("sh", "-c", ENABLE_PASSWORD_AUTHENTICATION) + // The default superuser role is created on a delay after the ring is joined, which is + // later than the CQL port opening; waiting for the port alone would let a test connect + // before there is anything to authenticate against. + .waitingFor(new WaitAllStrategy() + .withStrategy(Wait.forListeningPort()) + .withStrategy(new LogMessageWaitStrategy() + .withRegEx(".*Created default superuser role.*\\n")) + .withStartupTimeout(Duration.ofMinutes(3))); + cassandraContainer.start(); + + try (Cluster cluster = authenticatedCluster(); + Session session = cluster.connect()) { + session.execute("CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class':'SimpleStrategy', 'replication_factor':'1'}"); + session.execute("CREATE TABLE " + KEYSPACE + "." + TABLE + + " (" + KEY_COLUMN + " text PRIMARY KEY, " + VALUE_COLUMN + " text)"); + } + } + + @AfterClass(alwaysRun = true) + public void tearDown() { + if (cassandraContainer != null) { + cassandraContainer.stop(); + cassandraContainer = null; + } + } + + /** + * Establishes that the server under test enforces authentication at all, using the driver + * directly so that nothing about the sink is involved. Every other assertion in this class is + * only worth something while this one holds. + */ + @Test + public void serverRejectsUnauthenticatedConnection() { + try (Cluster cluster = cassandraContainer.getCluster()) { + cluster.connect(); + fail("Expected the server to reject a connection carrying no credentials"); + } catch (Exception e) { + assertNotNull(authenticationFailure(e), "Not an authentication failure: " + e); + } + } + + @Test + public void sinkFailsToOpenWithoutCredentials() { + assertOpenIsRejected(baseConfig()); + } + + @Test + public void sinkFailsToOpenWithWrongPassword() { + Map config = baseConfig(); + config.put("userName", SUPERUSER); + config.put("password", "not-the-password"); + + assertOpenIsRejected(config); + } + + @Test + public void sinkFailsToOpenWithUnknownUser() { + Map config = baseConfig(); + config.put("userName", "no-such-user"); + config.put("password", SUPERUSER_PASSWORD); + + assertOpenIsRejected(config); + } + + @Test + public void sinkWritesWhenCredentialsAreCorrect() throws Exception { + Map config = baseConfig(); + config.put("userName", SUPERUSER); + config.put("password", SUPERUSER_PASSWORD); + + String key = "authenticated"; + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + + CompletableFuture acked = new CompletableFuture<>(); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.of(key); + } + + @Override + public byte[] getValue() { + return ("value-" + key).getBytes(); + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + acked.completeExceptionally(new RuntimeException("Record failed")); + } + }); + acked.get(); + } finally { + sink.close(); + } + + try (Cluster cluster = authenticatedCluster(); + Session session = cluster.connect(KEYSPACE)) { + assertEquals(session + .execute("SELECT * FROM " + TABLE + " WHERE " + KEY_COLUMN + " = '" + key + "'") + .one() + .getString(VALUE_COLUMN), "value-" + key); + } + } + + private void assertOpenIsRejected(Map config) { + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + fail("Expected open() to be rejected by the server"); + } catch (Exception e) { + assertNotNull(authenticationFailure(e), "Not an authentication failure: " + e); + } + } + + /** + * Finds the {@link AuthenticationException} the driver raised, or {@code null} if the failure was + * something else. The driver reports an unusable contact point either directly or wrapped in a + * {@link NoHostAvailableException} holding the per-host cause, so both shapes are unwrapped here + * rather than asserting on one and hoping. + */ + private static AuthenticationException authenticationFailure(Throwable t) { + for (Throwable current = t; current != null; current = current.getCause()) { + if (current instanceof AuthenticationException) { + return (AuthenticationException) current; + } + if (current instanceof NoHostAvailableException) { + for (Throwable hostError : ((NoHostAvailableException) current).getErrors().values()) { + AuthenticationException found = authenticationFailure(hostError); + if (found != null) { + return found; + } + } + } + } + return null; + } + + private Cluster authenticatedCluster() { + return Cluster.builder() + .addContactPoint(cassandraContainer.getHost()) + .withPort(cassandraContainer.getMappedPort(CassandraContainer.CQL_PORT)) + .withCredentials(SUPERUSER, SUPERUSER_PASSWORD) + .withoutJMXReporting() + .build(); + } + + private Map baseConfig() { + Map config = new HashMap<>(); + config.put("roots", cassandraContainer.getHost() + ":" + + cassandraContainer.getMappedPort(CassandraContainer.CQL_PORT)); + config.put("keyspace", KEYSPACE); + config.put("keyname", KEY_COLUMN); + config.put("columnFamily", TABLE); + config.put("columnName", VALUE_COLUMN); + return config; + } +} diff --git a/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java new file mode 100644 index 0000000000..ca9b034d20 --- /dev/null +++ b/cassandra/src/test/java/org/apache/pulsar/io/cassandra/CassandraSinkAuthTest.java @@ -0,0 +1,214 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.cassandra; + +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Session; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SinkContext; +import org.testcontainers.containers.CassandraContainer; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +/** + * Covers the {@code userName} / {@code password} settings on {@link CassandraSinkConfig} against a + * cluster that does not require authentication — the image default, + * {@code AllowAllAuthenticator}. + * + *

What that arrangement is good for is the compatibility half of these settings: credentials + * survive config loading, an unset pair is still unset, and neither supplying credentials nor + * omitting them stops the sink writing to a cluster that never asked for them. The last of those is + * the regression this connector most needs guarded — an existing unauthenticated deployment must be + * unaffected by the settings existing. + * + *

It also covers the one case the connector rejects on its own account, without asking a server + * anything: a pair with only one half set. + * + *

That a server actually refuses a connection without credentials, and that the configured pair + * is what gets past the refusal, is asserted in {@link CassandraSinkAuthEnforcementTest}, which runs + * a container with {@code PasswordAuthenticator} enabled. + */ +public class CassandraSinkAuthTest { + + private static final String KEYSPACE = "auth_test_ks"; + private static final String TABLE = "auth_test_table"; + private static final String KEY_COLUMN = "key"; + private static final String VALUE_COLUMN = "value"; + + private CassandraContainer cassandraContainer; + + @BeforeClass + public void setUp() { + cassandraContainer = new CassandraContainer<>("cassandra:4.1") + .withStartupTimeout(Duration.ofMinutes(3)); + cassandraContainer.start(); + + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect()) { + session.execute("CREATE KEYSPACE " + KEYSPACE + + " WITH replication = {'class':'SimpleStrategy', 'replication_factor':'1'}"); + session.execute("CREATE TABLE " + KEYSPACE + "." + TABLE + + " (" + KEY_COLUMN + " text PRIMARY KEY, " + VALUE_COLUMN + " text)"); + } + } + + @AfterClass(alwaysRun = true) + public void tearDown() { + if (cassandraContainer != null) { + cassandraContainer.stop(); + cassandraContainer = null; + } + } + + @Test + public void credentialsSurviveConfigLoading() throws Exception { + Map config = baseConfig(); + config.put("userName", "cassandra"); + config.put("password", "cassandra"); + + CassandraSinkConfig loaded = CassandraSinkConfig.load(config); + + assertEquals(loaded.getUserName(), "cassandra"); + assertEquals(loaded.getPassword(), "cassandra"); + } + + @Test + public void credentialsAreUnsetWhenNotConfigured() throws Exception { + CassandraSinkConfig loaded = CassandraSinkConfig.load(baseConfig()); + + assertNull(loaded.getUserName()); + assertNull(loaded.getPassword()); + } + + @Test + public void sinkWritesWithCredentialsSupplied() throws Exception { + Map config = baseConfig(); + config.put("userName", cassandraContainer.getUsername()); + config.put("password", cassandraContainer.getPassword()); + + assertWriteSucceeds(config, "with-credentials"); + } + + @Test + public void sinkWritesWithoutCredentials() throws Exception { + assertWriteSucceeds(baseConfig(), "no-credentials"); + } + + @Test + public void openRejectsUsernameWithoutPassword() { + Map config = unroutableConfig(); + config.put("userName", "cassandra"); + + assertHalfConfiguredPairRejected(config); + } + + @Test + public void openRejectsPasswordWithoutUsername() { + Map config = unroutableConfig(); + config.put("password", "cassandra"); + + assertHalfConfiguredPairRejected(config); + } + + private void assertHalfConfiguredPairRejected(Map config) { + try { + new CassandraStringSink().open(config, mock(SinkContext.class)); + fail("Expected open() to reject a credential pair with only one half set"); + } catch (IllegalArgumentException e) { + assertTrue(e.getMessage().contains("userName and password must be supplied together"), + "Rejected for some other reason: " + e.getMessage()); + } catch (Exception e) { + fail("Expected IllegalArgumentException, got: " + e); + } + } + + /** + * Config whose {@code roots} point at a port nothing listens on. The rejection above has to + * happen before any connection is attempted, and pointing somewhere unreachable is what makes + * the test say so: drop the validation and this fails connecting rather than passing for the + * wrong reason. + */ + private Map unroutableConfig() { + Map config = baseConfig(); + config.put("roots", "127.0.0.1:1"); + return config; + } + + private void assertWriteSucceeds(Map config, String key) throws Exception { + CassandraStringSink sink = new CassandraStringSink(); + try { + sink.open(config, mock(SinkContext.class)); + + CompletableFuture acked = new CompletableFuture<>(); + sink.write(new Record() { + @Override + public Optional getKey() { + return Optional.of(key); + } + + @Override + public byte[] getValue() { + return ("value-" + key).getBytes(); + } + + @Override + public void ack() { + acked.complete(null); + } + + @Override + public void fail() { + acked.completeExceptionally(new RuntimeException("Record failed")); + } + }); + acked.get(); + } finally { + sink.close(); + } + + try (Cluster cluster = cassandraContainer.getCluster(); + Session session = cluster.connect(KEYSPACE)) { + assertEquals(session + .execute("SELECT * FROM " + TABLE + " WHERE " + KEY_COLUMN + " = '" + key + "'") + .one() + .getString(VALUE_COLUMN), "value-" + key); + } + } + + private Map baseConfig() { + Map config = new HashMap<>(); + config.put("roots", cassandraContainer.getHost() + ":" + cassandraContainer.getMappedPort(9042)); + config.put("keyspace", KEYSPACE); + config.put("keyname", KEY_COLUMN); + config.put("columnFamily", TABLE); + config.put("columnName", VALUE_COLUMN); + return config; + } +}