🐙 octavia-cli: implement secrement management

Author: alafanechere

octavia-cli/README.md

@@ -21,13 +21,22 @@ SUB_BUILD=OCTAVIA_CLI ./gradlew build #from the root of the repo
 2. Run the CLI from docker:
 ```bash
 docker run airbyte/octavia-cli:dev 
-````
+```
 3. Create an `octavia` alias in your `.bashrc` or `.zshrc`: 
-````bash
+```bash
 echo 'alias octavia="docker run airbyte/octavia-cli:dev"'  >> ~/.zshrc
 source ~/.zshrc
 octavia
-````
+```
+
+# Secret management
+Sources and destinations configurations have credential fields that you **do not want to store as plain text and version on Git**.
+`octavia` offers secret management through environment variables expansion:
+```yaml
+configuration:
+  password: ${MY_PASSWORD}
+```
+If you have set a  `MY_PASSWORD` environment variable, `octavia apply` will load its value into the `password` field. 
 
 # Current development status
 Octavia is currently under development. 
@@ -38,6 +47,7 @@ We welcome community contributions!
 
 | Date       | Milestone                           |
 |------------|-------------------------------------|
+| 2022-03-06 | Implement secret management through environment variable expansion |
 | 2022-03-02 | Implement `octavia apply` (sources and destination only)|
 | 2022-02-06 | Implement `octavia generate` (sources and destination only)|
 | 2022-01-25 | Implement `octavia init` + some context checks|

octavia-cli/integration_tests/test_generate/expected_rendered_yaml/destination_postgres/expected.yaml

@@ -13,7 +13,7 @@ configuration:
   database: # REQUIRED | string | Name of the database.
   schema: "public" # REQUIRED | string | The default schema tables are written to if the source does not specify a namespace. The usual value for this field is "public". | Example: public
   username: # REQUIRED | string | Username to use to access the database.
-  password: # SECRET | OPTIONAL | string | Password associated with the username.
+  password: ${PASSWORD} # SECRET (please store in environment variables) | OPTIONAL | string | Password associated with the username.
   ssl: # OPTIONAL | boolean | Encrypt data using SSL.
   tunnel_method:
     ## -------- Pick one valid structure among the examples below: --------
@@ -23,10 +23,10 @@ configuration:
     # tunnel_host: # REQUIRED | string | Hostname of the jump server host that allows inbound ssh tunnel.
     # tunnel_port: 22 # REQUIRED | integer | Port on the proxy/jump server that accepts inbound ssh connections. | Example: 22
     # tunnel_user: # REQUIRED | string | OS-level username for logging into the jump server host.
-    # ssh_key: # SECRET | REQUIRED | string | OS-level user account ssh key credentials in RSA PEM format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )
+    # ssh_key: ${SSH_KEY} # SECRET (please store in environment variables) | REQUIRED | string | OS-level user account ssh key credentials in RSA PEM format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )
     ## -------- Another valid structure for tunnel_method: --------
     # tunnel_method: "SSH_PASSWORD_AUTH" # REQUIRED | string | Connect through a jump server tunnel host using username and password authentication
     # tunnel_host: # REQUIRED | string | Hostname of the jump server host that allows inbound ssh tunnel.
     # tunnel_port: 22 # REQUIRED | integer | Port on the proxy/jump server that accepts inbound ssh connections. | Example: 22
     # tunnel_user: # REQUIRED | string | OS-level username for logging into the jump server host
-    # tunnel_user_password: # SECRET | REQUIRED | string | OS-level password for logging into the jump server host
+    # tunnel_user_password: ${TUNNEL_USER_PASSWORD} # SECRET (please store in environment variables) | REQUIRED | string | OS-level password for logging into the jump server host

octavia-cli/integration_tests/test_generate/expected_rendered_yaml/destination_s3/expected.yaml

@@ -12,8 +12,8 @@ configuration:
   s3_bucket_name: # REQUIRED | string | The name of the S3 bucket. | Example: airbyte_sync
   s3_bucket_path: # REQUIRED | string | Directory under the S3 bucket where data will be written. | Example: data_sync/test
   s3_bucket_region: # REQUIRED | string | The region of the S3 bucket.
-  access_key_id: # SECRET | OPTIONAL | string | The access key id to access the S3 bucket. Airbyte requires Read and Write permissions to the given bucket, if not set, Airbyte will rely on Instance Profile. | Example: A012345678910EXAMPLE
-  secret_access_key: # SECRET | OPTIONAL | string | The corresponding secret to the access key id, if S3 Key Id is set, then S3 Access Key must also be provided | Example: a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY
+  access_key_id: ${ACCESS_KEY_ID} # SECRET (please store in environment variables) | OPTIONAL | string | The access key id to access the S3 bucket. Airbyte requires Read and Write permissions to the given bucket, if not set, Airbyte will rely on Instance Profile. | Example: A012345678910EXAMPLE
+  secret_access_key: ${SECRET_ACCESS_KEY} # SECRET (please store in environment variables) | OPTIONAL | string | The corresponding secret to the access key id, if S3 Key Id is set, then S3 Access Key must also be provided | Example: a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY
   format:
     ## -------- Pick one valid structure among the examples below: --------
     format_type: "Avro" # REQUIRED | string}
