From d39b111d17338b25655fbaf23628d4ee8fc0bd7a Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 22:33:49 +0000
Subject: [PATCH 1/7] Initial plan


From 0c0896b189cced3dd7bde3c2b47c874b40f65903 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 22:40:35 +0000
Subject: [PATCH 2/7] Add Vault secret provider with JWT authentication support

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 pyproject.toml                |   1 +
 src/nyl/secrets/__init__.py   |   2 +-
 src/nyl/secrets/vault.py      | 339 ++++++++++++++++++++++++++++++++++
 src/nyl/secrets/vault_test.py | 259 ++++++++++++++++++++++++++
 uv.lock                       |  14 ++
 5 files changed, 614 insertions(+), 1 deletion(-)
 create mode 100644 src/nyl/secrets/vault.py
 create mode 100644 src/nyl/secrets/vault_test.py

diff --git a/pyproject.toml b/pyproject.toml
index 172690fb..093993ce 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -11,6 +11,7 @@ dependencies = [
     "bcrypt>=4.2.0",
     "databind>=4.5.2",
     "filelock>=3.15.4",
+    "hvac>=2.3.0",
     "jinja2>=3.1.4",
     "kubernetes>=30.1.0",
     "loguru>=0.7.2",
diff --git a/src/nyl/secrets/__init__.py b/src/nyl/secrets/__init__.py
index 1d5d5dab..0321b076 100644
--- a/src/nyl/secrets/__init__.py
+++ b/src/nyl/secrets/__init__.py
@@ -74,4 +74,4 @@ def unset(self, key: str, /) -> None:
         """
 
 
-from . import config, kubernetes, sops  # noqa
+from . import config, kubernetes, sops, vault  # noqa
diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
new file mode 100644
index 00000000..d0d95e54
--- /dev/null
+++ b/src/nyl/secrets/vault.py
@@ -0,0 +1,339 @@
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Iterable
+
+import hvac
+from databind.core import Union
+from loguru import logger
+
+from nyl.secrets import SecretProvider, SecretValue
+from nyl.tools.di import DependenciesProvider
+
+
+@Union.register(SecretProvider, name="vault")
+@dataclass
+class VaultSecretProvider(SecretProvider):
+    """
+    This secrets provider retrieves secrets from HashiCorp Vault using the KV secrets engine.
+
+    When running in ArgoCD, it authenticates using the Kubernetes service account JWT token.
+    When running locally, it uses the token from `~/.vault-token` (obtained via `vault login`).
+
+    The provider supports nested structures through dot notation, similar to the SOPS provider.
+    """
+
+    url: str
+    """
+    The URL of the Vault server (e.g., "https://vault.example.com:8200").
+    """
+
+    mount_point: str = "secret"
+    """
+    The mount point of the KV secrets engine in Vault. Defaults to "secret".
+    """
+
+    path: str = ""
+    """
+    The path prefix within the KV secrets engine where secrets are stored.
+    For example, if path is "myapp/", secrets will be retrieved from "secret/data/myapp/...".
+    """
+
+    jwt_role: str | None = None
+    """
+    The Vault role to use for JWT authentication when running in ArgoCD.
+    If not specified, JWT authentication will not be attempted.
+    """
+
+    namespace: str | None = None
+    """
+    The Vault namespace to use (Vault Enterprise feature).
+    """
+
+    _client: hvac.Client | None = field(init=False, repr=False, default=None)
+    _cache: dict[str, SecretValue] | None = field(init=False, repr=False, default=None)
+
+    def _get_client(self) -> hvac.Client:
+        """Get or create a Vault client with appropriate authentication."""
+        if self._client is not None:
+            return self._client
+
+        self._client = hvac.Client(url=self.url, namespace=self.namespace)
+
+        # Try to authenticate
+        if self._is_argocd_context() and self.jwt_role:
+            self._authenticate_with_jwt()
+        else:
+            self._authenticate_with_token()
+
+        if not self._client.is_authenticated():
+            raise RuntimeError("Failed to authenticate with Vault")
+
+        logger.info("Successfully authenticated with Vault at {}", self.url)
+        return self._client
+
+    def _is_argocd_context(self) -> bool:
+        """Check if we're running in an ArgoCD context by looking for ArgoCD environment variables."""
+        return "ARGOCD_APP_NAME" in os.environ
+
+    def _authenticate_with_jwt(self) -> None:
+        """Authenticate with Vault using Kubernetes JWT token (for ArgoCD context)."""
+        jwt_path = Path("/var/run/secrets/kubernetes.io/serviceaccount/token")
+        if not jwt_path.exists():
+            logger.warning(
+                "JWT authentication requested but token file not found at {}. Falling back to token auth.", jwt_path
+            )
+            self._authenticate_with_token()
+            return
+
+        jwt_token = jwt_path.read_text().strip()
+        logger.info("Authenticating with Vault using Kubernetes JWT (role: {})", self.jwt_role)
+
+        try:
+            self._client.auth.kubernetes.login(role=self.jwt_role, jwt=jwt_token)
+            logger.debug("Successfully authenticated with Vault using JWT")
+        except Exception as exc:
+            logger.error("Failed to authenticate with Vault using JWT: {}", exc)
+            raise
+
+    def _authenticate_with_token(self) -> None:
+        """Authenticate with Vault using token from ~/.vault-token (for local context)."""
+        token_path = Path.home() / ".vault-token"
+        if not token_path.exists():
+            raise RuntimeError(
+                f"Vault token file not found at {token_path}. Please run 'vault login' first or set VAULT_TOKEN."
+            )
+
+        token = token_path.read_text().strip()
+        self._client.token = token
+        logger.debug("Authenticated with Vault using token from {}", token_path)
+
+    def _normalize_path(self, key: str) -> str:
+        """Normalize a secret key to a full Vault path."""
+        # Combine the path prefix with the key
+        if self.path:
+            full_path = f"{self.path.rstrip('/')}/{key}"
+        else:
+            full_path = key
+
+        return full_path
+
+    def _split_key_path(self, key: str) -> tuple[str, str | None]:
+        """
+        Split a dotted key into the Vault secret path and the field within that secret.
+
+        For example, "database.password" splits into ("database", "password").
+        A key without dots like "api-key" returns ("api-key", None).
+        """
+        parts = key.split(".", 1)
+        if len(parts) == 1:
+            return parts[0], None
+        return parts[0], parts[1]
+
+    def _get_secret_data(self, secret_path: str) -> dict[str, Any]:
+        """Retrieve the data from a Vault secret at the given path."""
+        client = self._get_client()
+        full_path = self._normalize_path(secret_path)
+
+        try:
+            # For KV v2, we need to read from the /data/ path
+            response = client.secrets.kv.v2.read_secret_version(
+                path=full_path, mount_point=self.mount_point, raise_on_deleted_version=True
+            )
+            return response["data"]["data"]
+        except hvac.exceptions.InvalidPath:
+            raise KeyError(f"Secret not found at path: {full_path}")
+        except Exception as exc:
+            logger.error("Failed to read secret from Vault at path '{}': {}", full_path, exc)
+            raise
+
+    def _get_nested_value(self, data: dict[str, Any], field_path: str) -> SecretValue:
+        """Get a nested value from a dictionary using dot notation."""
+        parts = field_path.split(".")
+        value = data
+        for part in parts:
+            if not isinstance(value, dict):
+                raise KeyError(f"Cannot access field '{part}' in non-dict value")
+            if part not in value:
+                raise KeyError(f"Field '{part}' not found")
+            value = value[part]
+        return value
+
+    def load(self, force: bool = False) -> dict[str, SecretValue]:
+        """Load all secrets from Vault (not implemented - use individual get calls)."""
+        # Note: For Vault, it's not efficient to load all secrets at once
+        # as they may be spread across multiple paths. We'll populate the cache on-demand.
+        if self._cache is None or force:
+            self._cache = {}
+        return self._cache
+
+    # SecretProvider
+
+    def init(self, config_file: Path, dependencies: DependenciesProvider) -> None:
+        """Initialize the Vault provider."""
+        # No path resolution needed for Vault URLs
+        pass
+
+    def keys(self) -> Iterable[str]:
+        """
+        Return an iterator over all keys in the provider.
+
+        Note: This is not efficiently implementable for Vault without listing all secrets,
+        which may not be feasible in a multi-tenant environment. Returns cached keys only.
+        """
+        if self._cache:
+            return self._cache.keys()
+        return []
+
+    def get(self, key: str, /) -> SecretValue:
+        """
+        Retrieve a secret by key from Vault.
+
+        The key can use dot notation for nested access:
+        - "database" retrieves the entire secret at path "database"
+        - "database.password" retrieves the "password" field from the "database" secret
+        - "database.credentials.username" retrieves nested fields
+
+        Args:
+            key: The key of the secret to retrieve, with optional dot notation for nested access.
+        Returns:
+            The secret value.
+        Raises:
+            KeyError: If the key does not exist.
+        """
+        # Check cache first
+        if self._cache and key in self._cache:
+            return self._cache[key]
+
+        # Split the key into secret path and field path
+        secret_path, field_path = self._split_key_path(key)
+
+        # Get the secret data from Vault
+        data = self._get_secret_data(secret_path)
+
+        if field_path is None:
+            # Return the entire secret
+            result = data
+        else:
+            # Navigate to the nested field
+            result = self._get_nested_value(data, field_path)
+
+        # Cache the result
+        if self._cache is None:
+            self._cache = {}
+        self._cache[key] = result
+
+        return result
+
+    def set(self, key: str, value: SecretValue, /) -> None:
+        """
+        Set the value of a key in Vault.
+
+        For dot-notation keys, this will update the specific field within the secret,
+        preserving other fields. For top-level keys, this creates or replaces the entire secret.
+
+        Args:
+            key: The key of the secret to set.
+            value: The value to set.
+        Raises:
+            KeyError: If the key is invalid.
+            ValueError: If the value is invalid.
+            RuntimeError: If the key cannot be set for systematic reasons.
+        """
+        client = self._get_client()
+        secret_path, field_path = self._split_key_path(key)
+        full_path = self._normalize_path(secret_path)
+
+        if field_path is None:
+            # Setting the entire secret
+            if not isinstance(value, dict):
+                # Wrap non-dict values in a dict for KV v2
+                data = {"value": value}
+            else:
+                data = value
+        else:
+            # Setting a specific field - need to read existing data first
+            try:
+                existing_data = self._get_secret_data(secret_path)
+            except KeyError:
+                # Secret doesn't exist yet
+                existing_data = {}
+
+            # Navigate to the nested location and set the value
+            parts = field_path.split(".")
+            current = existing_data
+            for part in parts[:-1]:
+                if part not in current:
+                    current[part] = {}
+                elif not isinstance(current[part], dict):
+                    raise ValueError(f"Cannot set nested field '{field_path}' - parent is not a dict")
+                current = current[part]
+
+            current[parts[-1]] = value
+            data = existing_data
+
+        try:
+            client.secrets.kv.v2.create_or_update_secret(path=full_path, secret=data, mount_point=self.mount_point)
+            logger.info("Set secret in Vault at path '{}'", full_path)
+
+            # Update cache
+            if self._cache is None:
+                self._cache = {}
+            self._cache[key] = value
+        except Exception as exc:
+            logger.error("Failed to set secret in Vault at path '{}': {}", full_path, exc)
+            raise RuntimeError(f"Failed to set secret: {exc}") from exc
+
+    def unset(self, key: str, /) -> None:
+        """
+        Unset a secret by its key in Vault.
+
+        For dot-notation keys, this removes the specific field from the secret.
+        For top-level keys, this deletes the entire secret.
+
+        Args:
+            key: The key of the secret to unset.
+        """
+        client = self._get_client()
+        secret_path, field_path = self._split_key_path(key)
+        full_path = self._normalize_path(secret_path)
+
+        try:
+            if field_path is None:
+                # Delete the entire secret
+                client.secrets.kv.v2.delete_metadata_and_all_versions(path=full_path, mount_point=self.mount_point)
+                logger.info("Deleted secret from Vault at path '{}'", full_path)
+            else:
+                # Remove a specific field
+                try:
+                    existing_data = self._get_secret_data(secret_path)
+                except KeyError:
+                    logger.warning("Secret '{}' not found in Vault, nothing to unset", secret_path)
+                    return
+
+                # Navigate to the nested location and delete the value
+                parts = field_path.split(".")
+                current = existing_data
+                for part in parts[:-1]:
+                    if part not in current or not isinstance(current[part], dict):
+                        logger.warning("Field '{}' not found in secret '{}'", field_path, secret_path)
+                        return
+                    current = current[part]
+
+                if parts[-1] in current:
+                    del current[parts[-1]]
+                    # Write back the modified secret
+                    client.secrets.kv.v2.create_or_update_secret(
+                        path=full_path, secret=existing_data, mount_point=self.mount_point
+                    )
+                    logger.info("Removed field '{}' from secret at path '{}'", field_path, full_path)
+                else:
+                    logger.warning("Field '{}' not found in secret '{}'", field_path, secret_path)
+
+            # Update cache
+            if self._cache and key in self._cache:
+                del self._cache[key]
+
+        except Exception as exc:
+            logger.error("Failed to unset secret in Vault at path '{}': {}", full_path, exc)
+            raise RuntimeError(f"Failed to unset secret: {exc}") from exc
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
new file mode 100644
index 00000000..dba4a4b7
--- /dev/null
+++ b/src/nyl/secrets/vault_test.py
@@ -0,0 +1,259 @@
+import os
+import unittest.mock
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import hvac
+import pytest
+
+from nyl.secrets.vault import VaultSecretProvider
+from nyl.tools.di import DependenciesProvider
+
+
+@pytest.fixture
+def mock_vault_client():
+    """Create a mock Vault client for testing."""
+    with unittest.mock.patch("hvac.Client") as mock_client_class:
+        mock_client = unittest.mock.MagicMock()
+        mock_client_class.return_value = mock_client
+        mock_client.is_authenticated.return_value = True
+        yield mock_client
+
+
+@pytest.fixture
+def provider_with_mock(mock_vault_client):
+    """Create a VaultSecretProvider with mocked client."""
+    provider = VaultSecretProvider(url="https://vault.example.com:8200", mount_point="secret", path="myapp/")
+    provider._client = mock_vault_client
+    return provider
+
+
+def test_VaultSecretProvider_init() -> None:
+    """Test basic initialization of VaultSecretProvider."""
+    provider = VaultSecretProvider(
+        url="https://vault.example.com:8200", mount_point="secret", path="myapp/", jwt_role="my-role"
+    )
+    provider.init(config_file=Path("/tmp/nyl-secrets.yaml"), dependencies=DependenciesProvider.default())
+
+    assert provider.url == "https://vault.example.com:8200"
+    assert provider.mount_point == "secret"
+    assert provider.path == "myapp/"
+    assert provider.jwt_role == "my-role"
+
+
+def test_VaultSecretProvider_is_argocd_context() -> None:
+    """Test detection of ArgoCD context."""
+    provider = VaultSecretProvider(url="https://vault.example.com:8200")
+
+    # Without ArgoCD env vars
+    assert not provider._is_argocd_context()
+
+    # With ArgoCD env vars
+    with unittest.mock.patch.dict(os.environ, {"ARGOCD_APP_NAME": "test-app"}):
+        assert provider._is_argocd_context()
+
+
+def test_VaultSecretProvider_normalize_path() -> None:
+    """Test path normalization."""
+    provider = VaultSecretProvider(url="https://vault.example.com:8200", path="myapp/")
+    assert provider._normalize_path("database") == "myapp/database"
+
+    provider_no_prefix = VaultSecretProvider(url="https://vault.example.com:8200", path="")
+    assert provider_no_prefix._normalize_path("database") == "database"
+
+
+def test_VaultSecretProvider_split_key_path() -> None:
+    """Test splitting of dotted keys."""
+    provider = VaultSecretProvider(url="https://vault.example.com:8200")
+
+    assert provider._split_key_path("database") == ("database", None)
+    assert provider._split_key_path("database.password") == ("database", "password")
+    assert provider._split_key_path("database.credentials.username") == ("database", "credentials.username")
+
+
+def test_VaultSecretProvider_get_simple(provider_with_mock, mock_vault_client) -> None:
+    """Test getting a simple secret value."""
+    # Mock the Vault response for a simple secret
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"username": "admin", "password": "secret123"}}
+    }
+
+    result = provider_with_mock.get("database")
+    assert result == {"username": "admin", "password": "secret123"}
+
+    # Verify the call was made with correct parameters
+    mock_vault_client.secrets.kv.v2.read_secret_version.assert_called_once_with(
+        path="myapp/database", mount_point="secret", raise_on_deleted_version=True
+    )
+
+
+def test_VaultSecretProvider_get_nested(provider_with_mock, mock_vault_client) -> None:
+    """Test getting a nested secret value using dot notation."""
+    # Mock the Vault response
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"username": "admin", "password": "secret123"}}
+    }
+
+    result = provider_with_mock.get("database.password")
+    assert result == "secret123"
+
+
+def test_VaultSecretProvider_get_deeply_nested(provider_with_mock, mock_vault_client) -> None:
+    """Test getting a deeply nested secret value."""
+    # Mock the Vault response with nested structure
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"credentials": {"primary": {"username": "admin", "password": "secret123"}}}}
+    }
+
+    result = provider_with_mock.get("database.credentials.primary.username")
+    assert result == "admin"
+
+
+def test_VaultSecretProvider_get_not_found(provider_with_mock, mock_vault_client) -> None:
+    """Test handling of non-existent secrets."""
+    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
+
+    with pytest.raises(KeyError, match="Secret not found at path"):
+        provider_with_mock.get("nonexistent")
+
+
+def test_VaultSecretProvider_get_cache(provider_with_mock, mock_vault_client) -> None:
+    """Test that secrets are cached after first retrieval."""
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"username": "admin", "password": "secret123"}}
+    }
+
+    # First call
+    result1 = provider_with_mock.get("database")
+    assert result1 == {"username": "admin", "password": "secret123"}
+
+    # Second call should use cache
+    result2 = provider_with_mock.get("database")
+    assert result2 == {"username": "admin", "password": "secret123"}
+
+    # Vault should only be called once
+    assert mock_vault_client.secrets.kv.v2.read_secret_version.call_count == 1
+
+
+def test_VaultSecretProvider_set_entire_secret(provider_with_mock, mock_vault_client) -> None:
+    """Test setting an entire secret."""
+    provider_with_mock.set("database", {"username": "admin", "password": "secret123"})
+
+    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
+        path="myapp/database", secret={"username": "admin", "password": "secret123"}, mount_point="secret"
+    )
+
+
+def test_VaultSecretProvider_set_nested_field(provider_with_mock, mock_vault_client) -> None:
+    """Test setting a nested field in an existing secret."""
+    # Mock existing secret
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"username": "admin", "password": "old_password"}}
+    }
+
+    provider_with_mock.set("database.password", "new_password")
+
+    # Should have read the existing secret first
+    mock_vault_client.secrets.kv.v2.read_secret_version.assert_called_once()
+
+    # Should have written back with updated value
+    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
+        path="myapp/database", secret={"username": "admin", "password": "new_password"}, mount_point="secret"
+    )
+
+
+def test_VaultSecretProvider_set_nested_field_new_secret(provider_with_mock, mock_vault_client) -> None:
+    """Test setting a nested field when the secret doesn't exist yet."""
+    # Mock that secret doesn't exist
+    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
+
+    provider_with_mock.set("database.password", "secret123")
+
+    # Should have written a new secret
+    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
+        path="myapp/database", secret={"password": "secret123"}, mount_point="secret"
+    )
+
+
+def test_VaultSecretProvider_unset_entire_secret(provider_with_mock, mock_vault_client) -> None:
+    """Test deleting an entire secret."""
+    provider_with_mock.unset("database")
+
+    mock_vault_client.secrets.kv.v2.delete_metadata_and_all_versions.assert_called_once_with(
+        path="myapp/database", mount_point="secret"
+    )
+
+
+def test_VaultSecretProvider_unset_nested_field(provider_with_mock, mock_vault_client) -> None:
+    """Test removing a nested field from a secret."""
+    # Mock existing secret
+    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
+        "data": {"data": {"username": "admin", "password": "secret123", "api_key": "key123"}}
+    }
+
+    provider_with_mock.unset("database.password")
+
+    # Should have read the existing secret
+    mock_vault_client.secrets.kv.v2.read_secret_version.assert_called_once()
+
+    # Should have written back without the password field
+    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
+        path="myapp/database", secret={"username": "admin", "api_key": "key123"}, mount_point="secret"
+    )
+
+
+def test_VaultSecretProvider_keys_empty() -> None:
+    """Test keys method with empty cache."""
+    provider = VaultSecretProvider(url="https://vault.example.com:8200")
+    assert list(provider.keys()) == []
+
+
+def test_VaultSecretProvider_keys_with_cache(provider_with_mock, mock_vault_client) -> None:
+    """Test keys method returns cached keys."""
+    # Add some items to cache
+    provider_with_mock._cache = {"database": {"user": "admin"}, "database.password": "secret"}
+
+    keys = list(provider_with_mock.keys())
+    assert set(keys) == {"database", "database.password"}
+
+
+def test_VaultSecretProvider_authenticate_with_token() -> None:
+    """Test authentication with token file."""
+    with TemporaryDirectory() as tmpdir:
+        token_path = Path(tmpdir) / ".vault-token"
+        token_path.write_text("test-token-123\n")
+
+        with unittest.mock.patch("hvac.Client") as mock_client_class:
+            mock_client = unittest.mock.MagicMock()
+            mock_client_class.return_value = mock_client
+            mock_client.is_authenticated.return_value = True
+
+            provider = VaultSecretProvider(url="https://vault.example.com:8200")
+            provider._client = mock_client  # Set the client before calling authenticate
+
+            with unittest.mock.patch("pathlib.Path.home", return_value=Path(tmpdir)):
+                provider._authenticate_with_token()
+
+            assert mock_client.token == "test-token-123"
+
+
+def test_VaultSecretProvider_authenticate_with_jwt() -> None:
+    """Test authentication with JWT token (ArgoCD context)."""
+    with TemporaryDirectory() as tmpdir:
+        jwt_path = Path(tmpdir) / "token"
+        jwt_path.write_text("test-jwt-token\n")
+
+        with unittest.mock.patch("hvac.Client") as mock_client_class:
+            mock_client = unittest.mock.MagicMock()
+            mock_client_class.return_value = mock_client
+            mock_client.is_authenticated.return_value = True
+
+            provider = VaultSecretProvider(url="https://vault.example.com:8200", jwt_role="my-role")
+            provider._client = mock_client
+
+            with unittest.mock.patch("pathlib.Path.exists", return_value=True), unittest.mock.patch(
+                "pathlib.Path.read_text", return_value="test-jwt-token\n"
+            ):
+                provider._authenticate_with_jwt()
+
+            mock_client.auth.kubernetes.login.assert_called_once_with(role="my-role", jwt="test-jwt-token")
diff --git a/uv.lock b/uv.lock
index 69036de1..3cd155bb 100644
--- a/uv.lock
+++ b/uv.lock
@@ -323,6 +323,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/92/05/adeb6c495aec4f9d93f9e2fc29eeef6e14d452bba11d15bdb874ce1d5b10/google_auth-2.42.1-py2.py3-none-any.whl", hash = "sha256:eb73d71c91fc95dbd221a2eb87477c278a355e7367a35c0d84e6b0e5f9b4ad11", size = 222550, upload-time = "2025-10-30T16:42:17.878Z" },
 ]
 
