Return a detailed list response for unified-revoked API endpoint (#18862)

- Return a detailed reponse within the list api that an end-user can
   use to determine what clusters revoked the certificate on from the
   pki/certs/unified-revoked LIST api.

 - Return colon delimited serial numbers from the certs/revocation-queue
   LIST api

builtin/logical/pki/path_revoke.go

@@ -9,6 +9,7 @@ import (
 	"crypto/x509"
 	"encoding/pem"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/vault/sdk/framework"
@@ -593,6 +594,7 @@ func (b *backend) pathListRevocationQueueHandler(ctx context.Context, request *l
 			if hasSlash {
 				serial = serial[0 : len(serial)-1]
 			}
+			serial = denormalizeSerial(serial)
 
 			var data map[string]interface{}
 			rawData, isPresent := responseInfo[serial]
@@ -619,18 +621,40 @@ func (b *backend) pathListRevocationQueueHandler(ctx context.Context, request *l
 
 func (b *backend) pathListUnifiedRevokedCertsHandler(ctx context.Context, request *logical.Request, _ *framework.FieldData) (*logical.Response, error) {
 	sc := b.makeStorageContext(ctx, request.Storage)
+	responseKeys := []string{}
+	responseInfo := make(map[string]interface{})
 
-	revokedCerts, err := listUnifiedRevokedCerts(sc)
+	clusterPathsById, err := lookupUnifiedClusterPaths(sc)
 	if err != nil {
 		return nil, err
 	}
 
-	// Normalize serial back to a format people are expecting.
-	for i, serial := range revokedCerts {
-		revokedCerts[i] = denormalizeSerial(serial)
+	for clusterId := range clusterPathsById {
+		clusterSerials, err := listClusterSpecificUnifiedRevokedCerts(sc, clusterId)
+		if err != nil {
+			return nil, err
+		}
+		for _, serial := range clusterSerials {
+			if strings.HasSuffix(serial, "/") {
+				// Skip folders as they wouldn't be a proper revocation
+				continue
+			}
+			colonSerial := denormalizeSerial(serial)
+			var data map[string][]string
+			rawData, isPresent := responseInfo[colonSerial]
+			if !isPresent {
+				responseKeys = append(responseKeys, colonSerial)
+				data = map[string][]string{}
+			} else {
+				data = rawData.(map[string][]string)
+			}
+
+			data["revoking_clusters"] = append(data["revoking_clusters"], clusterId)
+			responseInfo[colonSerial] = data
+		}
 	}
 
-	return logical.ListResponse(revokedCerts), nil
+	return logical.ListResponseWithInfo(responseKeys, responseInfo), nil
 }
 
 const pathRevokeHelpSyn = `

builtin/logical/pki/storage_unified.go

@@ -2,6 +2,7 @@ package pki
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/hashicorp/vault/sdk/logical"
@@ -20,7 +21,7 @@ type unifiedRevocationEntry struct {
 }
 
 func getUnifiedRevocationBySerial(sc *storageContext, serial string) (*unifiedRevocationEntry, error) {
-	clusterPaths, err := lookupClusterPaths(sc)
+	clusterPaths, err := lookupUnifiedClusterPaths(sc)
 	if err != nil {
 		return nil, err
 	}
@@ -54,35 +55,33 @@ func writeUnifiedRevocationEntry(sc *storageContext, ure *unifiedRevocationEntry
 	return sc.Storage.Put(sc.Context, json)
 }
 
-func listUnifiedRevokedCerts(sc *storageContext) ([]string, error) {
-	allSerials := []string{}
-
-	clusterPaths, err := lookupClusterPaths(sc)
+// listClusterSpecificUnifiedRevokedCerts returns a list of revoked certificates from a given cluster
+func listClusterSpecificUnifiedRevokedCerts(sc *storageContext, clusterId string) ([]string, error) {
+	path := unifiedRevocationReadPathPrefix + clusterId + "/"
+	serials, err := sc.Storage.List(sc.Context, path)
 	if err != nil {
 		return nil, err
 	}
 
-	for _, path := range clusterPaths {
-		clusterSerials, err := sc.Storage.List(sc.Context, path)
-		if err != nil {
-			return nil, fmt.Errorf("failed listing revoked certs for path %s: %w", path, err)
-		}
-
-		allSerials = append(allSerials, clusterSerials...)
-	}
-	return allSerials, nil
+	return serials, nil
 }
 
-func lookupClusterPaths(sc *storageContext) ([]string, error) {
-	fullPaths := []string{}
+// lookupUnifiedClusterPaths returns a map of cluster id to the prefix storage path for that given cluster's
+// unified revoked certificates
+func lookupUnifiedClusterPaths(sc *storageContext) (map[string]string, error) {
+	fullPaths := map[string]string{}
 
 	clusterPaths, err := sc.Storage.List(sc.Context, unifiedRevocationReadPathPrefix)
 	if err != nil {
-		return fullPaths, err
+		return nil, err
 	}
 
-	for _, clusterId := range clusterPaths {
-		fullPaths = append(fullPaths, unifiedRevocationReadPathPrefix+clusterId)
+	for _, clusterIdWithSlash := range clusterPaths {
+		// Only include folder listings, if a file were to be stored under this path ignore it.
+		if strings.HasSuffix(clusterIdWithSlash, "/") {
+			clusterId := clusterIdWithSlash[:len(clusterIdWithSlash)-1] // remove trailing /
+			fullPaths[clusterId] = unifiedRevocationReadPathPrefix + clusterIdWithSlash
+		}
 	}
 
 	return fullPaths, nil