@@ -31,7 +31,7 @@ configuration:
       ## -------- Another valid structure for compression_codec: --------
       # codec: "zstandard" # REQUIRED | string
       # compression_level: 3 # REQUIRED | integer | Negative levels are 'fast' modes akin to lz4 or snappy, levels above 9 are generally for archival purposes, and levels above 18 use a lot of memory.
-      # include_checksum: # OPTIONAL | boolean | If true, include a checksum with each data block.
+      # include_hash: # OPTIONAL | boolean | If true, include a hash with each data block.
       ## -------- Another valid structure for compression_codec: --------
       # codec: "snappy" # REQUIRED | string
     part_size_mb: 5 # OPTIONAL | integer | This is the size of a "Part" being buffered in memory. It limits the memory usage when writing. Larger values will allow to upload a bigger files and improve the speed, but consumes9 more memory. Allowed values: min=5MB, max=525MB Default: 5MB. | Example: 5

octavia-cli/integration_tests/test_generate/expected_rendered_yaml/destination_s3/input_spec.yaml

@@ -180,9 +180,9 @@ spec:
                         default: 3
                         minimum: -5
                         maximum: 22
-                      include_checksum:
-                        title: "Include checksum"
-                        description: "If true, include a checksum with each data block."
+                      include_hash:
+                        title: "Include hash"
+                        description: "If true, include a hash with each data block."
                         type: "boolean"
                         default: false
                   - title: "snappy"

octavia-cli/integration_tests/test_generate/expected_rendered_yaml/source_postgres/expected.yaml

@@ -13,7 +13,7 @@ configuration:
   database: # REQUIRED | string | Name of the database.
   schemas: ["public"] # OPTIONAL | array | The list of schemas to sync from. Defaults to user. Case sensitive.
   username: # REQUIRED | string | Username to use to access the database.
-  password: # SECRET | OPTIONAL | string | Password associated with the username.
+  password: ${PASSWORD} # SECRET (please store in environment variables) | OPTIONAL | string | Password associated with the username.
   ssl: # OPTIONAL | boolean | Encrypt client/server communications for increased security.
   replication_method:
     ## -------- Pick one valid structure among the examples below: --------
@@ -31,10 +31,10 @@ configuration:
     # tunnel_host: # REQUIRED | string | Hostname of the jump server host that allows inbound ssh tunnel.
     # tunnel_port: 22 # REQUIRED | integer | Port on the proxy/jump server that accepts inbound ssh connections. | Example: 22
     # tunnel_user: # REQUIRED | string | OS-level username for logging into the jump server host.
-    # ssh_key: # SECRET | REQUIRED | string | OS-level user account ssh key credentials in RSA PEM format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )
+    # ssh_key: ${SSH_KEY} # SECRET (please store in environment variables) | REQUIRED | string | OS-level user account ssh key credentials in RSA PEM format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )
     ## -------- Another valid structure for tunnel_method: --------
     # tunnel_method: "SSH_PASSWORD_AUTH" # REQUIRED | string | Connect through a jump server tunnel host using username and password authentication
     # tunnel_host: # REQUIRED | string | Hostname of the jump server host that allows inbound ssh tunnel.
     # tunnel_port: 22 # REQUIRED | integer | Port on the proxy/jump server that accepts inbound ssh connections. | Example: 22
     # tunnel_user: # REQUIRED | string | OS-level username for logging into the jump server host
-    # tunnel_user_password: # SECRET | REQUIRED | string | OS-level password for logging into the jump server host
+    # tunnel_user_password: ${TUNNEL_USER_PASSWORD} # SECRET (please store in environment variables) | REQUIRED | string | OS-level password for logging into the jump server host

octavia-cli/octavia_cli/apply/diff_helpers.py

@@ -3,6 +3,7 @@
 #
 
 import hashlib
+import json
 from typing import Any
 
 import click
@@ -11,23 +12,17 @@
 SECRET_MASK = "**********"
 
 
-def compute_checksum(file_path: str) -> str:
-    """Compute SHA256 checksum from a file
+def hash_config(configuration: dict) -> str:
+    """Computes a SHA256 hash from a dictionnary.
 
     Args:
-        file_path (str): Path for the file for which you want to compute a checksum.
+        configuration (dict): The configuration to hash
 
     Returns:
