Skip to content

Commit 2492f31

Browse files
andriikorotkovschlattk
andriikorotkov
authored and
schlattk
committed
🎉 Source DB2: added ssl support (airbytehq#7355)
* added ssl conection to source-db2 and added new acceptance tests * updated ssl source-db2 version * updated ssl source-db2 version * updated ssl source-db2 version * updated ssl source-db2 version * updated spec * fixed code style * marked encryption as required and add NPE checker * fixed remarks * fixed remarks * bump new version
1 parent 0b01a28 commit 2492f31

File tree

10 files changed

+362
-14
lines changed

10 files changed

+362
-14
lines changed

‎airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/447e0381-3780-4b46-bb62-00a4e3c8b8e2.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,6 @@
22
"sourceDefinitionId": "447e0381-3780-4b46-bb62-00a4e3c8b8e2",
33
"name": "IBM Db2",
44
"dockerRepository": "airbyte/source-db2",
5-
"dockerImageTag": "0.1.1",
5+
"dockerImageTag": "0.1.2",
66
"documentationUrl": "https://docs.airbyte.io/integrations/sources/db2"
77
}

‎airbyte-config/init/src/main/resources/seed/source_definitions.yaml

+1-1
Original file line numberDiff line numberDiff line change
@@ -249,7 +249,7 @@
249249
- name: IBM Db2
250250
sourceDefinitionId: 447e0381-3780-4b46-bb62-00a4e3c8b8e2
251251
dockerRepository: airbyte/source-db2
252-
dockerImageTag: 0.1.1
252+
dockerImageTag: 0.1.2
253253
documentationUrl: https://docs.airbyte.io/integrations/sources/db2
254254
sourceType: database
255255
- name: Instagram

‎airbyte-integrations/connectors/source-db2/Dockerfile

+1-1
Original file line numberDiff line numberDiff line change
@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
88

99
RUN tar xf ${APPLICATION}.tar --strip-components=1
1010

11-
LABEL io.airbyte.version=0.1.1
11+
LABEL io.airbyte.version=0.1.2
1212
LABEL io.airbyte.name=airbyte/source-db2

‎airbyte-integrations/connectors/source-db2/src/main/java/io.airbyte.integrations.source.db2/Db2Source.java

+80-5
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,13 @@
1010
import io.airbyte.integrations.base.IntegrationRunner;
1111
import io.airbyte.integrations.base.Source;
1212
import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
13+
import java.io.IOException;
14+
import java.io.PrintWriter;
15+
import java.util.ArrayList;
16+
import java.util.List;
1317
import java.util.Set;
18+
import java.util.concurrent.TimeUnit;
19+
import org.apache.commons.lang3.RandomStringUtils;
1420
import org.slf4j.Logger;
1521
import org.slf4j.LoggerFactory;
1622

