Secrets Manager and secrets specification for tasks

dev-requirements.txt

@@ -28,7 +28,7 @@ flake8-black==0.2.1
     # via -r dev-requirements.in
 flake8-isort==4.0.0
     # via -r dev-requirements.in
-flake8==3.8.4
+flake8==3.9.0
     # via
     #   -r dev-requirements.in
     #   flake8-black
@@ -63,17 +63,17 @@ py==1.10.0
     # via
     #   -c requirements.txt
     #   pytest
-pycodestyle==2.6.0
+pycodestyle==2.7.0
     # via flake8
-pyflakes==2.2.0
+pyflakes==2.3.0
     # via flake8
 pyparsing==2.4.7
     # via
     #   -c requirements.txt
     #   packaging
 pytest==6.2.2
     # via -r dev-requirements.in
-regex==2020.11.13
+regex==2021.3.17
     # via
     #   -c requirements.txt
     #   black

doc-requirements.txt

@@ -41,9 +41,9 @@ black==19.10b0
     #   papermill
 bleach==3.3.0
     # via nbconvert
-boto3==1.17.26
+boto3==1.17.32
     # via sagemaker-training
-botocore==1.20.26
+botocore==1.20.32
     # via
     #   boto3
     #   s3transfer
@@ -78,7 +78,7 @@ decorator==4.4.2
     #   retry
 defusedxml==0.7.1
     # via nbconvert
-deprecated==1.2.11
+deprecated==1.2.12
     # via flytekit
 dirhash==0.2.1
     # via flytekit
@@ -90,7 +90,7 @@ entrypoints==0.3
     # via
     #   nbconvert
     #   papermill
-flyteidl==0.18.17
+flyteidl==0.18.24
     # via flytekit
 future==0.18.2
     # via croniter
@@ -108,7 +108,7 @@ idna==2.10
     # via requests
 imagesize==1.2.0
     # via sphinx
-importlib-metadata==3.7.2
+importlib-metadata==3.7.3
     # via keyring
 inotify_simple==1.2.1
     # via sagemaker-training
@@ -133,7 +133,7 @@ jmespath==0.10.0
     #   botocore
 jsonschema==3.2.0
     # via nbformat
-jupyter-client==6.1.11
+jupyter-client==6.1.12
     # via
     #   ipykernel
     #   nbclient
@@ -267,7 +267,7 @@ pyyaml==5.4.1
     #   sphinx-autoapi
 pyzmq==22.0.3
     # via jupyter-client
-regex==2020.11.13
+regex==2021.3.17
     # via
     #   black
     #   docker-image-py
@@ -277,13 +277,13 @@ requests==2.25.1
     #   papermill
     #   responses
     #   sphinx
-responses==0.12.1
+responses==0.13.1
     # via flytekit
 retry==0.9.2
     # via flytekit
 retrying==1.3.3
     # via sagemaker-training
-s3transfer==0.3.4
+s3transfer==0.3.6
     # via boto3
 sagemaker-training==3.7.3
     # via flytekit
@@ -312,7 +312,7 @@ snowballstemmer==2.1.0
     # via sphinx
 sortedcontainers==2.3.0
     # via flytekit
-soupsieve==2.2
+soupsieve==2.2.1
     # via beautifulsoup4
 sphinx-autoapi==1.7.0
     # via -r doc-requirements.in

flytekit/__init__.py

@@ -108,12 +108,22 @@
    PythonInstanceTask
    LaunchPlan
 
