Hi everyone. I have a problem where my flow is unintentionally exposing secrets in the UI. I am passing secrets to the

env

argument of a class derived from

ShellTask

.

bash = LoggedShellTask(stream_output=True)

# more stuff here ...

with Flow(
    FLOW_NAME, storage=docker_storage, run_config=run_config, result=PrefectResult()
) as flow:

    run_results = bash(
        helper_script=get_helper_script(repo_path),
        command=get_command(ml_script_path, ml_script_arguments, conda_env=conda_env),
        env=dict(
            SNOWFLAKE_USER=snowflake_user,
            SNOWFLAKE_PASS=snowflake_pass,
            SNOWFLAKE_DATABASE=snowflake_database,
            SNOWFLAKE_OUTPUT_SCHEMA=snowflake_output_schema,
        ),
        log_file_path=get_log_file_path(output_dir),
    )

The issue is Prefect automatically creates a

List

and

Dict

task that have

PrefectResults

by default and this exposes the secrets in the UI. A couple of ideas that come to mind: 1. Assign a specific results objects to each task (not at the flow level) 2. Create a wrapper task that receives all the secrets and returns a dict that is passed to the

ShellTask

3. Create a

ShellTask

that accepts each secret as a parameter Any suggestions? One feature I'd like to preserve is the ability to restart the flow and I'd rather not persist the secrets anywhere. Thanks!

I think you might need to lose the ability to restart here because if you did number 2, you’d need to turn off checkpointing for it to not persist. Or you could edit your pass to pull the Secrets directly with

Secret.get()

?

Thanks Kevin. I had not thought about that. How do secret tasks work when the flow restarts? Do they run every time even if they were successful before? Could I leverage that behavior somehow in a custom task?

Not a Secret task:

from prefect.client import Secret

@task
def abc():
    val = Secret("NAME").get()