@@ -19,6 +25,9 @@ public class Db2Source extends AbstractJdbcSource implements Source {
1925
private static final Logger LOGGER = LoggerFactory.getLogger(Db2Source.class);
2026
public static final String DRIVER_CLASS = "com.ibm.db2.jcc.DB2Driver";
2127

28+
private static final String KEY_STORE_PASS = RandomStringUtils.randomAlphanumeric(8);
29+
private static final String KEY_STORE_FILE_PATH = "clientkeystore.jks";
30+
2231
public Db2Source() {
2332
super(DRIVER_CLASS, new Db2JdbcStreamingQueryConfiguration());
2433
}
@@ -32,14 +41,31 @@ public static void main(final String[] args) throws Exception {
3241

3342
@Override
3443
public JsonNode toDatabaseConfig(final JsonNode config) {
35-
return Jsons.jsonNode(ImmutableMap.builder()
36-
.put("jdbc_url", String.format("jdbc:db2://%s:%s/%s",
37-
config.get("host").asText(),
38-
config.get("port").asText(),
39-
config.get("db").asText()))
44+
final StringBuilder jdbcUrl = new StringBuilder(String.format("jdbc:db2://%s:%s/%s",
45+
config.get("host").asText(),
46+
config.get("port").asText(),
47+
config.get("db").asText()));
48+
49+
var result = Jsons.jsonNode(ImmutableMap.builder()
50+
.put("jdbc_url", jdbcUrl.toString())
4051
.put("username", config.get("username").asText())
4152
.put("password", config.get("password").asText())
4253
.build());
54+
55+
// assume ssl if not explicitly mentioned.
56+
var additionalParams = obtainConnectionOptions(config.get("encryption"));
57+
if (!additionalParams.isEmpty()) {
58+
jdbcUrl.append(":").append(String.join(";", additionalParams));
59+
jdbcUrl.append(";");
60+
result = Jsons.jsonNode(ImmutableMap.builder()
61+
.put("jdbc_url", jdbcUrl.toString())
62+
.put("username", config.get("username").asText())
63+
.put("password", config.get("password").asText())
64+
.put("connection_properties", additionalParams)
65+
.build());
66+
}
67+
68+
return result;
4369
}
4470

4571
@Override
@@ -49,4 +75,53 @@ public Set<String> getExcludedInternalNameSpaces() {
4975
"SYSPROC", "SYSPUBLIC", "SYSSTAT", "SYSTOOLS");
5076
}
5177

78+
/* Helpers */
79+
80+
private List<String> obtainConnectionOptions(JsonNode encryption) {
81+
List<String> additionalParameters = new ArrayList<>();
82+
if (!encryption.isNull()) {
83+
String encryptionMethod = encryption.get("encryption_method").asText();
84+
if ("encrypted_verify_certificate".equals(encryptionMethod)) {
85+
var keyStorePassword = getKeyStorePassword(encryption.get("key_store_password"));
86+
try {
87+
convertAndImportCertificate(encryption.get("ssl_certificate").asText(), keyStorePassword);
88+
} catch (IOException | InterruptedException e) {
89+
throw new RuntimeException("Failed to import certificate into Java Keystore");
90+
}
91+
additionalParameters.add("sslConnection=true");
92+
additionalParameters.add("sslTrustStoreLocation=" + KEY_STORE_FILE_PATH);
93+
additionalParameters.add("sslTrustStorePassword=" + keyStorePassword);
94+
}
95+
}
96+
return additionalParameters;
97+
}
98+
99+
private static String getKeyStorePassword(JsonNode encryptionKeyStorePassword) {
100+
var keyStorePassword = KEY_STORE_PASS;
101+
if (!encryptionKeyStorePassword.isNull() || !encryptionKeyStorePassword.isEmpty()) {
102+
keyStorePassword = encryptionKeyStorePassword.asText();
103+
}
104+
return keyStorePassword;
105+
}
106+
107+
private static void convertAndImportCertificate(String certificate, String keyStorePassword)
108+
throws IOException, InterruptedException {
109+
Runtime run = Runtime.getRuntime();
110+
try (PrintWriter out = new PrintWriter("certificate.pem")) {
111+
out.print(certificate);
112+
}
113+
runProcess("openssl x509 -outform der -in certificate.pem -out certificate.der", run);
114+
runProcess(
115+
"keytool -import -alias rds-root -keystore " + KEY_STORE_FILE_PATH + " -file certificate.der -storepass " + keyStorePassword + " -noprompt",
116+
run);
117+
}
118+
119+
private static void runProcess(String cmd, Runtime run) throws IOException, InterruptedException {
120+
Process pr = run.exec(cmd);
121+
if (!pr.waitFor(30, TimeUnit.SECONDS)) {
122+
pr.destroy();
123+
throw new RuntimeException("Timeout while executing: " + cmd);
124+
}
125+
}
126+
52127
}

‎airbyte-integrations/connectors/source-db2/src/main/resources/spec.json

+60-6
Original file line numberDiff line numberDiff line change
@@ -4,34 +4,88 @@
44
"$schema": "http://json-schema.org/draft-07/schema#",
55
"title": "IBM Db2 Source Spec",
66
"type": "object",
7-
"required": ["host", "port", "db", "username", "password"],
7+
"required": ["host", "port", "db", "username", "password", "encryption"],
88
"additionalProperties": false,
99
"properties": {
1010
"host": {
1111
"description": "Host of the Db2.",
12-
"type": "string"
12+
"type": "string",
13+
"order": 0
1314
},
1415
"port": {
1516
"description": "Port of the database.",
1617
"type": "integer",
1718
"minimum": 0,
1819
"maximum": 65536,
1920
"default": 8123,
20-
"examples": ["8123"]
21+
"examples": ["8123"],
22+
"order": 1
2123
},
2224
"db": {
2325
"description": "Name of the database.",
2426
"type": "string",
25-
"examples": ["default"]
27+
"examples": ["default"],
28+
"order": 2
2629
},
2730
"username": {
2831
"description": "Username to use to access the database.",
29-
"type": "string"
32+
"type": "string",
33+
"order": 3
3034
},
3135
"password": {
3236
"description": "Password associated with the username.",
3337
"type": "string",
34-
"airbyte_secret": true
38+
"airbyte_secret": true,
39+
"order": 4
40+
},
41+
"encryption": {
42+
"title": "Encryption",
43+
"type": "object",
44+
"description": "Encryption method to use when communicating with the database",
45+
"order": 5,
46+
"oneOf": [
47+
{
48+
"title": "Unencrypted",
49+
"additionalProperties": false,
50+
"description": "Data transfer will not be encrypted.",
51+
"required": ["encryption_method"],
52+
"properties": {
53+
"encryption_method": {
54+
"type": "string",
55+
"const": "unencrypted",
56+
"enum": ["unencrypted"],
57+
"default": "unencrypted"
58+
}
59+
}
60+
},
61+
{
62+
"title": "TLS Encrypted (verify certificate)",
63+
"additionalProperties": false,
64+
"description": "Verify and use the cert provided by the server.",
65+
"required": ["encryption_method", "ssl_certificate"],
66+
"properties": {
67+
"encryption_method": {
68+
"type": "string",
69+
"const": "encrypted_verify_certificate",
70+
"enum": ["encrypted_verify_certificate"],
71+
"default": "encrypted_verify_certificate"
72+
},
73+
"ssl_certificate": {
74+
"title": "SSL PEM file",
75+
"description": "Privacy Enhanced Mail (PEM) files are concatenated certificate containers frequently used in certificate installations",
76+
"type": "string",
77+
"airbyte_secret": true,
78+
"multiline": true
79+
},
80+
"key_store_password": {
81+
"title": "Key Store Password. This field is optional. If you do not fill in this field, the password will be randomly generated.",
82+
"description": "Key Store Password",
83+
"type": "string",
84+
"airbyte_secret": true
85+
}
86+
}
87+
}
88+
]
3589
}
3690
}
3791
}

‎airbyte-integrations/connectors/source-db2/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/Db2SourceAcceptanceTest.java

+3
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,9 @@ protected void setupEnvironment(final TestDestinationEnv environment) throws Exc
9797
.put("db", db.getDatabaseName())
9898
.put("username", db.getUsername())
9999
.put("password", db.getPassword())
100+
.put("encryption", Jsons.jsonNode(ImmutableMap.builder()
101+
.put("encryption_method", "unencrypted")
102+
.build()))
100103
.build());
101104

102105
database = Databases.createJdbcDatabase(

0 commit comments

Comments
 (0)