Explicitly Hidden (from the UI) Configuration Values

[Daniel-Vetter-Coverwhale]

Is there any way to mark a configuration value as secret to avoid showing in the UI without it being an envvar?

My current use case is pulling database configuration including passwords from AWS secrets manager and building sling connections from that info, and if I open the launchpad for one of the sling assets using the sling resource it shows everything. I think it would be awesome if for things like SlingResources or other third party published configurable resources to accept things like passwords for fully specified instances, but mark them so the user doesn't have to worry about whether they show up in the UI.

I realize that I can set an environment variable and then reference it as part of loading the secret (which is what I'm doing for now), but it would feel better and more generally useful I think to just have some other way of specifying that a particular configuration value should never be shown in plaintext.

Some sample code showing how I'm building the sling connections from the secrets, currently using dagster.EnvVar which requires of course that we set the value from the secret into a visible (from the system) environment variable. As mentioned above I would prefer to not have to use the environment variable to hide the password from the UI. Also I wonder at the value of the having separate Source and Target connection classes, but that's a different issue.

def sling_connection_from_secret(
    source_or_target: typing.Literal["source", "target"], secret_name: str
) -> typing.Union[ds.SlingSourceConnection, ds.SlingTargetConnection]:
    sm_resource = dagster_aws.secretsmanager.SecretsManagerResource()
    secret = json.loads(sm_resource.get_client().get_secret_value(SecretId=secret_name)["SecretString"])
    try:
        var_name = f"{secret_name}.password"
        os.environ[var_name] = secret["password"]
        config = {
            "type": secret["engine"],
            "host": secret["host"],
            "user": secret["username"],
            "password": dagster.EnvVar(var_name),
            "database": secret["dbname"],
            "port": secret["port"],
        }
    except KeyError as ke:
        raise KeyError(f"{ke} not found in {secret_name}")

    if source_or_target == "source":
        conn = ds.SlingSourceConnection(**config)
    else:
        conn = ds.SlingTargetConnection(**config)

    return conn

[darktempla]

Could someone please chime in on this one in regards to if will be considered for the roadmap.
I too would prefer a more graceful flexible solution to this.

In my example I am grabbing secrets from hashicorp vault but it's the same concept (see code/ui below for context).

To make testing easier and keep objects decoupled I would prefer to just set the api_key on the resource rather than inject the vault client. If I move to CyberArk etc later it would require less refactoring etc not to mention it makes unit tests cleaner/easier as your passing simple type (string) vs complex (object etc).

The workaround above is likely the path I would go down at this stage. Thanks for sharing @Daniel-Vetter-Coverwhale.

Suggestion
Introduce a new type for example SecretVar that lets you just pass a value to it SecretVar('my-secret') and that too is not displayed in the UI etc. Internally inside EnvVar you could utilise this NEW object type and that would likely simplify display logic in the UI as your only looking for one type. I am not familiar with the code so much so just a general idea sure someone familiar with the code would have a better insights if this makes sense.

Thanks in advance.

Really liking Dagster I have looked at ALOT of other tools and the assets as code, lineage and materialization really make a huge difference.

CODE

import os

import hvac
from dagster import Definitions, load_assets_from_modules

from .assets import stocks
from .resources import FinancialModelingPrepApi

stock_assets = load_assets_from_modules([stocks], group_name="stocks")

vault_client = hvac.Client(url=os.getenv("VAULT_URL"), verify=True)
vault_client.auth.approle.login(role_id=os.getenv("VAULT_APP_ROLE_ID"), secret_id=os.getenv("VAULT_APP_ROLE_SECRET"))
fmp_api_key = vault_client.secrets.kv.v2.read_secret_version(path="financial-modeling-prep-api", mount_point="kv")['data']['data']['API_KEY']
 
defs = Definitions(
    assets=[*stock_assets],
    resources={
        "fmp_api": FinancialModelingPrepApi(api_key=fmp_api_key),
    },
)

UI

[caelan-schneider]

Also would love to see this.

[jtbricker]

I imagine that subclassing the EnvVar class might accomplish this, maybe something like:

class SecretsMgrEnvVar(EnvVar):
    """
    This class wraps the EnvVar class and overrides the `get_value` method to:

        1. First check if parent class's `get_value` method returns a value.
        2. If not, attempt to fetch the value from the Secrets Manager.

    """
    def get_value(self, default: Optional[str] = None) -> Optional[str]:
        value = super().get_value(default)
        if value is not None:
            return value

        # Call some method to get your secret from a secrets manager
        return SecretsManager.get_secret(self.env_var_name)