crl-updater: query by explicit shard too

Add querying by explicit shard (SA.GetRevokedCertsByShard) in addition to
querying by temporal shard (SA.GetRevokedCerts).

Merge results from both kinds of shard. De-duplicate by serial within a shard,
because the same certificate could wind up in a temporal shard that matches its
explicit shard.

When de-duplicating, validate that revocation reasons are the same or (very
unlikely) represent a re-revocation based on demonstrating key compromise. This
can happen because the two different SA queries occur at slightly different times.

Add unit testing that CRL entries make it through the whole pipeline from SA, to
CA, to uploader.

Rename some types in the unittest to be more accessible.

crl/updater/batch_test.go

@@ -26,9 +26,9 @@ func TestRunOnce(t *testing.T) {
 		[]*issuance.Certificate{e1, r3},
 		2, 18*time.Hour, 24*time.Hour,
 		6*time.Hour, time.Minute, 1, 1,
-		&fakeSAC{grcc: fakeGRCC{err: errors.New("db no worky")}, maxNotAfter: clk.Now().Add(90 * 24 * time.Hour)},
-		&fakeCGC{gcc: fakeGCC{}},
-		&fakeCSC{ucc: fakeUCC{}},
+		&fakeSAC{revokedCerts: revokedCertsStream{err: errors.New("db no worky")}, maxNotAfter: clk.Now().Add(90 * 24 * time.Hour)},
+		&fakeCA{gcc: generateCRLStream{}},
+		&fakeStorer{uploaderStream: &noopUploader{}},
 		metrics.NoopRegisterer, mockLog, clk,
 	)
 	test.AssertNotError(t, err, "building test crlUpdater")

crl/updater/updater.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/jmhodges/clock"
 	"github.com/prometheus/client_golang/prometheus"
+	"golang.org/x/crypto/ocsp"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -183,11 +184,72 @@ func (cu *crlUpdater) updateShardWithRetry(ctx context.Context, atTime time.Time
 	return nil
 }
 
