Secrets and Environment Variables

Secrets are a mechanism for securely injecting sensitive information, such as API keys, database credentials, or access tokens, into your tasks or image build contexts. This ensures that sensitive data is not hardcoded in your code or exposed in logs. Secrets can be mounted into the execution environment either as environment variables or as files.

Defining and Using Secrets

The core component for managing secrets is the Secret class. This class allows you to specify the name of the secret in your secret store, how it should be mounted, and its visibility within your tasks.

Basic Secret Definition

The simplest way to define a secret is by providing its key as a string directly to a task decorator. When defined this way, the secret's value will be automatically mounted as an environment variable.

import os
from typing import Optional
import pathlib
import re
from dataclasses import dataclass

@dataclass
class Secret:
    key: str
    group: Optional[str] = None
    mount: pathlib.Path | None = None
    as_env_var: Optional[str] = None

    def __post_init__(self):
        if not self.mount and not self.as_env_var:
            self.as_env_var = f"{self.group}_{self.key}" if self.group else self.key
            self.as_env_var = self.as_env_var.replace("-", "_").upper()
        if self.as_env_var is not None:
            pattern = r"^[A-Z_][A-Z0-9_]*$"
            if not re.match(pattern, self.as_env_var):
                raise ValueError(f"Invalid environment variable name: {self.as_env_var}, must match {pattern}")

    def stable_hash(self) -> str:
        import hashlib
        data = (
            self.key,
            self.group or "",
            str(self.mount) if self.mount else "",
            self.as_env_var or "",
        )
        joined = "|".join(data)
        return hashlib.sha256(joined.encode("utf-8")).hexdigest()

    def __hash__(self) -> int:
        return int(self.stable_hash()[:16], 16)

# Assume @task decorator is available
def task(secrets):
    def decorator(func):
        func._secrets = secrets
        return func
    return decorator

@task(secrets="my-secret")
async def my_task():
    # The secret named "my-secret" will be available as an environment variable.
    # By default, the environment variable name is derived from the secret key:
    # it's converted to uppercase, and hyphens are replaced with underscores.
    # So, "my-secret" becomes "MY_SECRET".
    secret_value = os.environ.get("MY_SECRET")
    print(f"My secret value: {secret_value}")

# Example usage (conceptual, as os.environ is not modified here directly)
# If 'my-secret' had value 'super-secret-value' in the secret store,
# then inside my_task, os.environ["MY_SECRET"] would be 'super-secret-value'.

Customizing Environment Variable Names

For more control, instantiate the Secret class directly. This allows you to specify a custom environment variable name using the as_env_var parameter. This is useful when integrating with libraries or tools that expect a specific environment variable name.

The as_env_var name must adhere to standard environment variable naming conventions, specifically matching the regex ^[A-Z_][A-Z0-9_]*$.

# Assume @task decorator is available
def task(secrets):
    def decorator(func):
        func._secrets = secrets
        return func
    return decorator

@task(secrets=Secret("my-openai-api-key", as_env_var="OPENAI_API_KEY"))
async def my_task2():
    # The secret "my-openai-api-key" is explicitly mounted as OPENAI_API_KEY.
    openai_key = os.environ.get("OPENAI_API_KEY")
    print(f"OpenAI API Key: {openai_key}")

# Example usage (conceptual)
# If 'my-openai-api-key' had value 'sk-12345' in the secret store,
# then inside my_task2, os.environ["OPENAI_API_KEY"] would be 'sk-12345'.

Secret Grouping

The group parameter in the Secret class allows for organizing secrets within a secret store. While its primary use depends on the underlying secret store implementation, it also influences the default environment variable name when as_env_var is not explicitly set.

If a group is provided and as_env_var is not, the environment variable name is derived as "{GROUP}_{KEY}" (uppercase, hyphens replaced with underscores).

# Assume @task decorator is available
def task(secrets):
    def decorator(func):
        func._secrets = secrets
        return func
    return decorator

@task(secrets=Secret(key="db-password", group="production-db"))
async def connect_to_db():
    # The secret "db-password" from group "production-db"
    # will be mounted as an environment variable named "PRODUCTION_DB_DB_PASSWORD".
    db_password = os.environ.get("PRODUCTION_DB_DB_PASSWORD")
    print(f"Database password: {db_password}")

Mounting Secrets as Files

Secrets can also be mounted as files within the task's execution environment. This is specified using the mount parameter of the Secret class.

Currently, the system supports mounting secrets only to the path /etc/flyte/secrets. Future enhancements will allow specifying arbitrary mount paths.

import os
import pathlib

# Assume @task decorator is available
def task(secrets):
    def decorator(func):
        func._secrets = secrets
        return func
    return decorator

@task(secrets=Secret("my-certificate", mount=pathlib.Path("/etc/flyte/secrets/my-certificate.pem")))
async def use_certificate():
    # The secret "my-certificate" will be mounted as a file at
    # /etc/flyte/secrets/my-certificate.pem.
    cert_path = "/etc/flyte/secrets/my-certificate.pem"
    if os.path.exists(cert_path):
        with open(cert_path, "r") as f:
            certificate_content = f.read()
        print(f"Certificate content loaded from file.")
    else:
        print(f"Certificate file not found at {cert_path}.")

Important Considerations and Best Practices

• Security: Never embed sensitive information directly in your code. Always use the secret management system to inject credentials.
• Immutability: Once a task starts, the secrets mounted into its environment are static. Changes to the secret in the secret store will not affect running tasks. New task executions will pick up the updated values.
• Environment Variable Naming: When specifying as_env_var, ensure it follows the ^[A-Z_][A-Z0-9_]*$ pattern. Invalid names will raise a ValueError.
• Access Control: Access to secrets is managed by the underlying platform's identity and access management (IAM) system. Ensure that the service account executing your tasks has the necessary permissions to retrieve the specified secrets.
• Limitations:
  • Currently, file-based secret mounts are restricted to the /etc/flyte/secrets directory.
  • Advanced features like secret versioning are not yet directly exposed through the Secret class but are planned for future support.
• Hashing: The Secret class provides stable_hash() and __hash__() methods for deterministic hashing, which is useful for internal system operations like caching and identifying unique secret configurations.