diff --git a/client/allocrunner/taskrunner/secrets/nomad_provider.go b/client/allocrunner/taskrunner/secrets/nomad_provider.go
new file mode 100644
index 000000000..02ffcc282
--- /dev/null
+++ b/client/allocrunner/taskrunner/secrets/nomad_provider.go
@@ -0,0 +1,80 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package secrets
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/hashicorp/go-envparse"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/mitchellh/mapstructure"
+)
+
+type nomadProviderConfig struct {
+	Namespace string `mapstructure:"namespace"`
+}
+
+func defaultNomadConfig(namespace string) *nomadProviderConfig {
+	return &nomadProviderConfig{
+		Namespace: namespace,
+	}
+}
+
+type NomadProvider struct {
+	secret   *structs.Secret
+	tmplPath string
+	config   *nomadProviderConfig
+}
+
+// NewNomadProvider takes a task secret and decodes the config, overwriting the default config fields
+// with any provided fields, returning an error if the secret or secret's config is invalid.
+func NewNomadProvider(secret *structs.Secret, path string, namespace string) (*NomadProvider, error) {
+	if secret == nil {
+		return nil, fmt.Errorf("empty secret for nomad provider")
+	}
+
+	conf := defaultNomadConfig(namespace)
+	if err := mapstructure.Decode(secret.Config, conf); err != nil {
+		return nil, err
+	}
+
+	return &NomadProvider{
+		config:   conf,
+		secret:   secret,
+		tmplPath: path,
+	}, nil
+}
+
+func (n *NomadProvider) BuildTemplate() *structs.Template {
+	data := fmt.Sprintf(`
+		{{ with nomadVar "%s@%s" }}
+		{{ range $k, $v := . }}
+		secret.%s.{{ $k }}={{ $v }}
+		{{ end }}
+		{{ end }}`,
+		n.secret.Path, n.config.Namespace, n.secret.Name)
+
+	return &structs.Template{
+		EmbeddedTmpl: data,
+		DestPath:     n.tmplPath,
+		ChangeMode:   structs.TemplateChangeModeNoop,
+		Once:         true,
+	}
+}
+
+func (n *NomadProvider) Parse() (map[string]string, error) {
+	dest := filepath.Clean(n.tmplPath)
+	f, err := os.Open(dest)
+	if err != nil {
+		return nil, fmt.Errorf("error opening env template: %v", err)
+	}
+	defer func() {
+		f.Close()
+		os.Remove(dest)
+	}()
+
+	return envparse.Parse(f)
+}
diff --git a/client/allocrunner/taskrunner/secrets/nomad_provider_test.go b/client/allocrunner/taskrunner/secrets/nomad_provider_test.go
new file mode 100644
index 000000000..78078fb91
--- /dev/null
+++ b/client/allocrunner/taskrunner/secrets/nomad_provider_test.go
@@ -0,0 +1,78 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package secrets
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/shoenig/test/must"
+)
+
+func TestNomadProvider_BuildTemplate(t *testing.T) {
+	t.Run("variable template succeeds", func(t *testing.T) {
+		testDir := t.TempDir()
+		testSecret := &structs.Secret{
+			Name:     "foo",
+			Provider: "nomad",
+			Path:     "/test/path",
+			Config: map[string]any{
+				"namespace": "dev",
+			},
+		}
+		p, err := NewNomadProvider(testSecret, testDir, "default")
+		must.NoError(t, err)
+
+		tmpl := p.BuildTemplate()
+		must.NotNil(t, tmpl)
+
+		// expected template should have correct path and name
+		expectedTmpl := `
+		{{ with nomadVar "/test/path@dev" }}
+		{{ range $k, $v := . }}
+		secret.foo.{{ $k }}={{ $v }}
+		{{ end }}
+		{{ end }}`
+		// validate template string contains expected data
+		must.Eq(t, tmpl.EmbeddedTmpl, expectedTmpl)
+	})
+
+	t.Run("invalid config options errors", func(t *testing.T) {
+		testDir := t.TempDir()
+		testSecret := &structs.Secret{
+			Name:     "foo",
+			Provider: "nomad",
+			Path:     "/test/path",
+			Config: map[string]any{
+				"namespace": 123,
+			},
+		}
+		_, err := NewNomadProvider(testSecret, testDir, "default")
+		must.Error(t, err)
+
+		// tmpl := p.BuildTemplate()
+		// must.Nil(t, tmpl)
+	})
+}
+
+func TestNomadProvider_Parse(t *testing.T) {
+	testDir := t.TempDir()
+
+	tmplPath := filepath.Join(testDir, "foo")
+
+	data := "foo=bar"
+	err := os.WriteFile(tmplPath, []byte(data), 0777)
+	must.NoError(t, err)
+
+	p, err := NewNomadProvider(&structs.Secret{}, tmplPath, "default")
+	must.NoError(t, err)
+
+	vars, err := p.Parse()
+	must.Eq(t, vars, map[string]string{"foo": "bar"})
+
+	_, err = os.Stat(tmplPath)
+	must.ErrorContains(t, err, "no such file")
+}
diff --git a/client/allocrunner/taskrunner/secrets_hook.go b/client/allocrunner/taskrunner/secrets_hook.go
new file mode 100644
index 000000000..c53cccadf
--- /dev/null
+++ b/client/allocrunner/taskrunner/secrets_hook.go
@@ -0,0 +1,183 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package taskrunner
+
+import (
+	"context"
+	"fmt"
+	"maps"
+	"path/filepath"
+
+	log "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/nomad/client/allocrunner/interfaces"
+	ti "github.com/hashicorp/nomad/client/allocrunner/taskrunner/interfaces"
+	"github.com/hashicorp/nomad/client/allocrunner/taskrunner/secrets"
+	"github.com/hashicorp/nomad/client/allocrunner/taskrunner/template"
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/client/taskenv"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+// SecretProvider currently only supports Vault and Nomad which use CT. Future
+// work can modify this interface to include custom providers using a plugin
+// interface.
+type SecretProvider interface {
+	// BuildTemplate should construct a template appropriate for that provider
+	// and append it to the templateManager's templates.
+	BuildTemplate() *structs.Template
+
+	// Parse allows each provider implementation to parse its "response" object.
+	Parse() (map[string]string, error)
+}
+
+type secretsHookConfig struct {
+	// logger is used to log
+	logger log.Logger
+
+	// lifecycle is used to interact with the task's lifecycle
+	lifecycle ti.TaskLifecycle
+
+	// events is used to emit events
+	events ti.EventEmitter
+
+	// clientConfig is the Nomad Client configuration
+	clientConfig *config.Config
+
+	// envBuilder is the environment variable builder for the task.
+	envBuilder *taskenv.Builder
+
+	// nomadNamespace is the job's Nomad namespace
+	nomadNamespace string
+}
+
+type secretsHook struct {
+	// logger is used to log
+	logger log.Logger
+
+	// lifecycle is used to interact with the task's lifecycle
+	lifecycle ti.TaskLifecycle
+
+	// events is used to emit events
+	events ti.EventEmitter
+
+	// clientConfig is the Nomad Client configuration
+	clientConfig *config.Config
+
+	// envBuilder is the environment variable builder for the task
+	envBuilder *taskenv.Builder
+
+	// nomadNamespace is the job's Nomad namespace
+	nomadNamespace string
+
+	// secrets to be fetched and populated for interpolation
+	secrets []*structs.Secret
+
+	// taskrunner secrets map
+	taskSecrets map[string]string
+}
+
+func newSecretsHook(conf *secretsHookConfig, secrets []*structs.Secret) *secretsHook {
+	return &secretsHook{
+		logger:         conf.logger,
+		lifecycle:      conf.lifecycle,
+		events:         conf.events,
+		clientConfig:   conf.clientConfig,
+		envBuilder:     conf.envBuilder,
+		nomadNamespace: conf.nomadNamespace,
+		secrets:        secrets,
+		// Future work will inject taskSecrets from the taskRunner, so that the taskrunner
+		// can make these secrets available to other hooks.
+		taskSecrets: make(map[string]string),
+	}
+}
+
+func (h *secretsHook) Name() string {
+	return "secrets"
+}
+
+func (h *secretsHook) Prestart(ctx context.Context, req *interfaces.TaskPrestartRequest, _ *interfaces.TaskPrestartResponse) error {
+	templates := []*structs.Template{}
+
+	providers, err := h.buildSecretProviders(req.TaskDir.SecretsDir)
+	if err != nil {
+		return err
+	}
+
+	for _, p := range providers {
+		templates = append(templates, p.BuildTemplate())
+	}
+
+	vaultCluster := req.Task.GetVaultClusterName()
+	vaultConfig := h.clientConfig.GetVaultConfigs(h.logger)[vaultCluster]
+
+	unblock := make(chan struct{})
+	tm, err := template.NewTaskTemplateManager(&template.TaskTemplateManagerConfig{
+		UnblockCh:            unblock,
+		Lifecycle:            h.lifecycle,
+		Events:               h.events,
+		Templates:            templates,
+		ClientConfig:         h.clientConfig,
+		VaultToken:           req.VaultToken,
+		VaultConfig:          vaultConfig,
+		VaultNamespace:       req.Alloc.Job.VaultNamespace,
+		TaskDir:              req.TaskDir.Dir,
+		EnvBuilder:           h.envBuilder,
+		MaxTemplateEventRate: template.DefaultMaxTemplateEventRate,
+		NomadNamespace:       h.nomadNamespace,
+		NomadToken:           req.NomadToken,
+		TaskID:               req.Alloc.ID + "-" + req.Task.Name,
+		Logger:               h.logger,
+	})
+	if err != nil {
+		return err
+	}
+
+	// Run the template manager to render templates.
+	go tm.Run()
+
+	// Safeguard against the template manager continuing to run.
+	defer tm.Stop()
+
+	select {
+	case <-ctx.Done():
+		return nil
+	case <-unblock:
+	}
+
+	// parse and copy variables to taskSecrets
+	for _, p := range providers {
+		vars, err := p.Parse()
+		if err != nil {
+			return err
+		}
+		maps.Copy(h.taskSecrets, vars)
+	}
+
+	return nil
+}
+
+func (h *secretsHook) buildSecretProviders(secretDir string) ([]SecretProvider, error) {
+	// Any configuration errors will be found when calling the secret providers constructor,
+	// so use a multierror to collect all errors and return them to the user at the same time.
+	providers, mErr := []SecretProvider{}, new(multierror.Error)
+
+	for idx, s := range h.secrets {
+		tmplPath := filepath.Join(secretDir, fmt.Sprintf("temp-%d", idx))
+		switch s.Provider {
+		case "nomad":
+			if p, err := secrets.NewNomadProvider(s, tmplPath, h.nomadNamespace); err != nil {
+				multierror.Append(mErr, err)
+			} else {
+				providers = append(providers, p)
+			}
+		case "vault":
+			// Unimplemented
+		default:
+			multierror.Append(mErr, fmt.Errorf("unknown secret provider type: %s", s.Provider))
+		}
+	}
+
+	return providers, mErr.ErrorOrNil()
+}
diff --git a/client/allocrunner/taskrunner/secrets_hook_test.go b/client/allocrunner/taskrunner/secrets_hook_test.go
new file mode 100644
index 000000000..ba6c47d0d
--- /dev/null
+++ b/client/allocrunner/taskrunner/secrets_hook_test.go
@@ -0,0 +1,217 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package taskrunner
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/nomad/ci"
+	"github.com/hashicorp/nomad/client/allocdir"
+	"github.com/hashicorp/nomad/client/allocrunner/interfaces"
+	trtesting "github.com/hashicorp/nomad/client/allocrunner/taskrunner/testing"
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/client/taskenv"
+	"github.com/hashicorp/nomad/helper/bufconndialer"
+	"github.com/hashicorp/nomad/helper/testlog"
+	"github.com/hashicorp/nomad/nomad/mock"
+	"github.com/hashicorp/nomad/nomad/structs"
+	"github.com/shoenig/test/must"
+)
+
+func TestSecretsHook_Prestart_Nomad(t *testing.T) {
+	ci.Parallel(t)
+
+	t.Run("nomad provider successfully renders valid secret", func(t *testing.T) {
+		secretsResp := `
+		{
+		  "CreateIndex": 812,
+		  "CreateTime": 1750782609539170600,
+		  "Items": {
+		    "key2": "value2",
+		    "key1": "value1"
+		  },
+		  "ModifyIndex": 812,
+		  "ModifyTime": 1750782609539170600,
+		  "Namespace": "default",
+		  "Path": "testnomadvar"
+		}
+		`
+		count := 0 // CT expects a nomad index header that increments, or else it continues polling
+		nomadServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("X-Nomad-Index", strconv.Itoa(count))
+			fmt.Fprintln(w, secretsResp)
+			count += 1
+		}))
+		t.Cleanup(nomadServer.Close)
+
+		l, d := bufconndialer.New()
+		nomadServer.Listener = l
+
+		nomadServer.Start()
+
+		clientConfig := config.DefaultConfig()
+		clientConfig.TemplateDialer = d
+		clientConfig.TemplateConfig.DisableSandbox = true
+
+		taskDir := t.TempDir()
+		alloc := mock.MinAlloc()
+		task := alloc.Job.TaskGroups[0].Tasks[0]
+
+		conf := &secretsHookConfig{
+			logger:       testlog.HCLogger(t),
+			lifecycle:    trtesting.NewMockTaskHooks(),
+			events:       &trtesting.MockEmitter{},
+			clientConfig: clientConfig,
+			envBuilder:   taskenv.NewBuilder(mock.Node(), alloc, task, clientConfig.Region),
+		}
+		secretHook := newSecretsHook(conf, []*structs.Secret{
+			{
+				Name:     "test_secret",
+				Provider: "nomad",
+				Path:     "testnomadvar",
+				Config: map[string]any{
+					"namespace": "default",
+				},
+			},
+		})
+
+		req := &interfaces.TaskPrestartRequest{
+			Alloc:   alloc,
+			Task:    task,
+			TaskDir: &allocdir.TaskDir{Dir: taskDir, SecretsDir: taskDir},
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		t.Cleanup(cancel)
+
+		err := secretHook.Prestart(ctx, req, nil)
+		must.NoError(t, err)
+
+		expected := map[string]string{
+			"secret.test_secret.key1": "value1",
+			"secret.test_secret.key2": "value2",
+		}
+		must.Eq(t, expected, secretHook.taskSecrets)
+	})
+
+	t.Run("returns early if context is cancelled", func(t *testing.T) {
+
+		secretsResp := `
+		{
+		  "CreateIndex": 812,
+		  "CreateTime": 1750782609539170600,
+		  "Items": {
+		    "key2": "value2",
+		    "key1": "value1"
+		  },
+		  "ModifyIndex": 812,
+		  "ModifyTime": 1750782609539170600,
+		  "Namespace": "default",
+		  "Path": "testnomadvar"
+		}
+		`
+		count := 0 // CT expects a nomad index header that increments, or else it continues polling
+		nomadServer := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Add("X-Nomad-Index", strconv.Itoa(count))
+			fmt.Fprintln(w, secretsResp)
+			count += 1
+		}))
+		t.Cleanup(nomadServer.Close)
+
+		l, d := bufconndialer.New()
+		nomadServer.Listener = l
+
+		nomadServer.Start()
+
+		clientConfig := config.DefaultConfig()
+		clientConfig.TemplateDialer = d
+		clientConfig.TemplateConfig.DisableSandbox = true
+
+		taskDir := t.TempDir()
+		alloc := mock.MinAlloc()
+		task := alloc.Job.TaskGroups[0].Tasks[0]
+
+		conf := &secretsHookConfig{
+
+			logger:       testlog.HCLogger(t),
+			lifecycle:    trtesting.NewMockTaskHooks(),
+			events:       &trtesting.MockEmitter{},
+			clientConfig: clientConfig,
+			envBuilder:   taskenv.NewBuilder(mock.Node(), alloc, task, clientConfig.Region),
+		}
+		secretHook := newSecretsHook(conf, []*structs.Secret{
+			{
+				Name:     "test_secret",
+				Provider: "nomad",
+				Path:     "testnomadvar",
+				Config: map[string]any{
+					"namespace": "default",
+				},
+			},
+		})
+
+		req := &interfaces.TaskPrestartRequest{
+			Alloc:   alloc,
+			Task:    task,
+			TaskDir: &allocdir.TaskDir{Dir: taskDir, SecretsDir: taskDir},
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		cancel() // cancel context to simulate task being stopped
+
+		err := secretHook.Prestart(ctx, req, nil)
+		must.NoError(t, err)
+
+		expected := map[string]string{}
+		must.Eq(t, expected, secretHook.taskSecrets)
+	})
+
+	t.Run("errors when failure building secret providers", func(t *testing.T) {
+		clientConfig := config.DefaultConfig()
+
+		taskDir := t.TempDir()
+		alloc := mock.MinAlloc()
+		task := alloc.Job.TaskGroups[0].Tasks[0]
+
+		conf := &secretsHookConfig{
+
+			logger:       testlog.HCLogger(t),
+			lifecycle:    trtesting.NewMockTaskHooks(),
+			events:       &trtesting.MockEmitter{},
+			clientConfig: clientConfig,
+			envBuilder:   taskenv.NewBuilder(mock.Node(), alloc, task, clientConfig.Region),
+		}
+
+		// give an invalid secret, in this case a nomad secret with bad namespace
+		secretHook := newSecretsHook(conf, []*structs.Secret{
+			{
+				Name:     "test_secret",
+				Provider: "nomad",
+				Path:     "testnomadvar",
+				Config: map[string]any{
+					"namespace": 123,
+				},
+			},
+		})
+
+		req := &interfaces.TaskPrestartRequest{
+			Alloc:   alloc,
+			Task:    task,
+			TaskDir: &allocdir.TaskDir{Dir: taskDir, SecretsDir: taskDir},
+		}
+
+		// Prestart should error and return after building secrets
+		err := secretHook.Prestart(context.Background(), req, nil)
+		must.Error(t, err)
+
+		expected := map[string]string{}
+		must.Eq(t, expected, secretHook.taskSecrets)
+	})
+}
diff --git a/client/allocrunner/taskrunner/task_runner_hooks.go b/client/allocrunner/taskrunner/task_runner_hooks.go
index 04faa353e..bbd57f661 100644
--- a/client/allocrunner/taskrunner/task_runner_hooks.go
+++ b/client/allocrunner/taskrunner/task_runner_hooks.go
@@ -55,40 +55,16 @@ func (tr *TaskRunner) initHooks() {
 
 	tr.logmonHookConfig = newLogMonHookConfig(task.Name, task.LogConfig, tr.taskDir.LogDir)
 
-	// Add the hook resources
-	tr.hookResources = &hookResources{}
-
-	// Create the task directory hook. This is run first to ensure the
+	// Create the task directory hook first. This is run first to ensure the
 	// directory path exists for other hooks.
-	alloc := tr.Alloc()
+	tr.hookResources = &hookResources{}
 	tr.runnerHooks = []interfaces.TaskHook{
 		newValidateHook(tr.clientConfig, hookLogger),
 		newDynamicUsersHook(tr.killCtx, tr.driverCapabilities.DynamicWorkloadUsers, tr.logger, tr.users),
 		newTaskDirHook(tr, hookLogger),
 		newIdentityHook(tr, hookLogger),
-		newLogMonHook(tr, hookLogger),
-		newDispatchHook(alloc, hookLogger),
-		newVolumeHook(tr, hookLogger),
-		newArtifactHook(tr, tr.getter, hookLogger),
-		newStatsHook(tr, tr.clientConfig.StatsCollectionInterval, tr.clientConfig.PublishAllocationMetrics, hookLogger),
-		newDeviceHook(tr.devicemanager, hookLogger),
-		newAPIHook(tr.shutdownCtx, tr.clientConfig.APIListenerRegistrar, hookLogger),
-		newWranglerHook(tr.wranglers, task.Name, alloc.ID, task.UsesCores(), hookLogger),
+		newConsulHook(hookLogger, tr),
 	}
-
-	// If the task has a CSI block, add the hook.
-	if task.CSIPluginConfig != nil {
-		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(
-			&csiPluginSupervisorHookConfig{
-				clientStateDirPath: tr.clientConfig.StateDir,
-				events:             tr,
-				runner:             tr,
-				lifecycle:          tr,
-				capabilities:       tr.driverCapabilities,
-				logger:             hookLogger,
-			}))
-	}
-
 	// If Vault is enabled, add the hook
 	if task.Vault != nil && tr.vaultClientFunc != nil {
 		tr.runnerHooks = append(tr.runnerHooks, newVaultHook(&vaultHookConfig{
@@ -105,13 +81,45 @@ func (tr *TaskRunner) initHooks() {
 		}))
 	}
 
+	if len(task.Secrets) > 0 {
+		tr.runnerHooks = append(tr.runnerHooks, newSecretsHook(&secretsHookConfig{
+			logger:         tr.logger,
+			lifecycle:      tr,
+			events:         tr,
+			clientConfig:   tr.clientConfig,
+			envBuilder:     tr.envBuilder,
+			nomadNamespace: tr.alloc.Job.Namespace,
+		}, task.Secrets))
+	}
+
+	alloc := tr.Alloc()
+	tr.runnerHooks = append(tr.runnerHooks, []interfaces.TaskHook{
+		newLogMonHook(tr, hookLogger),
+		newDispatchHook(alloc, hookLogger),
+		newVolumeHook(tr, hookLogger),
+		newArtifactHook(tr, tr.getter, hookLogger),
+		newStatsHook(tr, tr.clientConfig.StatsCollectionInterval, tr.clientConfig.PublishAllocationMetrics, hookLogger),
+		newDeviceHook(tr.devicemanager, hookLogger),
+		newAPIHook(tr.shutdownCtx, tr.clientConfig.APIListenerRegistrar, hookLogger),
+		newWranglerHook(tr.wranglers, task.Name, alloc.ID, task.UsesCores(), hookLogger),
+	}...)
+
+	// If the task has a CSI block, add the hook.
+	if task.CSIPluginConfig != nil {
+		tr.runnerHooks = append(tr.runnerHooks, newCSIPluginSupervisorHook(
+			&csiPluginSupervisorHookConfig{
+				clientStateDirPath: tr.clientConfig.StateDir,
+				events:             tr,
+				runner:             tr,
+				lifecycle:          tr,
+				capabilities:       tr.driverCapabilities,
+				logger:             hookLogger,
+			}))
+	}
+
 	// Get the consul namespace for the TG of the allocation.
 	consulNamespace := tr.alloc.ConsulNamespaceForTask(tr.taskName)
 
-	// Add the consul hook (populates task secret dirs and sets the environment if
-	// consul tokens are present for the task).
-	tr.runnerHooks = append(tr.runnerHooks, newConsulHook(hookLogger, tr))
-
 	// If there are templates is enabled, add the hook
 	if len(task.Templates) != 0 {
 		tr.runnerHooks = append(tr.runnerHooks, newTemplateHook(&templateHookConfig{