From e8f8702955ece9cfeff6e5ee6d13709ea24ac9d1 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Fri, 25 Sep 2026 10:01:58 +0100 Subject: [PATCH 01/11] svid/service: add support for WIT-SVID TTL Signed-off-by: Sorin Dumitru --- pkg/server/api/svid/v1/service.go | 3 +- pkg/server/api/svid/v1/service_test.go | 43 +++++++++++++++++++++++--- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/pkg/server/api/svid/v1/service.go b/pkg/server/api/svid/v1/service.go index e7f1bcd6d8..624cb44d16 100644 --- a/pkg/server/api/svid/v1/service.go +++ b/pkg/server/api/svid/v1/service.go @@ -577,8 +577,7 @@ func (s *Service) newWITSVID(ctx context.Context, param *svidv1.NewWITSVIDParams Algorithm: param.SigningAlgorithm, Key: publicKey, }, - // TODO: add WIT specific TTL (https://github.com/spiffe/spire/issues/6535) - TTL: time.Duration(entry.GetX509SvidTtl()) * time.Second, + TTL: time.Duration(entry.GetWitSvidTtl()) * time.Second, }) if err != nil { return &svidv1.BatchNewWITSVIDResponse_Result{ diff --git a/pkg/server/api/svid/v1/service_test.go b/pkg/server/api/svid/v1/service_test.go index aafc1554a0..2d4828131b 100644 --- a/pkg/server/api/svid/v1/service_test.go +++ b/pkg/server/api/svid/v1/service_test.go @@ -2238,24 +2238,33 @@ func TestServiceBatchNewWITSVID(t *testing.T) { ParentId: api.ProtoFromID(agentID), SpiffeId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/workload2"}, } + workloadEntryWithWITTTL := &types.Entry{ + Id: "workload-with-wit-ttl", + ParentId: api.ProtoFromID(agentID), + SpiffeId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/workload-with-wit-ttl"}, + WitSvidTtl: 90, + } invalidEntry := &types.Entry{ Id: "invalid", SpiffeId: &types.SPIFFEID{}, ParentId: api.ProtoFromID(agentID), } - test.ef.entries = []*types.Entry{workloadEntry1, workloadEntry2, invalidEntry} + test.ef.entries = []*types.Entry{workloadEntry1, workloadEntry2, workloadEntryWithWITTTL, invalidEntry} now := test.ca.Clock().Now().UTC() expiresAtFromCA := now.Add(test.ca.WITSVIDTTL()).Unix() expiresAtFromCAStr := strconv.FormatInt(expiresAtFromCA, 10) + expiresAtFromEntry := now.Add(time.Duration(workloadEntryWithWITTTL.WitSvidTtl) * time.Second).Unix() + expiresAtFromEntryStr := strconv.FormatInt(expiresAtFromEntry, 10) _, invalidPublicKeyErr := x509.ParsePKIXPublicKey([]byte{1, 2, 3}) require.Error(t, invalidPublicKeyErr) type expectResult struct { - entry *types.Entry - status *types.Status + entry *types.Entry + status *types.Status + expiresAt int64 } for _, tt := range []struct { @@ -2294,6 +2303,28 @@ func TestServiceBatchNewWITSVID(t *testing.T) { }, }, }, + }, { + name: "success with entry WIT-SVID TTL", + reqs: []string{workloadEntryWithWITTTL.Id}, + expectResults: []*expectResult{ + { + entry: workloadEntryWithWITTTL, + expiresAt: expiresAtFromEntry, + }, + }, + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "success", + telemetry.Type: "audit", + telemetry.RegistrationID: workloadEntryWithWITTTL.Id, + telemetry.ExpiresAt: expiresAtFromEntryStr, + telemetry.SPIFFEID: "spiffe://example.org/workload-with-wit-ttl", + }, + }, + }, }, { name: "WIT is disabled", disableWITSVIDs: true, @@ -2775,7 +2806,11 @@ func TestServiceBatchNewWITSVID(t *testing.T) { require.Equal(t, entry.SpiffeId.Path, svid.Id.Path) require.NotEmpty(t, svid.Token) require.Equal(t, now.Unix(), svid.IssuedAt) - require.Equal(t, expiresAtFromCA, svid.ExpiresAt) + expectedExpiresAt := expiresAtFromCA + if expect.expiresAt != 0 { + expectedExpiresAt = expect.expiresAt + } + require.Equal(t, expectedExpiresAt, svid.ExpiresAt) } }) } From 755a8939d671a32be989a9b24ee3122b85642d63 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Tue, 29 Sep 2026 20:22:02 +0100 Subject: [PATCH 02/11] Update spire-plugin-sdk version Signed-off-by: Sorin Dumitru --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index c4a0e66ad2..03be18b3c4 100644 --- a/go.mod +++ b/go.mod @@ -82,7 +82,7 @@ require ( github.com/smallstep/pkcs7 v0.2.3 github.com/spiffe/go-spiffe/v2 v2.8.2 github.com/spiffe/spire-api-sdk v1.2.5-0.20260924193931-38b883fcd6d1 - github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260617144146-5dcde407c4d1 + github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260929185137-e961422db712 github.com/stretchr/testify v1.12.1 github.com/uber-go/tally/v4 v4.1.17 github.com/valyala/fastjson v1.6.10 diff --git a/go.sum b/go.sum index 65e21273cd..85a9999223 100644 --- a/go.sum +++ b/go.sum @@ -736,6 +736,8 @@ github.com/spiffe/spire-api-sdk v1.2.5-0.20260924193931-38b883fcd6d1 h1:O0ZhDye/ github.com/spiffe/spire-api-sdk v1.2.5-0.20260924193931-38b883fcd6d1/go.mod h1:9hXJcMzatM1KwAtBDO3s6HccDCic++/5c2yOc5Iln8Y= github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260617144146-5dcde407c4d1 h1:gN8nWPdRSxYIACrgdPBuVwhNe6ABzpvUVza7cP8Yy38= github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260617144146-5dcde407c4d1/go.mod h1:QvrRDiBlXiJ7kNd176ZHsF5eklxxeTRgJSu2CXe0MKw= +github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260929185137-e961422db712 h1:dMxwHOpAc6jH4NtU4mgqHLE/daLwa2gDW04Wz4lIAIQ= +github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260929185137-e961422db712/go.mod h1:QvrRDiBlXiJ7kNd176ZHsF5eklxxeTRgJSu2CXe0MKw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= From 3e95e3c8f820ca3c0eadbe33aa6401d707e1e3c9 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Tue, 29 Sep 2026 20:44:26 +0100 Subject: [PATCH 03/11] Return Unimplemented for all plugins except spire Signed-off-by: Sorin Dumitru --- pkg/server/plugin/upstreamauthority/awspca/pca.go | 4 ++++ pkg/server/plugin/upstreamauthority/awspca/pca_test.go | 5 +++++ pkg/server/plugin/upstreamauthority/awssecret/awssecret.go | 4 ++++ .../plugin/upstreamauthority/awssecret/awssecret_test.go | 5 +++++ .../plugin/upstreamauthority/certmanager/certmanager.go | 4 ++++ .../plugin/upstreamauthority/certmanager/certmanager_test.go | 5 +++++ pkg/server/plugin/upstreamauthority/disk/disk.go | 4 ++++ pkg/server/plugin/upstreamauthority/disk/disk_test.go | 5 +++++ pkg/server/plugin/upstreamauthority/ejbca/ejbca.go | 4 ++++ pkg/server/plugin/upstreamauthority/gcpcas/gcpcas.go | 4 ++++ pkg/server/plugin/upstreamauthority/vault/vault.go | 4 ++++ pkg/server/plugin/upstreamauthority/vault/vault_test.go | 5 +++++ 12 files changed, 53 insertions(+) diff --git a/pkg/server/plugin/upstreamauthority/awspca/pca.go b/pkg/server/plugin/upstreamauthority/awspca/pca.go index 654233fea5..016b3b073d 100644 --- a/pkg/server/plugin/upstreamauthority/awspca/pca.go +++ b/pkg/server/plugin/upstreamauthority/awspca/pca.go @@ -311,6 +311,10 @@ func (*PCAPlugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyRe return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (*PCAPlugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *PCAPlugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/awspca/pca_test.go b/pkg/server/plugin/upstreamauthority/awspca/pca_test.go index d5b8bc6639..f7963e0b97 100644 --- a/pkg/server/plugin/upstreamauthority/awspca/pca_test.go +++ b/pkg/server/plugin/upstreamauthority/awspca/pca_test.go @@ -496,6 +496,11 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(aws_pca): publishing upstream is unsupported") assert.Nil(t, jwtAuthorities) assert.Nil(t, stream) + + witAuthorities, witStream, err := ua.PublishWITKey(context.Background(), &common.PublicKey{Kid: "ID", PkixBytes: pkixBytes}) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(aws_pca): publishing upstream is unsupported") + assert.Nil(t, witAuthorities) + assert.Nil(t, witStream) } func setupDescribeCertificateAuthority(client *pcaClientFake, status string, err error) { diff --git a/pkg/server/plugin/upstreamauthority/awssecret/awssecret.go b/pkg/server/plugin/upstreamauthority/awssecret/awssecret.go index c75af3840f..5849f8a93c 100644 --- a/pkg/server/plugin/upstreamauthority/awssecret/awssecret.go +++ b/pkg/server/plugin/upstreamauthority/awssecret/awssecret.go @@ -196,6 +196,10 @@ func (p *Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReq return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (p *Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/awssecret/awssecret_test.go b/pkg/server/plugin/upstreamauthority/awssecret/awssecret_test.go index a2e907cec4..abbf95f335 100644 --- a/pkg/server/plugin/upstreamauthority/awssecret/awssecret_test.go +++ b/pkg/server/plugin/upstreamauthority/awssecret/awssecret_test.go @@ -420,4 +420,9 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(awssecret): publishing upstream is unsupported") assert.Nil(t, jwtAuthorities) assert.Nil(t, stream) + + witAuthorities, witStream, err := ua.PublishWITKey(context.Background(), &common.PublicKey{Kid: "ID", PkixBytes: pkixBytes}) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(awssecret): publishing upstream is unsupported") + assert.Nil(t, witAuthorities) + assert.Nil(t, witStream) } diff --git a/pkg/server/plugin/upstreamauthority/certmanager/certmanager.go b/pkg/server/plugin/upstreamauthority/certmanager/certmanager.go index 2bcd2e9a6a..9e317cddf8 100644 --- a/pkg/server/plugin/upstreamauthority/certmanager/certmanager.go +++ b/pkg/server/plugin/upstreamauthority/certmanager/certmanager.go @@ -261,6 +261,10 @@ func (*Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReque return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (*Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/certmanager/certmanager_test.go b/pkg/server/plugin/upstreamauthority/certmanager/certmanager_test.go index 6d9a7ceaf8..b91e83047c 100644 --- a/pkg/server/plugin/upstreamauthority/certmanager/certmanager_test.go +++ b/pkg/server/plugin/upstreamauthority/certmanager/certmanager_test.go @@ -371,4 +371,9 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(cert-manager): publishing upstream is unsupported") assert.Nil(t, jwtAuthorities) assert.Nil(t, stream) + + witAuthorities, witStream, err := ua.PublishWITKey(context.Background(), &common.PublicKey{Kid: "ID", PkixBytes: pkixBytes}) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(cert-manager): publishing upstream is unsupported") + assert.Nil(t, witAuthorities) + assert.Nil(t, witStream) } diff --git a/pkg/server/plugin/upstreamauthority/disk/disk.go b/pkg/server/plugin/upstreamauthority/disk/disk.go index 199649b9d3..ef39395a86 100644 --- a/pkg/server/plugin/upstreamauthority/disk/disk.go +++ b/pkg/server/plugin/upstreamauthority/disk/disk.go @@ -159,6 +159,10 @@ func (*Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReque return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (*Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/disk/disk_test.go b/pkg/server/plugin/upstreamauthority/disk/disk_test.go index 46279df189..48402d9155 100644 --- a/pkg/server/plugin/upstreamauthority/disk/disk_test.go +++ b/pkg/server/plugin/upstreamauthority/disk/disk_test.go @@ -192,6 +192,11 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(disk): publishing upstream is unsupported") assert.Nil(t, jwtAuthorities) assert.Nil(t, stream) + + witAuthorities, witStream, err := ua.PublishWITKey(context.Background(), &common.PublicKey{Kid: "ID", PkixBytes: pkixBytes}) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(disk): publishing upstream is unsupported") + assert.Nil(t, witAuthorities) + assert.Nil(t, witStream) } func TestConfigure(t *testing.T) { diff --git a/pkg/server/plugin/upstreamauthority/ejbca/ejbca.go b/pkg/server/plugin/upstreamauthority/ejbca/ejbca.go index 01f6fdeadf..305d5ccb2c 100644 --- a/pkg/server/plugin/upstreamauthority/ejbca/ejbca.go +++ b/pkg/server/plugin/upstreamauthority/ejbca/ejbca.go @@ -323,6 +323,10 @@ func (p *Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReq return status.Error(codes.Unimplemented, "publishing JWT keys is not supported by the EJBCA UpstreamAuthority plugin") } +func (p *Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing WIT keys is not supported by the EJBCA UpstreamAuthority plugin") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/gcpcas/gcpcas.go b/pkg/server/plugin/upstreamauthority/gcpcas/gcpcas.go index 4437238386..7db65a87cd 100644 --- a/pkg/server/plugin/upstreamauthority/gcpcas/gcpcas.go +++ b/pkg/server/plugin/upstreamauthority/gcpcas/gcpcas.go @@ -167,6 +167,10 @@ func (p *Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReq return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (p *Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/vault/vault.go b/pkg/server/plugin/upstreamauthority/vault/vault.go index d14be10e2c..e721e7978d 100644 --- a/pkg/server/plugin/upstreamauthority/vault/vault.go +++ b/pkg/server/plugin/upstreamauthority/vault/vault.go @@ -236,6 +236,10 @@ func (*Plugin) PublishJWTKeyAndSubscribe(*upstreamauthorityv1.PublishJWTKeyReque return status.Error(codes.Unimplemented, "publishing upstream is unsupported") } +func (*Plugin) PublishWITKeyAndSubscribe(*upstreamauthorityv1.PublishWITKeyRequest, upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + return status.Error(codes.Unimplemented, "publishing upstream is unsupported") +} + func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") } diff --git a/pkg/server/plugin/upstreamauthority/vault/vault_test.go b/pkg/server/plugin/upstreamauthority/vault/vault_test.go index 5f1e6ba587..31e1adff67 100644 --- a/pkg/server/plugin/upstreamauthority/vault/vault_test.go +++ b/pkg/server/plugin/upstreamauthority/vault/vault_test.go @@ -792,6 +792,11 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(vault): publishing upstream is unsupported") assert.Nil(t, jwtAuthorities) assert.Nil(t, stream) + + witAuthorities, witStream, err := ua.PublishWITKey(context.Background(), &common.PublicKey{Kid: "ID", PkixBytes: pkixBytes}) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(vault): publishing upstream is unsupported") + assert.Nil(t, witAuthorities) + assert.Nil(t, witStream) } func getTestConfigureRequest(t *testing.T, addr string, tpl string) string { From f0bfb99c930503a89574e12c5ba254210c3bd63f Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Tue, 29 Sep 2026 20:51:13 +0100 Subject: [PATCH 04/11] Implement new API for spire upstreamauthority Signed-off-by: Sorin Dumitru --- doc/plugin_server_upstreamauthority_spire.md | 2 +- .../spire/fake_handlers_test.go | 18 +++ .../plugin/upstreamauthority/spire/spire.go | 95 ++++++++++----- .../spire/spire_server_client.go | 15 +++ .../upstreamauthority/spire/spire_test.go | 109 ++++++++++++++++-- 5 files changed, 200 insertions(+), 39 deletions(-) diff --git a/doc/plugin_server_upstreamauthority_spire.md b/doc/plugin_server_upstreamauthority_spire.md index 780aa89b35..d440b82ca8 100644 --- a/doc/plugin_server_upstreamauthority_spire.md +++ b/doc/plugin_server_upstreamauthority_spire.md @@ -4,7 +4,7 @@ The `spire` plugin uses credentials fetched from the Workload API to call an ups The SVIDs minted in a nested configuration are valid in the entire trust domain, not only in the scope of the server that originated the SVID. -In the case of X509-SVID, this is easily achieved because of the chaining semantics that X.509 has. On the other hand, for JWT-SVID, this capability is accomplished by propagating every JWT-SVID public signing key to the whole topology. +In the case of X509-SVID, this is easily achieved because of the chaining semantics that X.509 has. On the other hand, for JWT-SVID and WIT-SVID, this capability is accomplished by propagating every JWT-SVID and WIT-SVID public signing key to the whole topology. The plugin accepts the following configuration options: diff --git a/pkg/server/plugin/upstreamauthority/spire/fake_handlers_test.go b/pkg/server/plugin/upstreamauthority/spire/fake_handlers_test.go index 68ed2f91b9..4cf830d70d 100644 --- a/pkg/server/plugin/upstreamauthority/spire/fake_handlers_test.go +++ b/pkg/server/plugin/upstreamauthority/spire/fake_handlers_test.go @@ -161,6 +161,13 @@ func (h *handler) appendKey(key *types.JWTKey) *types.Bundle { return cloneBundle(h.bundle) } +func (h *handler) appendWITKey(key *types.WITKey) *types.Bundle { + h.mtx.Lock() + defer h.mtx.Unlock() + h.bundle.WitAuthorities = append(h.bundle.WitAuthorities, key) + return cloneBundle(h.bundle) +} + func (h *handler) appendRootCA(rootCA *types.X509Certificate) *types.Bundle { //nolint: unparam // Keeping return for future use h.mtx.Lock() defer h.mtx.Unlock() @@ -230,6 +237,17 @@ func (h *handler) PublishJWTAuthority(_ context.Context, req *bundlev1.PublishJW }, nil } +func (h *handler) PublishWITAuthority(_ context.Context, req *bundlev1.PublishWITAuthorityRequest) (*bundlev1.PublishWITAuthorityResponse, error) { + if err := h.getError(); err != nil { + return nil, err + } + + b := h.appendWITKey(req.WitAuthority) + return &bundlev1.PublishWITAuthorityResponse{ + WitAuthorities: b.WitAuthorities, + }, nil +} + func (h *handler) setDownstreamResponse(downstreamResponse *svidv1.NewDownstreamX509CAResponse) { h.mtx.Lock() defer h.mtx.Unlock() diff --git a/pkg/server/plugin/upstreamauthority/spire/spire.go b/pkg/server/plugin/upstreamauthority/spire/spire.go index c9debf84b9..9bb992b6a5 100644 --- a/pkg/server/plugin/upstreamauthority/spire/spire.go +++ b/pkg/server/plugin/upstreamauthority/spire/spire.go @@ -3,6 +3,7 @@ package spireplugin import ( "context" "fmt" + "slices" "sync" "time" @@ -17,6 +18,7 @@ import ( "github.com/spiffe/spire/pkg/common/catalog" "github.com/spiffe/spire/pkg/common/coretypes/bundle" "github.com/spiffe/spire/pkg/common/coretypes/jwtkey" + "github.com/spiffe/spire/pkg/common/coretypes/witkey" "github.com/spiffe/spire/pkg/common/coretypes/x509certificate" "github.com/spiffe/spire/pkg/common/idutil" "github.com/spiffe/spire/pkg/common/pluginconf" @@ -228,9 +230,15 @@ func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLoca jwtKeys = append(jwtKeys, pluginKey) } + witKeys, err := witkey.ToPluginFromAPIProtos(serverBundle.WitAuthorities) + if err != nil { + return err + } + err = stream.Send(&upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: rootCAs, UpstreamJwtKeys: jwtKeys, + UpstreamWitKeys: witKeys, }) if err != nil { return err @@ -238,6 +246,7 @@ func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLoca p.setBundleX509Authorities(rootCAs) p.setBundleJWTAuthorities(jwtKeys) + p.setBundleWITAuthorities(witKeys) ticker := p.clk.Ticker(internalPollFreq) defer ticker.Stop() @@ -249,16 +258,20 @@ func (p *Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLoca // the next internalPollFreq tick. updateCh := p.getBundleUpdateCh() - newRootCAs := p.getBundle().X509Authorities - newJWTKeys := p.getBundle().JwtAuthorities - if !areRootsEqual(rootCAs, newRootCAs) || !arePublicKeysEqual(jwtKeys, newJWTKeys) { + currentBundle := p.getBundle() + newRootCAs := currentBundle.X509Authorities + newJWTKeys := currentBundle.JwtAuthorities + newWITKeys := currentBundle.WitAuthorities + if !areProtosEqual(rootCAs, newRootCAs) || !areProtosEqual(jwtKeys, newJWTKeys) || !areProtosEqual(witKeys, newWITKeys) { err := stream.Send(&upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: newRootCAs, UpstreamJwtKeys: newJWTKeys, + UpstreamWitKeys: newWITKeys, }) if err == nil { rootCAs = newRootCAs jwtKeys = newJWTKeys + witKeys = newWITKeys } } select { @@ -310,6 +323,40 @@ func (p *Plugin) PublishJWTKeyAndSubscribe(req *upstreamauthorityv1.PublishJWTKe return nil } +func (p *Plugin) PublishWITKeyAndSubscribe(req *upstreamauthorityv1.PublishWITKeyRequest, stream upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + err := p.subscribeToPolling(stream.Context()) + if err != nil { + return err + } + defer p.unsubscribeToPolling() + + witKey, err := witkey.ToAPIFromPluginProto(req.WitKey) + if err != nil { + return status.Errorf(codes.Internal, "unable to parse WITKey into api WITKey: %v", err) + } + + resp, err := p.serverClient.publishWITAuthority(stream.Context(), witKey) + if err != nil { + return err + } + + witKeys, err := witkey.ToPluginFromAPIProtos(resp) + if err != nil { + return err + } + + p.setBundleWITAuthorities(witKeys) + + err = stream.Send(&upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: witKeys, + }) + if err != nil { + p.log.Error("Cannot send upstream WIT keys", "error", err) + return err + } + return nil +} + func (p *Plugin) pollBundleUpdates(ctx context.Context) { ticker := p.clk.Ticker(upstreamPollFreq) defer ticker.Stop() @@ -348,10 +395,10 @@ func (p *Plugin) pollBundleUpdates(ctx context.Context) { // setBundleIfVersionMatches updates currentBundle only when bundleVersion // still equals expectedVersion. This prevents a fetch that started before a -// local mutation (setBundleJWTAuthorities / setBundleX509Authorities) from -// overwriting the newer local state. bundleVersion is intentionally not -// incremented here; it is only incremented by the local-mutation helpers so -// that it remains a reliable guard against concurrent upstream fetches. +// local mutation (setBundle*Authorities) from overwriting the newer local +// state. bundleVersion is intentionally not incremented here; it is only +// incremented by the local-mutation helpers so that it remains a reliable +// guard against concurrent upstream fetches. func (p *Plugin) setBundleIfVersionMatches(b *types.Bundle, expectedVersion uint64) error { p.bundleMtx.Lock() defer p.bundleMtx.Unlock() @@ -398,6 +445,14 @@ func (p *Plugin) setBundleJWTAuthorities(keys []*plugintypes.JWTKey) { p.signalBundleUpdate() } +func (p *Plugin) setBundleWITAuthorities(keys []*plugintypes.WITKey) { + p.bundleMtx.Lock() + defer p.bundleMtx.Unlock() + p.currentBundle.WitAuthorities = keys + p.bundleVersion++ + p.signalBundleUpdate() +} + func (p *Plugin) setBundleX509Authorities(rootCAs []*plugintypes.X509Certificate) { p.bundleMtx.Lock() defer p.bundleMtx.Unlock() @@ -455,26 +510,8 @@ func (p *Plugin) startPolling(streamCtx context.Context) error { return nil } -func areRootsEqual(a, b []*plugintypes.X509Certificate) bool { - if len(a) != len(b) { - return false - } - for i, root := range a { - if !proto.Equal(root, b[i]) { - return false - } - } - return true -} - -func arePublicKeysEqual(a, b []*plugintypes.JWTKey) bool { - if len(a) != len(b) { - return false - } - for i, pk := range a { - if !proto.Equal(pk, b[i]) { - return false - } - } - return true +func areProtosEqual[T proto.Message](a, b []T) bool { + return slices.EqualFunc(a, b, func(x, y T) bool { + return proto.Equal(x, y) + }) } diff --git a/pkg/server/plugin/upstreamauthority/spire/spire_server_client.go b/pkg/server/plugin/upstreamauthority/spire/spire_server_client.go index e827ef9504..f1aa7b9e4a 100644 --- a/pkg/server/plugin/upstreamauthority/spire/spire_server_client.go +++ b/pkg/server/plugin/upstreamauthority/spire/spire_server_client.go @@ -151,6 +151,21 @@ func (c *serverClient) publishJWTAuthority(ctx context.Context, key *types.JWTKe return resp.JwtAuthorities, nil } +// publishWITAuthority publishes a WIT key to the server +func (c *serverClient) publishWITAuthority(ctx context.Context, key *types.WITKey) ([]*types.WITKey, error) { + c.mtx.RLock() + defer c.mtx.RUnlock() + + resp, err := c.bundleClient.PublishWITAuthority(ctx, &bundlev1.PublishWITAuthorityRequest{ + WitAuthority: key, + }) + if err != nil { + return nil, status.Errorf(codes.Internal, "failed to push WIT authority: %v", err) + } + + return resp.WitAuthorities, nil +} + // getBundle gets the bundle for the trust domain of the server func (c *serverClient) getBundle(ctx context.Context) (*types.Bundle, error) { c.mtx.RLock() diff --git a/pkg/server/plugin/upstreamauthority/spire/spire_test.go b/pkg/server/plugin/upstreamauthority/spire/spire_test.go index 2bdec6499d..2b301dd0cb 100644 --- a/pkg/server/plugin/upstreamauthority/spire/spire_test.go +++ b/pkg/server/plugin/upstreamauthority/spire/spire_test.go @@ -299,7 +299,7 @@ func TestMintX509CA(t *testing.T) { return } - x509Authorities, _, stream, err := ua.SubscribeToLocalBundle(ctx) + x509Authorities, _, _, stream, err := ua.SubscribeToLocalBundle(ctx) require.NoError(t, err) require.NotNil(t, stream) require.NotNil(t, x509Authorities) @@ -329,7 +329,7 @@ func TestMintX509CA(t *testing.T) { mockClock.Add(upstreamPollFreq) // Get bundle update - bundleUpdateResp, _, err := stream.RecvLocalBundleUpdate() + bundleUpdateResp, _, _, err := stream.RecvLocalBundleUpdate() require.NoError(t, err) require.Equal(t, append(expectedX509Authorities, expectedServerUpdateAuthority...), bundleUpdateResp) @@ -338,7 +338,7 @@ func TestMintX509CA(t *testing.T) { cancel() // Verify stream is closed - resp, _, err := stream.RecvLocalBundleUpdate() + resp, _, _, err := stream.RecvLocalBundleUpdate() spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Canceled, "upstreamauthority(spire): context canceled") require.Nil(t, resp) }) @@ -379,7 +379,7 @@ func TestPublishJWTKey(t *testing.T) { require.NoError(t, err) require.NotNil(t, upstreamJwtKeysFromPublish) - _, upstreamJwtKeys, stream, err := ua.SubscribeToLocalBundle(ctx) + _, upstreamJwtKeys, _, stream, err := ua.SubscribeToLocalBundle(ctx) require.NoError(t, err) require.NotNil(t, stream) require.NotNil(t, upstreamJwtKeys) @@ -398,7 +398,7 @@ func TestPublishJWTKey(t *testing.T) { mockClock.Add(upstreamPollFreq) // Get bundle update - _, resp, err := stream.RecvLocalBundleUpdate() + _, resp, _, err := stream.RecvLocalBundleUpdate() require.NoError(t, err) require.Len(t, resp, 4) require.Equal(t, resp[3].Kid, "kid-3") @@ -408,7 +408,7 @@ func TestPublishJWTKey(t *testing.T) { cancel() // Verify stream is closed - _, resp, err = stream.RecvLocalBundleUpdate() + _, resp, _, err = stream.RecvLocalBundleUpdate() require.Nil(t, resp) spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Canceled, "upstreamauthority(spire): context canceled") @@ -424,6 +424,81 @@ func TestPublishJWTKey(t *testing.T) { spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Internal, "upstreamauthority(spire): failed to push JWT authority: rpc error: code = Unknown desc = some erro") } +func TestPublishWITKey(t *testing.T) { + ca := testca.New(t, trustDomain) + serverCert, serverKey := ca.CreateX509Certificate( + testca.WithID(spiffeid.RequireFromPath(trustDomain, "/spire/server")), + ) + s := ca.CreateX509SVID( + spiffeid.RequireFromPath(trustDomain, "/workload"), + ) + svidCert, svidKey, err := s.MarshalRaw() + require.NoError(t, err) + + key := testkey.NewEC256(t) + pkixBytes, err := x509.MarshalPKIXPublicKey(key.Public()) + require.NoError(t, err) + + key2 := testkey.NewEC256(t) + pkixBytes2, err := x509.MarshalPKIXPublicKey(key2.Public()) + require.NoError(t, err) + + // Setup servers + mockClock := clock.NewMock(t) + server := testHandler{} + server.startTestServers(t, mockClock, ca, serverCert, serverKey, svidCert, svidKey) + ua := newWithDefault(t, mockClock, server.sAPIServer.addr, server.wAPIServer.workloadAPIAddr) + + // Get first response + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + upstreamWitKeysFromPublish, _, err := ua.PublishWITKey(ctx, &common.PublicKey{ + Kid: "wit-kid-1", + PkixBytes: pkixBytes, + }) + require.NoError(t, err) + require.Len(t, upstreamWitKeysFromPublish, 1) + assert.Equal(t, "wit-kid-1", upstreamWitKeysFromPublish[0].Kid) + + _, upstreamJwtKeys, upstreamWitKeys, stream, err := ua.SubscribeToLocalBundle(ctx) + require.NoError(t, err) + require.NotNil(t, stream) + require.Len(t, upstreamJwtKeys, 2) + require.Equal(t, upstreamWitKeysFromPublish, upstreamWitKeys) + + // Update bundle to trigger another response. Advance the clock past the + // upstream poll frequency to trigger a fetch; the plugin notifies + // SubscribeToLocalBundle immediately via bundleUpdated. + server.sAPIServer.appendWITKey(&types.WITKey{KeyId: "wit-kid-2", PublicKey: pkixBytes2}) + mockClock.Add(upstreamPollFreq) + mockClock.Add(upstreamPollFreq) + + // Get bundle update + _, _, resp, err := stream.RecvLocalBundleUpdate() + require.NoError(t, err) + require.Len(t, resp, 2) + require.Equal(t, "wit-kid-2", resp[1].Kid) + require.Equal(t, pkixBytes2, resp[1].PkixBytes) + + // Cancel ctx to stop getting updates + cancel() + + // Verify stream is closed + _, _, resp, err = stream.RecvLocalBundleUpdate() + require.Nil(t, resp) + spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Canceled, "upstreamauthority(spire): context canceled") + + // Fail to push WIT authority + ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + server.sAPIServer.setError(errors.New("some error")) + upstreamWitKeys, _, err = ua.PublishWITKey(ctx, &common.PublicKey{ + Kid: "wit-kid-3", + PkixBytes: pkixBytes, + }) + require.Nil(t, upstreamWitKeys) + spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Internal, "upstreamauthority(spire): failed to push WIT authority: rpc error: code = Unknown desc = some error") +} + func TestGetTrustBundle(t *testing.T) { ca := testca.New(t, trustDomain) serverCert, serverKey := ca.CreateX509Certificate( @@ -443,12 +518,13 @@ func TestGetTrustBundle(t *testing.T) { // Get first response ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - upstreamX509Roots, upstreamJwtKeys, stream, err := ua.SubscribeToLocalBundle(ctx) + upstreamX509Roots, upstreamJwtKeys, upstreamWitKeys, stream, err := ua.SubscribeToLocalBundle(ctx) require.NoError(t, err) require.NotNil(t, stream) require.Len(t, upstreamX509Roots, 1) require.Len(t, upstreamJwtKeys, 2) + require.Empty(t, upstreamWitKeys) assert.Equal(t, upstreamJwtKeys[0].Kid, "C6vs25welZOx6WksNYfbMfiw9l96pMnD") assert.Equal(t, upstreamJwtKeys[1].Kid, "gHTCunJbefYtnZnTctd84xeRWyMrEsWD") @@ -464,19 +540,34 @@ func TestGetTrustBundle(t *testing.T) { mockClock.Add(upstreamPollFreq) // Get bundle update - upstreamX509Roots, upstreamJwtKeys, err = stream.RecvLocalBundleUpdate() + upstreamX509Roots, upstreamJwtKeys, upstreamWitKeys, err = stream.RecvLocalBundleUpdate() require.NoError(t, err) require.Len(t, upstreamX509Roots, 1) require.Len(t, upstreamJwtKeys, 3) require.Equal(t, upstreamJwtKeys[2].Kid, "kid") require.Equal(t, upstreamJwtKeys[2].PkixBytes, pkixBytes) + require.Empty(t, upstreamWitKeys) + + // Update WIT authorities to trigger another response + server.sAPIServer.appendWITKey(&types.WITKey{KeyId: "wit-kid", PublicKey: pkixBytes}) + mockClock.Add(upstreamPollFreq) + mockClock.Add(upstreamPollFreq) + + upstreamX509Roots, upstreamJwtKeys, upstreamWitKeys, err = stream.RecvLocalBundleUpdate() + require.NoError(t, err) + require.Len(t, upstreamX509Roots, 1) + require.Len(t, upstreamJwtKeys, 3) + require.Len(t, upstreamWitKeys, 1) + require.Equal(t, "wit-kid", upstreamWitKeys[0].Kid) + require.Equal(t, pkixBytes, upstreamWitKeys[0].PkixBytes) cancel() // Verify stream is closed - upstreamX509Roots, upstreamJwtKeys, err = stream.RecvLocalBundleUpdate() + upstreamX509Roots, upstreamJwtKeys, upstreamWitKeys, err = stream.RecvLocalBundleUpdate() require.Nil(t, upstreamX509Roots) require.Nil(t, upstreamJwtKeys) + require.Nil(t, upstreamWitKeys) spiretest.RequireGRPCStatusHasPrefix(t, err, codes.Canceled, "upstreamauthority(spire): context canceled") } From d2c12d4e918be46ba39f37329d2425375b523f22 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Wed, 30 Sep 2026 09:23:00 +0100 Subject: [PATCH 05/11] upstreamauthority: make use of new RPC Signed-off-by: Sorin Dumitru --- .../upstreamauthority/upstreamauthority.go | 27 +- pkg/server/plugin/upstreamauthority/v1.go | 108 +++++++- .../plugin/upstreamauthority/v1_test.go | 232 +++++++++++++++++- 3 files changed, 353 insertions(+), 14 deletions(-) diff --git a/pkg/server/plugin/upstreamauthority/upstreamauthority.go b/pkg/server/plugin/upstreamauthority/upstreamauthority.go index 74ac442685..b945b75634 100644 --- a/pkg/server/plugin/upstreamauthority/upstreamauthority.go +++ b/pkg/server/plugin/upstreamauthority/upstreamauthority.go @@ -33,6 +33,16 @@ type UpstreamAuthority interface { // will return io.EOF when called. PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) (jwtAuthorities []*common.PublicKey, stream UpstreamJWTAuthorityStream, err error) + // PublishWITKey publishes the given WIT key with the upstream authority. + // Support for this method is optional. Implementations that do not support + // publishing WIT keys upstream return NotImplemented. + // The function returns the latest set of upstream WIT authorities and a + // stream for streaming upstream WIT authority updates. The returned stream + // MUST be closed when the caller is no longer interested in updates. If + // the upstream authority does not support streaming updates, the stream + // will return io.EOF when called. + PublishWITKey(ctx context.Context, witKey *common.PublicKey) (witAuthorities []*common.PublicKey, stream UpstreamWITAuthorityStream, err error) + // SubscribeToLocalBundle can be used to sync the local trust bundle with // the upstream trust bundle. // Support for this method is optional but strongly recommended. @@ -41,7 +51,7 @@ type UpstreamAuthority interface { // MUST be closed when the caller is no longer interested in updates. If // the upstream authority does not support streaming updates, the stream // will return io.EOF when called. - SubscribeToLocalBundle(ctx context.Context) (x509CAs []*x509certificate.X509Authority, jwtAuthorities []*common.PublicKey, stream LocalBundleUpdateStream, err error) + SubscribeToLocalBundle(ctx context.Context) (x509CAs []*x509certificate.X509Authority, jwtAuthorities []*common.PublicKey, witAuthorities []*common.PublicKey, stream LocalBundleUpdateStream, err error) } type UpstreamX509AuthorityStream interface { @@ -70,13 +80,26 @@ type UpstreamJWTAuthorityStream interface { Close() } +type UpstreamWITAuthorityStream interface { + // RecvUpstreamWITAuthorities returns the latest set of upstream WIT + // authorities. The call blocks until the update is received, the Close() + // method is called, or the context originally passed into PublishWITKey is + // canceled. If the function returns an error, no more updates will be + // available over the stream. + RecvUpstreamWITAuthorities() ([]*common.PublicKey, error) + + // Close() closes the stream. It MUST be called by callers of PublishWITKey + // when they are done with the stream. + Close() +} + type LocalBundleUpdateStream interface { // RecvLocalBundleUpdate returns the latest local trust domain bundle // The call blocks until the update is received, the Close() // method is called, or the context originally passed into GetTrustBundle is // canceled. If the function returns an error, no more updates will be // available over the stream. - RecvLocalBundleUpdate() ([]*x509certificate.X509Authority, []*common.PublicKey, error) + RecvLocalBundleUpdate() (x509CAs []*x509certificate.X509Authority, jwtAuthorities []*common.PublicKey, witAuthorities []*common.PublicKey, err error) // Close() closes the stream. It MUST be called by callers of GetTrustBundle // when they are done with the stream. diff --git a/pkg/server/plugin/upstreamauthority/v1.go b/pkg/server/plugin/upstreamauthority/v1.go index 53f9a9eede..c5adf86f9f 100644 --- a/pkg/server/plugin/upstreamauthority/v1.go +++ b/pkg/server/plugin/upstreamauthority/v1.go @@ -10,6 +10,7 @@ import ( upstreamauthorityv1 "github.com/spiffe/spire-plugin-sdk/proto/spire/plugin/server/upstreamauthority/v1" "github.com/spiffe/spire-plugin-sdk/proto/spire/plugin/types" "github.com/spiffe/spire/pkg/common/coretypes/jwtkey" + "github.com/spiffe/spire/pkg/common/coretypes/witkey" "github.com/spiffe/spire/pkg/common/coretypes/x509certificate" "github.com/spiffe/spire/pkg/common/plugin" "github.com/spiffe/spire/pkg/common/util" @@ -93,7 +94,44 @@ func (v1 *V1) PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) (_ [] return jwtKeys, &v1UpstreamJWTAuthorityStream{v1: v1, stream: stream, cancel: cancel}, nil } -func (v1 *V1) SubscribeToLocalBundle(ctx context.Context) (_ []*x509certificate.X509Authority, _ []*common.PublicKey, _ LocalBundleUpdateStream, err error) { +// PublishWITKey provides the V1 implementation of the UpstreamAuthority +// interface method of the same name. +func (v1 *V1) PublishWITKey(ctx context.Context, witKey *common.PublicKey) (_ []*common.PublicKey, _ UpstreamWITAuthorityStream, err error) { + ctx, cancel := context.WithCancel(ctx) + defer func() { + // Only cancel the context if the function fails. Otherwise, the + // returned stream will be in charge of cancellation. + if err != nil { + defer cancel() + } + }() + + pluginWITKey, err := witkey.ToPluginFromCommonProto(witKey) + if err != nil { + return nil, nil, err + } + + stream, err := v1.UpstreamAuthorityPluginClient.PublishWITKeyAndSubscribe(ctx, &upstreamauthorityv1.PublishWITKeyRequest{ + WitKey: pluginWITKey, + }) + if err != nil { + return nil, nil, v1.WrapErr(err) + } + + resp, err := stream.Recv() + if err != nil { + return nil, nil, v1.streamError(err) + } + + witKeys, err := v1.toCommonWITProtos(resp.UpstreamWitKeys) + if err != nil { + return nil, nil, err + } + + return witKeys, &v1UpstreamWITAuthorityStream{v1: v1, stream: stream, cancel: cancel}, nil +} + +func (v1 *V1) SubscribeToLocalBundle(ctx context.Context) (_ []*x509certificate.X509Authority, _ []*common.PublicKey, _ []*common.PublicKey, _ LocalBundleUpdateStream, err error) { ctx, cancel := context.WithCancel(ctx) defer func() { // Only cancel the context if the function fails. Otherwise, the @@ -105,25 +143,30 @@ func (v1 *V1) SubscribeToLocalBundle(ctx context.Context) (_ []*x509certificate. stream, err := v1.UpstreamAuthorityPluginClient.SubscribeToLocalBundle(ctx, &upstreamauthorityv1.SubscribeToLocalBundleRequest{}) if err != nil { - return nil, nil, nil, v1.WrapErr(err) + return nil, nil, nil, nil, v1.WrapErr(err) } resp, err := stream.Recv() if err != nil { - return nil, nil, nil, v1.streamError(err) + return nil, nil, nil, nil, v1.streamError(err) } jwtKeys, err := v1.toCommonProtos(resp.UpstreamJwtKeys) if err != nil { - return nil, nil, nil, err + return nil, nil, nil, nil, err + } + + witKeys, err := v1.toCommonWITProtos(resp.UpstreamWitKeys) + if err != nil { + return nil, nil, nil, nil, err } x509Authorities, err := v1.parseX509Authorities(resp.UpstreamX509Roots) if err != nil { - return nil, nil, nil, err + return nil, nil, nil, nil, err } - return x509Authorities, jwtKeys, &v1LocalBundleStream{v1: v1, stream: stream, cancel: cancel}, nil + return x509Authorities, jwtKeys, witKeys, &v1LocalBundleStream{v1: v1, stream: stream, cancel: cancel}, nil } func (v1 *V1) parseMintX509CAFirstResponse(resp *upstreamauthorityv1.MintX509CAResponse) ([]*x509.Certificate, []*x509certificate.X509Authority, error) { @@ -180,6 +223,14 @@ func (v1 *V1) toCommonProtos(pbs []*types.JWTKey) ([]*common.PublicKey, error) { return jwtKeys, nil } +func (v1 *V1) toCommonWITProtos(pbs []*types.WITKey) ([]*common.PublicKey, error) { + witKeys, err := witkey.ToCommonFromPluginProtos(pbs) + if err != nil { + return nil, v1.Errorf(codes.Internal, "invalid plugin response: %v", err) + } + return witKeys, nil +} + type v1UpstreamX509AuthorityStream struct { v1 *V1 stream upstreamauthorityv1.UpstreamAuthority_MintX509CAAndSubscribeClient @@ -242,22 +293,53 @@ func (s *v1UpstreamJWTAuthorityStream) Close() { s.cancel() } +type v1UpstreamWITAuthorityStream struct { + v1 *V1 + stream upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeClient + cancel context.CancelFunc +} + +func (s *v1UpstreamWITAuthorityStream) RecvUpstreamWITAuthorities() ([]*common.PublicKey, error) { + for { + resp, err := s.stream.Recv() + switch { + case errors.Is(err, io.EOF): + // This is expected if the plugin does not support streaming + // authority updates. + return nil, io.EOF + case err != nil: + return nil, s.v1.WrapErr(err) + } + + witKeys, err := s.v1.toCommonWITProtos(resp.UpstreamWitKeys) + if err != nil { + s.v1.Log.WithError(err).Warn("Failed to parse a WIT key update from the upstream authority plugin. Please report this bug.") + continue + } + return witKeys, nil + } +} + +func (s *v1UpstreamWITAuthorityStream) Close() { + s.cancel() +} + type v1LocalBundleStream struct { v1 *V1 stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleClient cancel context.CancelFunc } -func (s *v1LocalBundleStream) RecvLocalBundleUpdate() ([]*x509certificate.X509Authority, []*common.PublicKey, error) { +func (s *v1LocalBundleStream) RecvLocalBundleUpdate() ([]*x509certificate.X509Authority, []*common.PublicKey, []*common.PublicKey, error) { for { resp, err := s.stream.Recv() switch { case errors.Is(err, io.EOF): // This is expected if the plugin does not support streaming // authority updates. - return nil, nil, err + return nil, nil, nil, err case err != nil: - return nil, nil, s.v1.WrapErr(err) + return nil, nil, nil, s.v1.WrapErr(err) } x509Authorities, err := s.v1.parseX509Authorities(resp.UpstreamX509Roots) @@ -272,7 +354,13 @@ func (s *v1LocalBundleStream) RecvLocalBundleUpdate() ([]*x509certificate.X509Au continue } - return x509Authorities, jwtKeys, nil + witKeys, err := s.v1.toCommonWITProtos(resp.UpstreamWitKeys) + if err != nil { + s.v1.Log.WithError(err).Warn("Failed to parse a WIT key update from the upstream authority plugin. Please report this bug.") + continue + } + + return x509Authorities, jwtKeys, witKeys, nil } } diff --git a/pkg/server/plugin/upstreamauthority/v1_test.go b/pkg/server/plugin/upstreamauthority/v1_test.go index f18cb9e416..3d68588467 100644 --- a/pkg/server/plugin/upstreamauthority/v1_test.go +++ b/pkg/server/plugin/upstreamauthority/v1_test.go @@ -17,6 +17,7 @@ import ( "github.com/spiffe/spire-plugin-sdk/proto/spire/plugin/types" "github.com/spiffe/spire/pkg/common/catalog" "github.com/spiffe/spire/pkg/common/coretypes/jwtkey" + "github.com/spiffe/spire/pkg/common/coretypes/witkey" "github.com/spiffe/spire/pkg/common/coretypes/x509certificate" "github.com/spiffe/spire/pkg/server/plugin/upstreamauthority" "github.com/spiffe/spire/proto/spire/common" @@ -38,6 +39,8 @@ const ( var ( jwtKeyPKIX, _ = x509.MarshalPKIXPublicKey(testkey.MustEC256().Public()) jwtKey = &common.PublicKey{Kid: "KEYID", PkixBytes: jwtKeyPKIX, NotAfter: 12345} + witKeyPKIX, _ = x509.MarshalPKIXPublicKey(testkey.MustEC256().Public()) + witKey = &common.PublicKey{Kid: "WITKEYID", PkixBytes: witKeyPKIX, NotAfter: 12345} ) func TestV1MintX509CA(t *testing.T) { @@ -396,6 +399,155 @@ func TestV1PublishJWTKey(t *testing.T) { } } +func TestV1PublishWITKey(t *testing.T) { + key := testkey.NewEC256(t) + pkixBytes, err := x509.MarshalPKIXPublicKey(key.Public()) + require.NoError(t, err) + + expectedUpstreamWITKeys := []*common.PublicKey{ + { + Kid: "UPSTREAM KEY", + PkixBytes: pkixBytes, + }, + } + + withoutID := &upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: []*types.WITKey{ + {PublicKey: pkixBytes}, + }, + } + withoutPKIXData := &upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: []*types.WITKey{ + {KeyId: "UPSTREAM KEY"}, + }, + } + withMalformedPKIXData := &upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: []*types.WITKey{ + {KeyId: "UPSTREAM KEY", PublicKey: []byte("JUNK")}, + }, + } + withIDAndPKIXData := &upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: witkey.RequireToPluginFromCommonProtos(expectedUpstreamWITKeys), + } + + builder := BuildV1() + + for _, tt := range []struct { + test string + builder *V1Builder + expectCode codes.Code + expectMessage string + expectStreamUpdates bool + expectStreamCode codes.Code + expectStreamMessage string + expectLogs []spiretest.LogEntry + }{ + { + test: "plugin does not implement RPC", + builder: builder.WithoutPublishWITKey(), + expectCode: codes.Unimplemented, + expectMessage: "upstreamauthority(test): method PublishWITKeyAndSubscribe not implemented", + }, + { + test: "plugin returns before sending first response", + builder: builder.WithPreSendError(nil), + expectCode: codes.Internal, + expectMessage: "upstreamauthority(test): plugin closed stream unexpectedly", + }, + { + test: "plugin fails before sending first response", + builder: builder.WithPreSendError(errors.New("ohno")), + expectCode: codes.Unknown, + expectMessage: "upstreamauthority(test): ohno", + }, + { + test: "plugin response missing WIT key ID", + builder: builder.WithPublishWITKeyResponse(withoutID), + expectCode: codes.Internal, + expectMessage: "upstreamauthority(test): invalid plugin response: missing key ID for WIT key", + }, + { + test: "plugin response missing PKIX data", + builder: builder.WithPublishWITKeyResponse(withoutPKIXData), + expectCode: codes.Internal, + expectMessage: `upstreamauthority(test): invalid plugin response: missing public key for WIT key "UPSTREAM KEY"`, + }, + { + test: "plugin response has malformed PKIX data", + builder: builder.WithPublishWITKeyResponse(withMalformedPKIXData), + expectCode: codes.Internal, + expectMessage: `upstreamauthority(test): invalid plugin response: failed to unmarshal public key for WIT key "UPSTREAM KEY"`, + }, + { + test: "success but plugin does not support streaming updates", + builder: builder.WithPublishWITKeyResponse(withIDAndPKIXData), + }, + { + test: "success and plugin supports streaming updates", + builder: builder. + WithPublishWITKeyResponse(withIDAndPKIXData). + WithPublishWITKeyResponse(withIDAndPKIXData), + expectStreamUpdates: true, + }, + { + test: "second plugin response is bad (missing ID)", + builder: builder. + WithPublishWITKeyResponse(withIDAndPKIXData). + WithPublishWITKeyResponse(withoutID), + expectStreamUpdates: false, // because the second response is bad and ignored + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.WarnLevel, + Message: "Failed to parse a WIT key update from the upstream authority plugin. Please report this bug.", + Data: logrus.Fields{ + logrus.ErrorKey: "rpc error: code = Internal desc = upstreamauthority(test): invalid plugin response: missing key ID for WIT key", + }, + }, + }, + }, + { + test: "plugin fails to stream updates", + builder: builder. + WithPublishWITKeyResponse(withIDAndPKIXData). + WithPostSendError(errors.New("ohno")), + expectStreamUpdates: true, + expectStreamCode: codes.Unknown, + expectStreamMessage: "upstreamauthority(test): ohno", + }, + } { + t.Run(tt.test, func(t *testing.T) { + log, logHook := test.NewNullLogger() + + ua := tt.builder.WithLog(log).Load(t) + upstreamWITKeys, upstreamWITKeysStream, err := ua.PublishWITKey(context.Background(), witKey) + spiretest.RequireGRPCStatusHasPrefix(t, err, tt.expectCode, tt.expectMessage) + if tt.expectCode != codes.OK { + return + } + require.NotNil(t, upstreamWITKeysStream, "stream should have been returned") + defer upstreamWITKeysStream.Close() + spiretest.AssertProtoListEqual(t, expectedUpstreamWITKeys, upstreamWITKeys) + + switch { + case !tt.expectStreamUpdates: + upstreamWITKeys, err := upstreamWITKeysStream.RecvUpstreamWITAuthorities() + assert.Equal(t, io.EOF, err, "stream should have returned EOF") + assert.Nil(t, upstreamWITKeys, "no WIT keys should be received") + case tt.expectStreamCode == codes.OK: + upstreamWITKeys, err := upstreamWITKeysStream.RecvUpstreamWITAuthorities() + assert.NoError(t, err, "stream should have returned update") + spiretest.AssertProtoListEqual(t, expectedUpstreamWITKeys, upstreamWITKeys) + default: + upstreamWITKeys, err = upstreamWITKeysStream.RecvUpstreamWITAuthorities() + spiretest.RequireGRPCStatusHasPrefix(t, err, tt.expectStreamCode, tt.expectStreamMessage) + assert.Nil(t, upstreamWITKeys) + } + + spiretest.AssertLogs(t, logHook.AllEntries(), tt.expectLogs) + }) + } +} + func TestV1SubscribeToLocalBundle(t *testing.T) { upstreamCA := testca.New(t, spiffeid.RequireTrustDomainFromString("example.org")) @@ -417,6 +569,12 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { PkixBytes: pkixBytes, }, } + expectedUpstreamWITKeys := []*common.PublicKey{ + { + Kid: "UPSTREAM WIT KEY", + PkixBytes: pkixBytes, + }, + } noJwtAuthorities := &upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: validUpstreamX509Roots, } @@ -424,6 +582,14 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { fullResponse := &upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: validUpstreamX509Roots, UpstreamJwtKeys: jwtkey.RequireToPluginFromCommonProtos(expectedUpstreamJWTKeys), + UpstreamWitKeys: witkey.RequireToPluginFromCommonProtos(expectedUpstreamWITKeys), + } + + malformedWITKeys := &upstreamauthorityv1.SubscribeToLocalBundleResponse{ + UpstreamX509Roots: validUpstreamX509Roots, + UpstreamWitKeys: []*types.WITKey{ + {PublicKey: pkixBytes}, + }, } builder := BuildV1() @@ -450,6 +616,29 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { expectCode: codes.Unknown, expectMessage: "upstreamauthority(test): ohno", }, + { + test: "plugin response has malformed WIT key", + builder: builder.WithSubscribeToLocalBundleResponse(malformedWITKeys), + expectCode: codes.Internal, + expectMessage: "upstreamauthority(test): invalid plugin response: missing key ID for WIT key", + }, + { + test: "second plugin response has malformed WIT key", + builder: builder. + WithSubscribeToLocalBundleResponse(fullResponse). + WithSubscribeToLocalBundleResponse(malformedWITKeys), + expectCode: codes.OK, + expectStreamUpdates: false, // because the second response is bad and ignored + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.WarnLevel, + Message: "Failed to parse a WIT key update from the upstream authority plugin. Please report this bug.", + Data: logrus.Fields{ + logrus.ErrorKey: "rpc error: code = Internal desc = upstreamauthority(test): invalid plugin response: missing key ID for WIT key", + }, + }, + }, + }, { test: "success with empty JWT authorities", builder: builder.WithSubscribeToLocalBundleResponse(noJwtAuthorities), @@ -493,7 +682,7 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { ua := tt.builder.WithLog(log).Load(t) - _, _, stream, err := ua.SubscribeToLocalBundle(t.Context()) + _, _, _, stream, err := ua.SubscribeToLocalBundle(t.Context()) spiretest.RequireGRPCStatusHasPrefix(t, err, tt.expectCode, tt.expectMessage) if tt.expectCode != codes.OK { return @@ -506,12 +695,13 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { expectUpstreamX509Roots = tt.expectUpstreamX509RootsResponse } - upstreamX509Roots, upstreamJWTKeys, err := stream.RecvLocalBundleUpdate() + upstreamX509Roots, upstreamJWTKeys, upstreamWITKeys, err := stream.RecvLocalBundleUpdate() switch { case !tt.expectStreamUpdates: assert.Equal(t, io.EOF, err, "stream should have returned EOF") assert.Nil(t, upstreamX509Roots, "no roots should be received") assert.Nil(t, upstreamJWTKeys, "no keys should be received") + assert.Nil(t, upstreamWITKeys, "no keys should be received") case tt.expectStreamCode == codes.OK: assert.NoError(t, err, "stream should have returned update") expected := expectUpstreamX509Roots @@ -520,10 +710,12 @@ func TestV1SubscribeToLocalBundle(t *testing.T) { } assert.Equal(t, expected, upstreamX509Roots) spiretest.AssertProtoListEqual(t, expectedUpstreamJWTKeys, upstreamJWTKeys) + spiretest.AssertProtoListEqual(t, expectedUpstreamWITKeys, upstreamWITKeys) default: spiretest.RequireGRPCStatusHasPrefix(t, err, tt.expectStreamCode, tt.expectStreamMessage) assert.Nil(t, upstreamX509Roots) assert.Nil(t, upstreamJWTKeys) + assert.Nil(t, upstreamWITKeys) } spiretest.AssertLogs(t, logHook.AllEntries(), tt.expectLogs) @@ -570,6 +762,18 @@ func (b *V1Builder) WithPublishJWTKeyResponse(response *upstreamauthorityv1.Publ return b } +func (b *V1Builder) WithPublishWITKeyResponse(response *upstreamauthorityv1.PublishWITKeyResponse) *V1Builder { + b = b.clone() + b.p.publishWITKeyResponses = append(b.p.publishWITKeyResponses, response) + return b +} + +func (b *V1Builder) WithoutPublishWITKey() *V1Builder { + b = b.clone() + b.p.publishWITKeyUnimplemented = true + return b +} + func (b *V1Builder) WithSubscribeToLocalBundleResponse(response *upstreamauthorityv1.SubscribeToLocalBundleResponse) *V1Builder { b = b.clone() b.p.subscribeToLocalBundleResponses = append(b.p.subscribeToLocalBundleResponses, response) @@ -603,6 +807,8 @@ type v1Plugin struct { postSendErr error mintX509CAResponses []*upstreamauthorityv1.MintX509CAResponse publishJWTKeyResponses []*upstreamauthorityv1.PublishJWTKeyResponse + publishWITKeyResponses []*upstreamauthorityv1.PublishWITKeyResponse + publishWITKeyUnimplemented bool subscribeToLocalBundleResponses []*upstreamauthorityv1.SubscribeToLocalBundleResponse } @@ -645,6 +851,28 @@ func (v1 *v1Plugin) PublishJWTKeyAndSubscribe(req *upstreamauthorityv1.PublishJW return v1.postSendErr } +func (v1 *v1Plugin) PublishWITKeyAndSubscribe(req *upstreamauthorityv1.PublishWITKeyRequest, stream upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + if v1.publishWITKeyUnimplemented { + return v1.UnimplementedUpstreamAuthorityServer.PublishWITKeyAndSubscribe(req, stream) + } + + if diff := cmp.Diff(witkey.RequireToPluginFromCommonProto(witKey), req.WitKey, protocmp.Transform()); diff != "" { + return fmt.Errorf("unexpected public key: %s", diff) + } + + if v1.preSendErr != nil { + return *v1.preSendErr + } + + for _, response := range v1.publishWITKeyResponses { + if err := stream.Send(response); err != nil { + return err + } + } + + return v1.postSendErr +} + func (v1 *v1Plugin) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { if v1.preSendErr != nil { return *v1.preSendErr From d24c8b8d5d15aa353c698a94316a60018be57819 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Wed, 30 Sep 2026 09:23:52 +0100 Subject: [PATCH 06/11] Update fakeupstreamauthority Signed-off-by: Sorin Dumitru --- .../upstreamauthority.go | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/test/fakes/fakeupstreamauthority/upstreamauthority.go b/test/fakes/fakeupstreamauthority/upstreamauthority.go index 00c301a7f2..dec8ccb0c3 100644 --- a/test/fakes/fakeupstreamauthority/upstreamauthority.go +++ b/test/fakes/fakeupstreamauthority/upstreamauthority.go @@ -15,6 +15,7 @@ import ( "github.com/spiffe/go-spiffe/v2/spiffeid" upstreamauthorityv1 "github.com/spiffe/spire-plugin-sdk/proto/spire/plugin/server/upstreamauthority/v1" "github.com/spiffe/spire/pkg/common/coretypes/jwtkey" + "github.com/spiffe/spire/pkg/common/coretypes/witkey" "github.com/spiffe/spire/pkg/common/coretypes/x509certificate" "github.com/spiffe/spire/pkg/common/x509svid" "github.com/spiffe/spire/pkg/common/x509util" @@ -36,10 +37,12 @@ type Config struct { TrustDomain spiffeid.TrustDomain UseIntermediate bool DisallowPublishJWTKey bool + DisallowPublishWITKey bool UseSubscribeToLocalBundle bool KeyUsage x509.KeyUsage MutateMintX509CAResponse func(*upstreamauthorityv1.MintX509CAResponse) MutatePublishJWTKeyResponse func(*upstreamauthorityv1.PublishJWTKeyResponse) + MutatePublishWITKeyResponse func(*upstreamauthorityv1.PublishWITKeyResponse) } type UpstreamAuthority struct { @@ -58,9 +61,13 @@ type UpstreamAuthority struct { jwtKeysMtx sync.RWMutex jwtKeys []*common.PublicKey + witKeysMtx sync.RWMutex + witKeys []*common.PublicKey + streamsMtx sync.Mutex mintX509CAStreams map[chan struct{}]struct{} publishJWTKeyStreams map[chan struct{}]struct{} + publishWITKeyStreams map[chan struct{}]struct{} } func New(t *testing.T, config Config) *UpstreamAuthority { @@ -72,6 +79,7 @@ func New(t *testing.T, config Config) *UpstreamAuthority { config: config, mintX509CAStreams: make(map[chan struct{}]struct{}), publishJWTKeyStreams: make(map[chan struct{}]struct{}), + publishWITKeyStreams: make(map[chan struct{}]struct{}), } ua.RotateX509CA() return ua @@ -142,6 +150,35 @@ func (ua *UpstreamAuthority) PublishJWTKeyAndSubscribe(req *upstreamauthorityv1. } } +func (ua *UpstreamAuthority) PublishWITKeyAndSubscribe(req *upstreamauthorityv1.PublishWITKeyRequest, stream upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer) error { + if ua.config.DisallowPublishWITKey { + return status.Error(codes.Unimplemented, "disallowed") + } + + streamCh := ua.newPublishWITKeyStream() + defer ua.removePublishWITKeyStream(streamCh) + + ua.AppendWITKey(witkey.RequireToCommonFromPluginProto(req.WitKey)) + + ctx := stream.Context() + for { + select { + case <-ctx.Done(): + return nil + case <-streamCh: + if err := ua.sendPublishWITKeyStream(stream, &upstreamauthorityv1.PublishWITKeyResponse{ + UpstreamWitKeys: witkey.RequireToPluginFromCommonProtos(ua.WITKeys()), + }); err != nil { + return err + } + + if ua.config.UseSubscribeToLocalBundle { + return nil + } + } + } +} + func (ua *UpstreamAuthority) SubscribeToLocalBundle(req *upstreamauthorityv1.SubscribeToLocalBundleRequest, stream upstreamauthorityv1.UpstreamAuthority_SubscribeToLocalBundleServer) error { if !ua.config.UseSubscribeToLocalBundle { return status.Error(codes.Unimplemented, "fetching upstream trust bundle is unsupported") @@ -153,10 +190,14 @@ func (ua *UpstreamAuthority) SubscribeToLocalBundle(req *upstreamauthorityv1.Sub jwtStreamCh := ua.newPublishJWTKeyStream() defer ua.removePublishJWTKeyStream(jwtStreamCh) + witStreamCh := ua.newPublishWITKeyStream() + defer ua.removePublishWITKeyStream(witStreamCh) + // Send a first update on the stream, as required. if err := stream.Send(&upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: x509certificate.RequireToPluginProtos(ua.X509Roots()), UpstreamJwtKeys: jwtkey.RequireToPluginFromCommonProtos(ua.JWTKeys()), + UpstreamWitKeys: witkey.RequireToPluginFromCommonProtos(ua.WITKeys()), }); err != nil { return err } @@ -168,11 +209,13 @@ func (ua *UpstreamAuthority) SubscribeToLocalBundle(req *upstreamauthorityv1.Sub return nil case <-x509StreamCh: case <-jwtStreamCh: + case <-witStreamCh: } if err := stream.Send(&upstreamauthorityv1.SubscribeToLocalBundleResponse{ UpstreamX509Roots: x509certificate.RequireToPluginProtos(ua.X509Roots()), UpstreamJwtKeys: jwtkey.RequireToPluginFromCommonProtos(ua.JWTKeys()), + UpstreamWitKeys: witkey.RequireToPluginFromCommonProtos(ua.WITKeys()), }); err != nil { return err } @@ -251,6 +294,19 @@ func (ua *UpstreamAuthority) AppendJWTKey(jwtKey *common.PublicKey) { ua.TriggerJWTKeysChanged() } +func (ua *UpstreamAuthority) WITKeys() []*common.PublicKey { + ua.witKeysMtx.RLock() + defer ua.witKeysMtx.RUnlock() + return ua.witKeys +} + +func (ua *UpstreamAuthority) AppendWITKey(witKey *common.PublicKey) { + ua.witKeysMtx.Lock() + defer ua.witKeysMtx.Unlock() + ua.witKeys = append(ua.witKeys, witKey) + ua.TriggerWITKeysChanged() +} + func (ua *UpstreamAuthority) TriggerX509RootsChanged() { ua.streamsMtx.Lock() defer ua.streamsMtx.Unlock() @@ -273,6 +329,17 @@ func (ua *UpstreamAuthority) TriggerJWTKeysChanged() { } } +func (ua *UpstreamAuthority) TriggerWITKeysChanged() { + ua.streamsMtx.Lock() + defer ua.streamsMtx.Unlock() + for streamCh := range ua.publishWITKeyStreams { + select { + case streamCh <- struct{}{}: + default: + } + } +} + func (ua *UpstreamAuthority) newMintX509CAStream() chan struct{} { streamCh := make(chan struct{}, 1) ua.streamsMtx.Lock() @@ -330,6 +397,27 @@ func (ua *UpstreamAuthority) sendPublishJWTKeyStream(stream upstreamauthorityv1. return stream.Send(resp) } +func (ua *UpstreamAuthority) newPublishWITKeyStream() chan struct{} { + streamCh := make(chan struct{}, 1) + ua.streamsMtx.Lock() + ua.publishWITKeyStreams[streamCh] = struct{}{} + ua.streamsMtx.Unlock() + return streamCh +} + +func (ua *UpstreamAuthority) removePublishWITKeyStream(streamCh chan struct{}) { + ua.streamsMtx.Lock() + delete(ua.publishWITKeyStreams, streamCh) + ua.streamsMtx.Unlock() +} + +func (ua *UpstreamAuthority) sendPublishWITKeyStream(stream upstreamauthorityv1.UpstreamAuthority_PublishWITKeyAndSubscribeServer, resp *upstreamauthorityv1.PublishWITKeyResponse) error { + if ua.config.MutatePublishWITKeyResponse != nil { + ua.config.MutatePublishWITKeyResponse(resp) + } + return stream.Send(resp) +} + func (ua *UpstreamAuthority) createRootCertificate() { template := createCATemplate(ua.config.Clock.Now(), "FAKEUPSTREAMAUTHORITY-ROOT", ua.nextX509CASN(), ua.config.KeyUsage) root := createCertificate(ua.t, template, template, &x509RootKey.PublicKey, x509RootKey) From 97af9e3dbba2fdb666099fcf7b32d9c4ccd7f82a Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Wed, 30 Sep 2026 09:24:30 +0100 Subject: [PATCH 07/11] upstreamclient: add support for new RPC Signed-off-by: Sorin Dumitru --- pkg/server/ca/upstream_client.go | 98 +++++++++++++++++++++++++-- pkg/server/ca/upstream_client_test.go | 68 ++++++++++++++++++- 2 files changed, 160 insertions(+), 6 deletions(-) diff --git a/pkg/server/ca/upstream_client.go b/pkg/server/ca/upstream_client.go index 9f5fb2f1d4..caeabb6796 100644 --- a/pkg/server/ca/upstream_client.go +++ b/pkg/server/ca/upstream_client.go @@ -20,6 +20,7 @@ import ( type BundleUpdater interface { SyncX509Roots(ctx context.Context, roots []*x509certificate.X509Authority) error AppendJWTKeys(ctx context.Context, keys []*common.PublicKey) ([]*common.PublicKey, error) + AppendWITKeys(ctx context.Context, keys []*common.PublicKey) ([]*common.PublicKey, error) LogError(err error, msg string) } @@ -43,6 +44,8 @@ type UpstreamClient struct { mintX509CAStream *streamState publishJWTKeyMtx sync.Mutex publishJWTKeyStream *streamState + publishWITKeyMtx sync.Mutex + publishWITKeyStream *streamState subscribeToLocalBundleStreamMtx sync.Mutex subscribeToLocalBundleStream *streamState } @@ -53,6 +56,7 @@ func NewUpstreamClient(config UpstreamClientConfig) *UpstreamClient { c: config, mintX509CAStream: newStreamState(), publishJWTKeyStream: newStreamState(), + publishWITKeyStream: newStreamState(), subscribeToLocalBundleStream: newStreamState(), } } @@ -70,6 +74,11 @@ func (u *UpstreamClient) Close() error { defer u.publishJWTKeyMtx.Unlock() u.publishJWTKeyStream.Stop() }() + func() { + u.publishWITKeyMtx.Lock() + defer u.publishWITKeyMtx.Unlock() + u.publishWITKeyStream.Stop() + }() func() { u.subscribeToLocalBundleStreamMtx.Lock() defer u.subscribeToLocalBundleStreamMtx.Unlock() @@ -130,6 +139,32 @@ func (u *UpstreamClient) PublishJWTKey(ctx context.Context, jwtKey *common.Publi } } +// PublishWITKey publishes the WIT key to the UpstreamAuthority. It maintains +// an open stream to the UpstreamAuthority plugin to receive and append WIT key +// updates to the bundle. The stream remains open until another call to +// PublishWITKey happens or the client is closed. +func (u *UpstreamClient) PublishWITKey(ctx context.Context, witKey *common.PublicKey) (_ []*common.PublicKey, err error) { + u.publishWITKeyMtx.Lock() + defer u.publishWITKeyMtx.Unlock() + + firstResultCh := make(chan publishWITKeyResult, 1) + u.publishWITKeyStream.Start(func(streamCtx context.Context) { + u.runPublishWITKeyStream(streamCtx, witKey, firstResultCh) + }) + defer func() { + if err != nil { + u.publishWITKeyStream.Stop() + } + }() + + select { + case result := <-firstResultCh: + return result.witKeys, result.err + case <-ctx.Done(): + return nil, ctx.Err() + } +} + func (u *UpstreamClient) SubscribeToLocalBundle(ctx context.Context) (err error) { u.subscribeToLocalBundleStreamMtx.Lock() defer u.subscribeToLocalBundleStreamMtx.Unlock() @@ -243,8 +278,46 @@ func (u *UpstreamClient) runPublishJWTKeyStream(ctx context.Context, jwtKey *com } } +func (u *UpstreamClient) runPublishWITKeyStream(ctx context.Context, witKey *common.PublicKey, firstResultCh chan<- publishWITKeyResult) { + witKeys, witKeysStream, err := u.c.UpstreamAuthority.PublishWITKey(ctx, witKey) + if err != nil { + firstResultCh <- publishWITKeyResult{err: err} + return + } + defer witKeysStream.Close() + + updatedKeys, err := u.c.BundleUpdater.AppendWITKeys(ctx, witKeys) + if err != nil { + firstResultCh <- publishWITKeyResult{err: err} + return + } + firstResultCh <- publishWITKeyResult{witKeys: updatedKeys} + + for { + witKeys, err := witKeysStream.RecvUpstreamWITAuthorities() + if err != nil { + switch { + case errors.Is(err, io.EOF): + // This is normal if the plugin does not support streaming + // bundle updates. + case status.Code(err) == codes.Canceled: + // This is normal. This client cancels this stream when opening + // a new stream. + default: + u.c.BundleUpdater.LogError(err, "The upstream authority plugin stopped streaming WIT key updates prematurely. Please report this bug. Will retry later.") + } + return + } + + if _, err := u.c.BundleUpdater.AppendWITKeys(ctx, witKeys); err != nil { + u.c.BundleUpdater.LogError(err, "Failed to store WIT keys received by the upstream authority plugin.") + continue + } + } +} + func (u *UpstreamClient) runSubscribeToLocalBundleStream(ctx context.Context, firstResultCh chan<- bundleUpdatesResult) { - x509CAs, jwtKeys, authorityStream, err := u.c.UpstreamAuthority.SubscribeToLocalBundle(ctx) + x509CAs, jwtKeys, witKeys, authorityStream, err := u.c.UpstreamAuthority.SubscribeToLocalBundle(ctx) if err != nil { firstResultCh <- bundleUpdatesResult{err: err} return @@ -256,7 +329,12 @@ func (u *UpstreamClient) runSubscribeToLocalBundleStream(ctx context.Context, fi firstResultCh <- bundleUpdatesResult{err: err} return } - updatedKeys, err := u.c.BundleUpdater.AppendJWTKeys(ctx, jwtKeys) + updatedJWTKeys, err := u.c.BundleUpdater.AppendJWTKeys(ctx, jwtKeys) + if err != nil { + firstResultCh <- bundleUpdatesResult{err: err} + return + } + updatedWITKeys, err := u.c.BundleUpdater.AppendWITKeys(ctx, witKeys) if err != nil { firstResultCh <- bundleUpdatesResult{err: err} return @@ -269,11 +347,12 @@ func (u *UpstreamClient) runSubscribeToLocalBundleStream(ctx context.Context, fi firstResultCh <- bundleUpdatesResult{ x509CA: x509CA, - jwtKeys: updatedKeys, + jwtKeys: updatedJWTKeys, + witKeys: updatedWITKeys, } for { - x509CA, jwtKeys, err := authorityStream.RecvLocalBundleUpdate() + x509CA, jwtKeys, witKeys, err := authorityStream.RecvLocalBundleUpdate() if err != nil { switch { case errors.Is(err, io.EOF): @@ -297,6 +376,11 @@ func (u *UpstreamClient) runSubscribeToLocalBundleStream(ctx context.Context, fi u.c.BundleUpdater.LogError(err, "Failed to store JWT keys received by the upstream authority plugin.") continue } + + if _, err := u.c.BundleUpdater.AppendWITKeys(ctx, witKeys); err != nil { + u.c.BundleUpdater.LogError(err, "Failed to store WIT keys received by the upstream authority plugin.") + continue + } } } @@ -310,9 +394,15 @@ type publishJWTKeyResult struct { err error } +type publishWITKeyResult struct { + witKeys []*common.PublicKey + err error +} + type bundleUpdatesResult struct { x509CA []*x509.Certificate jwtKeys []*common.PublicKey + witKeys []*common.PublicKey err error } diff --git a/pkg/server/ca/upstream_client_test.go b/pkg/server/ca/upstream_client_test.go index 6985869a01..70a4ac92f6 100644 --- a/pkg/server/ca/upstream_client_test.go +++ b/pkg/server/ca/upstream_client_test.go @@ -151,6 +151,38 @@ func TestUpstreamClientPublishJWTKey_NotImplemented(t *testing.T) { require.Nil(t, jwtKeys) } +func TestUpstreamClientPublishWITKey_HandlesBundleUpdates(t *testing.T) { + client, updater, ua := setupUpstreamClientTest(t, fakeupstreamauthority.Config{ + TrustDomain: trustDomain, + }) + + key1 := makePublicKey(t, "KEY1") + key2 := makePublicKey(t, "KEY2") + + witKeys, err := client.PublishWITKey(context.Background(), key1) + require.NoError(t, err) + spiretest.RequireProtoListEqual(t, witKeys, ua.WITKeys()) + + // Assert that the initial bundle update happened. + spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1}, updater.WaitForAppendedWITKeys(t)) + + // Now trigger an update to the bundle by appending another key and wait + // for the bundle to receive the update. + ua.AppendWITKey(key2) + spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1, key2}, updater.WaitForAppendedWITKeys(t)) +} + +func TestUpstreamClientPublishWITKey_NotImplemented(t *testing.T) { + client, _, _ := setupUpstreamClientTest(t, fakeupstreamauthority.Config{ + TrustDomain: trustDomain, + DisallowPublishWITKey: true, + }) + + witKeys, err := client.PublishWITKey(context.Background(), makePublicKey(t, "KEY")) + spiretest.RequireGRPCStatus(t, err, codes.Unimplemented, "upstreamauthority(fake): disallowed") + require.Nil(t, witKeys) +} + func TestUpstreamClientSubscribeToLocalBundle(t *testing.T) { client, updater, ua := setupUpstreamClientTest(t, fakeupstreamauthority.Config{ TrustDomain: trustDomain, @@ -160,32 +192,43 @@ func TestUpstreamClientSubscribeToLocalBundle(t *testing.T) { err := client.SubscribeToLocalBundle(t.Context()) require.NoError(t, err) - // We should get an update with the initial CA and a list of empty JWT keys since - // the fakeupstreamauthority does not create one by default. + // We should get an update with the initial CA and a list of empty JWT and + // WIT keys since the fakeupstreamauthority does not create them by default. require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) require.Empty(t, updater.WaitForAppendedJWTKeys(t)) + require.Empty(t, updater.WaitForAppendedWITKeys(t)) // Trigger an update to the upstream bundle by rotating the root // certificate and wait for the bundle updater to receive the update. ua.RotateX509CA() require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) require.Empty(t, updater.WaitForAppendedJWTKeys(t)) + require.Empty(t, updater.WaitForAppendedWITKeys(t)) key1 := makePublicKey(t, "KEY1") ua.AppendJWTKey(key1) require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1}, updater.WaitForAppendedJWTKeys(t)) + require.Empty(t, updater.WaitForAppendedWITKeys(t)) // Trigger an update to the upstream bundle by rotating the root // certificate and wait for the bundle updater to receive the update. ua.RotateX509CA() require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1}, updater.WaitForAppendedJWTKeys(t)) + require.Empty(t, updater.WaitForAppendedWITKeys(t)) key2 := makePublicKey(t, "KEY2") ua.AppendJWTKey(key2) require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1, key2}, updater.WaitForAppendedJWTKeys(t)) + require.Empty(t, updater.WaitForAppendedWITKeys(t)) + + witKey := makePublicKey(t, "WITKEY") + ua.AppendWITKey(witKey) + require.Equal(t, ua.X509Roots(), updater.WaitForAppendedX509Roots(t)) + spiretest.RequireProtoListEqual(t, []*common.PublicKey{key1, key2}, updater.WaitForAppendedJWTKeys(t)) + spiretest.RequireProtoListEqual(t, []*common.PublicKey{witKey}, updater.WaitForAppendedWITKeys(t)) } func setupUpstreamClientTest(t *testing.T, config fakeupstreamauthority.Config) (*ca.UpstreamClient, *fakeBundleUpdater, *fakeupstreamauthority.UpstreamAuthority) { @@ -211,6 +254,7 @@ type bundleUpdateErr struct { type fakeBundleUpdater struct { x509RootsCh chan []*x509certificate.X509Authority jwtKeysCh chan []*common.PublicKey + witKeysCh chan []*common.PublicKey errorCh chan bundleUpdateErr } @@ -218,6 +262,7 @@ func newFakeBundleUpdater() *fakeBundleUpdater { return &fakeBundleUpdater{ x509RootsCh: make(chan []*x509certificate.X509Authority, 1), jwtKeysCh: make(chan []*common.PublicKey, 1), + witKeysCh: make(chan []*common.PublicKey, 1), errorCh: make(chan bundleUpdateErr, 1), } } @@ -260,6 +305,25 @@ func (u *fakeBundleUpdater) WaitForAppendedJWTKeys(t *testing.T) []*common.Publi } } +func (u *fakeBundleUpdater) AppendWITKeys(ctx context.Context, witKeys []*common.PublicKey) ([]*common.PublicKey, error) { + select { + case u.witKeysCh <- witKeys: + return witKeys, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func (u *fakeBundleUpdater) WaitForAppendedWITKeys(t *testing.T) []*common.PublicKey { + select { + case <-time.After(time.Minute): + require.FailNow(t, "timed out waiting for WIT keys to be appended") + return nil // unreachable + case witKeys := <-u.witKeysCh: + return witKeys + } +} + func (u *fakeBundleUpdater) LogError(err error, msg string) { e := bundleUpdateErr{ err: err, From 184e4a0e48b0436c87217cc5c5672da5c388cb61 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Wed, 30 Sep 2026 09:25:17 +0100 Subject: [PATCH 08/11] bundle service: publish WIT keys support Signed-off-by: Sorin Dumitru --- pkg/server/api/bundle/v1/service.go | 43 +++- pkg/server/api/bundle/v1/service_test.go | 246 +++++++++++++++++++++++ 2 files changed, 280 insertions(+), 9 deletions(-) diff --git a/pkg/server/api/bundle/v1/service.go b/pkg/server/api/bundle/v1/service.go index 985546b0a7..724862dc15 100644 --- a/pkg/server/api/bundle/v1/service.go +++ b/pkg/server/api/bundle/v1/service.go @@ -24,14 +24,7 @@ import ( // UpstreamPublisher defines the publisher interface. type UpstreamPublisher interface { PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error) -} - -// UpstreamPublisherFunc defines the function. -type UpstreamPublisherFunc func(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error) - -// PublishJWTKey publishes the JWT key with the given function. -func (fn UpstreamPublisherFunc) PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error) { - return fn(ctx, jwtKey) + PublishWITKey(ctx context.Context, witKey *common.PublicKey) ([]*common.PublicKey, error) } // Config defines the bundle service configuration. @@ -189,8 +182,40 @@ func (s *Service) PublishJWTAuthority(ctx context.Context, req *bundlev1.Publish // PublishWITAuthority published the WIT key on the server. func (s *Service) PublishWITAuthority(ctx context.Context, req *bundlev1.PublishWITAuthorityRequest) (*bundlev1.PublishWITAuthorityResponse, error) { + parseRequest := func() logrus.Fields { + fields := logrus.Fields{} + if req.WitAuthority != nil { + fields[telemetry.WITAuthorityExpiresAt] = req.WitAuthority.ExpiresAt + fields[telemetry.WITAuthorityKeyID] = req.WitAuthority.KeyId + fields[telemetry.WITAuthorityPublicKeySHA256] = api.HashByte(req.WitAuthority.PublicKey) + } + return fields + } + rpccontext.AddRPCAuditFields(ctx, parseRequest()) log := rpccontext.Logger(ctx) - return nil, commonapi.MakeErr(log, codes.Unimplemented, "WIT-SVID functionality is not yet implemented", nil) + + if err := rpccontext.RateLimit(ctx, 1); err != nil { + return nil, commonapi.MakeErr(log, status.Code(err), "rejecting request due to key publishing rate limiting", err) + } + + if req.WitAuthority == nil { + return nil, commonapi.MakeErr(log, codes.InvalidArgument, "missing WIT authority", nil) + } + + keys, err := api.ParseWITAuthorities([]*types.WITKey{req.WitAuthority}) + if err != nil { + return nil, commonapi.MakeErr(log, codes.InvalidArgument, "invalid WIT authority", err) + } + + resp, err := s.up.PublishWITKey(ctx, keys[0]) + if err != nil { + return nil, commonapi.MakeErr(log, codes.Internal, "failed to publish WIT key", err) + } + rpccontext.AuditRPC(ctx) + + return &bundlev1.PublishWITAuthorityResponse{ + WitAuthorities: api.PublicKeysToWITKeys(resp), + }, nil } // ListFederatedBundles returns an optionally paginated list of federated bundles. diff --git a/pkg/server/api/bundle/v1/service_test.go b/pkg/server/api/bundle/v1/service_test.go index 9ff1776e5c..a06853d081 100644 --- a/pkg/server/api/bundle/v1/service_test.go +++ b/pkg/server/api/bundle/v1/service_test.go @@ -1409,6 +1409,242 @@ func TestPublishJWTAuthority(t *testing.T) { } } +func TestPublishWITAuthority(t *testing.T) { + test := setupServiceTest(t) + defer test.Cleanup() + + pkixBytes, err := base64.StdEncoding.DecodeString("MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEYSlUVLqTD8DEnA4F1EWMTf5RXc5lnCxw+5WKJwngEL3rPc9i4Tgzz9riR3I/NiSlkgRO1WsxBusqpC284j9dXA==") + pkixHashed := api.HashByte(pkixBytes) + require.NoError(t, err) + expiresAt := time.Now().Unix() + expiresAtStr := strconv.FormatInt(expiresAt, 10) + witKey1 := &types.WITKey{ + ExpiresAt: expiresAt, + KeyId: "key1", + PublicKey: pkixBytes, + } + + _, expectedWITErr := x509.ParsePKIXPublicKey([]byte("malformed key")) + require.Error(t, expectedWITErr) + + for _, tt := range []struct { + name string + + code codes.Code + err string + expectLogs []spiretest.LogEntry + resultKeys []*types.WITKey + fakeErr error + fakeExpectKey *common.PublicKey + witKey *types.WITKey + rateLimiterErr error + }{ + { + name: "success", + witKey: witKey1, + fakeExpectKey: &common.PublicKey{ + PkixBytes: pkixBytes, + Kid: "key1", + NotAfter: expiresAt, + }, + resultKeys: []*types.WITKey{ + { + ExpiresAt: expiresAt, + KeyId: "key1", + PublicKey: pkixBytes, + }, + }, + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "success", + telemetry.Type: "audit", + telemetry.WITAuthorityKeyID: "key1", + telemetry.WITAuthorityPublicKeySHA256: pkixHashed, + telemetry.WITAuthorityExpiresAt: expiresAtStr, + }, + }, + }, + }, + { + name: "rate limit fails", + witKey: witKey1, + rateLimiterErr: status.Error(codes.Internal, "limit error"), + code: codes.Internal, + err: "rejecting request due to key publishing rate limiting: limit error", + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Rejecting request due to key publishing rate limiting", + Data: logrus.Fields{ + logrus.ErrorKey: "rpc error: code = Internal desc = limit error", + }, + }, + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "error", + telemetry.StatusCode: "Internal", + telemetry.StatusMessage: "rejecting request due to key publishing rate limiting: limit error", + telemetry.Type: "audit", + telemetry.WITAuthorityKeyID: "key1", + telemetry.WITAuthorityPublicKeySHA256: pkixHashed, + telemetry.WITAuthorityExpiresAt: expiresAtStr, + }, + }, + }, + }, + { + name: "missing WIT authority", + code: codes.InvalidArgument, + err: "missing WIT authority", + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Invalid argument: missing WIT authority", + }, + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "error", + telemetry.StatusCode: "InvalidArgument", + telemetry.StatusMessage: "missing WIT authority", + telemetry.Type: "audit", + }, + }, + }, + }, + { + name: "malformed key", + code: codes.InvalidArgument, + err: "invalid WIT authority: asn1:", + witKey: &types.WITKey{ + ExpiresAt: expiresAt, + KeyId: "key1", + PublicKey: []byte("malformed key"), + }, + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Invalid argument: invalid WIT authority", + Data: logrus.Fields{ + logrus.ErrorKey: expectedWITErr.Error(), + }, + }, + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "error", + telemetry.StatusCode: "InvalidArgument", + telemetry.StatusMessage: fmt.Sprintf("invalid WIT authority: %v", expectedWITErr), + telemetry.Type: "audit", + telemetry.WITAuthorityKeyID: "key1", + telemetry.WITAuthorityPublicKeySHA256: api.HashByte([]byte("malformed key")), + telemetry.WITAuthorityExpiresAt: expiresAtStr, + }, + }, + }, + }, + { + name: "missing key ID", + code: codes.InvalidArgument, + err: "invalid WIT authority: missing key ID", + witKey: &types.WITKey{ + ExpiresAt: expiresAt, + PublicKey: witKey1.PublicKey, + }, + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Invalid argument: invalid WIT authority", + Data: logrus.Fields{ + logrus.ErrorKey: "missing key ID", + }, + }, + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "error", + telemetry.StatusCode: "InvalidArgument", + telemetry.StatusMessage: "invalid WIT authority: missing key ID", + telemetry.Type: "audit", + telemetry.WITAuthorityKeyID: "", + telemetry.WITAuthorityPublicKeySHA256: pkixHashed, + telemetry.WITAuthorityExpiresAt: expiresAtStr, + }, + }, + }, + }, + { + name: "fail to publish", + code: codes.Internal, + err: "failed to publish WIT key: publish error", + fakeErr: errors.New("publish error"), + witKey: witKey1, + expectLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Failed to publish WIT key", + Data: logrus.Fields{ + logrus.ErrorKey: "publish error", + }, + }, + { + Level: logrus.InfoLevel, + Message: "API accessed", + Data: logrus.Fields{ + telemetry.Status: "error", + telemetry.StatusCode: "Internal", + telemetry.StatusMessage: "failed to publish WIT key: publish error", + telemetry.Type: "audit", + telemetry.WITAuthorityKeyID: "key1", + telemetry.WITAuthorityPublicKeySHA256: pkixHashed, + telemetry.WITAuthorityExpiresAt: expiresAtStr, + }, + }, + }, + }, + } { + t.Run(tt.name, func(t *testing.T) { + test.logHook.Reset() + + // Setup fake + test.up.t = t + test.up.err = tt.fakeErr + test.up.expectKey = tt.fakeExpectKey + + // Setup rate limiter + test.rateLimiter.count = 1 + test.rateLimiter.err = tt.rateLimiterErr + + resp, err := test.client.PublishWITAuthority(ctx, &bundlev1.PublishWITAuthorityRequest{ + WitAuthority: tt.witKey, + }) + + spiretest.AssertLogs(t, test.logHook.AllEntries(), tt.expectLogs) + if err != nil { + spiretest.RequireGRPCStatusContains(t, err, tt.code, tt.err) + require.Nil(t, resp) + + return + } + + require.NoError(t, err) + require.NotNil(t, resp) + + spiretest.RequireProtoEqual(t, &bundlev1.PublishWITAuthorityResponse{ + WitAuthorities: tt.resultKeys, + }, resp) + }) + } +} + func TestListFederatedBundles(t *testing.T) { test := setupServiceTest(t) defer test.Cleanup() @@ -3081,6 +3317,16 @@ func (f *fakeUpstreamPublisher) PublishJWTKey(_ context.Context, jwtKey *common. return []*common.PublicKey{jwtKey}, nil } +func (f *fakeUpstreamPublisher) PublishWITKey(_ context.Context, witKey *common.PublicKey) ([]*common.PublicKey, error) { + if f.err != nil { + return nil, f.err + } + + spiretest.AssertProtoEqual(f.t, f.expectKey, witKey) + + return []*common.PublicKey{witKey}, nil +} + type fakeRateLimiter struct { count int err error From 35318d2679a369e93ae758a66d59da614a6a6363 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Wed, 30 Sep 2026 09:25:58 +0100 Subject: [PATCH 09/11] publish wit signign keys Signed-off-by: Sorin Dumitru --- pkg/server/ca/manager/manager.go | 54 ++++++++++++++++++++++++--- pkg/server/ca/manager/manager_test.go | 46 +++++++++++++++++++++++ pkg/server/endpoints/config.go | 3 +- pkg/server/endpoints/middleware.go | 6 --- 4 files changed, 95 insertions(+), 14 deletions(-) diff --git a/pkg/server/ca/manager/manager.go b/pkg/server/ca/manager/manager.go index 4afc31959d..fb9a6f6bb7 100644 --- a/pkg/server/ca/manager/manager.go +++ b/pkg/server/ca/manager/manager.go @@ -56,10 +56,6 @@ type ManagedCA interface { NotifyTaintedX509Authorities([]*x509.Certificate) } -type JwtKeyPublisher interface { - PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error) -} - type AuthorityManager interface { GetCurrentJWTKeySlot() Slot GetNextJWTKeySlot() Slot @@ -77,6 +73,7 @@ type AuthorityManager interface { IsJWTSVIDsDisabled() bool IsWITSVIDsDisabled() bool PublishJWTKey(ctx context.Context, jwtKey *common.PublicKey) ([]*common.PublicKey, error) + PublishWITKey(ctx context.Context, witKey *common.PublicKey) ([]*common.PublicKey, error) NotifyTaintedX509Authority(ctx context.Context, authorityID string) error SubscribeToLocalBundle(ctx context.Context) error } @@ -123,6 +120,9 @@ type Manager struct { // Used to log a warning only once when the UpstreamAuthority does not support JWT-SVIDs. jwtUnimplementedWarnOnce sync.Once + // Used to log a warning only once when the UpstreamAuthority does not support WIT-SVIDs. + witUnimplementedWarnOnce sync.Once + // Used for testing backoff, must not be set in regular code triggerBackOffCh chan error } @@ -568,8 +568,7 @@ func (m *Manager) PrepareWITKey(ctx context.Context) (err error) { return err } - _, err = m.appendBundle(ctx, nil, nil, []*common.PublicKey{publicKey}) - if err != nil { + if _, err := m.PublishWITKey(ctx, publicKey); err != nil { return err } @@ -620,6 +619,38 @@ func (m *Manager) RotateWITKey(ctx context.Context) { m.activateWITKey(ctx) } +// PublishWITKey publishes the passed WIT key to the upstream server using the +// configured UpstreamAuthority plugin, then appends to the bundle the WIT keys +// returned by the upstream server, and finally it returns the updated list of +// WIT keys contained in the bundle. It follows the same fallback rules as +// PublishJWTKey. +func (m *Manager) PublishWITKey(ctx context.Context, witKey *common.PublicKey) ([]*common.PublicKey, error) { + if m.upstreamClient != nil { + publishCtx, cancel := context.WithTimeout(ctx, publishJWKTimeout) + defer cancel() + upstreamWITKeys, err := m.upstreamClient.PublishWITKey(publishCtx, witKey) + switch { + case status.Code(err) == codes.Unimplemented: + m.witUnimplementedWarnOnce.Do(func() { + m.c.Log.WithField("plugin_name", m.upstreamPluginName).Warn("UpstreamAuthority plugin does not support WIT-SVIDs. Workloads managed " + + "by this server may have trouble communicating with workloads outside " + + "this cluster when using WIT-SVIDs.") + }) + case err != nil: + return nil, err + default: + return upstreamWITKeys, nil + } + } + + bundle, err := m.appendBundle(ctx, nil, nil, []*common.PublicKey{witKey}) + if err != nil { + return nil, err + } + + return bundle.WitSigningKeys, nil +} + func (m *Manager) SubscribeToLocalBundle(ctx context.Context) error { if m.upstreamClient == nil { return nil @@ -1195,6 +1226,17 @@ func (u *bundleUpdater) AppendJWTKeys(ctx context.Context, keys []*common.Public return bundle.JwtSigningKeys, nil } +func (u *bundleUpdater) AppendWITKeys(ctx context.Context, keys []*common.PublicKey) ([]*common.PublicKey, error) { + bundle, err := u.appendBundle(ctx, &common.Bundle{ + TrustDomainId: u.trustDomainID, + WitSigningKeys: keys, + }) + if err != nil { + return nil, err + } + return bundle.WitSigningKeys, nil +} + func (u *bundleUpdater) LogError(err error, msg string) { u.log.WithError(err).Error(msg) } diff --git a/pkg/server/ca/manager/manager_test.go b/pkg/server/ca/manager/manager_test.go index a3e9ec753b..6841d5093d 100644 --- a/pkg/server/ca/manager/manager_test.go +++ b/pkg/server/ca/manager/manager_test.go @@ -770,6 +770,51 @@ func TestUpstreamAuthorityWithPublishJWTKeyImplemented(t *testing.T) { ) } +func TestUpstreamAuthorityWithPublishWITKeyImplemented(t *testing.T) { + ctx := context.Background() + test := setupTest(t) + bundle := test.createBundle(ctx) + require.Len(t, bundle.WitSigningKeys, 0) + + upstreamAuthority, ua := test.newFakeUpstreamAuthority(t, fakeupstreamauthority.Config{ + TrustDomain: testTrustDomain, + }) + test.initAndActivateUpstreamSignedManager(ctx, upstreamAuthority) + + require.Len(t, ua.WITKeys(), 1) + require.Equal(t, test.currentWITKey().Kid, ua.WITKeys()[0].Kid) + spiretest.AssertProtoListEqual(t, ua.WITKeys(), test.fetchBundle(ctx).WitSigningKeys) + assert.Equal(t, + 0, + test.countLogEntries(logrus.WarnLevel, "UpstreamAuthority plugin does not support WIT-SVIDs. Workloads managed "+ + "by this server may have trouble communicating with workloads outside "+ + "this cluster when using WIT-SVIDs."), + ) +} + +func TestUpstreamAuthorityWithPublishWITKeyNotImplemented(t *testing.T) { + ctx := context.Background() + test := setupTest(t) + + upstreamAuthority, ua := test.newFakeUpstreamAuthority(t, fakeupstreamauthority.Config{ + TrustDomain: testTrustDomain, + DisallowPublishWITKey: true, + }) + test.initAndActivateUpstreamSignedManager(ctx, upstreamAuthority) + + // The WIT key falls back to the local bundle + require.Empty(t, ua.WITKeys()) + witSigningKeys := test.fetchBundle(ctx).WitSigningKeys + require.Len(t, witSigningKeys, 1) + require.Equal(t, test.currentWITKey().Kid, witSigningKeys[0].Kid) + assert.Equal(t, + 1, + test.countLogEntries(logrus.WarnLevel, "UpstreamAuthority plugin does not support WIT-SVIDs. Workloads managed "+ + "by this server may have trouble communicating with workloads outside "+ + "this cluster when using WIT-SVIDs."), + ) +} + func TestUpstreamAuthorityWithSubscribeToBundleUpdate(t *testing.T) { ctx := context.Background() test := setupTest(t) @@ -799,6 +844,7 @@ func TestUpstreamAuthorityWithSubscribeToBundleUpdate(t *testing.T) { test.requireBundleRootCAs(ctx, t, ua.X509Root()) spiretest.AssertProtoListEqual(t, ua.JWTKeys(), test.fetchBundle(ctx).JwtSigningKeys) + spiretest.AssertProtoListEqual(t, ua.WITKeys(), test.fetchBundle(ctx).WitSigningKeys) } func TestX509CARotation(t *testing.T) { diff --git a/pkg/server/endpoints/config.go b/pkg/server/endpoints/config.go index 0fd02d94c7..3c7757bf3c 100644 --- a/pkg/server/endpoints/config.go +++ b/pkg/server/endpoints/config.go @@ -166,7 +166,6 @@ func (c *Config) maybeMakeBundleEndpointServer() (Server, func(context.Context) func (c *Config) makeAPIServers(entryFetcher api.AuthorizedEntryFetcher) APIServers { ds := c.Catalog.GetDataStore() - upstreamPublisher := UpstreamPublisher(c.AuthorityManager) return APIServers{ AgentServer: agentv1.New(agentv1.Config{ @@ -180,7 +179,7 @@ func (c *Config) makeAPIServers(entryFetcher api.AuthorizedEntryFetcher) APIServ BundleServer: bundlev1.New(bundlev1.Config{ TrustDomain: c.TrustDomain, DataStore: ds, - UpstreamPublisher: upstreamPublisher, + UpstreamPublisher: c.AuthorityManager, }), DebugServer: debugv1.New(debugv1.Config{ TrustDomain: c.TrustDomain, diff --git a/pkg/server/endpoints/middleware.go b/pkg/server/endpoints/middleware.go index 1ec3738508..11ecfb915f 100644 --- a/pkg/server/endpoints/middleware.go +++ b/pkg/server/endpoints/middleware.go @@ -11,12 +11,10 @@ import ( "github.com/spiffe/spire/pkg/common/errorutil" "github.com/spiffe/spire/pkg/common/telemetry" "github.com/spiffe/spire/pkg/server/api" - "github.com/spiffe/spire/pkg/server/api/bundle/v1" "github.com/spiffe/spire/pkg/server/api/limits" "github.com/spiffe/spire/pkg/server/api/middleware" "github.com/spiffe/spire/pkg/server/api/rpccontext" "github.com/spiffe/spire/pkg/server/authpolicy" - "github.com/spiffe/spire/pkg/server/ca/manager" "github.com/spiffe/spire/pkg/server/datastore" "github.com/spiffe/spire/proto/spire/common" "github.com/spiffe/spire/test/clock" @@ -54,10 +52,6 @@ func EntryFetcher(ds datastore.DataStore) middleware.EntryFetcher { }) } -func UpstreamPublisher(jwtKeyPublisher manager.JwtKeyPublisher) bundle.UpstreamPublisher { - return bundle.UpstreamPublisherFunc(jwtKeyPublisher.PublishJWTKey) -} - func AgentAuthorizer(ds datastore.DataStore, nodeCache api.AttestedNodeCache, maxAttestedNodeInfoStaleness time.Duration, clk clock.Clock) middleware.AgentAuthorizer { return middleware.AgentAuthorizerFunc(func(ctx context.Context, agentID spiffeid.ID, agentSVID *x509.Certificate) error { id := agentID.String() From 9f31d8371698f85ec073ae7428d23960a14846f3 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Fri, 2 Oct 2026 12:46:43 +0100 Subject: [PATCH 10/11] Ignore unparam for function it is only passed Warning now, but hardcoding to warning inside the function seems worse. Signed-off-by: Sorin Dumitru --- pkg/server/ca/manager/manager_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/server/ca/manager/manager_test.go b/pkg/server/ca/manager/manager_test.go index 6841d5093d..12a933b0fb 100644 --- a/pkg/server/ca/manager/manager_test.go +++ b/pkg/server/ca/manager/manager_test.go @@ -1939,7 +1939,7 @@ func (m *managerTest) waitForBundleUpdatedNotification(ctx context.Context, ch < } } -func (m *managerTest) countLogEntries(level logrus.Level, message string) int { +func (m *managerTest) countLogEntries(level logrus.Level, message string) int { //nolint:unparam count := 0 for _, e := range m.logHook.AllEntries() { if e.Message == message && level == e.Level { From 32d71662971859e1f7e590e66627ac03618bec19 Mon Sep 17 00:00:00 2001 From: Sorin Dumitru Date: Fri, 2 Oct 2026 13:18:11 +0100 Subject: [PATCH 11/11] Tidy up go mod Signed-off-by: Sorin Dumitru --- go.sum | 2 -- 1 file changed, 2 deletions(-) diff --git a/go.sum b/go.sum index 85a9999223..3e9ede4262 100644 --- a/go.sum +++ b/go.sum @@ -734,8 +734,6 @@ github.com/spiffe/go-spiffe/v2 v2.8.2 h1:jUEsvCMD6fH25J8K/w3q/XnIx8W1lb8+YLaEEHI github.com/spiffe/go-spiffe/v2 v2.8.2/go.mod h1:w2CLWKLMTX/PPYUEUPv3ltH0RXsw5S8suwNF46w9/Aw= github.com/spiffe/spire-api-sdk v1.2.5-0.20260924193931-38b883fcd6d1 h1:O0ZhDye/pIEZGKsaGnczAClIG0SK6jS5IwH+mfhIXdE= github.com/spiffe/spire-api-sdk v1.2.5-0.20260924193931-38b883fcd6d1/go.mod h1:9hXJcMzatM1KwAtBDO3s6HccDCic++/5c2yOc5Iln8Y= -github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260617144146-5dcde407c4d1 h1:gN8nWPdRSxYIACrgdPBuVwhNe6ABzpvUVza7cP8Yy38= -github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260617144146-5dcde407c4d1/go.mod h1:QvrRDiBlXiJ7kNd176ZHsF5eklxxeTRgJSu2CXe0MKw= github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260929185137-e961422db712 h1:dMxwHOpAc6jH4NtU4mgqHLE/daLwa2gDW04Wz4lIAIQ= github.com/spiffe/spire-plugin-sdk v1.4.4-0.20260929185137-e961422db712/go.mod h1:QvrRDiBlXiJ7kNd176ZHsF5eklxxeTRgJSu2CXe0MKw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=