What are best practices for using Prefect secrets within Prefect tasks?

[anna_geller]

View in #prefect-community on Slack

@Martin_T: Hi! Using PrefectSecret() to load secrets from Prefect Cloud. Is it possible to load secret from cloud into prefect.context.secrets during flow start so all tasks can access them when needed?
Most of our tasks variables are auth-related, and the code/layout is getting overly complex.

@Anna_Geller: In that case, you may retrieve the Secret using Client method as follows:

from prefect import task, Flow
from prefect.client.secrets import Secret


@task(log_stdout=True)
def use_secret_within_some_task_directly():
    secret_value = Secret("A_TEST_SECRET").get()
    print(secret_value)


with Flow("secret_flow_client") as flow:
    use_secret_within_some_task_directly()

if __name__ == "__main__":
    flow.run()

But best practice is to pass it to downstream tasks as data dependency as shown here:

from prefect import task, Flow
from prefect.tasks.secrets import PrefectSecret


@task(log_stdout=True)
def use_secret_from_data_dependency(secret_value: str):
    print(secret_value)


with Flow("secret_flow_client") as flow:
    secret_val = PrefectSecret("A_TEST_SECRET")
    use_secret_from_data_dependency(secret_val)

if __name__ == "__main__":
    flow.run()

In both cases, you don’t need to store secrets in the context - storing secrets in the context is only needed:

1. If you are using Prefect Server since Server doesn’t have the Secret backend
2. For local execution - even though local flow runs may retrieve Secrets from the Cloud backend, too.

@Martin_T: We’re currently using the latter method.
Yes Secret("A_TEST_SECRET").get() was what I was looking for. I wasn’t sure if this was retrieving from Prefect Cloud Secrets. https://docs.prefect.io/orchestration/concepts/secrets.html#using-secrets-elsewhere
Why is Secret().get() discouraged?

@Anna_Geller: It can lead to unexpected errors depending on when you set these secrets - for example, if you set this globally on a module scope:

from prefect import task, Flow
from prefect.client.secrets import Secret


secret_value = Secret("A_TEST_SECRET").get()

Then, this is problematic since secrets are generally considered a runtime information, while setting this as above retrieves the secret at build time during flow registration. As long as you use script storage, even this is fine, but if you use the default pickle storage and you retrieve the Secret globally like this, then your pickled flow would become a security risk as it would be possible to get the value of your Secret from that pickle file.

So doing this within a task is not an anti-pattern and is still fine because you retrieve this at runtime within the task run:

@task(log_stdout=True)
def use_secret_within_some_task_directly():
    secret_value = Secret("A_TEST_SECRET").get()
    print(secret_value)

So TL;DR - don’t use prefect.client.secrets.Secret globally when using pickle storage - in all other use cases, it should be fine.

@Kevin_Kho: I think you can try adding the secret with:

flow.storage = StorageClass(..., secrets=["SECRET_NAME"])