chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed pull #5147, reformat & add more configs

Browse Source
This commit is contained in:
jianghang.loujh
2024-06-26 15:33:14 +08:00
parent 5c61f2fb56
commit b0f3ff8dcc
13 changed files with 258 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
deployer/src/main/resources/example/instance.properties
View File

@@ -12,6 +12,14 @@ canal.instance.master.position=
canal.instance.master.timestamp= canal.instance.master.timestamp=
canal.instance.master.gtid= canal.instance.master.gtid=
# ssl
#canal.instance.master.trustCertificateKeyStoreType=
#canal.instance.master.trustCertificateKeyStoreUrl=
#canal.instance.master.trustCertificateKeyStorePassword=
#canal.instance.master.clientCertificateKeyStoreType=
#canal.instance.master.clientCertificateKeyStoreUrl=
#canal.instance.master.clientCertificateKeyStorePassword=
# rds oss binlog # rds oss binlog
canal.instance.rds.accesskey= canal.instance.rds.accesskey=
canal.instance.rds.secretkey= canal.instance.rds.secretkey=

24
deployer/src/main/resources/spring/file-instance.xml
View File

@@ -125,6 +125,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>
<property name="standbyInfo"> <property name="standbyInfo">
@@ -135,6 +147,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>

48
deployer/src/main/resources/spring/group-instance.xml
View File

@@ -122,6 +122,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>
<property name="standbyInfo"> <property name="standbyInfo">
@@ -132,6 +144,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>
@@ -229,6 +253,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>
<property name="standbyInfo"> <property name="standbyInfo">
@@ -239,6 +275,18 @@
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" /> <property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
<property name="enableDruid" value="${canal.instance.enableDruid:false}" /> <property name="enableDruid" value="${canal.instance.enableDruid:false}" />
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" /> <property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
<property name="sslInfo">
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
</bean>
</property>
</bean> </bean>
</property> </property>