+Secrets and SecurityContext
+============================
+
+.. autosummary::
+   :nosignatures:
+   :toctree: generated/
+
+   Secret
+   SecurityContext
+
 """
 
 
 import flytekit.plugins  # This will be deprecated, these are the old plugins, the new plugins live in plugins/
 from flytekit.core.base_sql_task import SQLTask
-from flytekit.core.base_task import TaskMetadata, kwtypes
+from flytekit.core.base_task import SecurityContext, TaskMetadata, kwtypes
 from flytekit.core.condition import conditional
 from flytekit.core.container_task import ContainerTask
 from flytekit.core.context_manager import ExecutionParameters, FlyteContext
@@ -126,7 +136,7 @@
 from flytekit.core.reference_entity import LaunchPlanReference, TaskReference, WorkflowReference
 from flytekit.core.resources import Resources
 from flytekit.core.schedule import CronSchedule, FixedRate
-from flytekit.core.task import reference_task, task
+from flytekit.core.task import Secret, reference_task, task
 from flytekit.core.workflow import WorkflowFailurePolicy, reference_workflow, workflow
 from flytekit.loggers import logger
 from flytekit.types import schema

flytekit/common/tasks/sdk_runnable.py

@@ -25,12 +25,69 @@
 from flytekit.configuration import internal as _internal_config
 from flytekit.configuration import resources as _resource_config
 from flytekit.configuration import sdk as _sdk_config
+from flytekit.configuration import secrets
 from flytekit.engines import loader as _engine_loader
 from flytekit.interfaces.stats import taggable
 from flytekit.models import literals as _literal_models
 from flytekit.models import task as _task_models
 
 
+class SecretsManager(object):
+    """
+    This provides a secrets resolution logic at runtime.
+    The resolution order is
+      - Try env var first. The env var should have the configuration.SECRETS_ENV_PREFIX. The env var will be all upper
+         cased
+      - If not then try the file where the name matches lower case
+        ``configuration.SECRETS_DEFAULT_DIR/<group>/configuration.SECRETS_FILE_PREFIX<key>``
+
+    All configuration values can always be overriden by injecting an environment variable
+    """
+
+    def __init__(self):
+        self._base_dir = str(secrets.SECRETS_DEFAULT_DIR.get()).strip()
+        self._file_prefix = str(secrets.SECRETS_FILE_PREFIX.get()).strip()
+        self._env_prefix = str(secrets.SECRETS_ENV_PREFIX.get()).strip()
+
+    def get(self, group: str, key: str) -> str:
+        """
+        Retrieves a secret using the resolution order -> Env followed by file. If not found raises a ValueError
+        """
+        self.check_group_key(group, key)
+        env_var = self.get_secrets_env_var(group, key)
+        fpath = self.get_secrets_file(group, key)
+        v = os.environ.get(env_var)
+        if v is not None:
+            return v
+        if os.path.exists(fpath):
+            with open(fpath, "r") as f:
+                return f.read().strip()
+        raise ValueError(
+            f"Unable to find secret for key {key} in group {group} " f"in Env Var:{env_var} and FilePath: {fpath}"
+        )
+
+    def get_secrets_env_var(self, group: str, key: str) -> str:
+        """
+        Returns a string that matches the ENV Variable to look for the secrets
+        """
+        self.check_group_key(group, key)
+        return f"{self._env_prefix}{group.upper()}_{key.upper()}"
+
+    def get_secrets_file(self, group: str, key: str) -> str:
+        """
+        Returns a path that matches the file to look for the secrets
+        """
+        self.check_group_key(group, key)
+        return os.path.join(self._base_dir, group.lower(), f"{self._file_prefix}{key.lower()}")
+
+    @staticmethod
+    def check_group_key(group: str, key: str):
+        if group is None or group == "":
+            raise ValueError("secrets group is a mandatory field.")
+        if key is None or key == "":
+            raise ValueError("secrets key is a mandatory field.")
+
+
 # TODO: Clean up working dir name
 class ExecutionParameters(object):
     """
