diff --git a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/pom.xml b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/pom.xml index 5c785535708e..49954d02e868 100644 --- a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/pom.xml +++ b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/pom.xml @@ -56,6 +56,10 @@ language governing permissions and limitations under the License. --> org.apache.nifi nifi-ssl-context-service-api + + org.apache.nifi + nifi-dbcp-service-api + org.apache.nifi nifi-distributed-cache-client-service diff --git a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java index ac4dca7bd828..091237b47393 100644 --- a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java +++ b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/main/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQL.java @@ -63,6 +63,7 @@ import org.apache.nifi.cdc.mysql.processors.ssl.ConnectionPropertiesProvider; import org.apache.nifi.cdc.mysql.processors.ssl.StandardConnectionPropertiesProvider; import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.DescribedValue; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.PropertyValue; import org.apache.nifi.components.ValidationContext; @@ -72,6 +73,8 @@ import org.apache.nifi.components.state.Scope; import org.apache.nifi.components.state.StateManager; import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.dbcp.api.DatabasePasswordProvider; +import org.apache.nifi.dbcp.api.DatabasePasswordRequestContext; import org.apache.nifi.expression.ExpressionLanguageScope; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.logging.ComponentLog; @@ -99,6 +102,7 @@ import java.sql.SQLFeatureNotSupportedException; import java.sql.Statement; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.List; @@ -109,6 +113,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; import java.util.logging.Logger; import java.util.regex.Pattern; import javax.net.ssl.SSLContext; @@ -156,6 +161,8 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { private static final int DEFAULT_MYSQL_PORT = 3306; + private static final String JDBC_URL_FORMAT = "jdbc:mysql://%s"; + // A regular expression matching multiline comments, used when parsing DDL statements private static final Pattern MULTI_COMMENT_PATTERN = Pattern.compile("/\\*.*?\\*/", Pattern.DOTALL); @@ -258,6 +265,42 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT) .build(); + public enum PasswordSource implements DescribedValue { + PASSWORD("Password", "Use the configured Password property for database authentication."), + PASSWORD_PROVIDER("Password Provider", "Obtain database passwords from a configured Database Password Provider."); + + private final String displayName; + private final String description; + + PasswordSource(final String displayName, final String description) { + this.displayName = displayName; + this.description = description; + } + + @Override + public String getDisplayName() { + return displayName; + } + + @Override + public String getValue() { + return name(); + } + + @Override + public String getDescription() { + return description; + } + } + + public static final PropertyDescriptor PASSWORD_SOURCE = new PropertyDescriptor.Builder() + .name("Password Source") + .description("Specifies whether to supply the database password directly or obtain it from a Database Password Provider.") + .allowableValues(PasswordSource.class) + .defaultValue(PasswordSource.PASSWORD) + .required(true) + .build(); + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() .name("Password") .description("Password to access the MySQL cluster") @@ -265,6 +308,15 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { .sensitive(true) .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT) + .dependsOn(PASSWORD_SOURCE, PasswordSource.PASSWORD) + .build(); + + public static final PropertyDescriptor DB_PASSWORD_PROVIDER = new PropertyDescriptor.Builder() + .name("Database Password Provider") + .description("Controller Service that supplies database passwords on demand. When configured, the Password property is ignored.") + .required(true) + .identifiesControllerService(DatabasePasswordProvider.class) + .dependsOn(PASSWORD_SOURCE, PasswordSource.PASSWORD_PROVIDER) .build(); public static final PropertyDescriptor EVENTS_PER_FLOWFILE_STRATEGY = new PropertyDescriptor.Builder() @@ -419,7 +471,9 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { DRIVER_NAME, DRIVER_LOCATION, USERNAME, + PASSWORD_SOURCE, PASSWORD, + DB_PASSWORD_PROVIDER, EVENTS_PER_FLOWFILE_STRATEGY, NUMBER_OF_EVENTS_PER_FLOWFILE, SERVER_ID, @@ -443,6 +497,10 @@ public class CaptureChangeMySQL extends AbstractSessionFactoryProcessor { private volatile BinlogLifecycleListener lifecycleListener; private volatile GtidSet gtidSet; + private volatile DatabasePasswordProvider passwordProvider; + private volatile DatabasePasswordRequestContext passwordRequestContext; + private volatile String password; + // Set queue capacity to avoid excessive memory consumption private final BlockingQueue queue = new LinkedBlockingQueue<>(1000); @@ -635,17 +693,30 @@ public void setup(ProcessContext context) { final SSLMode sslMode = SSLMode.valueOf(context.getProperty(SSL_MODE).getValue()); final SSLContextService sslContextService = sslMode == SSLMode.DISABLED ? null : context.getProperty(SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); + final PasswordSource passwordSource = context.getProperty(PASSWORD_SOURCE).asAllowableValue(PasswordSource.class); + switch (passwordSource) { + case PASSWORD -> { + passwordProvider = null; + passwordRequestContext = null; + password = StringUtils.defaultString(context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue()); + } + case PASSWORD_PROVIDER -> { + password = null; + passwordProvider = context.getProperty(DB_PASSWORD_PROVIDER).asControllerService(DatabasePasswordProvider.class); + passwordRequestContext = DatabasePasswordRequestContext.builder() + .jdbcUrl(JDBC_URL_FORMAT.formatted(context.getProperty(HOSTS).evaluateAttributeExpressions().getValue())) + .driverClassName(context.getProperty(DRIVER_NAME).evaluateAttributeExpressions().getValue()) + .databaseUser(context.getProperty(USERNAME).evaluateAttributeExpressions().getValue()) + .build(); + } + } + // Save off MySQL cluster and JDBC driver information, will be used to connect for event enrichment as well as for the binlog connector try { List hosts = getHosts(context.getProperty(HOSTS).evaluateAttributeExpressions().getValue()); String username = context.getProperty(USERNAME).evaluateAttributeExpressions().getValue(); - String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue(); - - // BinaryLogClient expects a non-null password, so set it to the empty string if it is not provided - if (password == null) { - password = ""; - } + String resolvedPassword = resolvePassword(); long connectTimeout = context.getProperty(CONNECT_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS); @@ -654,7 +725,7 @@ public void setup(ProcessContext context) { Long serverId = context.getProperty(SERVER_ID).evaluateAttributeExpressions().asLong(); - connect(hosts, username, password, serverId, driverLocation, driverName, connectTimeout, sslContextService, sslMode); + connect(hosts, username, resolvedPassword, serverId, driverLocation, driverName, connectTimeout, sslContextService, sslMode); } catch (IOException | IllegalStateException e) { if (eventListener != null) { eventListener.stop(); @@ -812,10 +883,7 @@ protected void connect(List hosts, String username, String pa } try { - if (connectTimeout == 0) { - connectTimeout = Long.MAX_VALUE; - } - binlogClient.connect(connectTimeout); + binlogClient.connect(connectTimeout == 0 ? Long.MAX_VALUE : connectTimeout); binlogResourceInfo.setTransitUri("mysql://" + connectedHost.getHostString() + ":" + connectedHost.getPort()); } catch (IOException | TimeoutException te) { @@ -842,11 +910,21 @@ protected void connect(List hosts, String username, String pa final TlsConfiguration tlsConfiguration = sslContextService == null ? null : sslContextService.createTlsConfiguration(); final ConnectionPropertiesProvider connectionPropertiesProvider = new StandardConnectionPropertiesProvider(sslMode, tlsConfiguration); final Map jdbcConnectionProperties = connectionPropertiesProvider.getConnectionProperties(); - jdbcConnectionHolder = new JDBCConnectionHolder(connectedHost, username, password, jdbcConnectionProperties, connectTimeout); + + if (passwordProvider != null && passwordRequestContext != null) { + passwordRequestContext = DatabasePasswordRequestContext.builder() + .jdbcUrl(passwordRequestContext.getJdbcUrl()) + .driverClassName(passwordRequestContext.getDriverClassName()) + .databaseUser(passwordRequestContext.getDatabaseUser()) + .connectionProperties(jdbcConnectionProperties) + .build(); + } + + jdbcConnectionHolder = new JDBCConnectionHolder(connectedHost, username, this::resolvePassword, jdbcConnectionProperties, connectTimeout); try { // Ensure connection can be created. getJdbcConnection(); - } catch (SQLException e) { + } catch (SQLException | ProcessException e) { getLogger().error("Error creating binlog enrichment JDBC connection to any of the specified hosts", e); if (eventListener != null) { eventListener.stop(); @@ -1157,6 +1235,10 @@ public void stop() throws CDCException { if (jdbcConnectionHolder != null) { jdbcConnectionHolder.close(); } + + password = null; + passwordProvider = null; + passwordRequestContext = null; } } @@ -1211,6 +1293,19 @@ protected BinaryLogClient createBinlogClient(String hostname, int port, String u return new BinaryLogClient(hostname, port, username, password); } + private String resolvePassword() { + if (passwordProvider == null) { + return StringUtils.defaultString(password); + } + final char[] passwordChars = passwordProvider.getPassword(passwordRequestContext); + if (passwordChars == null || passwordChars.length == 0) { + throw new ProcessException("Database Password Provider returned an empty password"); + } + final String resolvedPassword = new String(passwordChars); + Arrays.fill(passwordChars, '\0'); + return resolvedPassword; + } + /** * Retrieves the column information for the specified database and table. The column information can be used to enrich CDC events coming from the RDBMS. * @@ -1253,18 +1348,17 @@ protected Connection getJdbcConnection() throws SQLException { private class JDBCConnectionHolder { private final String connectionUrl; private final Properties connectionProps = new Properties(); + private final Supplier passwordSupplier; private final long connectionTimeoutMillis; private Connection connection; - private JDBCConnectionHolder(InetSocketAddress host, String username, String password, Map customProperties, long connectionTimeoutMillis) { - this.connectionUrl = "jdbc:mysql://" + host.getHostString() + ":" + host.getPort(); + private JDBCConnectionHolder(InetSocketAddress host, String username, Supplier passwordSupplier, Map customProperties, long connectionTimeoutMillis) { + this.connectionUrl = JDBC_URL_FORMAT.formatted(host.getHostString() + ":" + host.getPort()); + this.passwordSupplier = passwordSupplier; connectionProps.putAll(customProperties); if (username != null) { connectionProps.put("user", username); - if (password != null) { - connectionProps.put("password", password); - } } this.connectionTimeoutMillis = connectionTimeoutMillis; @@ -1280,7 +1374,13 @@ private Connection getConnection() throws SQLException { close(); getLogger().trace("Creating a new JDBC connection."); - connection = DriverManager.getConnection(connectionUrl, connectionProps); + final Properties props = new Properties(); + props.putAll(connectionProps); + final String password = passwordSupplier.get(); + if (password != null) { + props.put("password", password); + } + connection = DriverManager.getConnection(connectionUrl, props); return connection; } diff --git a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.java b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.java index a245243a0c24..b153c9b92f0b 100644 --- a/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.java +++ b/nifi-extension-bundles/nifi-cdc/nifi-cdc-mysql-bundle/nifi-cdc-mysql-processors/src/test/java/org/apache/nifi/cdc/mysql/processors/CaptureChangeMySQLTest.java @@ -43,6 +43,8 @@ import org.apache.nifi.cdc.mysql.event.BinlogEventInfo; import org.apache.nifi.cdc.mysql.processors.ssl.BinaryLogSSLSocketFactory; import org.apache.nifi.components.state.Scope; +import org.apache.nifi.dbcp.api.DatabasePasswordProvider; +import org.apache.nifi.dbcp.api.DatabasePasswordRequestContext; import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.provenance.ProvenanceEventType; @@ -83,9 +85,12 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** @@ -190,6 +195,86 @@ public void testSslModeRequiredSslContextServiceConnected(@Mock SSLContextServic assertEquals(BinaryLogSSLSocketFactory.class, sslSocketFactory.getClass(), "Binary Log SSLSocketFactory class not matched"); } + @Test + public void testPasswordSourceProviderWithoutControllerServiceNotValid() { + testRunner.setProperty(CaptureChangeMySQL.HOSTS, LOCAL_HOST_DEFAULT_PORT); + testRunner.setProperty(CaptureChangeMySQL.PASSWORD_SOURCE, CaptureChangeMySQL.PasswordSource.PASSWORD_PROVIDER); + testRunner.assertNotValid(); + } + + @Test + public void testPasswordSourceProviderWithControllerServiceValid(@Mock DatabasePasswordProvider passwordProvider) throws InitializationException { + testRunner.setProperty(CaptureChangeMySQL.HOSTS, LOCAL_HOST_DEFAULT_PORT); + testRunner.setProperty(CaptureChangeMySQL.PASSWORD_SOURCE, CaptureChangeMySQL.PasswordSource.PASSWORD_PROVIDER); + + final String identifier = DatabasePasswordProvider.class.getName(); + when(passwordProvider.getIdentifier()).thenReturn(identifier); + testRunner.addControllerService(identifier, passwordProvider); + testRunner.enableControllerService(passwordProvider); + testRunner.setProperty(CaptureChangeMySQL.DB_PASSWORD_PROVIDER, identifier); + testRunner.assertValid(); + } + + @Test + public void testPasswordProviderTokenUsedOnRun(@Mock DatabasePasswordProvider passwordProvider) throws InitializationException { + final String identifier = DatabasePasswordProvider.class.getName(); + when(passwordProvider.getIdentifier()).thenReturn(identifier); + when(passwordProvider.getPassword(any())).thenReturn("token".toCharArray()); + testRunner.addControllerService(identifier, passwordProvider); + testRunner.enableControllerService(passwordProvider); + + testRunner.setProperty(CaptureChangeMySQL.HOSTS, LOCAL_HOST_DEFAULT_PORT); + testRunner.setProperty(CaptureChangeMySQL.DRIVER_LOCATION, DRIVER_LOCATION); + testRunner.setProperty(CaptureChangeMySQL.USERNAME, ROOT_USER); + testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, CONNECT_TIMEOUT); + testRunner.setProperty(CaptureChangeMySQL.PASSWORD_SOURCE, CaptureChangeMySQL.PasswordSource.PASSWORD_PROVIDER); + testRunner.setProperty(CaptureChangeMySQL.DB_PASSWORD_PROVIDER, identifier); + + testRunner.run(1, false, true); + + verify(passwordProvider, atLeastOnce()).getPassword(any(DatabasePasswordRequestContext.class)); + } + + @Test + public void testPasswordProviderEmptyTokenThrowsProcessException(@Mock DatabasePasswordProvider passwordProvider) throws InitializationException { + final String identifier = DatabasePasswordProvider.class.getName(); + when(passwordProvider.getIdentifier()).thenReturn(identifier); + when(passwordProvider.getPassword(any())).thenReturn(new char[0]); + testRunner.addControllerService(identifier, passwordProvider); + testRunner.enableControllerService(passwordProvider); + + testRunner.setProperty(CaptureChangeMySQL.HOSTS, LOCAL_HOST_DEFAULT_PORT); + testRunner.setProperty(CaptureChangeMySQL.DRIVER_LOCATION, DRIVER_LOCATION); + testRunner.setProperty(CaptureChangeMySQL.USERNAME, ROOT_USER); + testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, CONNECT_TIMEOUT); + testRunner.setProperty(CaptureChangeMySQL.PASSWORD_SOURCE, CaptureChangeMySQL.PasswordSource.PASSWORD_PROVIDER); + testRunner.setProperty(CaptureChangeMySQL.DB_PASSWORD_PROVIDER, identifier); + + final AssertionError assertionError = assertThrows(AssertionError.class, () -> testRunner.run()); + assertInstanceOf(ProcessException.class, assertionError.getCause()); + assertEquals("Database Password Provider returned an empty password", assertionError.getCause().getMessage()); + } + + @Test + public void testPasswordProviderNullTokenThrowsProcessException(@Mock DatabasePasswordProvider passwordProvider) throws InitializationException { + final String identifier = DatabasePasswordProvider.class.getName(); + when(passwordProvider.getIdentifier()).thenReturn(identifier); + when(passwordProvider.getPassword(any())).thenReturn(null); + testRunner.addControllerService(identifier, passwordProvider); + testRunner.enableControllerService(passwordProvider); + + testRunner.setProperty(CaptureChangeMySQL.HOSTS, LOCAL_HOST_DEFAULT_PORT); + testRunner.setProperty(CaptureChangeMySQL.DRIVER_LOCATION, DRIVER_LOCATION); + testRunner.setProperty(CaptureChangeMySQL.USERNAME, ROOT_USER); + testRunner.setProperty(CaptureChangeMySQL.CONNECT_TIMEOUT, CONNECT_TIMEOUT); + testRunner.setProperty(CaptureChangeMySQL.PASSWORD_SOURCE, CaptureChangeMySQL.PasswordSource.PASSWORD_PROVIDER); + testRunner.setProperty(CaptureChangeMySQL.DB_PASSWORD_PROVIDER, identifier); + + final AssertionError assertionError = assertThrows(AssertionError.class, () -> testRunner.run()); + assertInstanceOf(ProcessException.class, assertionError.getCause()); + assertEquals("Database Password Provider returned an empty password", assertionError.getCause().getMessage()); + } + @Test public void testConnectionFailures() { testRunner.setProperty(CaptureChangeMySQL.DRIVER_LOCATION, DRIVER_LOCATION);