34
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java
View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.parse.driver.mysql; package com.alibaba.otter.canal.parse.driver.mysql;
import static com.alibaba.otter.canal.parse.driver.mysql.packets.Capability.CLIENT_SSL;
import java.io.IOException; import java.io.IOException;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.security.DigestException; import java.security.DigestException;
@@ -10,11 +12,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.AuthSwitchResponsePacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.*;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationSHA2Packet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.SslRequestCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.*; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.*;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool; import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool;
@@ -23,7 +21,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC; import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter; import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
import static com.alibaba.otter.canal.parse.driver.mysql.packets.Capability.CLIENT_SSL;
/** /**
* 基于mysql socket协议的链接实现 * 基于mysql socket协议的链接实现
@@ -37,7 +34,7 @@ public class MysqlConnector {
private InetSocketAddress address; private InetSocketAddress address;
private String username; private String username;
private String password; private String password;
private SslInfo sslInfo; private SslInfo sslInfo;
private String defaultSchema; private String defaultSchema;
private int soTimeout = 30 * 1000; private int soTimeout = 30 * 1000;
@@ -67,15 +64,14 @@ public class MysqlConnector {
this.password = password; this.password = password;
} }
public MysqlConnector(InetSocketAddress address, String username, String password, public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema){
String defaultSchema){
this(address, username, password); this(address, username, password);
this.defaultSchema = defaultSchema; this.defaultSchema = defaultSchema;
} }
public MysqlConnector(InetSocketAddress address, String username, String password, public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema,
String defaultSchema, SslInfo sslInfo) { SslInfo sslInfo){
this(address, username, password, defaultSchema); this(address, username, password, defaultSchema);
this.sslInfo = sslInfo; this.sslInfo = sslInfo;
} }
@@ -110,7 +106,9 @@ public class MysqlConnector {
sslCipher = result.getFieldValues().get(1); sslCipher = result.getFieldValues().get(1);
} }
logger.info("connect MysqlConnection in sslMode {}, Ssl_version:{}, Ssl_cipher:{}", logger.info("connect MysqlConnection in sslMode {}, Ssl_version:{}, Ssl_cipher:{}",
(sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED), sslVersion, sslCipher); (sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED),
sslVersion,
sslCipher);
} catch (Exception e) { } catch (Exception e) {
logger.info("Can't show SSL status, server may not standard MySQL server: {}", e.toString()); logger.info("Can't show SSL status, server may not standard MySQL server: {}", e.toString());
logger.debug("show SSL status exception", e); logger.debug("show SSL status exception", e);
@@ -206,21 +204,21 @@ public class MysqlConnector {
if (sslMode != SslMode.DISABLED) { if (sslMode != SslMode.DISABLED) {
boolean serverSupportSsl = (handshakePacket.serverCapabilities & CLIENT_SSL) > 0; boolean serverSupportSsl = (handshakePacket.serverCapabilities & CLIENT_SSL) > 0;
if (!serverSupportSsl) { if (!serverSupportSsl) {
throw new IOException("MySQL Server does not support SSL: " + address throw new IOException("MySQL Server does not support SSL: " + address + " serverCapabilities: "
+ " serverCapabilities: " + handshakePacket.serverCapabilities); + handshakePacket.serverCapabilities);
} }
byte[] sslPacket = new SslRequestCommandPacket(handshakePacket.serverCharsetNumber).toBytes(); byte[] sslPacket = new SslRequestCommandPacket(handshakePacket.serverCharsetNumber).toBytes();
HeaderPacket sslHeader = new HeaderPacket(); HeaderPacket sslHeader = new HeaderPacket();
sslHeader.setPacketBodyLength(sslPacket.length); sslHeader.setPacketBodyLength(sslPacket.length);
sslHeader.setPacketSequenceNumber((byte)(header.getPacketSequenceNumber() + 1)); sslHeader.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
header.setPacketSequenceNumber((byte)(header.getPacketSequenceNumber() + 1)); header.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
PacketManager.writePkg(channel, sslHeader.toBytes(), sslPacket); PacketManager.writePkg(channel, sslHeader.toBytes(), sslPacket);
channel = SocketChannelPool.connectSsl(channel, sslInfo); channel = SocketChannelPool.connectSsl(channel, sslInfo);
this.channel = channel; this.channel = channel;
} }
if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) { if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) {
// HandshakeV9 // HandshakeV9
auth323(channel, (byte)(header.getPacketSequenceNumber() + 1), handshakePacket.seed); auth323(channel, (byte) (header.getPacketSequenceNumber() + 1), handshakePacket.seed);
return; return;
} }
@@ -248,7 +246,7 @@ public class MysqlConnector {
byte[] clientAuthPkgBody = clientAuth.toBytes(); byte[] clientAuthPkgBody = clientAuth.toBytes();
HeaderPacket h = new HeaderPacket(); HeaderPacket h = new HeaderPacket();
h.setPacketBodyLength(clientAuthPkgBody.length); h.setPacketBodyLength(clientAuthPkgBody.length);
h.setPacketSequenceNumber((byte)(header.getPacketSequenceNumber() + 1)); h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
PacketManager.writePkg(channel, h.toBytes(), clientAuthPkgBody); PacketManager.writePkg(channel, h.toBytes(), clientAuthPkgBody);
logger.info("client authentication packet is sent out."); logger.info("client authentication packet is sent out.");

12
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/SslRequestCommandPacket.java
View File

@@ -15,7 +15,9 @@ public class SslRequestCommandPacket implements IPacket {
private final int serverCharsetNumber; private final int serverCharsetNumber;
public SslRequestCommandPacket(int serverCharsetNumber) {this.serverCharsetNumber = serverCharsetNumber;} public SslRequestCommandPacket(int serverCharsetNumber){
this.serverCharsetNumber = serverCharsetNumber;
}
@Override @Override
public void fromBytes(byte[] data) throws IOException { public void fromBytes(byte[] data) throws IOException {
@@ -25,11 +27,9 @@ public class SslRequestCommandPacket implements IPacket {
@Override @Override
public byte[] toBytes() throws IOException { public byte[] toBytes() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream(); ByteArrayOutputStream out = new ByteArrayOutputStream();
int clientCapabilities = Capability.CLIENT_LONG_FLAG int clientCapabilities = Capability.CLIENT_LONG_FLAG | Capability.CLIENT_PROTOCOL_41
| Capability.CLIENT_PROTOCOL_41 | Capability.CLIENT_SECURE_CONNECTION | Capability.CLIENT_PLUGIN_AUTH
| Capability.CLIENT_SECURE_CONNECTION | Capability.CLIENT_SSL;
| Capability.CLIENT_PLUGIN_AUTH
| Capability.CLIENT_SSL;
ByteHelper.writeUnsignedIntLittleEndian(clientCapabilities, out); ByteHelper.writeUnsignedIntLittleEndian(clientCapabilities, out);
ByteHelper.writeUnsignedIntLittleEndian(0, out); ByteHelper.writeUnsignedIntLittleEndian(0, out);
out.write(serverCharsetNumber); out.write(serverCharsetNumber);

18
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
View File

@@ -19,11 +19,11 @@ import javax.net.ssl.SSLSocket;
*/ */
public class BioSocketChannel implements SocketChannel { public class BioSocketChannel implements SocketChannel {
static final int DEFAULT_CONNECT_TIMEOUT = 10 * 1000; static final int DEFAULT_CONNECT_TIMEOUT = 10 * 1000;
static final int SO_TIMEOUT = 1000; static final int SO_TIMEOUT = 1000;
private Socket socket; private Socket socket;
private InputStream input; private InputStream input;
private OutputStream output; private OutputStream output;
private final boolean ssl; private final boolean ssl;
BioSocketChannel(Socket socket) throws IOException{ BioSocketChannel(Socket socket) throws IOException{
@@ -92,9 +92,9 @@ public class BioSocketChannel implements SocketChannel {
} }
} }
if (remain > 0 && accTimeout >= timeout) { if (remain > 0 && accTimeout >= timeout) {
throw new SocketTimeoutException("Timeout occurred, failed to read total " + readSize + " bytes in " throw new SocketTimeoutException(
+ timeout + " milliseconds, actual read only " + (readSize - remain) "Timeout occurred, failed to read total " + readSize + " bytes in " + timeout
+ " bytes"); + " milliseconds, actual read only " + (readSize - remain) + " bytes");
} }
return data; return data;
} }
@@ -137,9 +137,11 @@ public class BioSocketChannel implements SocketChannel {
} }
return false; return false;
} }
public boolean isSsl() { public boolean isSsl() {
return ssl; return ssl;
} }
public Socket getSocket() { public Socket getSocket() {
return socket; return socket;
} }

116
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java
View File

