Add AllowedHosts to Actor Definitions and Config Database (#21363)

* Add AllowedHosts to actor_definitions and database

* use objects for better null-ness handling

* Tables.ACTOR_DEFINITIO

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderTest.java

@@ -80,6 +80,10 @@ class BootloaderTest {
   private static final String VERSION_0321_ALPHA = "0.32.1-alpha";
   private static final String VERSION_0170_ALPHA = "0.17.0-alpha";
 
+  // ⚠️ This line should change with every new migration to show that you meant to make a new
+  // migration to the prod database
+  private static final String CURRENT_MIGRATION_VERSION = "0.40.27.001";
+
   @BeforeEach
   void setup() {
     container = new PostgreSQLContainer<>("postgres:13-alpine")
@@ -146,9 +150,7 @@ void testBootloaderAppBlankDb() throws Exception {
       assertEquals("0.40.26.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
 
       val configsMigrator = new ConfigsDatabaseMigrator(configDatabase, configsFlyway);
-      // this line should change with every new migration
-      // to show that you meant to make a new migration to the prod database
-      assertEquals("0.40.23.002", configsMigrator.getLatestMigration().getVersion().getVersion());
+      assertEquals(CURRENT_MIGRATION_VERSION, configsMigrator.getLatestMigration().getVersion().getVersion());
 
       assertEquals(VERSION_0330_ALPHA, jobsPersistence.getVersion().get());
       assertEquals(new Version(PROTOCOL_VERSION_123), jobsPersistence.getAirbyteProtocolVersionMin().get());

airbyte-config/config-models/src/main/resources/types/AllowedHosts.yaml

@@ -0,0 +1,14 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/AllowedHosts.yaml
+title: AllowedHosts
+description: A connector's allowed hosts.  If present, the platform will limit communication to only hosts which are listed in `AllowedHosts.hosts`.
+type: object
+required:
+additionalProperties: true
+properties:
+  hosts:
+    type: array
+    description: An array of hosts that this connector can connect to.  AllowedHosts not being present for the source or destination means that access to all hosts is allowed.  An empty list here means that no network access is granted.
+    items:
+      type: string

airbyte-config/config-models/src/main/resources/types/StandardDestinationDefinition.yaml

@@ -63,3 +63,5 @@ properties:
   supportsDbt:
     type: boolean
     description: an optional flag indicating whether DBT is used in the normalization. If the flag value is NULL - DBT is not used.
+  allowedHosts:
+    "$ref": AllowedHosts.yaml

airbyte-config/config-models/src/main/resources/types/StandardSourceDefinition.yaml

@@ -65,3 +65,5 @@ properties:
   protocolVersion:
     type: string
     description: the Airbyte Protocol version supported by the connector
+  allowedHosts:
+    "$ref": AllowedHosts.yaml

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/ConfigWriter.java

@@ -105,6 +105,8 @@ static void writeStandardSourceDefinition(final List<StandardSourceDefinition> c
             .set(Tables.ACTOR_DEFINITION.RESOURCE_REQUIREMENTS,
                 standardSourceDefinition.getResourceRequirements() == null ? null
                     : JSONB.valueOf(Jsons.serialize(standardSourceDefinition.getResourceRequirements())))
+            .set(Tables.ACTOR_DEFINITION.ALLOWED_HOSTS, standardSourceDefinition.getAllowedHosts() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardSourceDefinition.getAllowedHosts())))
             .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .where(Tables.ACTOR_DEFINITION.ID.eq(standardSourceDefinition.getSourceDefinitionId()))
             .execute();
@@ -136,6 +138,8 @@ static void writeStandardSourceDefinition(final List<StandardSourceDefinition> c
             .set(Tables.ACTOR_DEFINITION.RESOURCE_REQUIREMENTS,
                 standardSourceDefinition.getResourceRequirements() == null ? null
                     : JSONB.valueOf(Jsons.serialize(standardSourceDefinition.getResourceRequirements())))
+            .set(ACTOR_DEFINITION.ALLOWED_HOSTS, standardSourceDefinition.getAllowedHosts() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardSourceDefinition.getAllowedHosts())))
             .set(Tables.ACTOR_DEFINITION.CREATED_AT, timestamp)
             .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .execute();
@@ -172,7 +176,6 @@ static void writeStandardDestinationDefinition(final List<StandardDestinationDef
             .set(Tables.ACTOR_DEFINITION.RESOURCE_REQUIREMENTS,
                 standardDestinationDefinition.getResourceRequirements() == null ? null
                     : JSONB.valueOf(Jsons.serialize(standardDestinationDefinition.getResourceRequirements())))
-            .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .set(Tables.ACTOR_DEFINITION.NORMALIZATION_REPOSITORY,
                 Objects.nonNull(standardDestinationDefinition.getNormalizationConfig())
                     ? standardDestinationDefinition.getNormalizationConfig().getNormalizationRepository()
@@ -186,6 +189,9 @@ static void writeStandardDestinationDefinition(final List<StandardDestinationDef
                 Objects.nonNull(standardDestinationDefinition.getNormalizationConfig())
                     ? standardDestinationDefinition.getNormalizationConfig().getNormalizationIntegrationType()
                     : null)
+            .set(ACTOR_DEFINITION.ALLOWED_HOSTS, standardDestinationDefinition.getAllowedHosts() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardDestinationDefinition.getAllowedHosts())))
+            .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .where(Tables.ACTOR_DEFINITION.ID.eq(standardDestinationDefinition.getDestinationDefinitionId()))
             .execute();
 
@@ -213,8 +219,6 @@ static void writeStandardDestinationDefinition(final List<StandardDestinationDef
             .set(Tables.ACTOR_DEFINITION.RESOURCE_REQUIREMENTS,
                 standardDestinationDefinition.getResourceRequirements() == null ? null
                     : JSONB.valueOf(Jsons.serialize(standardDestinationDefinition.getResourceRequirements())))
-            .set(Tables.ACTOR_DEFINITION.CREATED_AT, timestamp)
-            .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .set(Tables.ACTOR_DEFINITION.NORMALIZATION_REPOSITORY,
                 Objects.nonNull(standardDestinationDefinition.getNormalizationConfig())
                     ? standardDestinationDefinition.getNormalizationConfig().getNormalizationRepository()
@@ -228,6 +232,10 @@ static void writeStandardDestinationDefinition(final List<StandardDestinationDef
                 Objects.nonNull(standardDestinationDefinition.getNormalizationConfig())
                     ? standardDestinationDefinition.getNormalizationConfig().getNormalizationIntegrationType()
                     : null)
+            .set(ACTOR_DEFINITION.ALLOWED_HOSTS, standardDestinationDefinition.getAllowedHosts() == null ? null
+                : JSONB.valueOf(Jsons.serialize(standardDestinationDefinition.getAllowedHosts())))
+            .set(Tables.ACTOR_DEFINITION.CREATED_AT, timestamp)
+            .set(Tables.ACTOR_DEFINITION.UPDATED_AT, timestamp)
             .execute();
       }
     });

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

@@ -509,6 +509,8 @@
   icon: faker.svg
   sourceType: api
   releaseStage: alpha
+  allowedHosts:
+    hosts: []
 - name: Fastbill
   sourceDefinitionId: eb3e9c1c-0467-4eb7-a172-5265e04ccd0a
   dockerRepository: airbyte/source-fastbill

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/configs/migrations/V0_40_27_001__AddAllowedHosts.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.db.instance.configs.migrations;
+
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+import org.jooq.DSLContext;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class V0_40_27_001__AddAllowedHosts extends BaseJavaMigration {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(V0_40_27_001__AddAllowedHosts.class);
+
+  @Override
+  public void migrate(final Context context) throws Exception {
+    LOGGER.info("Running migration: {}", this.getClass().getSimpleName());
+
+    // Warning: please do not use any jOOQ generated code to write a migration.
+    // As database schema changes, the generated jOOQ code can be deprecated. So
+    // old migration may not compile if there is any generated code.
+    try (final DSLContext ctx = DSL.using(context.getConnection())) {
+      addAllowedHostsToActorDefinition(ctx);
+    }
+  }
+
+  private static void addAllowedHostsToActorDefinition(final DSLContext ctx) {
+    ctx.alterTable("actor_definition")
+        .addColumnIfNotExists(DSL.field(
+            "allowed_hosts",
+            SQLDataType.JSONB.nullable(true)))
+        .execute();
+  }
+
+}

airbyte-db/db-lib/src/main/resources/configs_database/schema_dump.txt

@@ -58,6 +58,7 @@ create table "public"."actor_definition"(
   "normalization_tag" varchar(255) null,
   "supports_dbt" bool null,
   "normalization_integration_type" varchar(255) null,
+  "allowed_hosts" jsonb null,
   constraint "actor_definition_pkey"
     primary key ("id")
 );