@@ -98,6 +155,8 @@ def __init__(self, execution_date, tmp_dir, stats, execution_id, logging, **kwar
         self._logging = logging
         # AutoDeletingTempDir's should be used with a with block, which creates upon entry
         self._attrs = kwargs
+        # It is safe to recreate the Secrets Manager
+        self._secrets_manager = SecretsManager()
 
     @property
     def stats(self) -> taggable.TaggableStats:
@@ -151,6 +210,10 @@ def execution_id(self) -> str:
         """
         return self._execution_id
 
+    @property
+    def secrets(self) -> SecretsManager:
+        return self._secrets_manager
+
     def __getattr__(self, attr_name: str) -> typing.Any:
         """
         This houses certain task specific context. For example in Spark, it houses the SparkSession, etc

flytekit/common/tasks/task.py

@@ -37,7 +37,9 @@ class SdkTask(
     _task_model.TaskTemplate,
     metaclass=_sdk_bases.ExtendedSdkType,
 ):
-    def __init__(self, type, metadata, interface, custom, container=None, task_type_version=0, config=None):
+    def __init__(
+        self, type, metadata, interface, custom, container=None, task_type_version=0, security_context=None, config=None
+    ):
         """
         :param Text type: This is used to define additional extensions for use by Propeller or SDK.
         :param TaskMetadata metadata: This contains information needed at runtime to determine behavior such as
@@ -48,6 +50,7 @@ def __init__(self, type, metadata, interface, custom, container=None, task_type_
             a Container might be specified with the necessary command line arguments.
         :param int task_type_version: Specific version of this task type used by plugins to potentially modify
             execution behavior or serialization.
+        :param _SecurityContext security_context:
         """
         # TODO: Remove the identifier portion and fill in with local values.
         super(SdkTask, self).__init__(
@@ -64,6 +67,7 @@ def __init__(self, type, metadata, interface, custom, container=None, task_type_
             custom,
             container=container,
             task_type_version=task_type_version,
+            security_context=security_context,
             config=config,
         )
 

flytekit/common/translator.py

@@ -123,6 +123,7 @@ def get_serializable_task(
         custom=entity.get_custom(settings),
         container=entity.get_container(settings),
         task_type_version=entity.task_type_version,
+        security_context=entity.security_context,
         config=entity.get_config(settings),
     )
     # Reset just to make sure it's what we give it

flytekit/configuration/secrets.py

@@ -0,0 +1,23 @@
+import os
+
+from flytekit.configuration import common as _common_config
+
+# Secrets management
+SECRETS_ENV_PREFIX = _common_config.FlyteStringConfigurationEntry("secrets", "env_prefix", default="_FSEC_")
+"""
+This is the prefix that will be used to lookup for injected secrets at runtime. This can be overriden to using
+FLYTE_SECRETS_ENV_PREFIX variable
+"""
+
+SECRETS_DEFAULT_DIR = _common_config.FlyteStringConfigurationEntry(
+    "secrets", "default_dir", default=os.path.join(os.sep, "etc", "secrets")
+)
+"""
+This is the default directory that will be used to find secrets as individual files under. This can be overriden using
+FLYTE_SECRETS_DEFAULT_DIR.
+"""
+
+SECRETS_FILE_PREFIX = _common_config.FlyteStringConfigurationEntry("secrets", "file_prefix", default="")
+"""
+This is the prefix for the file in the default dir.
+"""

flytekit/core/base_task.py

@@ -29,6 +29,7 @@
 from flytekit.models import literals as _literal_models
 from flytekit.models import task as _task_model
 from flytekit.models.interface import Variable
+from flytekit.models.security import SecurityContext
 
 
 def kwtypes(**kwargs) -> Dict[str, Type]:
@@ -120,13 +121,15 @@ def __init__(
         interface: Optional[_interface_models.TypedInterface] = None,
         metadata: Optional[TaskMetadata] = None,
         task_type_version=0,
+        security_ctx: Optional[SecurityContext] = None,
         **kwargs,
     ):
         self._task_type = task_type
         self._name = name
         self._interface = interface
         self._metadata = metadata if metadata else TaskMetadata()
         self._task_type_version = task_type_version
+        self._security_ctx = security_ctx
 
         FlyteEntities.entities.append(self)
 
@@ -154,6 +157,10 @@ def python_interface(self) -> Optional[Interface]:
     def task_type_version(self) -> int:
         return self._task_type_version
 
+    @property
+    def security_context(self) -> SecurityContext:
+        return self._security_ctx
+
     def get_type_for_input_var(self, k: str, v: Any) -> type:
         """
         Returns the python native type for the given input variable
@@ -308,16 +315,21 @@ def __init__(
         name: str,
         task_config: T,
         interface: Optional[Interface] = None,
-        environment=None,
-        task_type_version=0,
+        environment: Optional[Dict[str, str]] = None,
         **kwargs,
     ):
+        """
+        Args:
+            task_type: a string that defines a unique task-type for every new extension. If a backend plugin is required
+                       then this has to be done in-concert with the backend plugin identifier
+            name: A unique name for the task instantiation. This is unique for every instance of task.
+            task_config: Configuration for the task. This is used to configure the specific plugin that handles this task
+            interface: A python native typed interface ``(inputs) -> outputs`` that declares the signature of the task
+            environment: Any environment variables that should be supplied during the execution of the task. Supplied as
+                         a dictionary of key/value pairs
+        """
         super().__init__(
-            task_type=task_type,
-            name=name,
-            interface=transform_interface_to_typed_interface(interface),
-            task_type_version=task_type_version,
-            **kwargs,
+            task_type=task_type, name=name, interface=transform_interface_to_typed_interface(interface), **kwargs,
         )
         self._python_interface = interface if interface else Interface()
         self._environment = environment if environment else {}