+[[package]]
+name = "hvac"
+version = "2.4.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "requests" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/e2/57/b46c397fb3842cfb02a44609aa834c887f38dd75f290c2fc5a34da4b2fee/hvac-2.4.0.tar.gz", hash = "sha256:e0056ad9064e7923e874e6769015b032580b639e29246f5ab1044f7959c1c7e0", size = 332543, upload-time = "2025-10-30T12:57:47.512Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/55/33/71e45a6bd6875f44a26f99da31c63b6840123e88bedf2c0b1ce429b8be12/hvac-2.4.0-py3-none-any.whl", hash = "sha256:008db5efd8c2f77bd37d2368ea5f713edceae1c65f11fd608393179478649e0f", size = 155921, upload-time = "2025-10-30T12:57:46.253Z" },
+]
+
 [[package]]
 name = "idna"
 version = "3.11"
@@ -564,6 +576,7 @@ dependencies = [
     { name = "bcrypt" },
     { name = "databind" },
     { name = "filelock" },
+    { name = "hvac" },
     { name = "jinja2" },
     { name = "kubernetes" },
     { name = "loguru" },
@@ -593,6 +606,7 @@ requires-dist = [
     { name = "bcrypt", specifier = ">=4.2.0" },
     { name = "databind", specifier = ">=4.5.2" },
     { name = "filelock", specifier = ">=3.15.4" },
+    { name = "hvac", specifier = ">=2.3.0" },
     { name = "jinja2", specifier = ">=3.1.4" },
     { name = "kubernetes", specifier = ">=30.1.0" },
     { name = "loguru", specifier = ">=0.7.2" },

From cee76c676ee26cd3da7a65c7c706c33cac18ea25 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 22:43:27 +0000
Subject: [PATCH 3/7] Add documentation and fix type checking for Vault
 provider

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 README.md                                     |   1 +
 .../reference/configuration/secrets.md        |  86 ++++++++++++++
 src/nyl/secrets/vault.py                      |  64 ++++++++---
 src/nyl/secrets/vault_test.py                 | 107 ++++++++++++++----
 4 files changed, 218 insertions(+), 40 deletions(-)

diff --git a/README.md b/README.md
index 0a505a98..ac29c5e9 100644
--- a/README.md
+++ b/README.md
@@ -15,6 +15,7 @@ For some features, additional programs must be available:
 - [helm](https://helm.sh/) for rendering Helm charts
 - [kyverno](https://kyverno.io/docs/kyverno-cli/) ^1.13.x when using the Nyl `PostProcessor` resource
 - [sops](https://github.com/getsops/sops) when using the SOPS secrets provider
+- [vault](https://www.vaultproject.io/) when using the Vault secrets provider (optional - only for local development)
 
 ## Local development
 
diff --git a/docs/content/reference/configuration/secrets.md b/docs/content/reference/configuration/secrets.md
index 79855b39..52d0532b 100644
--- a/docs/content/reference/configuration/secrets.md
+++ b/docs/content/reference/configuration/secrets.md
@@ -111,6 +111,92 @@ files apply. The `path` field is relative to the location of the `nyl-secrets.ya
 
 ---
 
+## Provider: [Vault](https://www.vaultproject.io/)
+
+Allows you to retrieve secrets from a HashiCorp Vault server using the KV secrets engine v2. This provider is designed
+to work securely in multi-tenant environments and integrates with both local development and ArgoCD deployments.
+
+### Authentication
+
+The provider supports two authentication methods:
+
+- **JWT Authentication (ArgoCD context)**: When running in ArgoCD, the provider authenticates using the Kubernetes
+  service account JWT token mounted at `/var/run/secrets/kubernetes.io/serviceaccount/token`. This enables secure
+  multi-tenant deployments where each application's workload identity is used to access only its authorized secrets.
+
+- **Token Authentication (Local development)**: When running locally, the provider uses the token stored in
+  `~/.vault-token` (obtained via `vault login`).
+
+The provider automatically detects the execution context by checking for ArgoCD environment variables.
+
+### Nested Keys and Dot Notation
+
+The Vault provider supports nested structures through dot notation, similar to the SOPS provider:
+
+- `"database"` retrieves the entire secret stored at the `database` path
+- `"database.password"` retrieves the `password` field from the `database` secret
+- `"database.credentials.username"` retrieves deeply nested fields
+
+### Configuration Options
+
+- `url` (required): The URL of the Vault server (e.g., `https://vault.example.com:8200`)
+- `mount_point` (optional): The mount point of the KV secrets engine. Default: `secret`
+- `path` (optional): Path prefix within the KV secrets engine. For example, if `path` is `myapp/`, secrets will be
+  retrieved from `secret/data/myapp/...`
+- `jwt_role` (optional): The Vault role to use for JWT authentication when running in ArgoCD. Required for JWT
+  authentication.
+- `namespace` (optional): The Vault namespace to use (Vault Enterprise feature only)
+
+__Example__
+
+=== "TOML"
+
+    ```toml title="nyl-secrets.toml"
+    [default]
+    type = "vault"
+    url = "https://vault.example.com:8200"
+    mount_point = "secret"
+    path = "myapp/"
+    jwt_role = "nyl-argocd-role"
+    ```
+
+=== "YAML"
+
+    ```yaml title="nyl-secrets.yaml"
+    default:
+      type: vault
+      url: https://vault.example.com:8200
+      mount_point: secret
+      path: myapp/
+      jwt_role: nyl-argocd-role
+    ```
+
+=== "JSON"
+
+    ```json title="nyl-secrets.json"
+    {
+      "default": {
+        "type": "vault",
+        "url": "https://vault.example.com:8200",
+        "mount_point": "secret",
+        "path": "myapp/",
+        "jwt_role": "nyl-argocd-role"
+      }
+    }
+    ```
+
+### Security Considerations
+
+When using Vault in a multi-tenant ArgoCD environment:
+
+1. Configure Vault to trust the Kubernetes cluster's JWT issuer
+2. Create a Vault role for each application or team with appropriate policies
+3. Bind the Vault role to the ArgoCD application's service account
+4. Ensure the `jwt_role` in the Nyl configuration matches the Vault role
+5. Use Vault policies to restrict access to only the necessary secrets
+
+---
+
 ## Provider: [KubernetesSecret](https://kubernetes.io/docs/concepts/configuration/secret/)
 
 Allows you to point to a Kubernetes Secret as a source for secrets. Since Kubernetes secret values must be strings,
diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
index d0d95e54..68dfb7f4 100644
--- a/src/nyl/secrets/vault.py
+++ b/src/nyl/secrets/vault.py
@@ -3,7 +3,7 @@
 from pathlib import Path
 from typing import Any, Iterable
 
-import hvac
+import hvac  # type: ignore[import-untyped]
 from databind.core import Union
 from loguru import logger
 
@@ -81,15 +81,19 @@ def _authenticate_with_jwt(self) -> None:
         jwt_path = Path("/var/run/secrets/kubernetes.io/serviceaccount/token")
         if not jwt_path.exists():
             logger.warning(
-                "JWT authentication requested but token file not found at {}. Falling back to token auth.", jwt_path
+                "JWT authentication requested but token file not found at {}. Falling back to token auth.",
+                jwt_path,
             )
             self._authenticate_with_token()
             return
 
         jwt_token = jwt_path.read_text().strip()
-        logger.info("Authenticating with Vault using Kubernetes JWT (role: {})", self.jwt_role)
+        logger.info(
+            "Authenticating with Vault using Kubernetes JWT (role: {})", self.jwt_role
+        )
 
         try:
+            assert self._client is not None
             self._client.auth.kubernetes.login(role=self.jwt_role, jwt=jwt_token)
             logger.debug("Successfully authenticated with Vault using JWT")
         except Exception as exc:
@@ -105,6 +109,7 @@ def _authenticate_with_token(self) -> None:
             )
 
         token = token_path.read_text().strip()
+        assert self._client is not None
         self._client.token = token
         logger.debug("Authenticated with Vault using token from {}", token_path)
 
@@ -138,13 +143,17 @@ def _get_secret_data(self, secret_path: str) -> dict[str, Any]:
         try:
             # For KV v2, we need to read from the /data/ path
             response = client.secrets.kv.v2.read_secret_version(
-                path=full_path, mount_point=self.mount_point, raise_on_deleted_version=True
+                path=full_path,
+                mount_point=self.mount_point,
+                raise_on_deleted_version=True,
             )
             return response["data"]["data"]
         except hvac.exceptions.InvalidPath:
             raise KeyError(f"Secret not found at path: {full_path}")
         except Exception as exc:
-            logger.error("Failed to read secret from Vault at path '{}': {}", full_path, exc)
+            logger.error(
+                "Failed to read secret from Vault at path '{}': {}", full_path, exc
+            )
             raise
 
     def _get_nested_value(self, data: dict[str, Any], field_path: str) -> SecretValue:
@@ -211,6 +220,7 @@ def get(self, key: str, /) -> SecretValue:
         # Get the secret data from Vault
         data = self._get_secret_data(secret_path)
 
+        result: SecretValue
         if field_path is None:
             # Return the entire secret
             result = data
@@ -266,14 +276,18 @@ def set(self, key: str, value: SecretValue, /) -> None:
                 if part not in current:
                     current[part] = {}
                 elif not isinstance(current[part], dict):
-                    raise ValueError(f"Cannot set nested field '{field_path}' - parent is not a dict")
+                    raise ValueError(
+                        f"Cannot set nested field '{field_path}' - parent is not a dict"
+                    )
                 current = current[part]
 
             current[parts[-1]] = value
             data = existing_data
 
         try:
-            client.secrets.kv.v2.create_or_update_secret(path=full_path, secret=data, mount_point=self.mount_point)
+            client.secrets.kv.v2.create_or_update_secret(
+                path=full_path, secret=data, mount_point=self.mount_point
+            )
             logger.info("Set secret in Vault at path '{}'", full_path)
 
             # Update cache
@@ -281,7 +295,9 @@ def set(self, key: str, value: SecretValue, /) -> None:
                 self._cache = {}
             self._cache[key] = value
         except Exception as exc:
-            logger.error("Failed to set secret in Vault at path '{}': {}", full_path, exc)
+            logger.error(
+                "Failed to set secret in Vault at path '{}': {}", full_path, exc
+            )
             raise RuntimeError(f"Failed to set secret: {exc}") from exc
 
     def unset(self, key: str, /) -> None:
@@ -301,14 +317,18 @@ def unset(self, key: str, /) -> None:
         try:
             if field_path is None:
                 # Delete the entire secret
-                client.secrets.kv.v2.delete_metadata_and_all_versions(path=full_path, mount_point=self.mount_point)
+                client.secrets.kv.v2.delete_metadata_and_all_versions(
+                    path=full_path, mount_point=self.mount_point
+                )
                 logger.info("Deleted secret from Vault at path '{}'", full_path)
             else:
                 # Remove a specific field
                 try:
                     existing_data = self._get_secret_data(secret_path)
                 except KeyError:
-                    logger.warning("Secret '{}' not found in Vault, nothing to unset", secret_path)
+                    logger.warning(
+                        "Secret '{}' not found in Vault, nothing to unset", secret_path
+                    )
                     return
 
                 # Navigate to the nested location and delete the value
@@ -316,7 +336,11 @@ def unset(self, key: str, /) -> None:
                 current = existing_data
                 for part in parts[:-1]:
                     if part not in current or not isinstance(current[part], dict):
-                        logger.warning("Field '{}' not found in secret '{}'", field_path, secret_path)
+                        logger.warning(
+                            "Field '{}' not found in secret '{}'",
+                            field_path,
+                            secret_path,
+                        )
                         return
                     current = current[part]
 
@@ -324,16 +348,26 @@ def unset(self, key: str, /) -> None:
                     del current[parts[-1]]
                     # Write back the modified secret
                     client.secrets.kv.v2.create_or_update_secret(
-                        path=full_path, secret=existing_data, mount_point=self.mount_point
+                        path=full_path,
+                        secret=existing_data,
+                        mount_point=self.mount_point,
+                    )
+                    logger.info(
+                        "Removed field '{}' from secret at path '{}'",
+                        field_path,
+                        full_path,
                     )
-                    logger.info("Removed field '{}' from secret at path '{}'", field_path, full_path)
                 else:
-                    logger.warning("Field '{}' not found in secret '{}'", field_path, secret_path)
+                    logger.warning(
+                        "Field '{}' not found in secret '{}'", field_path, secret_path
+                    )
 
             # Update cache
             if self._cache and key in self._cache:
                 del self._cache[key]
 
         except Exception as exc:
-            logger.error("Failed to unset secret in Vault at path '{}': {}", full_path, exc)
+            logger.error(
+                "Failed to unset secret in Vault at path '{}': {}", full_path, exc
+            )
             raise RuntimeError(f"Failed to unset secret: {exc}") from exc
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
index dba4a4b7..29224b84 100644
--- a/src/nyl/secrets/vault_test.py
+++ b/src/nyl/secrets/vault_test.py
@@ -23,7 +23,9 @@ def mock_vault_client():
 @pytest.fixture
 def provider_with_mock(mock_vault_client):
     """Create a VaultSecretProvider with mocked client."""
-    provider = VaultSecretProvider(url="https://vault.example.com:8200", mount_point="secret", path="myapp/")
+    provider = VaultSecretProvider(
+        url="https://vault.example.com:8200", mount_point="secret", path="myapp/"
+    )
     provider._client = mock_vault_client
     return provider
 
@@ -31,9 +33,15 @@ def provider_with_mock(mock_vault_client):
 def test_VaultSecretProvider_init() -> None:
     """Test basic initialization of VaultSecretProvider."""
     provider = VaultSecretProvider(
-        url="https://vault.example.com:8200", mount_point="secret", path="myapp/", jwt_role="my-role"
+        url="https://vault.example.com:8200",
+        mount_point="secret",
+        path="myapp/",
+        jwt_role="my-role",
+    )
+    provider.init(
+        config_file=Path("/tmp/nyl-secrets.yaml"),
+        dependencies=DependenciesProvider.default(),
     )
-    provider.init(config_file=Path("/tmp/nyl-secrets.yaml"), dependencies=DependenciesProvider.default())
 
     assert provider.url == "https://vault.example.com:8200"
     assert provider.mount_point == "secret"
@@ -58,7 +66,9 @@ def test_VaultSecretProvider_normalize_path() -> None:
     provider = VaultSecretProvider(url="https://vault.example.com:8200", path="myapp/")
     assert provider._normalize_path("database") == "myapp/database"
 
-    provider_no_prefix = VaultSecretProvider(url="https://vault.example.com:8200", path="")
+    provider_no_prefix = VaultSecretProvider(
+        url="https://vault.example.com:8200", path=""
+    )
     assert provider_no_prefix._normalize_path("database") == "database"
 
 
@@ -68,7 +78,10 @@ def test_VaultSecretProvider_split_key_path() -> None:
 
     assert provider._split_key_path("database") == ("database", None)
     assert provider._split_key_path("database.password") == ("database", "password")
-    assert provider._split_key_path("database.credentials.username") == ("database", "credentials.username")
+    assert provider._split_key_path("database.credentials.username") == (
+        "database",
+        "credentials.username",
+    )
 
 
 def test_VaultSecretProvider_get_simple(provider_with_mock, mock_vault_client) -> None:
@@ -98,20 +111,32 @@ def test_VaultSecretProvider_get_nested(provider_with_mock, mock_vault_client) -
     assert result == "secret123"
 
 
-def test_VaultSecretProvider_get_deeply_nested(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_get_deeply_nested(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test getting a deeply nested secret value."""
     # Mock the Vault response with nested structure
     mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
-        "data": {"data": {"credentials": {"primary": {"username": "admin", "password": "secret123"}}}}
+        "data": {
+            "data": {
+                "credentials": {
+                    "primary": {"username": "admin", "password": "secret123"}
+                }
+            }
+        }
     }
 
     result = provider_with_mock.get("database.credentials.primary.username")
     assert result == "admin"
 
 
-def test_VaultSecretProvider_get_not_found(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_get_not_found(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test handling of non-existent secrets."""
-    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
+    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = (
+        hvac.exceptions.InvalidPath()
+    )
 
     with pytest.raises(KeyError, match="Secret not found at path"):
         provider_with_mock.get("nonexistent")
@@ -135,16 +160,22 @@ def test_VaultSecretProvider_get_cache(provider_with_mock, mock_vault_client) ->
     assert mock_vault_client.secrets.kv.v2.read_secret_version.call_count == 1
 
 
-def test_VaultSecretProvider_set_entire_secret(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_set_entire_secret(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test setting an entire secret."""
     provider_with_mock.set("database", {"username": "admin", "password": "secret123"})
 
     mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database", secret={"username": "admin", "password": "secret123"}, mount_point="secret"
+        path="myapp/database",
+        secret={"username": "admin", "password": "secret123"},
+        mount_point="secret",
     )
 
 
-def test_VaultSecretProvider_set_nested_field(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_set_nested_field(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test setting a nested field in an existing secret."""
     # Mock existing secret
     mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
@@ -158,14 +189,20 @@ def test_VaultSecretProvider_set_nested_field(provider_with_mock, mock_vault_cli
 
     # Should have written back with updated value
     mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database", secret={"username": "admin", "password": "new_password"}, mount_point="secret"
+        path="myapp/database",
+        secret={"username": "admin", "password": "new_password"},
+        mount_point="secret",
     )
 
 
-def test_VaultSecretProvider_set_nested_field_new_secret(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_set_nested_field_new_secret(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test setting a nested field when the secret doesn't exist yet."""
     # Mock that secret doesn't exist
-    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = hvac.exceptions.InvalidPath()
+    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = (
+        hvac.exceptions.InvalidPath()
+    )
 
     provider_with_mock.set("database.password", "secret123")
 
@@ -175,7 +212,9 @@ def test_VaultSecretProvider_set_nested_field_new_secret(provider_with_mock, moc
     )
 
 
-def test_VaultSecretProvider_unset_entire_secret(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_unset_entire_secret(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test deleting an entire secret."""
     provider_with_mock.unset("database")
 
@@ -184,11 +223,15 @@ def test_VaultSecretProvider_unset_entire_secret(provider_with_mock, mock_vault_
     )
 
 
-def test_VaultSecretProvider_unset_nested_field(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_unset_nested_field(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test removing a nested field from a secret."""
     # Mock existing secret
     mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
-        "data": {"data": {"username": "admin", "password": "secret123", "api_key": "key123"}}
+        "data": {
+            "data": {"username": "admin", "password": "secret123", "api_key": "key123"}
+        }
     }
 
     provider_with_mock.unset("database.password")
@@ -198,7 +241,9 @@ def test_VaultSecretProvider_unset_nested_field(provider_with_mock, mock_vault_c
 
     # Should have written back without the password field
     mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database", secret={"username": "admin", "api_key": "key123"}, mount_point="secret"
+        path="myapp/database",
+        secret={"username": "admin", "api_key": "key123"},
+        mount_point="secret",
     )
 
 
@@ -208,10 +253,15 @@ def test_VaultSecretProvider_keys_empty() -> None:
     assert list(provider.keys()) == []
 
 
-def test_VaultSecretProvider_keys_with_cache(provider_with_mock, mock_vault_client) -> None:
+def test_VaultSecretProvider_keys_with_cache(
+    provider_with_mock, mock_vault_client
+) -> None:
     """Test keys method returns cached keys."""
     # Add some items to cache
-    provider_with_mock._cache = {"database": {"user": "admin"}, "database.password": "secret"}
+    provider_with_mock._cache = {
+        "database": {"user": "admin"},
+        "database.password": "secret",
+    }
 
     keys = list(provider_with_mock.keys())
     assert set(keys) == {"database", "database.password"}
@@ -248,12 +298,19 @@ def test_VaultSecretProvider_authenticate_with_jwt() -> None:
             mock_client_class.return_value = mock_client
             mock_client.is_authenticated.return_value = True
 
-            provider = VaultSecretProvider(url="https://vault.example.com:8200", jwt_role="my-role")
+            provider = VaultSecretProvider(
+                url="https://vault.example.com:8200", jwt_role="my-role"
+            )
             provider._client = mock_client
 
-            with unittest.mock.patch("pathlib.Path.exists", return_value=True), unittest.mock.patch(
-                "pathlib.Path.read_text", return_value="test-jwt-token\n"
+            with (
+                unittest.mock.patch("pathlib.Path.exists", return_value=True),
+                unittest.mock.patch(
+                    "pathlib.Path.read_text", return_value="test-jwt-token\n"
+                ),
             ):
                 provider._authenticate_with_jwt()
 
-            mock_client.auth.kubernetes.login.assert_called_once_with(role="my-role", jwt="test-jwt-token")
+            mock_client.auth.kubernetes.login.assert_called_once_with(
+                role="my-role", jwt="test-jwt-token"
+            )

From c580ced448ec7e2fd5ad117a23c2cf34fa7ae029 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 22:47:46 +0000
Subject: [PATCH 4/7] Address code review feedback: add VAULT_TOKEN support and
 improve consistency

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 .../reference/configuration/secrets.md        |  8 +++-
 src/nyl/secrets/vault.py                      | 41 ++++++++++++-------
 src/nyl/secrets/vault_test.py                 | 27 ++++++++++++
 3 files changed, 60 insertions(+), 16 deletions(-)

diff --git a/docs/content/reference/configuration/secrets.md b/docs/content/reference/configuration/secrets.md
index 52d0532b..9d0595c7 100644
--- a/docs/content/reference/configuration/secrets.md
+++ b/docs/content/reference/configuration/secrets.md
@@ -133,10 +133,13 @@ The provider automatically detects the execution context by checking for ArgoCD
 
 The Vault provider supports nested structures through dot notation, similar to the SOPS provider:
 
-- `"database"` retrieves the entire secret stored at the `database` path
+- `"database"` retrieves the entire secret stored at the `database` path (returns a dict)
 - `"database.password"` retrieves the `password` field from the `database` secret
 - `"database.credentials.username"` retrieves deeply nested fields
 
+**Note**: Top-level secrets in Vault KV v2 must be dictionaries. To store simple values like strings or numbers, 
+use dot notation. For example, to store an API key, use `"api-key.value"` instead of `"api-key"`.
+
 ### Configuration Options
 
 - `url` (required): The URL of the Vault server (e.g., `https://vault.example.com:8200`)
@@ -147,6 +150,9 @@ The Vault provider supports nested structures through dot notation, similar to t
   authentication.
 - `namespace` (optional): The Vault namespace to use (Vault Enterprise feature only)
 
+The provider supports token authentication via the `VAULT_TOKEN` environment variable or `~/.vault-token` file
+(obtained via `vault login`).
+
 __Example__
 
 === "TOML"
diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
index 68dfb7f4..32490c27 100644
--- a/src/nyl/secrets/vault.py
+++ b/src/nyl/secrets/vault.py
@@ -101,17 +101,23 @@ def _authenticate_with_jwt(self) -> None:
             raise
 
     def _authenticate_with_token(self) -> None:
-        """Authenticate with Vault using token from ~/.vault-token (for local context)."""
-        token_path = Path.home() / ".vault-token"
-        if not token_path.exists():
-            raise RuntimeError(
-                f"Vault token file not found at {token_path}. Please run 'vault login' first or set VAULT_TOKEN."
-            )
+        """Authenticate with Vault using token from ~/.vault-token or VAULT_TOKEN env var (for local context)."""
+        # First check VAULT_TOKEN environment variable
+        token = os.environ.get("VAULT_TOKEN")
+        if token:
+            logger.debug("Using Vault token from VAULT_TOKEN environment variable")
+        else:
+            # Fall back to token file
+            token_path = Path.home() / ".vault-token"
+            if not token_path.exists():
+                raise RuntimeError(
+                    f"Vault token file not found at {token_path}. Please run 'vault login' first or set VAULT_TOKEN environment variable."
+                )
+            token = token_path.read_text().strip()
+            logger.debug("Authenticated with Vault using token from {}", token_path)
 
-        token = token_path.read_text().strip()
         assert self._client is not None
         self._client.token = token
-        logger.debug("Authenticated with Vault using token from {}", token_path)
 
     def _normalize_path(self, key: str) -> str:
         """Normalize a secret key to a full Vault path."""
@@ -242,12 +248,15 @@ def set(self, key: str, value: SecretValue, /) -> None:
         For dot-notation keys, this will update the specific field within the secret,
         preserving other fields. For top-level keys, this creates or replaces the entire secret.
 
+        Note: Top-level secrets in Vault KV v2 must be dictionaries. If you need to store a simple
+        value like a string or number, use dot notation (e.g., "api-key.value" instead of "api-key").
+
         Args:
             key: The key of the secret to set.
-            value: The value to set.
+            value: The value to set. For top-level keys, this must be a dict.
         Raises:
             KeyError: If the key is invalid.
-            ValueError: If the value is invalid.
+            ValueError: If the value is invalid (e.g., non-dict for top-level secret).
             RuntimeError: If the key cannot be set for systematic reasons.
         """
         client = self._get_client()
@@ -255,12 +264,14 @@ def set(self, key: str, value: SecretValue, /) -> None:
         full_path = self._normalize_path(secret_path)
 
         if field_path is None:
-            # Setting the entire secret
+            # Setting the entire secret - must be a dict
             if not isinstance(value, dict):
-                # Wrap non-dict values in a dict for KV v2
-                data = {"value": value}
-            else:
-                data = value
+                raise ValueError(
+                    f"Top-level secrets in Vault must be dictionaries. "
+                    f"To store a simple value, use dot notation (e.g., '{key}.value'). "
+                    f"Got {type(value).__name__}: {value!r}"
+                )
+            data = value
         else:
             # Setting a specific field - need to read existing data first
             try:
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
index 29224b84..4d716fa1 100644
--- a/src/nyl/secrets/vault_test.py
+++ b/src/nyl/secrets/vault_test.py
@@ -173,6 +173,17 @@ def test_VaultSecretProvider_set_entire_secret(
     )
 
 
+def test_VaultSecretProvider_set_entire_secret_non_dict_raises(
+    provider_with_mock, mock_vault_client
+) -> None:
+    """Test that setting a top-level secret with a non-dict value raises an error."""
+    with pytest.raises(
+        ValueError,
+        match="Top-level secrets in Vault must be dictionaries",
+    ):
+        provider_with_mock.set("api-key", "secret123")
+
+
 def test_VaultSecretProvider_set_nested_field(
     provider_with_mock, mock_vault_client
 ) -> None:
@@ -287,6 +298,22 @@ def test_VaultSecretProvider_authenticate_with_token() -> None:
             assert mock_client.token == "test-token-123"
 
 
+def test_VaultSecretProvider_authenticate_with_env_token() -> None:
+    """Test authentication with VAULT_TOKEN environment variable."""
+    with unittest.mock.patch("hvac.Client") as mock_client_class:
+        mock_client = unittest.mock.MagicMock()
+        mock_client_class.return_value = mock_client
+        mock_client.is_authenticated.return_value = True
+
+        provider = VaultSecretProvider(url="https://vault.example.com:8200")
+        provider._client = mock_client
+
+        with unittest.mock.patch.dict(os.environ, {"VAULT_TOKEN": "env-token-456"}):
+            provider._authenticate_with_token()
+
+        assert mock_client.token == "env-token-456"
+
+
 def test_VaultSecretProvider_authenticate_with_jwt() -> None:
     """Test authentication with JWT token (ArgoCD context)."""
     with TemporaryDirectory() as tmpdir:

From 3a21f7b0063a6da9c72c4855358f076e207e2ec6 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 22:52:35 +0000
Subject: [PATCH 5/7] Security improvements: reduce logging sensitivity and
 improve test mocking

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 src/nyl/secrets/vault.py      | 20 ++++++++--------
 src/nyl/secrets/vault_test.py | 43 ++++++++++++++++-------------------
 2 files changed, 29 insertions(+), 34 deletions(-)

diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
index 32490c27..dfe2d67c 100644
--- a/src/nyl/secrets/vault.py
+++ b/src/nyl/secrets/vault.py
@@ -299,16 +299,14 @@ def set(self, key: str, value: SecretValue, /) -> None:
             client.secrets.kv.v2.create_or_update_secret(
                 path=full_path, secret=data, mount_point=self.mount_point
             )
-            logger.info("Set secret in Vault at path '{}'", full_path)
+            logger.debug("Set secret in Vault for key '{}'", key)
 
             # Update cache
             if self._cache is None:
                 self._cache = {}
             self._cache[key] = value
         except Exception as exc:
-            logger.error(
-                "Failed to set secret in Vault at path '{}': {}", full_path, exc
-            )
+            logger.error("Failed to set secret in Vault for key '{}': {}", key, exc)
             raise RuntimeError(f"Failed to set secret: {exc}") from exc
 
     def unset(self, key: str, /) -> None:
@@ -331,14 +329,14 @@ def unset(self, key: str, /) -> None:
                 client.secrets.kv.v2.delete_metadata_and_all_versions(
                     path=full_path, mount_point=self.mount_point
                 )
-                logger.info("Deleted secret from Vault at path '{}'", full_path)
+                logger.debug("Deleted secret from Vault for key '{}'", key)
             else:
                 # Remove a specific field
                 try:
                     existing_data = self._get_secret_data(secret_path)
                 except KeyError:
                     logger.warning(
-                        "Secret '{}' not found in Vault, nothing to unset", secret_path
+                        "Secret '{}' not found in Vault, nothing to unset", key
                     )
                     return
 
@@ -350,7 +348,7 @@ def unset(self, key: str, /) -> None:
                         logger.warning(
                             "Field '{}' not found in secret '{}'",
                             field_path,
-                            secret_path,
+                            key,
                         )
                         return
                     current = current[part]
@@ -363,14 +361,14 @@ def unset(self, key: str, /) -> None:
                         secret=existing_data,
                         mount_point=self.mount_point,
                     )
-                    logger.info(
-                        "Removed field '{}' from secret at path '{}'",
+                    logger.debug(
+                        "Removed field '{}' from secret for key '{}'",
                         field_path,
-                        full_path,
+                        key,
                     )
                 else:
                     logger.warning(
-                        "Field '{}' not found in secret '{}'", field_path, secret_path
+                        "Field '{}' not found in secret '{}'", field_path, key
                     )
 
             # Update cache
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
index 4d716fa1..a5b1ecd0 100644
--- a/src/nyl/secrets/vault_test.py
+++ b/src/nyl/secrets/vault_test.py
@@ -316,28 +316,25 @@ def test_VaultSecretProvider_authenticate_with_env_token() -> None:
 
 def test_VaultSecretProvider_authenticate_with_jwt() -> None:
     """Test authentication with JWT token (ArgoCD context)."""
-    with TemporaryDirectory() as tmpdir:
-        jwt_path = Path(tmpdir) / "token"
-        jwt_path.write_text("test-jwt-token\n")
+    with unittest.mock.patch("hvac.Client") as mock_client_class:
+        mock_client = unittest.mock.MagicMock()
+        mock_client_class.return_value = mock_client
+        mock_client.is_authenticated.return_value = True
 
-        with unittest.mock.patch("hvac.Client") as mock_client_class:
-            mock_client = unittest.mock.MagicMock()
-            mock_client_class.return_value = mock_client
-            mock_client.is_authenticated.return_value = True
+        provider = VaultSecretProvider(
+            url="https://vault.example.com:8200", jwt_role="my-role"
+        )
+        provider._client = mock_client
+
+        # Mock the JWT token path directly in the module
+        with unittest.mock.patch("nyl.secrets.vault.Path") as mock_path_class:
+            mock_jwt_path = unittest.mock.MagicMock()
+            mock_jwt_path.exists.return_value = True
+            mock_jwt_path.read_text.return_value = "test-jwt-token\n"
+            mock_path_class.return_value = mock_jwt_path
+
+            provider._authenticate_with_jwt()
 
-            provider = VaultSecretProvider(
-                url="https://vault.example.com:8200", jwt_role="my-role"
-            )
-            provider._client = mock_client
-
-            with (
-                unittest.mock.patch("pathlib.Path.exists", return_value=True),
-                unittest.mock.patch(
-                    "pathlib.Path.read_text", return_value="test-jwt-token\n"
-                ),
-            ):
-                provider._authenticate_with_jwt()
-
-            mock_client.auth.kubernetes.login.assert_called_once_with(
-                role="my-role", jwt="test-jwt-token"
-            )
+        mock_client.auth.kubernetes.login.assert_called_once_with(
+            role="my-role", jwt="test-jwt-token"
+        )

From c2eea33ab2a7c7ca733c0deed02d9bb869789c26 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Sun, 23 Nov 2025 23:20:31 +0000
Subject: [PATCH 6/7] Address PR feedback: use hash separator, remove
 set/unset, add Nyl JWT auth

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 .../reference/configuration/secrets.md        |  54 +++-
 src/nyl/secrets/vault.py                      | 258 ++++++++----------
 src/nyl/secrets/vault_test.py                 | 184 ++++++-------
 3 files changed, 246 insertions(+), 250 deletions(-)

diff --git a/docs/content/reference/configuration/secrets.md b/docs/content/reference/configuration/secrets.md
index 9d0595c7..f7fa98d3 100644
--- a/docs/content/reference/configuration/secrets.md
+++ b/docs/content/reference/configuration/secrets.md
@@ -129,16 +129,17 @@ The provider supports two authentication methods:
 
 The provider automatically detects the execution context by checking for ArgoCD environment variables.
 
-### Nested Keys and Dot Notation
+### Key Format with Hash Separator
 
-The Vault provider supports nested structures through dot notation, similar to the SOPS provider:
+The Vault provider uses a hash (`#`) separator to distinguish between the secret path and field access:
 
 - `"database"` retrieves the entire secret stored at the `database` path (returns a dict)
-- `"database.password"` retrieves the `password` field from the `database` secret
-- `"database.credentials.username"` retrieves deeply nested fields
+- `"database#password"` retrieves the `password` field from the `database` secret
+- `"database#credentials.username"` retrieves nested fields using dot notation
+- `"db.prod#password"` retrieves the `password` field from a secret at path `db.prod`
 
-**Note**: Top-level secrets in Vault KV v2 must be dictionaries. To store simple values like strings or numbers, 
-use dot notation. For example, to store an API key, use `"api-key.value"` instead of `"api-key"`.
+**Why use hash separator?** This allows you to access secrets with dots in their path names, which was not possible 
+with pure dot notation. For example, a secret at path `my-app.v2` can be accessed with `"my-app.v2#fieldname"`.
 
 ### Configuration Options
 
@@ -148,11 +149,17 @@ use dot notation. For example, to store an API key, use `"api-key.value"` instea
   retrieved from `secret/data/myapp/...`
 - `jwt_role` (optional): The Vault role to use for JWT authentication when running in ArgoCD. Required for JWT
   authentication.
+- `jwt_auth_method` (optional): JWT authentication method to use. Options:
+  - `"kubernetes"` (default): Use Kubernetes service account token (simple, single-tenant)
+  - `"nyl"`: Use Nyl-issued ArgoCD application-specific token (multi-tenant, requires NYL_VAULT_JWT env var)
 - `namespace` (optional): The Vault namespace to use (Vault Enterprise feature only)
 
 The provider supports token authentication via the `VAULT_TOKEN` environment variable or `~/.vault-token` file
 (obtained via `vault login`).
 
+**Note**: The Vault provider is read-only. Setting or unsetting secrets is not supported - manage Vault secrets 
+through Vault's own interface or CLI tools.
+
 __Example__
 
 === "TOML"
@@ -164,6 +171,7 @@ __Example__
     mount_point = "secret"
     path = "myapp/"
     jwt_role = "nyl-argocd-role"
+    jwt_auth_method = "kubernetes"  # or "nyl" for multi-tenant
     ```
 
 === "YAML"
@@ -175,6 +183,7 @@ __Example__
       mount_point: secret
       path: myapp/
       jwt_role: nyl-argocd-role
+      jwt_auth_method: kubernetes  # or "nyl" for multi-tenant
     ```
 
 === "JSON"
@@ -186,21 +195,52 @@ __Example__
         "url": "https://vault.example.com:8200",
         "mount_point": "secret",
         "path": "myapp/",
-        "jwt_role": "nyl-argocd-role"
+        "jwt_role": "nyl-argocd-role",
+        "jwt_auth_method": "kubernetes"
       }
     }
     ```
 
+### Usage Examples
+
+```yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: db-credentials
+stringData:
+  # Access entire secret
+  config: ${{ secrets.get("database") | to_json }}
+  # Access specific field
+  password: ${{ secrets.get("database#password") }}
+  # Access nested field
+  username: ${{ secrets.get("database#credentials.username") }}
+  # Access secret with dots in path
+  api-key: ${{ secrets.get("my-app.v2#api_key") }}
+```
+
 ### Security Considerations
 
 When using Vault in a multi-tenant ArgoCD environment:
 
+#### For Kubernetes JWT Authentication (simple, single-tenant)
 1. Configure Vault to trust the Kubernetes cluster's JWT issuer
 2. Create a Vault role for each application or team with appropriate policies
 3. Bind the Vault role to the ArgoCD application's service account
 4. Ensure the `jwt_role` in the Nyl configuration matches the Vault role
 5. Use Vault policies to restrict access to only the necessary secrets
 
+#### For Nyl JWT Authentication (multi-tenant)
+1. Configure Vault to trust Nyl as a JWT issuer
+2. Create Vault roles that map to ArgoCD project and application identities
+3. Nyl will issue JWTs with claims including:
+   - `argocd_project`: The ArgoCD project name
+   - `argocd_app`: The ArgoCD application name
+   - `repository`: The Git repository URL
+4. Use Vault policies to restrict access based on these claims
+5. Set `jwt_auth_method: "nyl"` in the provider configuration
+6. The NYL_VAULT_JWT environment variable will be populated by Nyl with the appropriate token
+
 ---
 
 ## Provider: [KubernetesSecret](https://kubernetes.io/docs/concepts/configuration/secret/)
diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
index dfe2d67c..30c1ad64 100644
--- a/src/nyl/secrets/vault.py
+++ b/src/nyl/secrets/vault.py
@@ -17,10 +17,17 @@ class VaultSecretProvider(SecretProvider):
     """
     This secrets provider retrieves secrets from HashiCorp Vault using the KV secrets engine.
 
-    When running in ArgoCD, it authenticates using the Kubernetes service account JWT token.
-    When running locally, it uses the token from `~/.vault-token` (obtained via `vault login`).
+    Authentication methods:
+    - JWT (ArgoCD): Supports both Kubernetes service account tokens and Nyl-issued
+      ArgoCD-application-specific workload identity tokens
+    - Token (local): Uses token from VAULT_TOKEN env var or ~/.vault-token file
 
-    The provider supports nested structures through dot notation, similar to the SOPS provider.
+    Key format: Use # to separate secret path from field access with dot notation:
+    - "database" -> entire secret at path "database"
+    - "database#password" -> "password" field from "database" secret
+    - "db/prod#credentials.username" -> nested field from "db/prod" secret
+
+    This allows secrets with dots in their path names to be accessed correctly.
     """
 
     url: str
@@ -45,6 +52,14 @@ class VaultSecretProvider(SecretProvider):
     If not specified, JWT authentication will not be attempted.
     """
 
+    jwt_auth_method: str = "kubernetes"
+    """
+    The JWT authentication method to use. Options:
+    - "kubernetes": Use Kubernetes service account token (simple, single-tenant)
+    - "nyl": Use Nyl-issued ArgoCD application-specific token (multi-tenant)
+    Defaults to "kubernetes".
+    """
+
     namespace: str | None = None
     """
     The Vault namespace to use (Vault Enterprise feature).
@@ -77,7 +92,20 @@ def _is_argocd_context(self) -> bool:
         return "ARGOCD_APP_NAME" in os.environ
 
     def _authenticate_with_jwt(self) -> None:
-        """Authenticate with Vault using Kubernetes JWT token (for ArgoCD context)."""
+        """
+        Authenticate with Vault using JWT token (for ArgoCD context).
+
+        Supports two authentication methods:
+        1. Kubernetes service account token (simple, single-tenant)
+        2. Nyl-issued ArgoCD application-specific token (multi-tenant)
+        """
+        if self.jwt_auth_method == "nyl":
+            self._authenticate_with_nyl_jwt()
+        else:
+            self._authenticate_with_kubernetes_jwt()
+
+    def _authenticate_with_kubernetes_jwt(self) -> None:
+        """Authenticate with Vault using Kubernetes service account JWT token."""
         jwt_path = Path("/var/run/secrets/kubernetes.io/serviceaccount/token")
         if not jwt_path.exists():
             logger.warning(
@@ -89,15 +117,59 @@ def _authenticate_with_jwt(self) -> None:
 
         jwt_token = jwt_path.read_text().strip()
         logger.info(
-            "Authenticating with Vault using Kubernetes JWT (role: {})", self.jwt_role
+            "Authenticating with Vault using Kubernetes JWT (role: {})",
+            self.jwt_role,
         )
 
         try:
             assert self._client is not None
             self._client.auth.kubernetes.login(role=self.jwt_role, jwt=jwt_token)
-            logger.debug("Successfully authenticated with Vault using JWT")
+            logger.debug("Successfully authenticated with Vault using Kubernetes JWT")
         except Exception as exc:
-            logger.error("Failed to authenticate with Vault using JWT: {}", exc)
+            logger.error(
+                "Failed to authenticate with Vault using Kubernetes JWT: {}", exc
+            )
+            raise
+
+    def _authenticate_with_nyl_jwt(self) -> None:
+        """
+        Authenticate with Vault using Nyl-issued ArgoCD application-specific JWT token.
+
+        This token is issued by Nyl and contains claims about the ArgoCD application
+        being deployed, enabling multi-tenant secure access to Vault secrets.
+
+        The token payload includes:
+        - iss: Issuer (e.g., "https://my-argocd.example.com/#nyl-v1")
+        - aud: Audience (Vault URL)
+        - sub: Subject (e.g., "project:default:application:my-argo-app")
+        - argocd_project: ArgoCD project name
+        - argocd_app: ArgoCD application name
+        - repository: Git repository URL
+        """
+        # Check for Nyl-issued JWT token in environment variable
+        jwt_token = os.environ.get("NYL_VAULT_JWT")
+        if not jwt_token:
+            logger.error(
+                "Nyl JWT authentication requested but NYL_VAULT_JWT environment variable not set. "
+                "This token should be issued by Nyl based on the ArgoCD application context."
+            )
+            raise RuntimeError(
+                "NYL_VAULT_JWT environment variable not set for Nyl JWT authentication"
+            )
+
+        logger.info(
+            "Authenticating with Vault using Nyl-issued JWT (role: {})", self.jwt_role
+        )
+
+        try:
+            assert self._client is not None
+            # Use JWT auth method (not kubernetes auth)
+            self._client.auth.jwt.login(role=self.jwt_role, jwt=jwt_token)
+            logger.debug("Successfully authenticated with Vault using Nyl-issued JWT")
+        except Exception as exc:
+            logger.error(
+                "Failed to authenticate with Vault using Nyl-issued JWT: {}", exc
+            )
             raise
 
     def _authenticate_with_token(self) -> None:
@@ -131,15 +203,21 @@ def _normalize_path(self, key: str) -> str:
 
     def _split_key_path(self, key: str) -> tuple[str, str | None]:
         """
-        Split a dotted key into the Vault secret path and the field within that secret.
+        Split a key into the Vault secret path and the field within that secret.
+
+        The key format is: "path/to/secret#field.nested.path"
+        - Everything before the hash (#) is the secret path in Vault
+        - Everything after the hash is the field path using dot notation for nested access
 
-        For example, "database.password" splits into ("database", "password").
-        A key without dots like "api-key" returns ("api-key", None).
+        For example:
+        - "database#password" -> ("database", "password")
+        - "path/to/secret#credentials.username" -> ("path/to/secret", "credentials.username")
+        - "api-key" -> ("api-key", None) - returns entire secret
         """
-        parts = key.split(".", 1)
-        if len(parts) == 1:
-            return parts[0], None
-        return parts[0], parts[1]
+        if "#" in key:
+            parts = key.split("#", 1)
+            return parts[0], parts[1]
+        return key, None
 
     def _get_secret_data(self, secret_path: str) -> dict[str, Any]:
         """Retrieve the data from a Vault secret at the given path."""
@@ -204,13 +282,17 @@ def get(self, key: str, /) -> SecretValue:
         """
         Retrieve a secret by key from Vault.
 
-        The key can use dot notation for nested access:
+        The key format is "path/to/secret#field.path" where:
+        - Everything before # is the Vault secret path
+        - Everything after # is the field path using dot notation for nested access
+
+        Examples:
         - "database" retrieves the entire secret at path "database"
-        - "database.password" retrieves the "password" field from the "database" secret
-        - "database.credentials.username" retrieves nested fields
+        - "database#password" retrieves the "password" field from the "database" secret
+        - "db/prod#credentials.username" retrieves nested field from "db/prod" secret
 
         Args:
-            key: The key of the secret to retrieve, with optional dot notation for nested access.
+            key: The key of the secret to retrieve. Use # to separate path from field access.
         Returns:
             The secret value.
         Raises:
@@ -245,138 +327,28 @@ def set(self, key: str, value: SecretValue, /) -> None:
         """
         Set the value of a key in Vault.
 
-        For dot-notation keys, this will update the specific field within the secret,
-        preserving other fields. For top-level keys, this creates or replaces the entire secret.
+        This operation is not supported for the Vault provider. Vault secrets should be
+        managed through Vault's own interface or CLI tools.
 
-        Note: Top-level secrets in Vault KV v2 must be dictionaries. If you need to store a simple
-        value like a string or number, use dot notation (e.g., "api-key.value" instead of "api-key").
-
-        Args:
-            key: The key of the secret to set.
-            value: The value to set. For top-level keys, this must be a dict.
         Raises:
-            KeyError: If the key is invalid.
-            ValueError: If the value is invalid (e.g., non-dict for top-level secret).
-            RuntimeError: If the key cannot be set for systematic reasons.
+            NotImplementedError: Always raised as this operation is not supported.
         """
-        client = self._get_client()
-        secret_path, field_path = self._split_key_path(key)
-        full_path = self._normalize_path(secret_path)
-
-        if field_path is None:
-            # Setting the entire secret - must be a dict
-            if not isinstance(value, dict):
-                raise ValueError(
-                    f"Top-level secrets in Vault must be dictionaries. "
-                    f"To store a simple value, use dot notation (e.g., '{key}.value'). "
-                    f"Got {type(value).__name__}: {value!r}"
-                )
-            data = value
-        else:
-            # Setting a specific field - need to read existing data first
-            try:
-                existing_data = self._get_secret_data(secret_path)
-            except KeyError:
-                # Secret doesn't exist yet
-                existing_data = {}
-
-            # Navigate to the nested location and set the value
-            parts = field_path.split(".")
-            current = existing_data
-            for part in parts[:-1]:
-                if part not in current:
-                    current[part] = {}
-                elif not isinstance(current[part], dict):
-                    raise ValueError(
-                        f"Cannot set nested field '{field_path}' - parent is not a dict"
-                    )
-                current = current[part]
-
-            current[parts[-1]] = value
-            data = existing_data
-
-        try:
-            client.secrets.kv.v2.create_or_update_secret(
-                path=full_path, secret=data, mount_point=self.mount_point
-            )
-            logger.debug("Set secret in Vault for key '{}'", key)
-
-            # Update cache
-            if self._cache is None:
-                self._cache = {}
-            self._cache[key] = value
-        except Exception as exc:
-            logger.error("Failed to set secret in Vault for key '{}': {}", key, exc)
-            raise RuntimeError(f"Failed to set secret: {exc}") from exc
+        raise NotImplementedError(
+            "Setting secrets is not supported for the Vault provider. "
+            "Please manage Vault secrets through Vault's interface or CLI."
+        )
 
     def unset(self, key: str, /) -> None:
         """
         Unset a secret by its key in Vault.
 
-        For dot-notation keys, this removes the specific field from the secret.
-        For top-level keys, this deletes the entire secret.
+        This operation is not supported for the Vault provider. Vault secrets should be
+        managed through Vault's own interface or CLI tools.
 
-        Args:
-            key: The key of the secret to unset.
+        Raises:
+            NotImplementedError: Always raised as this operation is not supported.
         """
-        client = self._get_client()
-        secret_path, field_path = self._split_key_path(key)
-        full_path = self._normalize_path(secret_path)
-
-        try:
-            if field_path is None:
-                # Delete the entire secret
-                client.secrets.kv.v2.delete_metadata_and_all_versions(
-                    path=full_path, mount_point=self.mount_point
-                )
-                logger.debug("Deleted secret from Vault for key '{}'", key)
-            else:
-                # Remove a specific field
-                try:
-                    existing_data = self._get_secret_data(secret_path)
-                except KeyError:
-                    logger.warning(
-                        "Secret '{}' not found in Vault, nothing to unset", key
-                    )
-                    return
-
-                # Navigate to the nested location and delete the value
-                parts = field_path.split(".")
-                current = existing_data
-                for part in parts[:-1]:
-                    if part not in current or not isinstance(current[part], dict):
-                        logger.warning(
-                            "Field '{}' not found in secret '{}'",
-                            field_path,
-                            key,
-                        )
-                        return
-                    current = current[part]
-
-                if parts[-1] in current:
-                    del current[parts[-1]]
-                    # Write back the modified secret
-                    client.secrets.kv.v2.create_or_update_secret(
-                        path=full_path,
-                        secret=existing_data,
-                        mount_point=self.mount_point,
-                    )
-                    logger.debug(
-                        "Removed field '{}' from secret for key '{}'",
-                        field_path,
-                        key,
-                    )
-                else:
-                    logger.warning(
-                        "Field '{}' not found in secret '{}'", field_path, key
-                    )
-
-            # Update cache
-            if self._cache and key in self._cache:
-                del self._cache[key]
-
-        except Exception as exc:
-            logger.error(
-                "Failed to unset secret in Vault at path '{}': {}", full_path, exc
-            )
-            raise RuntimeError(f"Failed to unset secret: {exc}") from exc
+        raise NotImplementedError(
+            "Unsetting secrets is not supported for the Vault provider. "
+            "Please manage Vault secrets through Vault's interface or CLI."
+        )
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
index a5b1ecd0..782cc480 100644
--- a/src/nyl/secrets/vault_test.py
+++ b/src/nyl/secrets/vault_test.py
@@ -73,16 +73,28 @@ def test_VaultSecretProvider_normalize_path() -> None:
 
 
 def test_VaultSecretProvider_split_key_path() -> None:
-    """Test splitting of dotted keys."""
+    """Test splitting of keys using hash separator."""
     provider = VaultSecretProvider(url="https://vault.example.com:8200")
 
+    # Test without hash - entire secret
     assert provider._split_key_path("database") == ("database", None)
-    assert provider._split_key_path("database.password") == ("database", "password")
-    assert provider._split_key_path("database.credentials.username") == (
+
+    # Test with hash - field access
+    assert provider._split_key_path("database#password") == ("database", "password")
+
+    # Test with nested field path
+    assert provider._split_key_path("database#credentials.username") == (
         "database",
         "credentials.username",
     )
 
+    # Test with dots in path (now supported!)
+    assert provider._split_key_path("db.prod#password") == ("db.prod", "password")
+    assert provider._split_key_path("path/to/secret.v2#field") == (
+        "path/to/secret.v2",
+        "field",
+    )
+
 
 def test_VaultSecretProvider_get_simple(provider_with_mock, mock_vault_client) -> None:
     """Test getting a simple secret value."""
@@ -101,13 +113,13 @@ def test_VaultSecretProvider_get_simple(provider_with_mock, mock_vault_client) -
 
 
 def test_VaultSecretProvider_get_nested(provider_with_mock, mock_vault_client) -> None:
-    """Test getting a nested secret value using dot notation."""
+    """Test getting a nested secret value using hash separator and dot notation."""
     # Mock the Vault response
     mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
         "data": {"data": {"username": "admin", "password": "secret123"}}
     }
 
-    result = provider_with_mock.get("database.password")
+    result = provider_with_mock.get("database#password")
     assert result == "secret123"
 
 
@@ -126,7 +138,7 @@ def test_VaultSecretProvider_get_deeply_nested(
         }
     }
 
-    result = provider_with_mock.get("database.credentials.primary.username")
+    result = provider_with_mock.get("database#credentials.primary.username")
     assert result == "admin"
 
 
@@ -160,102 +172,28 @@ def test_VaultSecretProvider_get_cache(provider_with_mock, mock_vault_client) ->
     assert mock_vault_client.secrets.kv.v2.read_secret_version.call_count == 1
 
 
-def test_VaultSecretProvider_set_entire_secret(
+def test_VaultSecretProvider_set_not_implemented(
     provider_with_mock, mock_vault_client
 ) -> None:
-    """Test setting an entire secret."""
-    provider_with_mock.set("database", {"username": "admin", "password": "secret123"})
-
-    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database",
-        secret={"username": "admin", "password": "secret123"},
-        mount_point="secret",
-    )
-
-
-def test_VaultSecretProvider_set_entire_secret_non_dict_raises(
-    provider_with_mock, mock_vault_client
-) -> None:
-    """Test that setting a top-level secret with a non-dict value raises an error."""
+    """Test that setting secrets raises NotImplementedError."""
     with pytest.raises(
-        ValueError,
-        match="Top-level secrets in Vault must be dictionaries",
+        NotImplementedError,
+        match="Setting secrets is not supported for the Vault provider",
     ):
-        provider_with_mock.set("api-key", "secret123")
-
-
-def test_VaultSecretProvider_set_nested_field(
-    provider_with_mock, mock_vault_client
-) -> None:
-    """Test setting a nested field in an existing secret."""
-    # Mock existing secret
-    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
-        "data": {"data": {"username": "admin", "password": "old_password"}}
-    }
-
-    provider_with_mock.set("database.password", "new_password")
-
-    # Should have read the existing secret first
-    mock_vault_client.secrets.kv.v2.read_secret_version.assert_called_once()
-
-    # Should have written back with updated value
-    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database",
-        secret={"username": "admin", "password": "new_password"},
-        mount_point="secret",
-    )
-
-
-def test_VaultSecretProvider_set_nested_field_new_secret(
-    provider_with_mock, mock_vault_client
-) -> None:
-    """Test setting a nested field when the secret doesn't exist yet."""
-    # Mock that secret doesn't exist
-    mock_vault_client.secrets.kv.v2.read_secret_version.side_effect = (
-        hvac.exceptions.InvalidPath()
-    )
-
-    provider_with_mock.set("database.password", "secret123")
-
-    # Should have written a new secret
-    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database", secret={"password": "secret123"}, mount_point="secret"
-    )
-
-
-def test_VaultSecretProvider_unset_entire_secret(
-    provider_with_mock, mock_vault_client
-) -> None:
-    """Test deleting an entire secret."""
-    provider_with_mock.unset("database")
-
-    mock_vault_client.secrets.kv.v2.delete_metadata_and_all_versions.assert_called_once_with(
-        path="myapp/database", mount_point="secret"
-    )
+        provider_with_mock.set(
+            "database", {"username": "admin", "password": "secret123"}
+        )
 
 
-def test_VaultSecretProvider_unset_nested_field(
+def test_VaultSecretProvider_unset_not_implemented(
     provider_with_mock, mock_vault_client
 ) -> None:
-    """Test removing a nested field from a secret."""
-    # Mock existing secret
-    mock_vault_client.secrets.kv.v2.read_secret_version.return_value = {
-        "data": {
-            "data": {"username": "admin", "password": "secret123", "api_key": "key123"}
-        }
-    }
-
-    provider_with_mock.unset("database.password")
-
-    # Should have read the existing secret
-    mock_vault_client.secrets.kv.v2.read_secret_version.assert_called_once()
-
-    # Should have written back without the password field
-    mock_vault_client.secrets.kv.v2.create_or_update_secret.assert_called_once_with(
-        path="myapp/database",
-        secret={"username": "admin", "api_key": "key123"},
-        mount_point="secret",
-    )
+    """Test that unsetting secrets raises NotImplementedError."""
+    with pytest.raises(
+        NotImplementedError,
+        match="Unsetting secrets is not supported for the Vault provider",
+    ):
+        provider_with_mock.unset("database")
 
 
 def test_VaultSecretProvider_keys_empty() -> None:
@@ -271,11 +209,11 @@ def test_VaultSecretProvider_keys_with_cache(
     # Add some items to cache
     provider_with_mock._cache = {
         "database": {"user": "admin"},
-        "database.password": "secret",
+        "database#password": "secret",
     }
 
     keys = list(provider_with_mock.keys())
-    assert set(keys) == {"database", "database.password"}
+    assert set(keys) == {"database", "database#password"}
 
 
 def test_VaultSecretProvider_authenticate_with_token() -> None:
@@ -314,15 +252,17 @@ def test_VaultSecretProvider_authenticate_with_env_token() -> None:
         assert mock_client.token == "env-token-456"
 
 
-def test_VaultSecretProvider_authenticate_with_jwt() -> None:
-    """Test authentication with JWT token (ArgoCD context)."""
+def test_VaultSecretProvider_authenticate_with_kubernetes_jwt() -> None:
+    """Test authentication with Kubernetes service account JWT token (ArgoCD context)."""
     with unittest.mock.patch("hvac.Client") as mock_client_class:
         mock_client = unittest.mock.MagicMock()
         mock_client_class.return_value = mock_client
         mock_client.is_authenticated.return_value = True
 
         provider = VaultSecretProvider(
-            url="https://vault.example.com:8200", jwt_role="my-role"
+            url="https://vault.example.com:8200",
+            jwt_role="my-role",
+            jwt_auth_method="kubernetes",
         )
         provider._client = mock_client
 
@@ -333,8 +273,52 @@ def test_VaultSecretProvider_authenticate_with_jwt() -> None:
             mock_jwt_path.read_text.return_value = "test-jwt-token\n"
             mock_path_class.return_value = mock_jwt_path
 
-            provider._authenticate_with_jwt()
+            provider._authenticate_with_kubernetes_jwt()
 
         mock_client.auth.kubernetes.login.assert_called_once_with(
             role="my-role", jwt="test-jwt-token"
         )
+
+
+def test_VaultSecretProvider_authenticate_with_nyl_jwt() -> None:
+    """Test authentication with Nyl-issued ArgoCD application-specific JWT token."""
+    with unittest.mock.patch("hvac.Client") as mock_client_class:
+        mock_client = unittest.mock.MagicMock()
+        mock_client_class.return_value = mock_client
+        mock_client.is_authenticated.return_value = True
+
+        provider = VaultSecretProvider(
+            url="https://vault.example.com:8200",
+            jwt_role="my-role",
+            jwt_auth_method="nyl",
+        )
+        provider._client = mock_client
+
+        with unittest.mock.patch.dict(
+            os.environ, {"NYL_VAULT_JWT": "nyl-issued-jwt-token"}
+        ):
+            provider._authenticate_with_nyl_jwt()
+
+        mock_client.auth.jwt.login.assert_called_once_with(
+            role="my-role", jwt="nyl-issued-jwt-token"
+        )
+
+
+def test_VaultSecretProvider_authenticate_with_nyl_jwt_missing_token() -> None:
+    """Test that Nyl JWT authentication fails when NYL_VAULT_JWT is not set."""
+    with unittest.mock.patch("hvac.Client") as mock_client_class:
+        mock_client = unittest.mock.MagicMock()
+        mock_client_class.return_value = mock_client
+        mock_client.is_authenticated.return_value = True
+
+        provider = VaultSecretProvider(
+            url="https://vault.example.com:8200",
+            jwt_role="my-role",
+            jwt_auth_method="nyl",
+        )
+        provider._client = mock_client
+
+        with pytest.raises(
+            RuntimeError, match="NYL_VAULT_JWT environment variable not set"
+        ):
+            provider._authenticate_with_nyl_jwt()

From f39900730abec2c333aac9e3063dafcb6a9bd128 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Mon, 24 Nov 2025 00:00:54 +0000
Subject: [PATCH 7/7] Generate Nyl JWT tokens internally instead of reading
 from env var

Co-authored-by: NiklasRosenstein <1318438+NiklasRosenstein@users.noreply.github.com>
---
 .../reference/configuration/secrets.md        |  16 +-
 pyproject.toml                                |   1 +
 src/nyl/commands/template.py                  | 158 ++++++++++++---
 src/nyl/secrets/vault.py                      |  30 +--
 src/nyl/secrets/vault_test.py                 |  13 +-
 src/nyl/tools/jwt.py                          | 145 ++++++++++++++
 src/nyl/tools/jwt_test.py                     | 184 ++++++++++++++++++
 uv.lock                                       |  11 ++
 8 files changed, 508 insertions(+), 50 deletions(-)
 create mode 100644 src/nyl/tools/jwt.py
 create mode 100644 src/nyl/tools/jwt_test.py

diff --git a/docs/content/reference/configuration/secrets.md b/docs/content/reference/configuration/secrets.md
index f7fa98d3..6158362f 100644
--- a/docs/content/reference/configuration/secrets.md
+++ b/docs/content/reference/configuration/secrets.md
@@ -151,7 +151,10 @@ with pure dot notation. For example, a secret at path `my-app.v2` can be accesse
   authentication.
 - `jwt_auth_method` (optional): JWT authentication method to use. Options:
   - `"kubernetes"` (default): Use Kubernetes service account token (simple, single-tenant)
-  - `"nyl"`: Use Nyl-issued ArgoCD application-specific token (multi-tenant, requires NYL_VAULT_JWT env var)
+  - `"nyl"`: Nyl generates ArgoCD application-specific JWT tokens for multi-tenant environments
+- `jwt_signing_key` (optional): Signing key for Nyl JWT tokens (required when `jwt_auth_method="nyl"`). Can also be
+  provided via the `NYL_VAULT_JWT_SIGNING_KEY` environment variable. This should be a shared secret between Nyl
+  and Vault.
 - `namespace` (optional): The Vault namespace to use (Vault Enterprise feature only)
 
 The provider supports token authentication via the `VAULT_TOKEN` environment variable or `~/.vault-token` file
@@ -231,15 +234,18 @@ When using Vault in a multi-tenant ArgoCD environment:
 5. Use Vault policies to restrict access to only the necessary secrets
 
 #### For Nyl JWT Authentication (multi-tenant)
-1. Configure Vault to trust Nyl as a JWT issuer
+1. Configure Vault to trust Nyl as a JWT issuer (using the `iss` claim from generated tokens)
 2. Create Vault roles that map to ArgoCD project and application identities
-3. Nyl will issue JWTs with claims including:
+3. Nyl automatically generates JWTs when in ArgoCD context with claims including:
+   - `iss`: Issuer (e.g., `https://argocd.example.com/#nyl-v1`)
+   - `aud`: Audience (Vault URL)
+   - `sub`: Subject (e.g., `project:default:application:my-app`)
    - `argocd_project`: The ArgoCD project name
    - `argocd_app`: The ArgoCD application name
    - `repository`: The Git repository URL
 4. Use Vault policies to restrict access based on these claims
-5. Set `jwt_auth_method: "nyl"` in the provider configuration
-6. The NYL_VAULT_JWT environment variable will be populated by Nyl with the appropriate token
+5. Set `jwt_auth_method: "nyl"` and provide `jwt_signing_key` in the provider configuration
+6. The JWT token is automatically generated by Nyl based on the ArgoCD environment
 
 ---
 
diff --git a/pyproject.toml b/pyproject.toml
index 093993ce..aa83f010 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,6 +16,7 @@ dependencies = [
     "kubernetes>=30.1.0",
     "loguru>=0.7.2",
     "nr-stream>=1.1.5",
+    "pyjwt>=2.8.0",
     "pyroscope-io>=0.8.11",
     "pyyaml>=6.0.1",
     "requests>=2.32.3",
diff --git a/src/nyl/commands/template.py b/src/nyl/commands/template.py
index e48ff980..69890693 100644
--- a/src/nyl/commands/template.py
+++ b/src/nyl/commands/template.py
@@ -27,7 +27,10 @@
 from nyl.templating import NylTemplateEngine
 from nyl.tools import yaml
 from nyl.tools.kubectl import Kubectl
-from nyl.tools.kubernetes import drop_empty_metadata_labels, populate_namespace_to_resources
+from nyl.tools.kubernetes import (
+    drop_empty_metadata_labels,
+    populate_namespace_to_resources,
+)
 from nyl.tools.logging import lazy_str
 from nyl.tools.types import Resource, ResourceList
 
@@ -60,7 +63,9 @@ def get_incluster_kubernetes_client() -> ApiClient:
     return ApiClient()
 
 
-def get_profile_kubernetes_client(profiles: ProfileManager, profile: str | None) -> ApiClient:
+def get_profile_kubernetes_client(
+    profiles: ProfileManager, profile: str | None
+) -> ApiClient:
     """
     Create a Kubernetes :class:`ApiClient` from the selected *profile*.
 
@@ -87,11 +92,21 @@ def get_profile_kubernetes_client(profiles: ProfileManager, profile: str | None)
 
 @app.command()
 def template(
-    paths: list[Path] = Argument(..., help="The YAML file(s) to render. Can be a directory."),
-    profile: Optional[str] = Option(None, envvar="NYL_PROFILE", help="The Nyl profile to use."),
-    secrets_provider: str = Option("default", "--secrets", envvar="NYL_SECRETS", help="The secrets provider to use."),
+    paths: list[Path] = Argument(
+        ..., help="The YAML file(s) to render. Can be a directory."
+    ),
+    profile: Optional[str] = Option(
+        None, envvar="NYL_PROFILE", help="The Nyl profile to use."
+    ),
+    secrets_provider: str = Option(
+        "default",
+        "--secrets",
+        envvar="NYL_SECRETS",
+        help="The secrets provider to use.",
+    ),
     in_cluster: bool = Option(
-        False, help="Use the in-cluster Kubernetes configuration. The --profile option is ignored."
+        False,
+        help="Use the in-cluster Kubernetes configuration. The --profile option is ignored.",
     ),
     apply: bool = Option(
         False,
@@ -135,7 +150,9 @@ def template(
         envvar="NYL_TEMPLATE_JOBS",
     ),
     state_dir: Optional[Path] = Option(
-        None, help="The directory to store state in (such as kubeconfig files).", envvar="NYL_STATE_DIR"
+        None,
+        help="The directory to store state in (such as kubeconfig files).",
+        envvar="NYL_STATE_DIR",
     ),
     cache_dir: Optional[Path] = Option(
         None,
@@ -158,10 +175,15 @@ def template(
         profile = os.environ["ARGOCD_ENV_NYL_PROFILE"]
         connect_with_profile = False
 
-    if paths == [Path(".")] and (env_paths := os.getenv("ARGOCD_ENV_NYL_CMP_TEMPLATE_INPUT")) is not None:
+    if (
+        paths == [Path(".")]
+        and (env_paths := os.getenv("ARGOCD_ENV_NYL_CMP_TEMPLATE_INPUT")) is not None
+    ):
         paths = [Path(p) for p in env_paths.split(",")]
         if not paths:
-            logger.error("<cyan>ARGOCD_ENV_NYL_CMP_TEMPLATE_INPUT</> is set, but empty.")
+            logger.error(
+                "<cyan>ARGOCD_ENV_NYL_CMP_TEMPLATE_INPUT</> is set, but empty."
+            )
             exit(1)
         logger.opt(colors=True).info(
             "Using paths from <cyan>ARGOCD_ENV_NYL_CMP_TEMPLATE_INPUT</>: <blue>{}</>",
@@ -191,7 +213,10 @@ def template(
     #       about the cluster are passed via the environment variables.
     #       See https://argo-cd.readthedocs.io/en/stable/user-guide/build-environment/
     PROVIDER.set(
-        ApiClientConfig, ApiClientConfig(in_cluster=in_cluster, profile=profile if connect_with_profile else None)
+        ApiClientConfig,
+        ApiClientConfig(
+            in_cluster=in_cluster, profile=profile if connect_with_profile else None
+        ),
     )
     client = PROVIDER.get(ApiClient)
 
@@ -207,6 +232,43 @@ def template(
 
     secrets = PROVIDER.get(SecretsConfig)
 
+    # Generate Nyl JWT tokens for Vault providers that use Nyl JWT authentication
+    # This needs to be done before the secrets are used, in the ArgoCD context
+    if "ARGOCD_APP_NAME" in os.environ:
+        from nyl.secrets.vault import VaultSecretProvider
+        from nyl.tools.jwt import generate_nyl_jwt_from_argocd_env
+
+        for provider_name, provider in secrets.providers.items():
+            if (
+                isinstance(provider, VaultSecretProvider)
+                and provider.jwt_auth_method == "nyl"
+            ):
+                # Get the signing key from the provider config or environment
+                signing_key = provider.jwt_signing_key or os.getenv(
+                    "NYL_VAULT_JWT_SIGNING_KEY"
+                )
+                if not signing_key:
+                    logger.error(
+                        "Vault provider '{}' is configured with jwt_auth_method='nyl' but no signing key is available. "
+                        "Please set jwt_signing_key in the provider configuration or NYL_VAULT_JWT_SIGNING_KEY environment variable.",
+                        provider_name,
+                    )
+                    exit(1)
+
+                try:
+                    token = generate_nyl_jwt_from_argocd_env(provider.url, signing_key)
+                    provider.set_nyl_jwt_token(token)
+                    logger.debug(
+                        "Generated Nyl JWT token for Vault provider '{}'", provider_name
+                    )
+                except Exception as exc:
+                    logger.error(
+                        "Failed to generate Nyl JWT token for Vault provider '{}': {}",
+                        provider_name,
+                        exc,
+                    )
+                    exit(1)
+
     generator = DispatchingGenerator.default(
         cache_dir=cache_dir,
         search_path=project.config.settings.search_path,
@@ -218,7 +280,9 @@ def template(
     )
 
     for source in load_manifests(paths):
-        logger.opt(colors=True).info("Rendering manifests from <blue>{}</>.", source.file)
+        logger.opt(colors=True).info(
+            "Rendering manifests from <blue>{}</>.", source.file
+        )
 
         template_engine = NylTemplateEngine(
             secrets.providers[secrets_provider],
@@ -233,7 +297,9 @@ def template(
         # However, if the default profile does not exist, we don't want to raise an error, as this would be a
         # breaking change for users who upgrade Nyl without having a default profile defined.
         # If a profile *was* specified and it doesn't exist, we *do* want to raise an error.
-        profile_config = PROVIDER.get(ProfileManager).config.profiles.get(profile or DEFAULT_PROFILE)
+        profile_config = PROVIDER.get(ProfileManager).config.profiles.get(
+            profile or DEFAULT_PROFILE
+        )
         if profile_config is not None:
             vars(template_engine.values).update(profile_config.values)
         elif profile is not None:
@@ -263,7 +329,9 @@ def template(
             source.resources.remove(resource)
 
         # Begin populating the default namespace to resources.
-        current_default_namespace = get_default_namespace_for_manifest(source, default_namespace)
+        current_default_namespace = get_default_namespace_for_manifest(
+            source, default_namespace
+        )
         populate_namespace_to_resources(source.resources, current_default_namespace)
 
         source.resources = template_engine.evaluate(source.resources)
@@ -273,7 +341,9 @@ def template(
                 def new_generation(resource: Resource) -> Future[ResourceList]:
                     def worker() -> ResourceList:
                         resources_ = template_engine.evaluate(ResourceList([resource]))
-                        populate_namespace_to_resources(resources_, current_default_namespace)
+                        populate_namespace_to_resources(
+                            resources_, current_default_namespace
+                        )
                         return resources_
 
                     return executor.submit(worker)
@@ -285,7 +355,9 @@ def worker() -> ResourceList:
                     skip_resources=[PostProcessor],
                 )
 
-        source.resources, post_processors = PostProcessor.extract_from_list(source.resources)
+        source.resources, post_processors = PostProcessor.extract_from_list(
+            source.resources
+        )
 
         # Find the namespaces that are defined in the file. If we find any resources without a namespace, we will
         # inject that namespace name into them. Also find the applyset defined in the file.
@@ -317,7 +389,9 @@ def worker() -> ResourceList:
             applyset_name = current_default_namespace
             applyset = ApplySet.new(applyset_name)
             logger.opt(colors=True).info(
-                "Automatically creating ApplySet for <blue>{}</> (name: <magenta>{}</>).", source.file, applyset_name
+                "Automatically creating ApplySet for <blue>{}</> (name: <magenta>{}</>).",
+                source.file,
+                applyset_name,
             )
 
         if applyset is not None:
@@ -347,7 +421,9 @@ def worker() -> ResourceList:
             # Inline resources often don't have metadata and they are not persisted to the cluster, hence
             # we don't need to process them here.
             if NylResource.matches(resource, API_VERSION_INLINE):
-                assert not inline, "Inline resources should have been processed by this timepdm lint."
+                assert not inline, (
+                    "Inline resources should have been processed by this timepdm lint."
+                )
                 continue
 
             if "metadata" not in resource:
@@ -361,17 +437,25 @@ def worker() -> ResourceList:
         # Tag resources as part of the current apply set, if any.
         if applyset is not None and applyset_part_of:
             for resource in source.resources:
-                if APPLYSET_LABEL_PART_OF not in (labels := resource["metadata"].setdefault("labels", {})):
+                if APPLYSET_LABEL_PART_OF not in (
+                    labels := resource["metadata"].setdefault("labels", {})
+                ):
                     labels[APPLYSET_LABEL_PART_OF] = applyset.id
 
         populate_namespace_to_resources(source.resources, current_default_namespace)
         drop_empty_metadata_labels(source.resources)
 
         # Now apply the post-processor.
-        source.resources = PostProcessor.apply_all(source.resources, post_processors, source.file)
+        source.resources = PostProcessor.apply_all(
+            source.resources, post_processors, source.file
+        )
 
         if apply:
-            logger.info("Kubectl-apply {} resource(s) from '{}'", len(source.resources), source.file)
+            logger.info(
+                "Kubectl-apply {} resource(s) from '{}'",
+                len(source.resources),
+                source.file,
+            )
             kubectl.apply(
                 manifests=source.resources,
                 applyset=applyset.reference if applyset else None,
@@ -379,7 +463,11 @@ def worker() -> ResourceList:
                 force_conflicts=True,
             )
         elif diff:
-            logger.info("Kubectl-diff {} resource(s) from '{}'", len(source.resources), source.file)
+            logger.info(
+                "Kubectl-diff {} resource(s) from '{}'",
+                len(source.resources),
+                source.file,
+            )
             kubectl.diff(manifests=source.resources, applyset=applyset)
         else:
             # If we're not going to be applying the resources immediately via `kubectl`, we print them to stdout.
@@ -396,7 +484,12 @@ def worker() -> ResourceList:
                 "data": {
                     "duration_seconds": time.perf_counter() - start_time,
                     "inputs": [
-                        str(p.absolute().relative_to(project.file.parent) if project.file else p) for p in paths
+                        str(
+                            p.absolute().relative_to(project.file.parent)
+                            if project.file
+                            else p
+                        )
+                        for p in paths
                     ],
                     # See https://argo-cd.readthedocs.io/en/stable/user-guide/build-environment/
                     "argocd_app_name": os.getenv("ARGOCD_APP_NAME"),
@@ -404,7 +497,9 @@ def worker() -> ResourceList:
                     "argocd_app_project_name": os.getenv("ARGOCD_APP_PROJECT_NAME"),
                     "argocd_app_revision": os.getenv("ARGOCD_APP_REVISION"),
                     "argocd_app_source_path": os.getenv("ARGOCD_APP_SOURCE_PATH"),
-                    "argocd_app_source_repo_url": os.getenv("ARGOCD_APP_SOURCE_REPO_URL"),
+                    "argocd_app_source_repo_url": os.getenv(
+                        "ARGOCD_APP_SOURCE_REPO_URL"
+                    ),
                 },
             }
         ),
@@ -444,7 +539,9 @@ def load_manifests(paths: list[Path]) -> list[ManifestsWithSource]:
 
     result = []
     for file in files:
-        resources = ResourceList(list(map(Resource, filter(None, yaml.loads_all(file.read_text())))))
+        resources = ResourceList(
+            list(map(Resource, filter(None, yaml.loads_all(file.read_text()))))
+        )
         result.append(ManifestsWithSource(resources, file))
 
     return result
@@ -501,7 +598,9 @@ def is_namespace_resource(resource: Resource) -> bool:
     return resource.get("apiVersion") == "v1" and resource.get("kind") == "Namespace"
 
 
-def get_default_namespace_for_manifest(source: ManifestsWithSource, fallback: str | None = None) -> str:
+def get_default_namespace_for_manifest(
+    source: ManifestsWithSource, fallback: str | None = None
+) -> str:
     """
     Given the contents of a manifest file, determine the fallback namespace to apply to resources that have been
     recorded without a namespace.
@@ -541,13 +640,18 @@ def get_default_namespace_for_manifest(source: ManifestsWithSource, fallback: st
         return use_namespace
 
     if len(namespace_resources) == 1:
-        logger.debug("Manifest '{}' defines exactly one Namespace resource. Using '{}' as the default namespace.")
+        logger.debug(
+            "Manifest '{}' defines exactly one Namespace resource. Using '{}' as the default namespace."
+        )
         return namespace_resources[0]["metadata"]["name"]  # type: ignore[no-any-return]
 
     default_namespaces = {
         x["metadata"]["name"]
         for x in namespace_resources
-        if x["metadata"].get("annotations", {}).get(DEFAULT_NAMESPACE_ANNOTATION, "false") == "true"
+        if x["metadata"]
+        .get("annotations", {})
+        .get(DEFAULT_NAMESPACE_ANNOTATION, "false")
+        == "true"
     }
 
     if len(default_namespaces) == 0:
diff --git a/src/nyl/secrets/vault.py b/src/nyl/secrets/vault.py
index 30c1ad64..e4e9d2c3 100644
--- a/src/nyl/secrets/vault.py
+++ b/src/nyl/secrets/vault.py
@@ -60,6 +60,13 @@ class VaultSecretProvider(SecretProvider):
     Defaults to "kubernetes".
     """
 
+    jwt_signing_key: str | None = None
+    """
+    The signing key for generating Nyl JWT tokens (required when jwt_auth_method="nyl").
+    This should be a shared secret between Nyl and Vault. Can also be provided via
+    the NYL_VAULT_JWT_SIGNING_KEY environment variable.
+    """
+
     namespace: str | None = None
     """
     The Vault namespace to use (Vault Enterprise feature).
@@ -67,6 +74,11 @@ class VaultSecretProvider(SecretProvider):
 
     _client: hvac.Client | None = field(init=False, repr=False, default=None)
     _cache: dict[str, SecretValue] | None = field(init=False, repr=False, default=None)
+    _nyl_jwt_token: str | None = field(init=False, repr=False, default=None)
+
+    def set_nyl_jwt_token(self, token: str) -> None:
+        """Set the Nyl-issued JWT token for authentication."""
+        self._nyl_jwt_token = token
 
     def _get_client(self) -> hvac.Client:
         """Get or create a Vault client with appropriate authentication."""
@@ -135,8 +147,9 @@ def _authenticate_with_nyl_jwt(self) -> None:
         """
         Authenticate with Vault using Nyl-issued ArgoCD application-specific JWT token.
 
-        This token is issued by Nyl and contains claims about the ArgoCD application
-        being deployed, enabling multi-tenant secure access to Vault secrets.
+        This token is generated by Nyl based on the ArgoCD application context
+        and contains claims about the ArgoCD application being deployed, enabling
+        multi-tenant secure access to Vault secrets.
 
         The token payload includes:
         - iss: Issuer (e.g., "https://my-argocd.example.com/#nyl-v1")
@@ -146,15 +159,10 @@ def _authenticate_with_nyl_jwt(self) -> None:
         - argocd_app: ArgoCD application name
         - repository: Git repository URL
         """
-        # Check for Nyl-issued JWT token in environment variable
-        jwt_token = os.environ.get("NYL_VAULT_JWT")
-        if not jwt_token:
-            logger.error(
-                "Nyl JWT authentication requested but NYL_VAULT_JWT environment variable not set. "
-                "This token should be issued by Nyl based on the ArgoCD application context."
-            )
+        if not self._nyl_jwt_token:
             raise RuntimeError(
-                "NYL_VAULT_JWT environment variable not set for Nyl JWT authentication"
+                "Nyl JWT authentication requested but token not set. "
+                "The token should be generated and set by Nyl before accessing secrets."
             )
 
         logger.info(
@@ -164,7 +172,7 @@ def _authenticate_with_nyl_jwt(self) -> None:
         try:
             assert self._client is not None
             # Use JWT auth method (not kubernetes auth)
-            self._client.auth.jwt.login(role=self.jwt_role, jwt=jwt_token)
+            self._client.auth.jwt.login(role=self.jwt_role, jwt=self._nyl_jwt_token)
             logger.debug("Successfully authenticated with Vault using Nyl-issued JWT")
         except Exception as exc:
             logger.error(
diff --git a/src/nyl/secrets/vault_test.py b/src/nyl/secrets/vault_test.py
index 782cc480..b840092a 100644
--- a/src/nyl/secrets/vault_test.py
+++ b/src/nyl/secrets/vault_test.py
@@ -294,10 +294,10 @@ def test_VaultSecretProvider_authenticate_with_nyl_jwt() -> None:
         )
         provider._client = mock_client
 
-        with unittest.mock.patch.dict(
-            os.environ, {"NYL_VAULT_JWT": "nyl-issued-jwt-token"}
-        ):
-            provider._authenticate_with_nyl_jwt()
+        # Set the Nyl JWT token (as would be done by the template command)
+        provider.set_nyl_jwt_token("nyl-issued-jwt-token")
+
+        provider._authenticate_with_nyl_jwt()
 
         mock_client.auth.jwt.login.assert_called_once_with(
             role="my-role", jwt="nyl-issued-jwt-token"
@@ -318,7 +318,6 @@ def test_VaultSecretProvider_authenticate_with_nyl_jwt_missing_token() -> None:
         )
         provider._client = mock_client
 
-        with pytest.raises(
-            RuntimeError, match="NYL_VAULT_JWT environment variable not set"
-        ):
+        with pytest.raises(RuntimeError, match="token not set"):
+            provider._authenticate_with_nyl_jwt()
             provider._authenticate_with_nyl_jwt()
diff --git a/src/nyl/tools/jwt.py b/src/nyl/tools/jwt.py
new file mode 100644
index 00000000..5ef2e284
--- /dev/null
+++ b/src/nyl/tools/jwt.py
@@ -0,0 +1,145 @@
+"""
+JWT token generation for Nyl-issued workload identity tokens.
+
+This module provides functionality to generate JWT tokens that assert the identity
+of an ArgoCD application being templated by Nyl, for use with Vault authentication
+in multi-tenant environments.
+"""
+
+import os
+import time
+from dataclasses import dataclass
+from typing import Any
+
+import jwt
+from loguru import logger
+
+
+@dataclass
+class NylJwtClaims:
+    """Claims for a Nyl-issued JWT token asserting ArgoCD application identity."""
+
+    issuer: str
+    """The issuer of the token (e.g., "https://my-argocd.example.com/#nyl-v1")."""
+
+    audience: str
+    """The audience for the token (e.g., Vault URL)."""
+
+    argocd_project: str
+    """The ArgoCD project name."""
+
+    argocd_app: str
+    """The ArgoCD application name."""
+
+    repository: str | None
+    """The Git repository URL (if available)."""
+
+    def to_payload(self) -> dict[str, Any]:
+        """Convert claims to JWT payload."""
+        payload = {
+            "iss": self.issuer,
+            "aud": self.audience,
+            "sub": f"project:{self.argocd_project}:application:{self.argocd_app}",
+            "argocd_project": self.argocd_project,
+            "argocd_app": self.argocd_app,
+            "iat": int(time.time()),
+            "exp": int(time.time()) + 3600,  # Token valid for 1 hour
+        }
+        if self.repository:
+            payload["repository"] = self.repository
+        return payload
+
+
+def generate_nyl_jwt_from_argocd_env(vault_url: str, signing_key: str) -> str:
+    """
+    Generate a Nyl-issued JWT token based on ArgoCD environment variables.
+
+    This function reads ArgoCD environment variables to extract the application
+    identity and generates a JWT token that can be used to authenticate with Vault.
+
+    Args:
+        vault_url: The Vault server URL (used as the audience claim).
+        signing_key: The private key to sign the JWT with (HS256 algorithm).
+
+    Returns:
+        A signed JWT token as a string.
+
+    Raises:
+        RuntimeError: If required ArgoCD environment variables are not set.
+    """
+    # Extract ArgoCD environment variables
+    argocd_app_name = os.getenv("ARGOCD_APP_NAME")
+    argocd_project = os.getenv("ARGOCD_APP_PROJECT_NAME") or "default"
+    argocd_repo = os.getenv("ARGOCD_APP_SOURCE_REPO_URL")
+
+    if not argocd_app_name:
+        raise RuntimeError(
+            "Cannot generate Nyl JWT token: ARGOCD_APP_NAME environment variable not set. "
+            "This token can only be generated when running in ArgoCD context."
+        )
+
+    # Determine the issuer (ArgoCD server URL with nyl-v1 fragment)
+    # In ArgoCD context, we might not have the server URL directly, so we construct it
+    # or use a configured value
+    argocd_server = os.getenv("ARGOCD_SERVER") or os.getenv(
+        "ARGOCD_APPLICATION_NAME", "argocd"
+    )
+    issuer = f"https://{argocd_server}/#nyl-v1"
+
+    claims = NylJwtClaims(
+        issuer=issuer,
+        audience=vault_url,
+        argocd_project=argocd_project,
+        argocd_app=argocd_app_name,
+        repository=argocd_repo,
+    )
+
+    logger.debug(
+        "Generating Nyl JWT token for ArgoCD app '{}' in project '{}'",
+        argocd_app_name,
+        argocd_project,
+    )
+
+    # Generate the JWT token using HS256 algorithm
+    token = jwt.encode(claims.to_payload(), signing_key, algorithm="HS256")
+
+    return token
+
+
+def generate_nyl_jwt(
+    argocd_project: str,
+    argocd_app: str,
+    vault_url: str,
+    signing_key: str,
+    repository: str | None = None,
+    issuer: str | None = None,
+) -> str:
+    """
+    Generate a Nyl-issued JWT token with explicit parameters.
+
+    This is useful for testing or when ArgoCD environment variables are not available.
+
+    Args:
+        argocd_project: The ArgoCD project name.
+        argocd_app: The ArgoCD application name.
+        vault_url: The Vault server URL (used as the audience claim).
+        signing_key: The private key to sign the JWT with (HS256 algorithm).
+        repository: Optional Git repository URL.
+        issuer: Optional custom issuer. If not provided, uses a default.
+
+    Returns:
+        A signed JWT token as a string.
+    """
+    if not issuer:
+        issuer = "https://argocd/#nyl-v1"
+
+    claims = NylJwtClaims(
+        issuer=issuer,
+        audience=vault_url,
+        argocd_project=argocd_project,
+        argocd_app=argocd_app,
+        repository=repository,
+    )
+
+    token = jwt.encode(claims.to_payload(), signing_key, algorithm="HS256")
+    return token
diff --git a/src/nyl/tools/jwt_test.py b/src/nyl/tools/jwt_test.py
new file mode 100644
index 00000000..8f40a462
--- /dev/null
+++ b/src/nyl/tools/jwt_test.py
@@ -0,0 +1,184 @@
+import os
+import time
+import unittest.mock
+
+import jwt as pyjwt
+import pytest
+
+from nyl.tools.jwt import (
+    NylJwtClaims,
+    generate_nyl_jwt,
+    generate_nyl_jwt_from_argocd_env,
+)
+
+
+def test_NylJwtClaims_to_payload() -> None:
+    """Test converting claims to JWT payload."""
+    claims = NylJwtClaims(
+        issuer="https://argocd.example.com/#nyl-v1",
+        audience="https://vault.example.com:8200",
+        argocd_project="default",
+        argocd_app="my-app",
+        repository="git@github.com:example/repo.git",
+    )
+
+    payload = claims.to_payload()
+
+    assert payload["iss"] == "https://argocd.example.com/#nyl-v1"
+    assert payload["aud"] == "https://vault.example.com:8200"
+    assert payload["sub"] == "project:default:application:my-app"
+    assert payload["argocd_project"] == "default"
+    assert payload["argocd_app"] == "my-app"
+    assert payload["repository"] == "git@github.com:example/repo.git"
+    assert "iat" in payload
+    assert "exp" in payload
+    assert payload["exp"] > payload["iat"]
+
+
+def test_NylJwtClaims_to_payload_without_repository() -> None:
+    """Test JWT payload without repository."""
+    claims = NylJwtClaims(
+        issuer="https://argocd.example.com/#nyl-v1",
+        audience="https://vault.example.com:8200",
+        argocd_project="default",
+        argocd_app="my-app",
+        repository=None,
+    )
+
+    payload = claims.to_payload()
+
+    assert "repository" not in payload
+
+
+def test_generate_nyl_jwt() -> None:
+    """Test generating a Nyl JWT with explicit parameters."""
+    token = generate_nyl_jwt(
+        argocd_project="test-project",
+        argocd_app="test-app",
+        vault_url="https://vault.example.com:8200",
+        signing_key="test-secret-key",
+        repository="git@github.com:test/repo.git",
+        issuer="https://argocd.example.com/#nyl-v1",
+    )
+
+    # Decode the token to verify its contents
+    decoded = pyjwt.decode(
+        token,
+        "test-secret-key",
+        algorithms=["HS256"],
+        audience="https://vault.example.com:8200",
+    )
+
+    assert decoded["iss"] == "https://argocd.example.com/#nyl-v1"
+    assert decoded["aud"] == "https://vault.example.com:8200"
+    assert decoded["sub"] == "project:test-project:application:test-app"
+    assert decoded["argocd_project"] == "test-project"
+    assert decoded["argocd_app"] == "test-app"
+    assert decoded["repository"] == "git@github.com:test/repo.git"
+
+
+def test_generate_nyl_jwt_default_issuer() -> None:
+    """Test generating a Nyl JWT with default issuer."""
+    token = generate_nyl_jwt(
+        argocd_project="test-project",
+        argocd_app="test-app",
+        vault_url="https://vault.example.com:8200",
+        signing_key="test-secret-key",
+    )
+
+    decoded = pyjwt.decode(
+        token,
+        "test-secret-key",
+        algorithms=["HS256"],
+        audience="https://vault.example.com:8200",
+    )
+
+    assert decoded["iss"] == "https://argocd/#nyl-v1"
+
+
+def test_generate_nyl_jwt_from_argocd_env() -> None:
+    """Test generating a Nyl JWT from ArgoCD environment variables."""
+    with unittest.mock.patch.dict(
+        os.environ,
+        {
+            "ARGOCD_APP_NAME": "my-app",
+            "ARGOCD_APP_PROJECT_NAME": "my-project",
+            "ARGOCD_APP_SOURCE_REPO_URL": "git@github.com:example/repo.git",
+            "ARGOCD_SERVER": "argocd.example.com",
+        },
+    ):
+        token = generate_nyl_jwt_from_argocd_env(
+            vault_url="https://vault.example.com:8200", signing_key="test-secret-key"
+        )
+
+        decoded = pyjwt.decode(
+            token,
+            "test-secret-key",
+            algorithms=["HS256"],
+            audience="https://vault.example.com:8200",
+        )
+
+        assert decoded["iss"] == "https://argocd.example.com/#nyl-v1"
+        assert decoded["aud"] == "https://vault.example.com:8200"
+        assert decoded["sub"] == "project:my-project:application:my-app"
+        assert decoded["argocd_project"] == "my-project"
+        assert decoded["argocd_app"] == "my-app"
+        assert decoded["repository"] == "git@github.com:example/repo.git"
+
+
+def test_generate_nyl_jwt_from_argocd_env_default_project() -> None:
+    """Test generating a Nyl JWT with default project when not specified."""
+    with unittest.mock.patch.dict(
+        os.environ,
+        {
+            "ARGOCD_APP_NAME": "my-app",
+            # No ARGOCD_APP_PROJECT_NAME set
+            "ARGOCD_APP_SOURCE_REPO_URL": "git@github.com:example/repo.git",
+        },
+        clear=False,
+    ):
+        token = generate_nyl_jwt_from_argocd_env(
+            vault_url="https://vault.example.com:8200", signing_key="test-secret-key"
+        )
+
+        decoded = pyjwt.decode(
+            token,
+            "test-secret-key",
+            algorithms=["HS256"],
+            audience="https://vault.example.com:8200",
+        )
+
+        assert decoded["argocd_project"] == "default"
+
+
+def test_generate_nyl_jwt_from_argocd_env_missing_app_name() -> None:
+    """Test that generating JWT fails when ARGOCD_APP_NAME is not set."""
+    with unittest.mock.patch.dict(os.environ, {}, clear=True):
+        with pytest.raises(
+            RuntimeError, match="ARGOCD_APP_NAME environment variable not set"
+        ):
+            generate_nyl_jwt_from_argocd_env(
+                vault_url="https://vault.example.com:8200",
+                signing_key="test-secret-key",
+            )
+
+
+def test_jwt_token_expiration() -> None:
+    """Test that JWT token has appropriate expiration."""
+    token = generate_nyl_jwt(
+        argocd_project="test-project",
+        argocd_app="test-app",
+        vault_url="https://vault.example.com:8200",
+        signing_key="test-secret-key",
+    )
+
+    decoded = pyjwt.decode(
+        token,
+        "test-secret-key",
+        algorithms=["HS256"],
+        audience="https://vault.example.com:8200",
+    )
+
+    # Token should expire approximately 1 hour from now
+    expected_exp = time.time() + 3600
+    assert abs(decoded["exp"] - expected_exp) < 5  # Allow 5 seconds tolerance
diff --git a/uv.lock b/uv.lock
index 3cd155bb..ba61464d 100644
--- a/uv.lock
+++ b/uv.lock
@@ -581,6 +581,7 @@ dependencies = [
     { name = "kubernetes" },
     { name = "loguru" },
     { name = "nr-stream" },
+    { name = "pyjwt" },
     { name = "pyroscope-io" },
     { name = "pyyaml" },
     { name = "requests" },
@@ -611,6 +612,7 @@ requires-dist = [
     { name = "kubernetes", specifier = ">=30.1.0" },
     { name = "loguru", specifier = ">=0.7.2" },
     { name = "nr-stream", specifier = ">=1.1.5" },
+    { name = "pyjwt", specifier = ">=2.8.0" },
     { name = "pyroscope-io", specifier = ">=0.8.11" },
     { name = "pyyaml", specifier = ">=6.0.1" },
     { name = "requests", specifier = ">=2.32.3" },
@@ -706,6 +708,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
 ]
 
+[[package]]
+name = "pyjwt"
+version = "2.10.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/e7/46/bd74733ff231675599650d3e47f361794b22ef3e3770998dda30d3b63726/pyjwt-2.10.1.tar.gz", hash = "sha256:3cc5772eb20009233caf06e9d8a0577824723b44e6648ee0a2aedb6cf9381953", size = 87785, upload-time = "2024-11-28T03:43:29.933Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/61/ad/689f02752eeec26aed679477e80e632ef1b682313be70793d798c1d5fc8f/PyJWT-2.10.1-py3-none-any.whl", hash = "sha256:dcdd193e30abefd5debf142f9adfcdd2b58004e644f25406ffaebd50bd98dacb", size = 22997, upload-time = "2024-11-28T03:43:27.893Z" },
+]
+
 [[package]]
 name = "pyroscope-io"
 version = "0.8.11"