-        str: The computed hash digest
+        str: _description_
     """
-    BLOCK_SIZE = 65536
-    file_hash = hashlib.sha256()
-    with open(file_path, "rb") as f:
-        fb = f.read(BLOCK_SIZE)
-        while len(fb) > 0:
-            file_hash.update(fb)
-            fb = f.read(BLOCK_SIZE)
-    return file_hash.hexdigest()
+    stringified = json.dumps(configuration, sort_keys=True)
+    return hashlib.sha256(stringified.encode("utf-8")).hexdigest()
 
 
 def exclude_secrets_from_diff(obj: Any, path: str) -> bool:

octavia-cli/octavia_cli/apply/resources.py

@@ -23,7 +23,8 @@
 from airbyte_api_client.model.source_update import SourceUpdate
 from click import ClickException
 
-from .diff_helpers import compute_checksum, compute_diff
+from .diff_helpers import compute_diff, hash_config
+from .yaml_loaders import EnvVarLoader
 
 
 class DuplicateResourceError(ClickException):
@@ -39,27 +40,27 @@ class InvalidConfigurationError(ClickException):
 
 
 class ResourceState:
-    def __init__(self, configuration_path: str, resource_id: str, generation_timestamp: int, configuration_checksum: str):
+    def __init__(self, configuration_path: str, resource_id: str, generation_timestamp: int, configuration_hash: str):
         """This constructor is meant to be private. Construction shall be made with create or from_file class methods.
 
         Args:
-            configuration_path (str): Path to the configuration path the state relates to.
+            configuration_path (str): Path to the configuration this state relates to.
             resource_id (str): Id of the resource the state relates to.
             generation_timestamp (int): State generation timestamp.
-            configuration_checksum (str): Checksum of the configuration file.
+            configuration_hash (str): Checksum of the configuration file.
         """
         self.configuration_path = configuration_path
         self.resource_id = resource_id
         self.generation_timestamp = generation_timestamp
-        self.configuration_checksum = configuration_checksum
+        self.configuration_hash = configuration_hash
         self.path = os.path.join(os.path.dirname(self.configuration_path), "state.yaml")
 
     def as_dict(self):
         return {
-            "configuration_path": self.configuration_path,
             "resource_id": self.resource_id,
             "generation_timestamp": self.generation_timestamp,
-            "configuration_checksum": self.configuration_checksum,
+            "configuration_path": self.configuration_path,
+            "configuration_hash": self.configuration_hash,
         }
 
     def _save(self) -> None:
@@ -68,19 +69,20 @@ def _save(self) -> None:
             yaml.dump(self.as_dict(), state_file)
 
     @classmethod
-    def create(cls, configuration_path: str, resource_id: str) -> "ResourceState":
+    def create(cls, configuration_path: str, configuration: dict, resource_id: str) -> "ResourceState":
         """Create a state for a resource configuration.
 
         Args:
             configuration_path (str): Path to the YAML file defining the resource.
+            configuration (dict): Configuration object that will be hashed.
             resource_id (str): UUID of the resource.
 
         Returns:
             ResourceState: state representing the resource.
         """
         generation_timestamp = int(time.time())
-        configuration_checksum = compute_checksum(configuration_path)
-        state = ResourceState(configuration_path, resource_id, generation_timestamp, configuration_checksum)
+        configuration_hash = hash_config(configuration)
+        state = ResourceState(configuration_path, resource_id, generation_timestamp, configuration_hash)
         state._save()
         return state
 
@@ -100,7 +102,7 @@ def from_file(cls, file_path: str) -> "ResourceState":
             raw_state["configuration_path"],
             raw_state["resource_id"],
             raw_state["generation_timestamp"],
-            raw_state["configuration_checksum"],
+            raw_state["configuration_hash"],
         )
 
 
@@ -194,9 +196,7 @@ def __init__(
         self.configuration_path = configuration_path
         self.api_instance = self.api(api_client)
         self.state = self._get_state_from_file()
-        self.local_file_changed = (
-            True if self.state is None else compute_checksum(self.configuration_path) != self.state.configuration_checksum
-        )
+        self.local_file_changed = True if self.state is None else hash_config(self.local_configuration) != self.state.configuration_hash
 
     @property
     def remote_resource(self):
@@ -274,7 +274,9 @@ def get_diff_with_remote_resource(self) -> str:
         return diff.pretty()
 
     def _create_or_update(
-        self, operation_fn: Callable, payload: Union[SourceCreate, SourceUpdate, DestinationCreate, DestinationUpdate]
+        self,
+        operation_fn: Callable,
+        payload: Union[SourceCreate, SourceUpdate, DestinationCreate, DestinationUpdate],
     ) -> Union[SourceRead, DestinationRead]:
         """Wrapper to trigger create or update of remote resource.
 
