From fe78c0a5683c5f10bae90f524d6e381bf4222784 Mon Sep 17 00:00:00 2001 From: oneshcheret Date: Tue, 28 Sep 2021 14:13:17 +0300 Subject: [PATCH 1/3] =?UTF-8?q?=F0=9F=8E=89=20Destination=20MSSQL:=20Added?= =?UTF-8?q?=20support=20for=20connection=20via=20SSH=20tunnels=20(#5970)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../connectors/destination-mssql/build.gradle | 1 + .../destination/mssql/MSSQLDestination.java | 5 +- .../SshKeyMSSQLDestinationAcceptanceTest.java | 15 ++ .../SshMSSQLDestinationAcceptanceTest.java | 178 ++++++++++++++++++ ...asswordMSSQLDestinationAcceptanceTest.java | 16 ++ 5 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshKeyMSSQLDestinationAcceptanceTest.java create mode 100644 airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshMSSQLDestinationAcceptanceTest.java create mode 100644 airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshPasswordMSSQLDestinationAcceptanceTest.java diff --git a/airbyte-integrations/connectors/destination-mssql/build.gradle b/airbyte-integrations/connectors/destination-mssql/build.gradle index 34c8a691f0cf..f558e2b8d8b8 100644 --- a/airbyte-integrations/connectors/destination-mssql/build.gradle +++ b/airbyte-integrations/connectors/destination-mssql/build.gradle @@ -21,6 +21,7 @@ dependencies { testImplementation "org.testcontainers:mssqlserver:1.15.3" integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test') + integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-mssql') implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs) integrationTestJavaImplementation files(project(':airbyte-integrations:bases:base-normalization').airbyteDocker.outputs) diff --git a/airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java b/airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java index f352b71880c2..6f05f1357e9c 100644 --- a/airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java +++ b/airbyte-integrations/connectors/destination-mssql/src/main/java/io/airbyte/integrations/destination/mssql/MSSQLDestination.java @@ -9,6 +9,7 @@ import io.airbyte.commons.json.Jsons; import io.airbyte.integrations.base.Destination; import io.airbyte.integrations.base.IntegrationRunner; +import io.airbyte.integrations.base.ssh.SshWrappedDestination; import io.airbyte.integrations.destination.jdbc.AbstractJdbcDestination; import java.io.File; import java.util.ArrayList; @@ -22,6 +23,8 @@ public class MSSQLDestination extends AbstractJdbcDestination implements Destina private static final Logger LOGGER = LoggerFactory.getLogger(MSSQLDestination.class); public static final String DRIVER_CLASS = "com.microsoft.sqlserver.jdbc.SQLServerDriver"; + public static final List HOST_KEY = List.of("host"); + public static final List PORT_KEY = List.of("port"); public MSSQLDestination() { super(DRIVER_CLASS, new MSSQLNameTransformer(), new SqlServerOperations()); @@ -89,7 +92,7 @@ private void readSsl(JsonNode config, List additionalParameters) { } public static void main(String[] args) throws Exception { - final Destination destination = new MSSQLDestination(); + final Destination destination = new SshWrappedDestination(new MSSQLDestination(), HOST_KEY, PORT_KEY); LOGGER.info("starting destination: {}", MSSQLDestination.class); new IntegrationRunner(destination).run(args); LOGGER.info("completed destination: {}", MSSQLDestination.class); diff --git a/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshKeyMSSQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshKeyMSSQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..75b329949267 --- /dev/null +++ b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshKeyMSSQLDestinationAcceptanceTest.java @@ -0,0 +1,15 @@ +/* + * Copyright (c) 2021 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.mssql; + +import io.airbyte.integrations.base.ssh.SshTunnel; + +public class SshKeyMSSQLDestinationAcceptanceTest extends SshMSSQLDestinationAcceptanceTest { + + @Override + public SshTunnel.TunnelMethod getTunnelMethod() { + return SshTunnel.TunnelMethod.SSH_KEY_AUTH; + } +} diff --git a/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshMSSQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshMSSQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..1923d238f49e --- /dev/null +++ b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshMSSQLDestinationAcceptanceTest.java @@ -0,0 +1,178 @@ +/* + * Copyright (c) 2021 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.mssql; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.collect.ImmutableMap; +import io.airbyte.commons.functional.CheckedFunction; +import io.airbyte.commons.json.Jsons; +import io.airbyte.db.Database; +import io.airbyte.db.Databases; +import io.airbyte.integrations.base.JavaBaseConstants; +import io.airbyte.integrations.base.ssh.SshBastionContainer; +import io.airbyte.integrations.base.ssh.SshTunnel; +import io.airbyte.integrations.destination.ExtendedNameTransformer; +import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; +import org.apache.commons.lang3.RandomStringUtils; +import org.jooq.JSONFormat; +import org.jooq.JSONFormat.RecordFormat; +import org.testcontainers.containers.JdbcDatabaseContainer; +import org.testcontainers.containers.MSSQLServerContainer; +import org.testcontainers.containers.Network; + +/** + * Abstract class that allows us to avoid duplicating testing logic for testing SSH with a key file or with a password. + */ +public abstract class SshMSSQLDestinationAcceptanceTest extends DestinationAcceptanceTest { + + private static final JSONFormat JSON_FORMAT = new JSONFormat().recordFormat(RecordFormat.OBJECT); + + private final ExtendedNameTransformer namingResolver = new ExtendedNameTransformer(); + + private final String schemaName = RandomStringUtils.randomAlphabetic(8).toLowerCase(); + private static final String database = "test"; + private static MSSQLServerContainer db; + private final SshBastionContainer bastion = new SshBastionContainer(); + + public abstract SshTunnel.TunnelMethod getTunnelMethod(); + + @Override + protected String getImageName() { + return "airbyte/destination-mssql:dev"; + } + + @Override + protected JsonNode getConfig() throws Exception { + return bastion.getTunnelConfig(getTunnelMethod(), getMSSQLDbConfigBuilder(db)); + } + + @Override + protected JsonNode getFailCheckConfig() throws Exception { + final JsonNode clone = Jsons.clone(getConfig()); + ((ObjectNode) clone).put("password", "wrong password"); + return clone; + } + + @Override + protected List retrieveRecords(final TestDestinationEnv env, + final String streamName, + final String namespace, + final JsonNode streamSchema) + throws Exception { + return retrieveRecordsFromTable(namingResolver.getRawTableName(streamName), namespace) + .stream() + .map(r -> Jsons.deserialize(r.get(JavaBaseConstants.COLUMN_NAME_DATA).asText())) + .collect(Collectors.toList()); + } + + @Override + protected boolean supportsDBT() { + return true; + } + + @Override + protected boolean implementsNamespaces() { + return true; + } + + @Override + protected List resolveIdentifier(final String identifier) { + final List result = new ArrayList<>(); + final String resolved = namingResolver.getIdentifier(identifier); + result.add(identifier); + result.add(resolved); + if (!resolved.startsWith("\"")) { + result.add(resolved.toLowerCase()); + result.add(resolved.toUpperCase()); + } + return result; + } + + public ImmutableMap.Builder getMSSQLDbConfigBuilder(JdbcDatabaseContainer db) { + return ImmutableMap.builder() + .put("host", Objects.requireNonNull(db.getContainerInfo().getNetworkSettings() + .getNetworks() + .get(((Network.NetworkImpl) bastion.getNetWork()).getName()) + .getIpAddress())) + .put("username", db.getUsername()) + .put("password", db.getPassword()) + .put("port", db.getExposedPorts().get(0)) + .put("database", database) + .put("schema", schemaName) + .put("ssl", false); + } + + private static Database getDatabaseFromConfig(final JsonNode config) { + return Databases.createDatabase( + config.get("username").asText(), + config.get("password").asText(), + String.format("jdbc:sqlserver://%s:%s", + config.get("host").asText(), + config.get("port").asInt()), + "com.microsoft.sqlserver.jdbc.SQLServerDriver", + null); + } + + private List retrieveRecordsFromTable(final String tableName, final String schemaName) throws Exception { + var schema = schemaName == null ? this.schemaName : schemaName; + final JsonNode config = getConfig(); + return SshTunnel.sshWrap( + config, + MSSQLDestination.HOST_KEY, + MSSQLDestination.PORT_KEY, + (CheckedFunction, Exception>) mangledConfig -> getDatabaseFromConfig(mangledConfig) + .query( + ctx -> ctx + .fetch(String.format("USE %s;" + + "SELECT * FROM %s.%s ORDER BY %s ASC;", + database, schema, tableName.toLowerCase(), JavaBaseConstants.COLUMN_NAME_EMITTED_AT)) + .stream() + .map(r -> r.formatJSON(JSON_FORMAT)) + .map(Jsons::deserialize) + .collect(Collectors.toList()))); + } + + @Override + protected void setup(final TestDestinationEnv testEnv) throws Exception { + startTestContainers(); + + SshTunnel.sshWrap( + getConfig(), + MSSQLDestination.HOST_KEY, + MSSQLDestination.PORT_KEY, + mangledConfig -> { + getDatabaseFromConfig(mangledConfig).query(ctx -> { + ctx.fetch(String.format("CREATE DATABASE %s;", database)); + ctx.fetch(String.format("USE %s;", database)); + ctx.fetch(String.format("CREATE SCHEMA %s;", schemaName)); + + return null; + }); + }); + } + + private void startTestContainers() { + bastion.initAndStartBastion(); + initAndStartJdbcContainer(); + } + + private void initAndStartJdbcContainer() { + db = new MSSQLServerContainer<>("mcr.microsoft.com/mssql/server:2017-latest") + .withNetwork(bastion.getNetWork()) + .acceptLicense(); + db.start(); + } + + @Override + protected void tearDown(final TestDestinationEnv testEnv) { + bastion.stopAndCloseContainers(db); + } + +} diff --git a/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshPasswordMSSQLDestinationAcceptanceTest.java b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshPasswordMSSQLDestinationAcceptanceTest.java new file mode 100644 index 000000000000..eb6332f7287d --- /dev/null +++ b/airbyte-integrations/connectors/destination-mssql/src/test-integration/java/io/airbyte/integrations/destination/mssql/SshPasswordMSSQLDestinationAcceptanceTest.java @@ -0,0 +1,16 @@ +/* + * Copyright (c) 2021 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.mssql; + +import io.airbyte.integrations.base.ssh.SshTunnel; + +public class SshPasswordMSSQLDestinationAcceptanceTest extends SshMSSQLDestinationAcceptanceTest { + + @Override + public SshTunnel.TunnelMethod getTunnelMethod() { + return SshTunnel.TunnelMethod.SSH_PASSWORD_AUTH; + } + +} From 534d254f62a3bfebf06d887c2a319b303ee7993d Mon Sep 17 00:00:00 2001 From: oneshcheret Date: Tue, 28 Sep 2021 15:25:35 +0300 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=8E=89=20Destination=20MSSQL:=20bumb?= =?UTF-8?q?=20image=20version?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../d4353156-9217-4cad-8dd7-c108fd4f74cf.json | 2 +- .../init/src/main/resources/seed/destination_definitions.yaml | 2 +- airbyte-integrations/connectors/destination-mssql/Dockerfile | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json index 9509e4f738c4..3374577b9f31 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/d4353156-9217-4cad-8dd7-c108fd4f74cf.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "d4353156-9217-4cad-8dd7-c108fd4f74cf", "name": "MS SQL Server", "dockerRepository": "airbyte/destination-mssql", - "dockerImageTag": "0.1.8", + "dockerImageTag": "0.1.9", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/mssql" } diff --git a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml index 88a6b662c225..809edf9c93c7 100644 --- a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml +++ b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml @@ -68,7 +68,7 @@ - destinationDefinitionId: d4353156-9217-4cad-8dd7-c108fd4f74cf name: MS SQL Server dockerRepository: airbyte/destination-mssql - dockerImageTag: 0.1.8 + dockerImageTag: 0.1.9 documentationUrl: https://docs.airbyte.io/integrations/destinations/mssql - destinationDefinitionId: 3986776d-2319-4de9-8af8-db14c0996e72 name: Oracle (Alpha) diff --git a/airbyte-integrations/connectors/destination-mssql/Dockerfile b/airbyte-integrations/connectors/destination-mssql/Dockerfile index 2d65b3dfb22c..611f12576d5f 100644 --- a/airbyte-integrations/connectors/destination-mssql/Dockerfile +++ b/airbyte-integrations/connectors/destination-mssql/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.1.8 +LABEL io.airbyte.version=0.1.9 LABEL io.airbyte.name=airbyte/destination-mssql From da34fb94f630d5726a456b19d13e86ca4ba9e0fa Mon Sep 17 00:00:00 2001 From: oneshcheret Date: Wed, 29 Sep 2021 09:45:25 +0300 Subject: [PATCH 3/3] Destination MSSQL: update mssql.md file --- docs/integrations/destinations/mssql.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/docs/integrations/destinations/mssql.md b/docs/integrations/destinations/mssql.md index cb299c2c4764..8b22391d917c 100644 --- a/docs/integrations/destinations/mssql.md +++ b/docs/integrations/destinations/mssql.md @@ -75,11 +75,34 @@ You should now have all the requirements needed to configure SQL Server as a des * **Encrypted (verify certificate)**: Use the server's SSL certificate, after standard certificate verification. * **Host Name In Certificate** (optional): When using certificate verification, this property can be set to specify an expected name for added security. If this value is present, and the server's certificate's host name does not match it, certificate verification will fail. +### Connection to MS SQL Server via an SSH Tunnel + +Airbyte has the ability to connect to the MS SQL Server instance via an SSH Tunnel. The reason you might want to do this because it is not possible +(or against security policy) to connect to the database directly (e.g. it does not have a public IP address). + +When using an SSH tunnel, you are configuring Airbyte to connect to an intermediate server (a.k.a. a bastion sever) that have direct access to the database. +Airbyte connects to the bastion and then asks the bastion to connect directly to the server. + +Using this feature requires additional configuration, when creating the source. We will talk through what each piece of configuration means. +1. Configure all fields for the source as you normally would, except `SSH Tunnel Method`. +2. `SSH Tunnel Method` defaults to `No Tunnel` (meaning a direct connection). If you want to use an SSH Tunnel choose `SSH Key Authentication` or `Password Authentication`. +3. Choose `Key Authentication` if you will be using an RSA private key as your secret for establishing the SSH Tunnel (see below for more information on generating this key). +4. Choose `Password Authentication` if you will be using a password as your secret for establishing the SSH Tunnel. +5. `SSH Tunnel Jump Server Host` refers to the intermediate (bastion) server that Airbyte will connect to. This should be a hostname or an IP Address. +6. `SSH Connection Port` is the port on the bastion server with which to make the SSH connection. The default port for SSH connections is `22`, +so unless you have explicitly changed something, go with the default. +7. `SSH Login Username` is the username that Airbyte should use when connection to the bastion server. This is NOT the MS SQL Server username. +8. If you are using `Password Authentication`, then `SSH Login Username` should be set to the password of the User from the previous step. +If you are using `SSH Key Authentication` leave this blank. Again, this is not the MS SQL Server password, but the password for the OS-user that +Airbyte is using to perform commands on the bastion. +9. If you are using `SSH Key Authentication`, then `SSH Private Key` should be set to the RSA Private Key that you are using to create the SSH connection. +This should be the full contents of the key file starting with `-----BEGIN RSA PRIVATE KEY-----` and ending with `-----END RSA PRIVATE KEY-----`. ## Changelog | Version | Date | Pull Request | Subject | | :------ | :-------- | :----- | :------ | +| 0.1.9 | 2021-09-29 | [#5970](https://github.com/airbytehq/airbyte/pull/5970) | Add support & test cases for MSSQL Destination via SSH tunnels | | 0.1.8 | 2021-08-07 | [#5272](https://github.com/airbytehq/airbyte/pull/5272) | Add batch method to insert records | | 0.1.7 | 2021-07-30 | [#5125](https://github.com/airbytehq/airbyte/pull/5125) | Enable `additionalPropertities` in spec.json | | 0.1.6 | 2021-06-21 | [#3555](https://github.com/airbytehq/airbyte/pull/3555) | Partial Success in BufferedStreamConsumer |