diff --git a/.changelog/15661.txt b/.changelog/15661.txt
new file mode 100644
index 0000000000..54915746ad
--- /dev/null
+++ b/.changelog/15661.txt
@@ -0,0 +1,3 @@
+```release-note:bug
+connect: Fixed issue where using Vault 1.11+ as CA provider in a secondary datacenter would eventually break Intermediate CAs
+```
diff --git a/.circleci/config.yml b/.circleci/config.yml
index a6c178e157..173a48f2b8 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -21,7 +21,6 @@ references:
     GIT_COMMITTER_NAME: circleci-consul
     S3_ARTIFACT_BUCKET: consul-dev-artifacts-v2
     BASH_ENV: .circleci/bash_env.sh
-    VAULT_BINARY_VERSION: 1.9.4
     GO_VERSION: 1.19.2
   envoy-versions: &supported_envoy_versions
     - &default_envoy_version "1.21.5"
@@ -32,6 +31,11 @@ references:
     - &default_nomad_version "1.3.3"
     - "1.2.10"
     - "1.1.16"
+  vault-versions: &supported_vault_versions
+    - &default_vault_version "1.12.2"
+    - "1.11.6"
+    - "1.10.9"
+    - "1.9.10"
   images:
     # When updating the Go version, remember to also update the versions in the
     # workflows section for go-test-lib jobs.
@@ -587,7 +591,6 @@ jobs:
       - setup_remote_docker
       - run: make ci.dev-docker
       - run: *notify-slack-failure
-
   nomad-integration-test: &NOMAD_TESTS
     docker:
       - image: docker.mirror.hashicorp.services/cimg/go:1.19
@@ -935,19 +938,26 @@ jobs:
           path: *TEST_RESULTS_DIR
       - run: *notify-slack-failure
 
-  # run integration tests for the connect ca providers
-  test-connect-ca-providers:
+  # run integration tests for the connect ca providers with vault
+  vault-integration-test:
     docker:
       - image: *GOLANG_IMAGE
+    parameters:
+      vault-version:
+        type: enum
+        enum: *supported_vault_versions
+        default: *default_vault_version
     environment:
       <<: *ENVIRONMENT
-    steps:
+      VAULT_BINARY_VERSION: << parameters.vault-version >>
+    steps: &VAULT_INTEGRATION_TEST_STEPS
       - run:
           name: Install vault
           command: |
             wget -q -O /tmp/vault.zip https://releases.hashicorp.com/vault/${VAULT_BINARY_VERSION}/vault_${VAULT_BINARY_VERSION}_linux_amd64.zip
             sudo unzip -d /usr/local/bin /tmp/vault.zip
             rm -rf /tmp/vault*
+            vault version
       - checkout
       - run: go mod download
       - run:
@@ -1067,7 +1077,6 @@ workflows:
           name: "lint-32bit"
           go-arch: "386"
           <<: *filter-ignore-non-go-branches
-      - test-connect-ca-providers: *filter-ignore-non-go-branches
       - go-test-arm64: *filter-ignore-non-go-branches
       - dev-build: *filter-ignore-non-go-branches
       - go-test:
@@ -1148,6 +1157,11 @@ workflows:
           matrix:
             parameters:
               nomad-version: *supported_nomad_versions
+      - vault-integration-test:
+          matrix:
+            parameters:
+              vault-version: *supported_vault_versions
+          <<: *filter-ignore-non-go-branches
       - envoy-integration-test:
           requires:
             - dev-build
diff --git a/agent/connect/ca/mock_Provider.go b/agent/connect/ca/mock_Provider.go
index bdc5d9c8e6..5c9bfcfcd2 100644
--- a/agent/connect/ca/mock_Provider.go
+++ b/agent/connect/ca/mock_Provider.go
@@ -107,7 +107,7 @@ func (_m *MockProvider) GenerateIntermediate() (string, error) {
 }
 
 // GenerateIntermediateCSR provides a mock function with given fields:
