Skip to content

Commit

Permalink
fix: regression in auth token configuration (#9)
Browse files Browse the repository at this point in the history
* fix: regression in auth token configuration

The regression was caused by masking the auth token so Kafka Connect
would not log it. For the avoidance of doubts: This issue could not
be abused to get unauthorized access to QuestDB.
  • Loading branch information
jerrinot authored Aug 8, 2023
1 parent 0d0c6a7 commit 3ee7afd
Show file tree
Hide file tree
Showing 6 changed files with 303 additions and 192 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.connect.errors.ConnectException;

import java.util.Arrays;
Expand Down Expand Up @@ -153,8 +154,8 @@ public String getUsername() {
return getString(USERNAME);
}

public String getToken() {
return getString(TOKEN);
public Password getToken() {
return getPassword(TOKEN);
}

public boolean isTls() {
Expand Down
18 changes: 4 additions & 14 deletions connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java
Original file line number Diff line number Diff line change
Expand Up @@ -9,25 +9,15 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.data.Date;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.data.Time;
import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.data.*;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.RetriableException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import java.util.concurrent.TimeUnit;

public final class QuestDBSinkTask extends SinkTask {
Expand Down Expand Up @@ -91,10 +81,10 @@ private Sender createSender() {
}
if (config.getToken() != null) {
String username = config.getUsername();
if (username == null || username.equals("")) {
if (username == null || username.isEmpty()) {
throw new ConnectException("Username cannot be empty when using ILP authentication");
}
builder.enableAuth(username).authToken(config.getToken());
builder.enableAuth(username).authToken(config.getToken().value());
}
Sender rawSender = builder.build();
String symbolColumns = config.getSymbolColumns();
Expand Down
77 changes: 77 additions & 0 deletions connector/src/test/java/io/questdb/kafka/ConnectTestUtils.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
package io.questdb.kafka;

import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.runtime.AbstractStatus;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.storage.StringConverter;
import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
import org.awaitility.Awaitility;
import org.testcontainers.containers.GenericContainer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG;
import static org.junit.jupiter.api.Assertions.fail;

public final class ConnectTestUtils {
public static final long CONNECTOR_START_TIMEOUT_MS = SECONDS.toMillis(60);
public static final String CONNECTOR_NAME = "questdb-sink-connector";
private static final AtomicInteger ID_GEN = new AtomicInteger(0);

private ConnectTestUtils() {
}

static void assertConnectorTaskRunningEventually(EmbeddedConnectCluster connect) {
assertConnectorTaskStateEventually(connect, AbstractStatus.State.RUNNING);
}

static void assertConnectorTaskFailedEventually(EmbeddedConnectCluster connect) {
assertConnectorTaskStateEventually(connect, AbstractStatus.State.FAILED);
}

static void assertConnectorTaskStateEventually(EmbeddedConnectCluster connect, AbstractStatus.State expectedState) {
Awaitility.await().atMost(CONNECTOR_START_TIMEOUT_MS, MILLISECONDS).untilAsserted(() -> assertConnectorTaskState(connect, CONNECTOR_NAME, expectedState));
}

static Map<String, String> baseConnectorProps(GenericContainer<?> questDBContainer, String topicName) {
Map<String, String> props = new HashMap<>();
props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, QuestDBSinkConnector.class.getName());
props.put("topics", topicName);
props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
props.put(VALUE_CONVERTER_CLASS_CONFIG, JsonConverter.class.getName());
props.put("host", questDBContainer.getHost() + ":" + questDBContainer.getMappedPort(QuestDBUtils.QUESTDB_ILP_PORT));
return props;
}

static void assertConnectorTaskState(EmbeddedConnectCluster connect, String connectorName, AbstractStatus.State expectedState) {
ConnectorStateInfo info = connect.connectorStatus(connectorName);
if (info == null) {
fail("Connector " + connectorName + " not found");
}
List<ConnectorStateInfo.TaskState> taskStates = info.tasks();
if (taskStates.size() == 0) {
fail("No tasks found for connector " + connectorName);
}
for (ConnectorStateInfo.TaskState taskState : taskStates) {
if (!Objects.equals(taskState.state(), expectedState.toString())) {
fail("Task " + taskState.id() + " for connector " + connectorName + " is in state " + taskState.state() + " but expected " + expectedState);
}
}
}

static String newTopicName() {
return "topic" + ID_GEN.getAndIncrement();
}

static String newTableName() {
return "table" + ID_GEN.getAndIncrement();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
package io.questdb.kafka;

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.storage.Converter;
import org.apache.kafka.connect.storage.ConverterConfig;
import org.apache.kafka.connect.storage.ConverterType;
import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.FixedHostPortGenericContainer;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.MountableFile;

import java.util.Map;

import static java.util.Collections.singletonMap;

@Testcontainers
public class QuestDBSinkConnectorEmbeddedAuthTest {
private EmbeddedConnectCluster connect;
private Converter converter;
private String topicName;

// must match the user in authDb.txt
private static final String TEST_USER_TOKEN = "UvuVb1USHGRRT08gEnwN2zGZrvM4MsLQ5brgF6SVkAw=";
private static final String TEST_USER_NAME = "testUser1";

@Container
private static GenericContainer<?> questDBContainer = newQuestDbConnector();

private static GenericContainer<?> newQuestDbConnector() {
FixedHostPortGenericContainer<?> container = new FixedHostPortGenericContainer<>("questdb/questdb:7.3");
container.addExposedPort(QuestDBUtils.QUESTDB_HTTP_PORT);
container.addExposedPort(QuestDBUtils.QUESTDB_ILP_PORT);
container.setWaitStrategy(new LogMessageWaitStrategy().withRegEx(".*server-main enjoy.*"));
container.withCopyFileToContainer(MountableFile.forClasspathResource("/authDb.txt"), "/var/lib/questdb/conf/authDb.txt");
container.withEnv("QDB_LINE_TCP_AUTH_DB_PATH", "conf/authDb.txt");
return container.withLogConsumer(new Slf4jLogConsumer(LoggerFactory.getLogger("questdb")));
}

@BeforeEach
public void setUp() {
topicName = ConnectTestUtils.newTopicName();
JsonConverter jsonConverter = new JsonConverter();
jsonConverter.configure(singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.VALUE.getName()));
converter = jsonConverter;

connect = new EmbeddedConnectCluster.Builder()
.name("questdb-connect-cluster")
.build();

connect.start();
}

@Test
public void testSmoke() {
connect.kafka().createTopic(topicName, 1);
Map<String, String> props = ConnectTestUtils.baseConnectorProps(questDBContainer, topicName);
props.put(QuestDBSinkConnectorConfig.USERNAME, TEST_USER_NAME);
props.put(QuestDBSinkConnectorConfig.TOKEN, TEST_USER_TOKEN);

connect.configureConnector(ConnectTestUtils.CONNECTOR_NAME, props);
ConnectTestUtils.assertConnectorTaskRunningEventually(connect);
Schema schema = SchemaBuilder.struct().name("com.example.Person")
.field("firstname", Schema.STRING_SCHEMA)
.field("lastname", Schema.STRING_SCHEMA)
.field("age", Schema.INT8_SCHEMA)
.build();

Struct struct = new Struct(schema)
.put("firstname", "John")
.put("lastname", "Doe")
.put("age", (byte) 42);

connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));

QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"age\"\r\n"
+ "\"John\",\"Doe\",42\r\n",
"select firstname,lastname,age from " + topicName);
}
}
Loading

0 comments on commit 3ee7afd

Please sign in to comment.