+type crlStream interface {
+	Recv() (*proto.CRLEntry, error)
+}
+
+// reRevoked returns the later of the two entries, only if the latter represents a valid
+// re-revocation of the former (reason == KeyCompromise).
+func reRevoked(a *proto.CRLEntry, b *proto.CRLEntry) (*proto.CRLEntry, error) {
+	first, second := a, b
+	if b.RevokedAt.AsTime().Before(a.RevokedAt.AsTime()) {
+		first, second = b, a
+	}
+	if first.Reason != ocsp.KeyCompromise && second.Reason == ocsp.KeyCompromise {
+		return second, nil
+	}
+	// The RA has logic to prevent re-revocation for any reason other than KeyCompromise,
+	// so this should be impossible. The best we can do is error out.
+	return nil, fmt.Errorf("certificate %s was revoked with reason %d at %s and re-revoked with invalid reason %d at %s",
+		first.Serial, first.Reason, first.RevokedAt.AsTime(), second.Reason, second.RevokedAt.AsTime())
+}
+
+// addFromStream pulls `proto.CRLEntry` objects from a stream, adding them to the crlEntries map.
+//
+// Consolidates duplicates and checks for internal consistency of the results.
+//
+// Returns the number of entries received from the stream, regardless of duplicate status.
+func addFromStream(crlEntries map[string]*proto.CRLEntry, stream crlStream) (int, error) {
+	var count int
+	for {
+		entry, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return 0, fmt.Errorf("retrieving entry from SA: %w", err)
+		}
+		count++
+		previousEntry := crlEntries[entry.Serial]
+		if previousEntry == nil {
+			crlEntries[entry.Serial] = entry
+			continue
+		}
+		if previousEntry.Reason == entry.Reason &&
+			previousEntry.RevokedAt.AsTime().Equal(entry.RevokedAt.AsTime()) {
+			continue
+		}
+
+		// There's a tiny possibility a certificate was re-revoked for KeyCompromise and
+		// we got a different view of it from temporal sharding vs explicit sharding.
+		// Prefer the re-revoked CRL entry, which must be the one with KeyCompromise.
+		second, err := reRevoked(entry, previousEntry)
+		if err != nil {
+			return 0, err
+		}
+		crlEntries[entry.Serial] = second
+	}
+	return count, nil
+}
+
 // updateShard processes a single shard. It computes the shard's boundaries, gets
 // the list of revoked certs in that shard from the SA, gets the CA to sign the
 // resulting CRL, and gets the crl-storer to upload it. It returns an error if
 // any of these operations fail.
 func (cu *crlUpdater) updateShard(ctx context.Context, atTime time.Time, issuerNameID issuance.NameID, shardIdx int, chunks []chunk) (err error) {
+	if shardIdx == 0 {
+		return fmt.Errorf("invalid shard 0")
+	}
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -207,8 +269,10 @@ func (cu *crlUpdater) updateShard(ctx context.Context, atTime time.Time, issuerN
 	cu.log.Infof(
 		"Generating CRL shard: id=[%s] numChunks=[%d]", crlID, len(chunks))
 
-	// Get the full list of CRL Entries for this shard from the SA.
-	var crlEntries []*proto.CRLEntry
+	// Deduplicate the CRL entries by serial number, since we can get the same certificate via
+	// both temporal sharding (GetRevokedCerts) and explicit sharding (GetRevokedCertsByShard).
+	crlEntries := make(map[string]*proto.CRLEntry)
+
 	for _, chunk := range chunks {
 		saStream, err := cu.sa.GetRevokedCerts(ctx, &sapb.GetRevokedCertsRequest{
 			IssuerNameID:  int64(issuerNameID),
@@ -217,25 +281,41 @@ func (cu *crlUpdater) updateShard(ctx context.Context, atTime time.Time, issuerN
 			RevokedBefore: timestamppb.New(atTime),
 		})
 		if err != nil {
-			return fmt.Errorf("connecting to SA: %w", err)
+			return fmt.Errorf("GetRevokedCerts: %w", err)
 		}
 
-		for {
-			entry, err := saStream.Recv()
-			if err != nil {
-				if err == io.EOF {
-					break
-				}
-				return fmt.Errorf("retrieving entry from SA: %w", err)
-			}
-			crlEntries = append(crlEntries, entry)
+		n, err := addFromStream(crlEntries, saStream)
+		if err != nil {
+			return fmt.Errorf("streaming GetRevokedCerts: %w", err)
 		}
 
 		cu.log.Infof(
 			"Queried SA for CRL shard: id=[%s] expiresAfter=[%s] expiresBefore=[%s] numEntries=[%d]",
-			crlID, chunk.start, chunk.end, len(crlEntries))
+			crlID, chunk.start, chunk.end, n)
+	}
+
+	// Query for unexpired certificates, with padding to ensure that revoked certificates show
+	// up in at least one CRL, even if they expire between revocation and CRL generation.
+	expiresAfter := cu.clk.Now().Add(cu.lookbackPeriod)
+
+	saStream, err := cu.sa.GetRevokedCertsByShard(ctx, &sapb.GetRevokedCertsByShardRequest{
+		IssuerNameID:  int64(issuerNameID),
+		ShardIdx:      int64(shardIdx),
+		ExpiresAfter:  timestamppb.New(expiresAfter),
+		RevokedBefore: timestamppb.New(atTime),
+	})
+	if err != nil {
+		return fmt.Errorf("GetRevokedCertsByShard: %w", err)
 	}
 
+	n, err := addFromStream(crlEntries, saStream)
+	if err != nil {
+		return fmt.Errorf("streaming GetRevokedCertsByShard: %w", err)
+	}
+
+	cu.log.Infof(
+		"Queried SA by CRL shard number: id=[%s] shardIdx=[%d] numEntries=[%d]", crlID, shardIdx, n)
+
 	// Send the full list of CRL Entries to the CA.
 	caStream, err := cu.ca.GenerateCRL(ctx)
 	if err != nil {