-func (_m *MockProvider) GenerateIntermediateCSR() (string, error) {
+func (_m *MockProvider) GenerateIntermediateCSR() (string, string, error) {
 	ret := _m.Called()
 
 	var r0 string
@@ -117,14 +117,21 @@ func (_m *MockProvider) GenerateIntermediateCSR() (string, error) {
 		r0 = ret.Get(0).(string)
 	}
 
-	var r1 error
-	if rf, ok := ret.Get(1).(func() error); ok {
+	var r1 string
+	if rf, ok := ret.Get(1).(func() string); ok {
 		r1 = rf()
 	} else {
-		r1 = ret.Error(1)
+		r1 = ret.Get(1).(string)
 	}
 
-	return r0, r1
+	var r2 error
+	if rf, ok := ret.Get(2).(func() error); ok {
+		r2 = rf()
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
 }
 
 // GenerateRoot provides a mock function with given fields:
@@ -149,12 +156,12 @@ func (_m *MockProvider) GenerateRoot() (RootResult, error) {
 }
 
 // SetIntermediate provides a mock function with given fields: intermediatePEM, rootPEM
-func (_m *MockProvider) SetIntermediate(intermediatePEM string, rootPEM string) error {
-	ret := _m.Called(intermediatePEM, rootPEM)
+func (_m *MockProvider) SetIntermediate(intermediatePEM string, rootPEM string, keyId string) error {
+	ret := _m.Called(intermediatePEM, rootPEM, keyId)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, string) error); ok {
-		r0 = rf(intermediatePEM, rootPEM)
+	if rf, ok := ret.Get(0).(func(string, string, string) error); ok {
+		r0 = rf(intermediatePEM, rootPEM, keyId)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/agent/connect/ca/provider.go b/agent/connect/ca/provider.go
index 6d585490fa..c80ccd82be 100644
--- a/agent/connect/ca/provider.go
+++ b/agent/connect/ca/provider.go
@@ -179,14 +179,17 @@ type SecondaryProvider interface {
 	//
 	// After the certificate is signed, SecondaryProvider.SetIntermediate will
 	// be called to store the intermediate CA.
-	GenerateIntermediateCSR() (string, error)
+	//
+	// The second return value is an opaque string meant to be passed back to
+	// the subsequent call to SetIntermediate.
+	GenerateIntermediateCSR() (string, string, error)
 
 	// SetIntermediate is called to store a newly signed leaf signing certificate and
 	// the chain of certificates back to the root CA certificate.
 	//
 	// The provider should save the certificates and use them to
 	// Provider.Sign leaf certificates.
-	SetIntermediate(intermediatePEM, rootPEM string) error
+	SetIntermediate(intermediatePEM, rootPEM, opaque string) error
 }
 
 // RootResult is the result returned by PrimaryProvider.GenerateRoot.
diff --git a/agent/connect/ca/provider_aws.go b/agent/connect/ca/provider_aws.go
index cef0c7ddbe..97e075b114 100644
--- a/agent/connect/ca/provider_aws.go
+++ b/agent/connect/ca/provider_aws.go
@@ -75,6 +75,8 @@ type AWSProvider struct {
 	logger          hclog.Logger
 }
 
+var _ Provider = (*AWSProvider)(nil)
+
 // NewAWSProvider returns a new AWSProvider
 func NewAWSProvider(logger hclog.Logger) *AWSProvider {
 	return &AWSProvider{logger: logger}
@@ -498,23 +500,24 @@ func (a *AWSProvider) signCSR(csrPEM string, templateARN string, ttl time.Durati
 }
 
 // GenerateIntermediateCSR implements Provider
-func (a *AWSProvider) GenerateIntermediateCSR() (string, error) {
+func (a *AWSProvider) GenerateIntermediateCSR() (string, string, error) {
 	if a.isPrimary {
-		return "", fmt.Errorf("provider is the root certificate authority, " +
+		return "", "", fmt.Errorf("provider is the root certificate authority, " +
 			"cannot generate an intermediate CSR")
 	}
 
 	err := a.ensureCA()
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	// We should have the CA created now and should be able to generate the CSR.
-	return a.getCACSR()
+	pem, err := a.getCACSR()
+	return pem, "", err
 }
 
 // SetIntermediate implements Provider
-func (a *AWSProvider) SetIntermediate(intermediatePEM string, rootPEM string) error {
+func (a *AWSProvider) SetIntermediate(intermediatePEM string, rootPEM string, _ string) error {
 	err := a.ensureCA()
 	if err != nil {
 		return err
diff --git a/agent/connect/ca/provider_aws_test.go b/agent/connect/ca/provider_aws_test.go
index b73d4a95d5..f6105d3dc8 100644
--- a/agent/connect/ca/provider_aws_test.go
+++ b/agent/connect/ca/provider_aws_test.go
@@ -216,7 +216,7 @@ func TestAWSBootstrapAndSignSecondary(t *testing.T) {
 			"ExistingARN": p2State[AWSStateCAARNKey],
 		})
 		p2 = testAWSProvider(t, cfg2)
-		require.NoError(t, p2.SetIntermediate(newIntPEM, newRootPEM))
+		require.NoError(t, p2.SetIntermediate(newIntPEM, newRootPEM, ""))
 
 		root, err = p1.GenerateRoot()
 		require.NoError(t, err)
diff --git a/agent/connect/ca/provider_consul.go b/agent/connect/ca/provider_consul.go
index 40eb2f4457..0493fffe75 100644
--- a/agent/connect/ca/provider_consul.go
+++ b/agent/connect/ca/provider_consul.go
@@ -49,6 +49,8 @@ type ConsulProvider struct {
 	sync.RWMutex
 }
 
+var _ Provider = (*ConsulProvider)(nil)
+
 // NewConsulProvider returns a new ConsulProvider that is ready to be used.
 func NewConsulProvider(delegate ConsulProviderStateDelegate, logger hclog.Logger) *ConsulProvider {
 	return &ConsulProvider{Delegate: delegate, logger: logger}
@@ -205,26 +207,26 @@ func (c *ConsulProvider) GenerateRoot() (RootResult, error) {
 
 // GenerateIntermediateCSR creates a private key and generates a CSR
 // for another datacenter's root to sign.
-func (c *ConsulProvider) GenerateIntermediateCSR() (string, error) {
+func (c *ConsulProvider) GenerateIntermediateCSR() (string, string, error) {
 	providerState, err := c.getState()
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	if c.isPrimary {
-		return "", fmt.Errorf("provider is the root certificate authority, " +
+		return "", "", fmt.Errorf("provider is the root certificate authority, " +
 			"cannot generate an intermediate CSR")
 	}
 
 	// Create a new private key and CSR.
 	signer, pk, err := connect.GeneratePrivateKeyWithConfig(c.config.PrivateKeyType, c.config.PrivateKeyBits)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	csr, err := connect.CreateCACSR(c.spiffeID, signer)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 
 	// Write the new provider state to the store.
@@ -235,15 +237,15 @@ func (c *ConsulProvider) GenerateIntermediateCSR() (string, error) {
 		ProviderState: &newState,
 	}
 	if _, err := c.Delegate.ApplyCARequest(args); err != nil {
-		return "", err
+		return "", "", err
 	}
 
-	return csr, nil
+	return csr, "", nil
 }
 
 // SetIntermediate validates that the given intermediate is for the right private key
 // and writes the given intermediate and root certificates to the state.
-func (c *ConsulProvider) SetIntermediate(intermediatePEM, rootPEM string) error {
+func (c *ConsulProvider) SetIntermediate(intermediatePEM, rootPEM, _ string) error {
 	providerState, err := c.getState()
 	if err != nil {
 		return err
diff --git a/agent/connect/ca/provider_consul_test.go b/agent/connect/ca/provider_consul_test.go
index a4df05b767..fea9aab450 100644
--- a/agent/connect/ca/provider_consul_test.go
+++ b/agent/connect/ca/provider_consul_test.go
@@ -417,7 +417,7 @@ func TestConsulProvider_SignIntermediate(t *testing.T) {
 func testSignIntermediateCrossDC(t *testing.T, provider1, provider2 Provider) {
 
 	// Get the intermediate CSR from provider2.
-	csrPEM, err := provider2.GenerateIntermediateCSR()
+	csrPEM, opaque, err := provider2.GenerateIntermediateCSR()
 	require.NoError(t, err)
 	csr, err := connect.ParseCSR(csrPEM)
 	require.NoError(t, err)
@@ -430,7 +430,7 @@ func testSignIntermediateCrossDC(t *testing.T, provider1, provider2 Provider) {
 	rootPEM := root.PEM
 
 	// Give the new intermediate to provider2 to use.
-	require.NoError(t, provider2.SetIntermediate(intermediatePEM, rootPEM))
+	require.NoError(t, provider2.SetIntermediate(intermediatePEM, rootPEM, opaque))
 
 	// Have provider2 sign a leaf cert and make sure the chain is correct.
 	spiffeService := &connect.SpiffeIDService{
diff --git a/agent/connect/ca/provider_vault.go b/agent/connect/ca/provider_vault.go
index 48eacb9da3..d42c9037cc 100644
--- a/agent/connect/ca/provider_vault.go
+++ b/agent/connect/ca/provider_vault.go
@@ -79,6 +79,8 @@ type VaultProvider struct {
 	isConsulMountedIntermediate bool
 }
 
+var _ Provider = (*VaultProvider)(nil)
+
 func NewVaultProvider(logger hclog.Logger) *VaultProvider {
 	return &VaultProvider{
 		stopWatcher: func() {},
@@ -365,14 +367,13 @@ func (v *VaultProvider) GenerateRoot() (RootResult, error) {
 // GenerateIntermediateCSR creates a private key and generates a CSR
 // for another datacenter's root to sign, overwriting the intermediate backend
 // in the process.
-func (v *VaultProvider) GenerateIntermediateCSR() (string, error) {
+func (v *VaultProvider) GenerateIntermediateCSR() (string, string, error) {
 	if v.isPrimary {
-		return "", fmt.Errorf("provider is the root certificate authority, " +
+		return "", "", fmt.Errorf("provider is the root certificate authority, " +
 			"cannot generate an intermediate CSR")
 	}
 
-	csr, _, err := v.generateIntermediateCSR()
-	return csr, err
+	return v.generateIntermediateCSR()
 }
 
 func (v *VaultProvider) setupIntermediatePKIPath() error {
@@ -486,7 +487,7 @@ func (v *VaultProvider) generateIntermediateCSR() (string, string, error) {
 
 // SetIntermediate writes the incoming intermediate and root certificates to the
 // intermediate backend (as a chain).
-func (v *VaultProvider) SetIntermediate(intermediatePEM, rootPEM string) error {
+func (v *VaultProvider) SetIntermediate(intermediatePEM, rootPEM, keyId string) error {
 	if v.isPrimary {
 		return fmt.Errorf("cannot set an intermediate using another root in the primary datacenter")
 	}
@@ -496,13 +497,21 @@ func (v *VaultProvider) SetIntermediate(intermediatePEM, rootPEM string) error {
 		return err
 	}
 
-	_, err = v.writeNamespaced(v.config.IntermediatePKINamespace, v.config.IntermediatePKIPath+"intermediate/set-signed", map[string]interface{}{
+	importResp, err := v.writeNamespaced(v.config.IntermediatePKINamespace, v.config.IntermediatePKIPath+"intermediate/set-signed", map[string]interface{}{
 		"certificate": intermediatePEM,
 	})
 	if err != nil {
 		return err
 	}
 
+	// Vault 1.11+ will return a non-nil response from intermediate/set-signed
+	if importResp != nil {
+		err := v.setDefaultIntermediateIssuer(importResp, keyId)
+		if err != nil {
+			return fmt.Errorf("failed to update default intermediate issuer: %w", err)
+		}
+	}
+
 	return nil
 }
 
diff --git a/agent/connect/ca/provider_vault_test.go b/agent/connect/ca/provider_vault_test.go
index 9fbe9f337d..13326150e4 100644
--- a/agent/connect/ca/provider_vault_test.go
+++ b/agent/connect/ca/provider_vault_test.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/hashicorp/consul/agent/connect"
 	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/sdk/testutil"
 	"github.com/hashicorp/consul/sdk/testutil/retry"
 )
 
@@ -903,7 +904,6 @@ func TestVaultProvider_ReconfigureIntermediateTTL(t *testing.T) {
 }
 
 func TestVaultCAProvider_GenerateIntermediate(t *testing.T) {
-
 	SkipIfVaultNotPresent(t)
 
 	provider, _ := testVaultProviderWithConfig(t, true, nil)
@@ -924,6 +924,76 @@ func TestVaultCAProvider_GenerateIntermediate(t *testing.T) {
 	require.NotEqual(t, orig, new)
 }
 
+func TestVaultCAProvider_GenerateIntermediate_inSecondary(t *testing.T) {
+	SkipIfVaultNotPresent(t)
+
+	// Primary DC will be a consul provider.
+	conf := testConsulCAConfig()
+	delegate := newMockDelegate(t, conf)
+	primaryProvider := TestConsulProvider(t, delegate)
+	require.NoError(t, primaryProvider.Configure(testProviderConfig(conf)))
+	_, err := primaryProvider.GenerateRoot()
+	require.NoError(t, err)
+
+	// Ensure that we don't configure vault to try and mint leafs that
+	// outlive their CA during the test (which hard fails in vault).
+	intermediateCertTTL := getIntermediateCertTTL(t, conf)
+	leafCertTTL := intermediateCertTTL - 4*time.Hour
+
+	provider, _ := testVaultProviderWithConfig(t, false, map[string]any{
+		"LeafCertTTL": []uint8(leafCertTTL.String()),
+	})
+
+	var origIntermediate string
+	testutil.RunStep(t, "initialize secondary provider", func(t *testing.T) {
+		// Get the intermediate CSR from provider.
+		csrPEM, issuerID, err := provider.GenerateIntermediateCSR()
+		require.NoError(t, err)
+		csr, err := connect.ParseCSR(csrPEM)
+		require.NoError(t, err)
+
+		// Sign the CSR with primaryProvider.
+		intermediatePEM, err := primaryProvider.SignIntermediate(csr)
+		require.NoError(t, err)
+		root, err := primaryProvider.GenerateRoot()
+		require.NoError(t, err)
+		rootPEM := root.PEM
+
+		// Give the new intermediate to provider to use.
+		require.NoError(t, provider.SetIntermediate(intermediatePEM, rootPEM, issuerID))
+
+		origIntermediate, err = provider.ActiveIntermediate()
+		require.NoError(t, err)
+	})
+
+	testutil.RunStep(t, "renew secondary provider", func(t *testing.T) {
+		// Get the intermediate CSR from provider.
+		csrPEM, issuerID, err := provider.GenerateIntermediateCSR()
+		require.NoError(t, err)
+		csr, err := connect.ParseCSR(csrPEM)
+		require.NoError(t, err)
+
+		// Sign the CSR with primaryProvider.
+		intermediatePEM, err := primaryProvider.SignIntermediate(csr)
+		require.NoError(t, err)
+		root, err := primaryProvider.GenerateRoot()
+		require.NoError(t, err)
+		rootPEM := root.PEM
+
+		// Give the new intermediate to provider to use.
+		require.NoError(t, provider.SetIntermediate(intermediatePEM, rootPEM, issuerID))
+
+		// This test was created to ensure that our calls to Vault
+		// returns a new Intermediate certificate and further calls
+		// to ActiveIntermediate return the same new cert.
+		newActiveIntermediate, err := provider.ActiveIntermediate()
+		require.NoError(t, err)
+
+		require.NotEqual(t, origIntermediate, newActiveIntermediate)
+		require.Equal(t, intermediatePEM, newActiveIntermediate)
+	})
+}
+
 func TestVaultCAProvider_VaultManaged(t *testing.T) {
 
 	SkipIfVaultNotPresent(t)
diff --git a/agent/consul/leader_connect_ca.go b/agent/consul/leader_connect_ca.go
index 8d2bf9cb86..848976da1e 100644
--- a/agent/consul/leader_connect_ca.go
+++ b/agent/consul/leader_connect_ca.go
@@ -1052,7 +1052,7 @@ func (c *CAManager) primaryRenewIntermediate(provider ca.Provider, newActiveRoot
 // provider.
 // Should only be called while the state lock is held by setting the state to non-ready.
 func (c *CAManager) secondaryRequestNewSigningCert(provider ca.Provider, newActiveRoot *structs.CARoot) error {
-	csr, err := provider.GenerateIntermediateCSR()
+	csr, opaque, err := provider.GenerateIntermediateCSR()
 	if err != nil {
 		return err
 	}
@@ -1064,7 +1064,7 @@ func (c *CAManager) secondaryRequestNewSigningCert(provider ca.Provider, newActi
 		return nil
 	}
 
-	if err := provider.SetIntermediate(intermediatePEM, newActiveRoot.RootCert); err != nil {
+	if err := provider.SetIntermediate(intermediatePEM, newActiveRoot.RootCert, opaque); err != nil {
 		return fmt.Errorf("Failed to set the intermediate certificate with the CA provider: %v", err)
 	}
 
diff --git a/agent/consul/leader_connect_ca_test.go b/agent/consul/leader_connect_ca_test.go
index d78f38d9ec..adc065627c 100644
--- a/agent/consul/leader_connect_ca_test.go
+++ b/agent/consul/leader_connect_ca_test.go
@@ -248,11 +248,11 @@ func (m *mockCAProvider) State() (map[string]string, error)     { return nil, ni
 func (m *mockCAProvider) GenerateRoot() (ca.RootResult, error) {
 	return ca.RootResult{PEM: m.rootPEM}, nil
 }
-func (m *mockCAProvider) GenerateIntermediateCSR() (string, error) {
+func (m *mockCAProvider) GenerateIntermediateCSR() (string, string, error) {
 	m.callbackCh <- "provider/GenerateIntermediateCSR"
-	return "", nil
+	return "", "", nil
 }
-func (m *mockCAProvider) SetIntermediate(intermediatePEM, rootPEM string) error {
+func (m *mockCAProvider) SetIntermediate(intermediatePEM, rootPEM, _ string) error {
 	m.callbackCh <- "provider/SetIntermediate"
 	return nil
 }
@@ -1033,9 +1033,14 @@ func setupPrimaryCA(t *testing.T, client *vaultapi.Client, path string, rootPEM
 	})
 	require.NoError(t, err, "failed to sign intermediate")
 
+	cert := intermediate.Data["certificate"].(string)
+
 	var buf strings.Builder
-	buf.WriteString(lib.EnsureTrailingNewline(intermediate.Data["certificate"].(string)))
-	buf.WriteString(lib.EnsureTrailingNewline(rootPEM))
+	buf.WriteString(lib.EnsureTrailingNewline(cert))
+	if !strings.Contains(strings.TrimSpace(cert), strings.TrimSpace(rootPEM)) {
+		// Vault < v1.11 included the root in the output of sign-intermediate.
+		buf.WriteString(lib.EnsureTrailingNewline(rootPEM))
+	}
 
 	_, err = client.Logical().Write(path+"/intermediate/set-signed", map[string]interface{}{
 		"certificate": buf.String(),