@@ -8,11 +8,7 @@ import java.net.URL;
import java.security.InvalidAlgorithmParameterException; import java.security.InvalidAlgorithmParameterException;
import java.security.KeyStore; import java.security.KeyStore;
import java.security.cert.*; import java.security.cert.*;
import java.util.ArrayList; import java.util.*;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import javax.naming.InvalidNameException; import javax.naming.InvalidNameException;
@@ -21,18 +17,19 @@ import javax.naming.ldap.Rdn;
import javax.net.ssl.*; import javax.net.ssl.*;
import javax.security.auth.x500.X500Principal; import javax.security.auth.x500.X500Principal;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
/** /**
* @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28 * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
* @author chuanyi 2018-3-3 保留<code>open</code>减少文件变更数量 * @author chuanyi 2018-3-3 保留<code>open</code>减少文件变更数量
*/ */
public abstract class BioSocketChannelPool { public abstract class BioSocketChannelPool {
private static final Logger logger = LoggerFactory.getLogger(BioSocketChannelPool.class); private static final Logger logger = LoggerFactory.getLogger(BioSocketChannelPool.class);
public static BioSocketChannel open(SocketAddress address) throws Exception { public static BioSocketChannel open(SocketAddress address) throws Exception {
@@ -42,7 +39,6 @@ public abstract class BioSocketChannelPool {
public static BioSocketChannel openSsl(Socket socket, SslInfo sslInfo) throws Exception { public static BioSocketChannel openSsl(Socket socket, SslInfo sslInfo) throws Exception {
SslMode sslMode = sslInfo.getSslMode(); SslMode sslMode = sslInfo.getSslMode();
switch (sslMode) { switch (sslMode) {
case REQUIRED: case REQUIRED:
case PREFERRED: case PREFERRED:
@@ -67,9 +63,7 @@ public abstract class BioSocketChannelPool {
} }
/** /**
* from JDBC driver * from JDBC driver com.mysql.cj.protocol.ExportControlled#performTlsHandshake
*
* com.mysql.cj.protocol.ExportControlled#performTlsHandshake
* com.mysql.cj.protocol.ExportControlled#getSSLContext * com.mysql.cj.protocol.ExportControlled#getSSLContext
* *
* @param socket * @param socket
@@ -82,14 +76,12 @@ public abstract class BioSocketChannelPool {
boolean verifyServerCert = sslMode == SslMode.VERIFY_CA || sslMode == SslMode.VERIFY_IDENTITY; boolean verifyServerCert = sslMode == SslMode.VERIFY_CA || sslMode == SslMode.VERIFY_IDENTITY;
String clientCertificateKeyStoreUrl = sslInfo.getClientCertificateKeyStoreUrl(); String clientCertificateKeyStoreUrl = sslInfo.getClientCertificateKeyStoreUrl();
String clientCertificateKeyStoreType = sslInfo.getClientCertificateKeyStoreType() != null String clientCertificateKeyStoreType = sslInfo.getClientCertificateKeyStoreType() != null ? sslInfo
? sslInfo.getClientCertificateKeyStoreType() .getClientCertificateKeyStoreType() : "JKS";
: "JKS";
String clientCertificateKeyStorePassword = sslInfo.getClientCertificateKeyStorePassword(); String clientCertificateKeyStorePassword = sslInfo.getClientCertificateKeyStorePassword();
String trustCertificateKeyStoreUrl = sslInfo.getTrustCertificateKeyStoreUrl(); String trustCertificateKeyStoreUrl = sslInfo.getTrustCertificateKeyStoreUrl();
String trustCertificateKeyStoreType = sslInfo.getTrustCertificateKeyStoreType() != null String trustCertificateKeyStoreType = sslInfo.getTrustCertificateKeyStoreType() != null ? sslInfo
? sslInfo.getTrustCertificateKeyStoreType() .getTrustCertificateKeyStoreType() : "JKS";
: "JKS";
String trustCertificateKeyStorePassword = sslInfo.getTrustCertificateKeyStorePassword(); String trustCertificateKeyStorePassword = sslInfo.getTrustCertificateKeyStorePassword();
boolean fallbackToDefaultTrustStore = true; boolean fallbackToDefaultTrustStore = true;
String hostName = sslMode == SslMode.VERIFY_IDENTITY ? socket.getInetAddress().getHostName() : null; String hostName = sslMode == SslMode.VERIFY_IDENTITY ? socket.getInetAddress().getHostName() : null;
@@ -97,36 +89,38 @@ public abstract class BioSocketChannelPool {
SSLContext sslContext = getSSLContext(clientCertificateKeyStoreUrl, SSLContext sslContext = getSSLContext(clientCertificateKeyStoreUrl,
clientCertificateKeyStoreType, clientCertificateKeyStoreType,
clientCertificateKeyStorePassword, clientCertificateKeyStorePassword,
trustCertificateKeyStoreUrl, trustCertificateKeyStoreType, trustCertificateKeyStoreUrl,
trustCertificateKeyStoreType,
trustCertificateKeyStorePassword, trustCertificateKeyStorePassword,
fallbackToDefaultTrustStore, fallbackToDefaultTrustStore,
verifyServerCert, hostName); verifyServerCert,
hostName);
SSLSocketFactory socketFactory = sslContext.getSocketFactory(); SSLSocketFactory socketFactory = sslContext.getSocketFactory();
SSLSocket sslSocket = (SSLSocket)socketFactory.createSocket( SSLSocket sslSocket = (SSLSocket) socketFactory
socket, socket.getInetAddress().getHostName(), socket.getPort(), true); .createSocket(socket, socket.getInetAddress().getHostName(), socket.getPort(), true);
String[] protocolArr = null; String[] protocolArr = null;
if (StringUtils.isNotEmpty(sslInfo.getTlsVersions())) { if (StringUtils.isNotEmpty(sslInfo.getTlsVersions())) {
protocolArr = StringUtils.split(sslInfo.getTlsVersions(), ","); protocolArr = StringUtils.split(sslInfo.getTlsVersions(), ",");
} }
if (protocolArr == null || protocolArr.length == 0) { if (protocolArr == null || protocolArr.length == 0) {
protocolArr = new String[] {"TLSv1.2", "TLSv1.3"}; protocolArr = new String[] { "TLSv1.2", "TLSv1.3" };
} }
logger.info("SSL protocol: {}", StringUtils.join(protocolArr, ",")); logger.info("SSL protocol: {}", StringUtils.join(protocolArr, ","));
sslSocket.setEnabledProtocols(protocolArr); sslSocket.setEnabledProtocols(protocolArr);
//sslSocket.setEnabledCipherSuites("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_DSS_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_DSS_WITH_AES_128_GCM_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384,TLS_RSA_WITH_AES_256_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA384,TLS_DHE_RSA_WITH_AES_256_CBC_SHA256,TLS_DHE_DSS_WITH_AES_256_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA,TLS_DHE_RSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256,TLS_RSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_DSS_WITH_AES_128_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_128_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_DSS_WITH_AES_128_CBC_SHA".split(",")); // sslSocket.setEnabledCipherSuites("TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_ECDSA_WITH_AES_256_GCM_SHA384,TLS_ECDH_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_RSA_WITH_AES_256_GCM_SHA384,TLS_DHE_DSS_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,TLS_RSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_GCM_SHA256,TLS_ECDH_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_RSA_WITH_AES_128_GCM_SHA256,TLS_DHE_DSS_WITH_AES_128_GCM_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384,TLS_RSA_WITH_AES_256_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA384,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA384,TLS_DHE_RSA_WITH_AES_256_CBC_SHA256,TLS_DHE_DSS_WITH_AES_256_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA,TLS_ECDH_RSA_WITH_AES_256_CBC_SHA,TLS_DHE_RSA_WITH_AES_256_CBC_SHA,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256,TLS_RSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA256,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_RSA_WITH_AES_128_CBC_SHA256,TLS_DHE_DSS_WITH_AES_128_CBC_SHA256,TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_128_CBC_SHA,TLS_ECDH_ECDSA_WITH_AES_128_CBC_SHA,TLS_ECDH_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_RSA_WITH_AES_128_CBC_SHA,TLS_DHE_DSS_WITH_AES_128_CBC_SHA".split(","));
sslSocket.startHandshake(); sslSocket.startHandshake();
logger.info("SSL socket handshake success."); logger.info("SSL socket handshake success.");
return sslSocket; return sslSocket;
} }
private static SSLContext getSSLContext(String clientCertificateKeyStoreUrl, String clientCertificateKeyStoreType, private static SSLContext getSSLContext(String clientCertificateKeyStoreUrl, String clientCertificateKeyStoreType,
String clientCertificateKeyStorePassword, String clientCertificateKeyStorePassword,
String trustCertificateKeyStoreUrl, String trustCertificateKeyStoreType, String trustCertificateKeyStoreUrl, String trustCertificateKeyStoreType,
String trustCertificateKeyStorePassword, String trustCertificateKeyStorePassword,
boolean fallbackToDefaultTrustStore, boolean verifyServerCert, String hostName) boolean fallbackToDefaultTrustStore, boolean verifyServerCert,
throws Exception { String hostName) throws Exception {
KeyManager[] kms = null; KeyManager[] kms = null;
List<TrustManager> tms = new ArrayList<>(); List<TrustManager> tms = new ArrayList<>();
@@ -140,8 +134,8 @@ public abstract class BioSocketChannelPool {
if (StringUtils.isNotEmpty(clientCertificateKeyStoreType)) { if (StringUtils.isNotEmpty(clientCertificateKeyStoreType)) {
KeyStore clientKeyStore = KeyStore.getInstance(clientCertificateKeyStoreType); KeyStore clientKeyStore = KeyStore.getInstance(clientCertificateKeyStoreType);
URL ksURL = new URL(clientCertificateKeyStoreUrl); URL ksURL = new URL(clientCertificateKeyStoreUrl);
char[] password = (clientCertificateKeyStorePassword == null) ? new char[0] char[] password = (clientCertificateKeyStorePassword == null) ? new char[0] : clientCertificateKeyStorePassword
: clientCertificateKeyStorePassword.toCharArray(); .toCharArray();
ksIS = ksURL.openStream(); ksIS = ksURL.openStream();
clientKeyStore.load(ksIS, password); clientKeyStore.load(ksIS, password);
kmf.init(clientKeyStore, password); kmf.init(clientKeyStore, password);
@@ -152,7 +146,8 @@ public abstract class BioSocketChannelPool {
try { try {
ksIS.close(); ksIS.close();
} catch (IOException e) { } catch (IOException e) {
// can't close input stream, but keystore can be properly initialized so we shouldn't throw // can't close input stream, but keystore can be properly initialized so we
// shouldn't throw
// this exception // this exception
} }
} }
@@ -165,11 +160,11 @@ public abstract class BioSocketChannelPool {
char[] trustStorePassword = null; char[] trustStorePassword = null;
KeyStore trustKeyStore = null; KeyStore trustKeyStore = null;
if (StringUtils.isNotEmpty(trustCertificateKeyStoreUrl) && StringUtils.isNotEmpty( if (StringUtils.isNotEmpty(trustCertificateKeyStoreUrl)
trustCertificateKeyStoreType)) { && StringUtils.isNotEmpty(trustCertificateKeyStoreType)) {
trustStoreType = trustCertificateKeyStoreType; trustStoreType = trustCertificateKeyStoreType;
trustStorePassword = (trustCertificateKeyStorePassword == null) ? new char[0] trustStorePassword = (trustCertificateKeyStorePassword == null) ? new char[0] : trustCertificateKeyStorePassword
: trustCertificateKeyStorePassword.toCharArray(); .toCharArray();
trustStoreIS = new URL(trustCertificateKeyStoreUrl).openStream(); trustStoreIS = new URL(trustCertificateKeyStoreUrl).openStream();
trustKeyStore = KeyStore.getInstance(trustStoreType); trustKeyStore = KeyStore.getInstance(trustStoreType);
@@ -178,15 +173,18 @@ public abstract class BioSocketChannelPool {
if (trustKeyStore != null || fallbackToDefaultTrustStore) { if (trustKeyStore != null || fallbackToDefaultTrustStore) {
tmf.init(trustKeyStore); tmf.init(trustKeyStore);
// (trustKeyStore == null) initializes the TrustManagerFactory with the default truststore. // (trustKeyStore == null) initializes the TrustManagerFactory with the default
// truststore.
// building the customized list of TrustManagers from original one if it's available // building the customized list of TrustManagers from original one if it's
// available
TrustManager[] origTms = tmf.getTrustManagers(); TrustManager[] origTms = tmf.getTrustManagers();
for (TrustManager tm : origTms) { for (TrustManager tm : origTms) {
// wrap X509TrustManager or put original if non-X509 TrustManager // wrap X509TrustManager or put original if non-X509 TrustManager
tms.add(tm instanceof X509TrustManager ? new X509TrustManagerWrapper((X509TrustManager)tm, tms.add(tm instanceof X509TrustManager ? new X509TrustManagerWrapper((X509TrustManager) tm,
verifyServerCert, hostName) : tm); verifyServerCert,
hostName) : tm);
} }
} }
@@ -195,13 +193,15 @@ public abstract class BioSocketChannelPool {
try { try {
trustStoreIS.close(); trustStoreIS.close();
} catch (IOException e) { } catch (IOException e) {
// can't close input stream, but keystore can be properly initialized so we shouldn't throw this // can't close input stream, but keystore can be properly initialized so we
// shouldn't throw this
// exception // exception
} }
} }
} }
// if original TrustManagers are not available then putting one X509TrustManagerWrapper which take care only // if original TrustManagers are not available then putting one
// X509TrustManagerWrapper which take care only
// about expiration check // about expiration check
if (tms.size() == 0) { if (tms.size() == 0) {
tms.add(new X509TrustManagerWrapper(verifyServerCert, hostName)); tms.add(new X509TrustManagerWrapper(verifyServerCert, hostName));
@@ -214,24 +214,24 @@ public abstract class BioSocketChannelPool {
public static class X509TrustManagerWrapper implements X509TrustManager { public static class X509TrustManagerWrapper implements X509TrustManager {
private X509TrustManager origTm = null; private X509TrustManager origTm = null;
private boolean verifyServerCert = false; private boolean verifyServerCert = false;
private String hostName = null; private String hostName = null;
private CertificateFactory certFactory = null; private CertificateFactory certFactory = null;
private PKIXParameters validatorParams = null; private PKIXParameters validatorParams = null;
private CertPathValidator validator = null; private CertPathValidator validator = null;
public X509TrustManagerWrapper(X509TrustManager tm, boolean verifyServerCertificate, String hostName) public X509TrustManagerWrapper(X509TrustManager tm, boolean verifyServerCertificate,
throws CertificateException { String hostName) throws CertificateException{
this.origTm = tm; this.origTm = tm;
this.verifyServerCert = verifyServerCertificate; this.verifyServerCert = verifyServerCertificate;
this.hostName = hostName; this.hostName = hostName;
if (verifyServerCertificate) { if (verifyServerCertificate) {
try { try {
Set<TrustAnchor> anch = Arrays.stream(tm.getAcceptedIssuers()).map(c -> new TrustAnchor(c, null)) Set<TrustAnchor> anch = Arrays.stream(tm.getAcceptedIssuers())
.collect( .map(c -> new TrustAnchor(c, null))
Collectors.toSet()); .collect(Collectors.toSet());
this.validatorParams = new PKIXParameters(anch); this.validatorParams = new PKIXParameters(anch);
this.validatorParams.setRevocationEnabled(false); this.validatorParams.setRevocationEnabled(false);
this.validator = CertPathValidator.getInstance("PKIX"); this.validator = CertPathValidator.getInstance("PKIX");
@@ -243,7 +243,7 @@ public abstract class BioSocketChannelPool {
} }
public X509TrustManagerWrapper(boolean verifyServerCertificate, String hostName) { public X509TrustManagerWrapper(boolean verifyServerCertificate, String hostName){
this.verifyServerCert = verifyServerCertificate; this.verifyServerCert = verifyServerCertificate;
this.hostName = hostName; this.hostName = hostName;
} }
@@ -266,7 +266,7 @@ public abstract class BioSocketChannelPool {
// Validate against truststore // Validate against truststore
CertPathValidatorResult result = this.validator.validate(certPath, this.validatorParams); CertPathValidatorResult result = this.validator.validate(certPath, this.validatorParams);
// Check expiration for the CA used to validate this path // Check expiration for the CA used to validate this path
((PKIXCertPathValidatorResult)result).getTrustAnchor().getTrustedCert().checkValidity(); ((PKIXCertPathValidatorResult) result).getTrustAnchor().getTrustedCert().checkValidity();
} catch (InvalidAlgorithmParameterException e) { } catch (InvalidAlgorithmParameterException e) {
throw new CertificateException(e); throw new CertificateException(e);
@@ -308,8 +308,10 @@ public abstract class BioSocketChannelPool {
if (!expectHostNames.contains(this.hostName)) { if (!expectHostNames.contains(this.hostName)) {
throw new CertificateException( throw new CertificateException(
"Server certificate identity check failed. The certificate Common Name " "Server certificate identity check failed. The certificate Common Name "
+ expectHostNames.stream().map(h -> "'" + h + "'").collect(Collectors.joining(", ")) + expectHostNames.stream()
+ " does not match with '" + this.hostName + "'."); .map(h -> "'" + h + "'")
.collect(Collectors.joining(", "))
+ " does not match with '" + this.hostName + "'.");
} }
} }

12
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
View File

@@ -3,18 +3,19 @@ package com.alibaba.otter.canal.parse.driver.mysql.socket;
import java.io.IOException; import java.io.IOException;
import java.net.SocketAddress; import java.net.SocketAddress;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
/** /**
* @author agapple 2018年3月12日 下午10:46:22 * @author agapple 2018年3月12日 下午10:46:22
* @since 1.0.26 * @since 1.0.26
*/ */
public abstract class SocketChannelPool { public abstract class SocketChannelPool {
private static final Logger logger = LoggerFactory.getLogger(SocketChannelPool.class); private static final Logger logger = LoggerFactory.getLogger(SocketChannelPool.class);
public static SocketChannel open(SocketAddress address) throws Exception { public static SocketChannel open(SocketAddress address) throws Exception {
@@ -35,8 +36,7 @@ public abstract class SocketChannelPool {
} else { } else {
SocketAddress remoteSocketAddress = channel.getRemoteSocketAddress(); SocketAddress remoteSocketAddress = channel.getRemoteSocketAddress();
try { try {
return BioSocketChannelPool.openSsl( return BioSocketChannelPool.openSsl(((BioSocketChannel) channel).getSocket(), sslInfo);
((BioSocketChannel)channel).getSocket(), sslInfo);
} catch (Exception e) { } catch (Exception e) {
if (sslMode == SslMode.PREFERRED) { if (sslMode == SslMode.PREFERRED) {
// still use non ssl channel // still use non ssl channel
@@ -45,7 +45,7 @@ public abstract class SocketChannelPool {
} }
IOException ioe; IOException ioe;
if (e instanceof IOException) { if (e instanceof IOException) {
ioe = (IOException)e; ioe = (IOException) e;
} else { } else {
ioe = new IOException(e); ioe = new IOException(e);
} }

20
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/ssl/SslInfo.java
View File

@@ -7,18 +7,18 @@ package com.alibaba.otter.canal.parse.driver.mysql.ssl;
public class SslInfo { public class SslInfo {
private SslMode sslMode = SslMode.DISABLED; private SslMode sslMode = SslMode.DISABLED;
private String tlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3 private String tlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3
private String trustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12 private String trustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12
private String trustCertificateKeyStoreUrl; // trustStore 证书 private String trustCertificateKeyStoreUrl; // trustStore 证书
private String trustCertificateKeyStorePassword; // trustStore 证书密码 private String trustCertificateKeyStorePassword; // trustStore 证书密码
private String clientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12 private String clientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12
private String clientCertificateKeyStoreUrl; // client 证书 private String clientCertificateKeyStoreUrl; // client 证书
private String clientCertificateKeyStorePassword; // client 证书密码 private String clientCertificateKeyStorePassword; // client 证书密码
public SslInfo(SslMode sslMode, String tlsVersions, String trustCertificateKeyStoreType, public SslInfo(SslMode sslMode, String tlsVersions, String trustCertificateKeyStoreType,
String trustCertificateKeyStoreUrl, String trustCertificateKeyStorePassword, String trustCertificateKeyStoreUrl, String trustCertificateKeyStorePassword,
String clientCertificateKeyStoreType, String clientCertificateKeyStoreType, String clientCertificateKeyStoreUrl,
String clientCertificateKeyStoreUrl, String clientCertificateKeyStorePassword) { String clientCertificateKeyStorePassword) {
this.sslMode = sslMode; this.sslMode = sslMode;
this.tlsVersions = tlsVersions; this.tlsVersions = tlsVersions;
this.trustCertificateKeyStoreType = trustCertificateKeyStoreType; this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;

92
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java
View File

@@ -66,14 +66,20 @@ public class CanalParameter implements Serializable {
private String dbUsername; // 数据库用户 private String dbUsername; // 数据库用户
private String dbPassword; // 数据库密码 private String dbPassword; // 数据库密码
private String sslMode = SslMode.DISABLED.name(); private String sslMode = SslMode.DISABLED.name();
private String tlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3 private String tlsVersions; // 和
private String trustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12 // enabledTLSProtocols
private String trustCertificateKeyStoreUrl; // trustStore 证书路径 // 同含义，TLSv1.2,TLSv1.3
private String trustCertificateKeyStorePassword; // trustStore 证书密码 private String trustCertificateKeyStoreType; // trustStore
private String clientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12 // 证书类型，支持 JKS (默认)
private String clientCertificateKeyStoreUrl; // client 证书路径 // 和 PKCS12
private String clientCertificateKeyStorePassword; // client 证书密码 private String trustCertificateKeyStoreUrl; // trustStore 证书路径
private String trustCertificateKeyStorePassword; // trustStore 证书密码
private String clientCertificateKeyStoreType; // client 证书类型，支持
// JKS (默认) 和
// PKCS12
private String clientCertificateKeyStoreUrl; // client 证书路径
private String clientCertificateKeyStorePassword; // client 证书密码
// binlog链接信息 // binlog链接信息
private IndexMode indexMode; private IndexMode indexMode;
@@ -117,26 +123,38 @@ public class CanalParameter implements Serializable {
private InetSocketAddress masterAddress; // 主库信息 private InetSocketAddress masterAddress; // 主库信息
private String masterUsername; // 帐号 private String masterUsername; // 帐号
private String masterPassword; // 密码 private String masterPassword; // 密码
private String masterSslMode = SslMode.DISABLED.name(); private String masterSslMode = SslMode.DISABLED.name();
private String masterTlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3 private String masterTlsVersions; // 和
private String masterTrustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12 // enabledTLSProtocols
private String masterTrustCertificateKeyStoreUrl; // trustStore 证书路径 // 同含义，TLSv1.2,TLSv1.3
private String masterTrustCertificateKeyStorePassword; // trustStore 证书密码 private String masterTrustCertificateKeyStoreType; // trustStore
private String masterClientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12 // 证书类型，支持 JKS (默认)
private String masterClientCertificateKeyStoreUrl; // client 证书路径 // 和 PKCS12
private String masterClientCertificateKeyStorePassword; // client 证书密码 private String masterTrustCertificateKeyStoreUrl; // trustStore 证书路径
private String masterTrustCertificateKeyStorePassword; // trustStore 证书密码
private String masterClientCertificateKeyStoreType; // client 证书类型，支持
// JKS (默认) 和
// PKCS12
private String masterClientCertificateKeyStoreUrl; // client 证书路径
private String masterClientCertificateKeyStorePassword; // client 证书密码
private InetSocketAddress standbyAddress; // 备库信息 private InetSocketAddress standbyAddress; // 备库信息
private String standbyUsername; // 帐号 private String standbyUsername; // 帐号
private String standbyPassword; private String standbyPassword;
private String standbySslMode = SslMode.DISABLED.name(); private String standbySslMode = SslMode.DISABLED.name();
private String standbyTlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3 private String standbyTlsVersions; // 和
private String standbyTrustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12 // enabledTLSProtocols
private String standbyTrustCertificateKeyStoreUrl; // trustStore 证书路径 // 同含义，TLSv1.2,TLSv1.3
private String standbyTrustCertificateKeyStorePassword; // trustStore 证书密码 private String standbyTrustCertificateKeyStoreType; // trustStore
private String standbyClientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12 // 证书类型，支持 JKS (默认)
private String standbyClientCertificateKeyStoreUrl; // client 证书路径 // 和 PKCS12
private String standbyClientCertificateKeyStorePassword; // client 证书密码 private String standbyTrustCertificateKeyStoreUrl; // trustStore 证书路径
private String standbyTrustCertificateKeyStorePassword; // trustStore 证书密码
private String standbyClientCertificateKeyStoreType; // client 证书类型，支持
// JKS (默认) 和
// PKCS12
private String standbyClientCertificateKeyStoreUrl; // client 证书路径
private String standbyClientCertificateKeyStorePassword; // client 证书密码
private String masterLogfileName = null; // master起始位置 private String masterLogfileName = null; // master起始位置
private Long masterLogfileOffest = null; private Long masterLogfileOffest = null;
@@ -146,11 +164,11 @@ public class CanalParameter implements Serializable {
private Long standbyTimestamp = null; private Long standbyTimestamp = null;
private Boolean parallel = Boolean.FALSE; private Boolean parallel = Boolean.FALSE;
//自定义alarmHandler类全路径 // 自定义alarmHandler类全路径
private String alarmHandlerClass = null; private String alarmHandlerClass = null;
//自定义alarmHandler插件文件夹路径 // 自定义alarmHandler插件文件夹路径
private String alarmHandlerPluginDir = null; private String alarmHandlerPluginDir = null;
//是否支持多流消费 // 是否支持多流消费
private Boolean multiStreamEnable = Boolean.FALSE; private Boolean multiStreamEnable = Boolean.FALSE;
public static enum RunMode { public static enum RunMode {
@@ -209,6 +227,7 @@ public class CanalParameter implements Serializable {
} }
public static enum StorageMode { public static enum StorageMode {
/** 内存存储模式 */ /** 内存存储模式 */
MEMORY, MEMORY,
/** 文件存储模式 */ /** 文件存储模式 */
@@ -231,6 +250,7 @@ public class CanalParameter implements Serializable {
} }
public static enum StorageScavengeMode { public static enum StorageScavengeMode {
/** 在存储满的时候触发 */ /** 在存储满的时候触发 */
ON_FULL, ON_FULL,
/** 在每次有ack请求时触发 */ /** 在每次有ack请求时触发 */
@@ -258,6 +278,7 @@ public class CanalParameter implements Serializable {
} }
public static enum SourcingType { public static enum SourcingType {
/** mysql DB */ /** mysql DB */
MYSQL, MYSQL,
/** localBinLog */ /** localBinLog */
@@ -285,6 +306,7 @@ public class CanalParameter implements Serializable {
} }
public static enum MetaMode { public static enum MetaMode {
/** 内存存储模式 */ /** 内存存储模式 */
MEMORY, MEMORY,
/** 文件存储模式 */ /** 文件存储模式 */
@@ -312,6 +334,7 @@ public class CanalParameter implements Serializable {
} }
public static enum IndexMode { public static enum IndexMode {
/** 内存存储模式 */ /** 内存存储模式 */
MEMORY, MEMORY,
/** 文件存储模式 */ /** 文件存储模式 */
@@ -345,6 +368,7 @@ public class CanalParameter implements Serializable {
} }
public static enum BatchMode { public static enum BatchMode {
/** 对象数量 */ /** 对象数量 */
ITEMSIZE, ITEMSIZE,
@@ -703,16 +727,16 @@ public class CanalParameter implements Serializable {
groupDbAddresses.add(groupAddresses); groupDbAddresses.add(groupAddresses);
} }
} else { } else {
if (masterAddress != null) { if (masterAddress != null) {
List<DataSourcing> groupAddresses = new ArrayList<>(); List<DataSourcing> groupAddresses = new ArrayList<>();
groupAddresses.add(new DataSourcing(sourcingType, masterAddress)); groupAddresses.add(new DataSourcing(sourcingType, masterAddress));
groupDbAddresses.add(groupAddresses); groupDbAddresses.add(groupAddresses);
} }
if (standbyAddress != null) { if (standbyAddress != null) {
List<DataSourcing> groupAddresses = new ArrayList<>(); List<DataSourcing> groupAddresses = new ArrayList<>();
groupAddresses.add(new DataSourcing(sourcingType, standbyAddress)); groupAddresses.add(new DataSourcing(sourcingType, standbyAddress));
groupDbAddresses.add(groupAddresses); groupDbAddresses.add(groupAddresses);
} }
} }
} }

7
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -74,8 +74,7 @@ public class MysqlConnection implements ErosaConnection {
connector.setConnTimeout(connTimeout); connector.setConnTimeout(connTimeout);
} }
public MysqlConnection(InetSocketAddress address, String username, String password, public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema){
String defaultSchema){
authInfo = new AuthenticationInfo(); authInfo = new AuthenticationInfo();
authInfo.setAddress(address); authInfo.setAddress(address);
authInfo.setUsername(username); authInfo.setUsername(username);
@@ -87,8 +86,8 @@ public class MysqlConnection implements ErosaConnection {
connector.setConnTimeout(connTimeout); connector.setConnTimeout(connTimeout);
} }
public MysqlConnection(InetSocketAddress address, String username, String password, public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema,
String defaultSchema, SslInfo sslInfo){ SslInfo sslInfo){
authInfo = new AuthenticationInfo(); authInfo = new AuthenticationInfo();
authInfo.setAddress(address); authInfo.setAddress(address);
authInfo.setUsername(username); authInfo.setUsername(username);

20
parse/src/main/java/com/alibaba/otter/canal/parse/support/AuthenticationInfo.java
View File

@@ -2,12 +2,12 @@ package com.alibaba.otter.canal.parse.support;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import com.alibaba.otter.canal.common.utils.CommonUtils;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
import org.apache.commons.lang.builder.ToStringBuilder; import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle; import org.apache.commons.lang.builder.ToStringStyle;
import com.alibaba.otter.canal.common.utils.CommonUtils;
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
/** /**
* 数据库认证信息 * 数据库认证信息
* *
@@ -16,17 +16,15 @@ import org.apache.commons.lang.builder.ToStringStyle;
*/ */
public class AuthenticationInfo { public class AuthenticationInfo {
private InetSocketAddress address; // 主库信息 private InetSocketAddress address; // 主库信息
private String username; // 帐号 private String username; // 帐号
private String password; // 密码 private String password; // 密码
private String defaultDatabaseName;// 默认链接的数据库 private String defaultDatabaseName;// 默认链接的数据库
private String pwdPublicKey; //公钥 private String pwdPublicKey; // 公钥
private boolean enableDruid; //是否使用druid加密解密数据库密码 private boolean enableDruid; // 是否使用druid加密解密数据库密码
private SslInfo sslInfo; private SslInfo sslInfo;
public void initPwd() throws Exception{ public void initPwd() throws Exception {
if (enableDruid) { if (enableDruid) {
this.password = CommonUtils.decryptDruidPassword(pwdPublicKey, password); this.password = CommonUtils.decryptDruidPassword(pwdPublicKey, password);
} }
@@ -47,8 +45,8 @@ public class AuthenticationInfo {
this.defaultDatabaseName = defaultDatabaseName; this.defaultDatabaseName = defaultDatabaseName;
} }
public AuthenticationInfo(InetSocketAddress address, String username, String password, public AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName,
String defaultDatabaseName, SslInfo sslInfo) { SslInfo sslInfo){
this(address, username, password, defaultDatabaseName); this(address, username, password, defaultDatabaseName);
this.sslInfo = sslInfo; this.sslInfo = sslInfo;
} }