@@ -291,7 +293,7 @@ def _create_or_update(
         """
         try:
             result = operation_fn(self.api_instance, payload)
-            return result, ResourceState.create(self.configuration_path, result[self.resource_id_field])
+            return result, ResourceState.create(self.configuration_path, self.local_configuration, result[self.resource_id_field])
         except airbyte_api_client.ApiException as api_error:
             if api_error.status == 422:
                 # This  API response error is really verbose, but it embodies all the details about why the config is not valid.
@@ -417,10 +419,11 @@ def factory(api_client: airbyte_api_client.ApiClient, workspace_id: str, configu
         Union[Source, Destination]: The resource object created from the YAML config.
     """
     with open(configuration_path, "r") as f:
-        local_configuration = yaml.load(f, yaml.FullLoader)
+        local_configuration = yaml.load(f, EnvVarLoader)
     if local_configuration["definition_type"] == "source":
         return Source(api_client, workspace_id, local_configuration, configuration_path)
     if local_configuration["definition_type"] == "destination":
         return Destination(api_client, workspace_id, local_configuration, configuration_path)
+
     else:
         raise NotImplementedError(f"Resource {local_configuration['definition_type']} was not yet implemented")

octavia-cli/octavia_cli/apply/yaml_loaders.py

@@ -0,0 +1,35 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import os
+import re
+from typing import Any
+
+import yaml
+
+ENV_VAR_MATCHER_PATTERN = re.compile(r".*\$\{([^}^{]+)\}.*")
+
+
+def env_var_replacer(loader: yaml.Loader, node: yaml.Node) -> Any:
+    """Convert a YAML node to a Python object, expanding variable.
+
+    Args:
+        loader (yaml.Loader): Not used
+        node (yaml.Node): Yaml node to convert to python object
+
+    Returns:
+        Any: Python object with expanded vars.
+    """
+    return os.path.expandvars(node.value)
+
+
+class EnvVarLoader(yaml.SafeLoader):
+    pass
+
+
+# All yaml nodes matching the regex will be tagged as !environment_variable.
+EnvVarLoader.add_implicit_resolver("!environment_variable", ENV_VAR_MATCHER_PATTERN, None)
+
+# All yaml nodes tagged as !environment_variable will be constructed with the env_var_replacer callback.
+EnvVarLoader.add_constructor("!environment_variable", env_var_replacer)

octavia-cli/octavia_cli/generate/renderer.py

@@ -89,7 +89,7 @@ def _get_type_comment(self) -> str:
         return self.type if self.type else None
 
     def _get_secret_comment(self) -> str:
-        return "SECRET" if self.airbyte_secret else None
+        return "SECRET (please store in environment variables)" if self.airbyte_secret else None
 
     def _get_description_comment(self) -> str:
         return self.description if self.description else None
@@ -109,6 +109,8 @@ def _get_example_comment(self) -> str:
     def _get_default(self) -> str:
         if self.const:
             return self.const
+        if self.airbyte_secret:
+            return f"${{{self.name.upper()}}}"
         return self.default
 
     @staticmethod

octavia-cli/octavia_cli/templates/source_or_destination.yaml.j2

@@ -7,7 +7,7 @@ definition_image: {{ definition.docker_repository }}
 definition_version: {{ definition.docker_image_tag }}
 
 {%- macro render_field(field, is_commented) %}
-{%- if is_commented %}# {% endif %}{{ field.name }}:{% if field.default %} {{ field.default | tojson() }}{% endif %} # {{ field.comment }}
+{%- if is_commented %}# {% endif %}{{ field.name }}:{% if field.default %} {% if field.airbyte_secret %}{{ field.default }}{% else %}{{ field.default | tojson() }}{% endif %}{% endif %} # {{ field.comment }}
 {%- endmacro %}
 
 {%- macro render_sub_fields(sub_fields, is_commented) %}

octavia-cli/unit_tests/test_apply/test_diff_helpers.py

@@ -2,17 +2,13 @@
 # Copyright (c) 2021 Airbyte, Inc., all rights reserved.
 #
 
-from unittest.mock import mock_open, patch
-
 import pytest
 from octavia_cli.apply import diff_helpers
 
 
-def test_compute_checksum(mocker):
-    with patch("builtins.open", mock_open(read_data=b"data")) as mock_file:
-        digest = diff_helpers.compute_checksum("test_file_path")
-        assert digest == "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7"
-    mock_file.assert_called_with("test_file_path", "rb")
+def test_hash_config():
+    data_to_hash = {"example": "foo"}
+    assert diff_helpers.hash_config(data_to_hash) == "8d621bd700ff9a864bc603f56b4ec73536110b37d814dd4629767e898da70bef"
 
 
 @pytest.mark.parametrize(