Add EnvVarSecret

CHANGELOG.md

@@ -12,6 +12,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 
 - Add a `save`/`load` interface to Flows - [#1685](https://github.com/PrefectHQ/prefect/pull/1685)
 - Add option to specify `aws_session_token` for the `FargateTaskEnvironment` - [#1688](https://github.com/PrefectHQ/prefect/pull/1688)
+- Add `EnvVarSecrets` for loading sensitive information from environment variables - [#1683](https://github.com/PrefectHQ/prefect/pull/1683)
 - Add an informative version header to all Cloud client requests - [#1690](https://github.com/PrefectHQ/prefect/pull/1690)
 
 ### Task Library

docs/.vuepress/config.js

@@ -201,13 +201,15 @@ module.exports = {
             "concepts/flows",
             "concepts/parameters",
             "concepts/states",
-            "concepts/mapping",
             "concepts/engine",
             "concepts/execution",
-            "concepts/persistence",
+            "concepts/logging",
+            "concepts/mapping",
             "concepts/notifications",
+            "concepts/persistence",
             "concepts/results",
             "concepts/schedules",
+            "concepts/secrets",
             "concepts/configuration",
             "concepts/best-practices",
             "concepts/common-pitfalls"

docs/core/concepts/secrets.md

@@ -0,0 +1,43 @@
+# Secrets
+
+## Overview
+
+Very often, workflows require sensitive information to run: API keys, passwords, tokens, credentials, etc. As a matter of best practice, such information should never be hardcoded into a workflow's source code, as the code itself will need to be guarded. Furthermore, sensitive information should not be provided via a Prefect `Parameter`, because Parameters, like many tasks, can have their results stored.
+
+Prefect provides a mechanism called `Secrets` for working with sensitive information.
+
+- `Secret` tasks, which are special tasks that can be used in your flow when working with sensitive information. Unlike regular tasks, `Secret` tasks are designed to access sensitive information at runtime and use a special `ResultHandler` to ensure the results are not stored.
+- The `prefect.client.secrets` API, which provides an interface for working with sensitive information. This API can be used where tasks are unavailable, including notifications, state handlers, and result handlers.
+
+::: tip Keep secrets secret!
+Though Prefect takes steps to ensure that `Secret` objects do not reveal sensitive information, other tasks may not be so careful. Once a secret value is loaded into your flow, it can be used for any purpose. Please use caution anytime you are working with sensitive data.
+:::
+
+## Mechanisms
+
+### Local Context
+
+The base `Secret` class first checks for secrets in local context, under `prefect.context.secrets`. This is useful for local testing, as secrets can be added to context by setting the environment variable `PREFECT__CONTEXT__SECRETS__FOO`, corresponding to `secrets.foo` (or `secrets.FOO`, if your OS is case-sensitive).
+
+### Prefect Cloud
+
+If the secret is not found in local context and `config.cloud.use_local_secrets=False`, the base `Secret` class queries the Prefect Cloud API for a stored secret. This call can only be made successfully by authenticated Prefect Cloud Agents.
+
+### Environment Variables
+
+The `EnvVarSecret` class reads secret values from environment variables.
+
+```python
+from prefect import task, Flow
+from prefect.tasks.secrets import EnvVarSecret
+
+@task
+def print_value(x):
+    print(x)
+
+with Flow("Example") as flow:
+    secret = EnvVarSecret("PATH")
+    print_value(secret)
+
+flow.run() # prints the value of the "PATH" environment variable
+```

docs/outline.toml

@@ -197,10 +197,10 @@ classes = ["S3Download", "S3Upload", "LambdaCreate", "LambdaDelete" , "LambdaInv
 [pages.tasks.azure]
 title = "Azure Tasks"
 module = "prefect.tasks.azure"
-classes = ["BlobStorageDownload", 
-           "BlobStorageUpload", 
-           "CosmosDBCreateItem", 
-           "CosmosDBReadItems", 
+classes = ["BlobStorageDownload",
+           "BlobStorageUpload",
+           "CosmosDBCreateItem",
+           "CosmosDBReadItems",
            "CosmosDBQueryItems"]
 
 [pages.tasks.azureml]
@@ -308,7 +308,7 @@ classes = ["ParseRSSFeed"]
 [pages.tasks.secrets]
 title = "Secret Tasks"
 module = "prefect.tasks.secrets"
-classes = ["Secret"]
+classes = ["Secret", "EnvVarSecret"]
 
 [pages.tasks.snowflake]
 title = "Snowflake Tasks"

src/prefect/tasks/secrets/__init__.py

@@ -6,3 +6,4 @@ class for interacting with other secret providers. Secrets always use a special
 prevents the persistence of sensitive information.
 """
 from .base import Secret
+from .env_var import EnvVarSecret

src/prefect/tasks/secrets/env_var.py

@@ -0,0 +1,48 @@
+import os
+from typing import Any, Callable
+
+from prefect.tasks.secrets import Secret
+
+
+class EnvVarSecret(Secret):
+    """
+    A `Secret` task that retrieves a value from an environment variable.
+
+    Args:
+        - env_var (str): the environment variable that contains the secret value
+        - name (str, optional): a name for the task. If not provided, `env_var` is used.
+        - cast (Callable[[Any], Any]): A function that will be called on the Parameter
+            value to coerce it to a type.
+        - raise_if_missing (bool): if True, an error will be raised if the env var is not found.
+        - **kwargs (Any, optional): additional keyword arguments to pass to the Task constructor
+    """
+
+    def __init__(
+        self,
+        env_var: str,
+        name: str = None,
+        cast: Callable[[Any], Any] = None,
+        raise_if_missing: bool = False,
+        **kwargs
+    ):
+        self.env_var = env_var
+        self.cast = cast
+        self.raise_if_missing = raise_if_missing
+        if name is None:
+            name = env_var
+
+        super().__init__(name=name, **kwargs)
+
+    def run(self):
+        """
+        Returns the value of an environment variable after applying an optional `cast` function.
+
+        Returns:
+            - Any: the (optionally type-cast) value of the environment variable
+        """
+        if self.raise_if_missing and self.env_var not in os.environ:
+            raise ValueError("Environment variable not set: {}".format(self.env_var))
+        value = os.getenv(self.env_var)
+        if value is not None and self.cast is not None:
+            value = self.cast(value)
+        return value

tests/tasks/secrets/test_env_var.py

@@ -0,0 +1,70 @@
+import pendulum
+import pytest
+import prefect
+from prefect.tasks.secrets import EnvVarSecret
+
+
+def test_create_envvarsecret_requires_env_var():
+    with pytest.raises(TypeError, match="required positional argument: 'env_var'"):
+        EnvVarSecret()
+
+
+def test_name_defaults_to_env_var():
+    e = EnvVarSecret(env_var="FOO")
+    assert e.env_var == "FOO"
+    assert e.name == "FOO"
+
+
+def test_name_can_be_customized():
+    e = EnvVarSecret(env_var="FOO", name="BAR")
+    assert e.env_var == "FOO"
+    assert e.name == "BAR"
+
+
+def test_default_cast_is_none():
+    e = EnvVarSecret(env_var="FOO")
+    assert e.cast is None
+
+
+def test_run_secret(monkeypatch):
+    monkeypatch.setenv("FOO", "1")
+    e = EnvVarSecret(env_var="FOO")
+    assert e.run() == "1"
+
+
+def test_run_secret_without_env_var_set_returns_none(monkeypatch):
+    monkeypatch.delenv("FOO", raising=False)
+    e = EnvVarSecret(env_var="FOO")
+    assert e.run() is None
+
+
+def test_run_secret_without_env_var_set_raises(monkeypatch):
+    monkeypatch.delenv("FOO", raising=False)
+    e = EnvVarSecret(env_var="FOO", raise_if_missing=True)
+    with pytest.raises(ValueError, match="variable not set"):
+        e.run()
+
+
+def test_run_secret_with_cast(monkeypatch):
+    monkeypatch.setenv("FOO", "1")
+    e = EnvVarSecret(env_var="FOO", cast=int)
+    assert e.run() == 1
+
+
+def test_run_secret_without_env_var_set_returns_none_even_if_cast_set(monkeypatch):
+    monkeypatch.delenv("FOO", raising=False)
+    e = EnvVarSecret(env_var="FOO", cast=int)
+    assert e.run() is None
+
+
+def test_run_secret_without_env_var_set_raises_with_cast(monkeypatch):
+    monkeypatch.delenv("FOO", raising=False)
+    e = EnvVarSecret(env_var="FOO", raise_if_missing=True, cast=int)
+    with pytest.raises(ValueError, match="variable not set"):
+        e.run()
+
+
+def test_run_secret_with_cast_datetime(monkeypatch):
+    monkeypatch.setenv("FOO", "2019-01-02 03:04:05")
+    e = EnvVarSecret(env_var="FOO", cast=pendulum.parse)
+    assert e.run() == pendulum.datetime(2019, 1, 2, 3, 4, 5)