Make CT log selection simpler and more robust (#8152)

Simplify the way we load and handle CT logs: rather than keeping them
grouped by operator, simply keep a flat list and annotate each log with
its operator's name. At submission time, instead of shuffling operator
groups and submitting to one log from each group, shuffle the whole set
of individual logs.

Support tiled logs by similarly annotating each log with whether it is
tiled or not.

Also make the way we know when to stop getting SCTs more robust.
Previously we would stop as soon as we had two, since we knew that they
would be from different operator groups and didn't care about tiled
logs. Instead, introduce an explicit CT policy compliance evaluation
function which tells us if the set of SCTs we have so far forms a
compliant set.

This is not our desired end-state for CT log submission. Ideally we'd
like to: simplify things even further (don't race all the logs, simply
try to submit to two at a time), improve selection (intelligently pick
the next log to submit to, rather than just a random shuffle), and
fine-tune latency (tiled logs should have longer timeouts than classic
ones). Those improvements will come in future PRs.

Part of #7872

Author: aarongable

ctpolicy/ctpolicy.go

@@ -2,6 +2,7 @@ package ctpolicy
 
 import (
 	"context"
+	"encoding/base64"
 	"fmt"
 	"strings"
 	"time"
@@ -23,15 +24,14 @@ const (
 // CTPolicy is used to hold information about SCTs required from various
 // groupings
 type CTPolicy struct {
-	pub                 pubpb.PublisherClient
-	sctLogs             loglist.List
-	infoLogs            loglist.List
-	finalLogs           loglist.List
-	stagger             time.Duration
-	log                 blog.Logger
-	winnerCounter       *prometheus.CounterVec
-	operatorGroupsGauge *prometheus.GaugeVec
-	shardExpiryGauge    *prometheus.GaugeVec
+	pub              pubpb.PublisherClient
+	sctLogs          loglist.List
+	infoLogs         loglist.List
+	finalLogs        loglist.List
+	stagger          time.Duration
+	log              blog.Logger
+	winnerCounter    *prometheus.CounterVec
+	shardExpiryGauge *prometheus.GaugeVec
 }
 
 // New creates a new CTPolicy struct
@@ -45,15 +45,6 @@ func New(pub pubpb.PublisherClient, sctLogs loglist.List, infoLogs loglist.List,
 	)
 	stats.MustRegister(winnerCounter)
 
-	operatorGroupsGauge := prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Name: "ct_operator_group_size_gauge",
-			Help: "Gauge for CT operators group size, by operator and log source (capable of providing SCT, informational logs, logs we submit final certs to).",
-		},
-		[]string{"operator", "source"},
-	)
-	stats.MustRegister(operatorGroupsGauge)
-
 	shardExpiryGauge := prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Name: "ct_shard_expiration_seconds",
@@ -63,43 +54,30 @@ func New(pub pubpb.PublisherClient, sctLogs loglist.List, infoLogs loglist.List,
 	)
 	stats.MustRegister(shardExpiryGauge)
 
-	for op, group := range sctLogs {
-		operatorGroupsGauge.WithLabelValues(op, "sctLogs").Set(float64(len(group)))
-
-		for _, log := range group {
-			if log.EndExclusive.IsZero() {
-				// Handles the case for non-temporally sharded logs too.
-				shardExpiryGauge.WithLabelValues(op, log.Name).Set(float64(0))
-			} else {
-				shardExpiryGauge.WithLabelValues(op, log.Name).Set(float64(log.EndExclusive.Unix()))
-			}
+	for _, log := range sctLogs {
+		if log.EndExclusive.IsZero() {
+			// Handles the case for non-temporally sharded logs too.
+			shardExpiryGauge.WithLabelValues(log.Operator, log.Name).Set(float64(0))
+		} else {
+			shardExpiryGauge.WithLabelValues(log.Operator, log.Name).Set(float64(log.EndExclusive.Unix()))
 		}
 	}
 
-	for op, group := range infoLogs {
-		operatorGroupsGauge.WithLabelValues(op, "infoLogs").Set(float64(len(group)))
-	}
-
-	for op, group := range finalLogs {
-		operatorGroupsGauge.WithLabelValues(op, "finalLogs").Set(float64(len(group)))
-	}
-
 	return &CTPolicy{
-		pub:                 pub,
-		sctLogs:             sctLogs,
-		infoLogs:            infoLogs,
-		finalLogs:           finalLogs,
-		stagger:             stagger,
-		log:                 log,
-		winnerCounter:       winnerCounter,
-		operatorGroupsGauge: operatorGroupsGauge,
-		shardExpiryGauge:    shardExpiryGauge,
+		pub:              pub,
+		sctLogs:          sctLogs,
+		infoLogs:         infoLogs,
+		finalLogs:        finalLogs,
+		stagger:          stagger,
+		log:              log,
+		winnerCounter:    winnerCounter,
+		shardExpiryGauge: shardExpiryGauge,
 	}
 }
 
 type result struct {
+	log loglist.Log
 	sct []byte
-	url string
 	err error
 }
 
@@ -115,73 +93,68 @@ func (ctp *CTPolicy) GetSCTs(ctx context.Context, cert core.CertDER, expiration
 	subCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	// This closure will be called in parallel once for each operator group.
-	getOne := func(i int, g string) ([]byte, string, error) {
-		// Sleep a little bit to stagger our requests to the later groups. Use `i-1`
-		// to compute the stagger duration so that the first two groups (indices 0
+	// This closure will be called in parallel once for each log.
+	getOne := func(i int, l loglist.Log) ([]byte, error) {
+		// Sleep a little bit to stagger our requests to the later logs. Use `i-1`
+		// to compute the stagger duration so that the first two logs (indices 0
 		// and 1) get negative or zero (i.e. instant) sleep durations. If the
-		// context gets cancelled (most likely because two logs from other operator
-		// groups returned SCTs already) before the sleep is complete, quit instead.
+		// context gets cancelled (most likely because we got enough SCTs from other
+		// logs already) before the sleep is complete, quit instead.
 		select {
 		case <-subCtx.Done():
-			return nil, "", subCtx.Err()
+			return nil, subCtx.Err()
 		case <-time.After(time.Duration(i-1) * ctp.stagger):
 		}
 
-		// Pick a random log from among those in the group. In practice, very few
-		// operator groups have more than one log, so this loses little flexibility.
-		url, key, err := ctp.sctLogs.PickOne(g, expiration)
-		if err != nil {
-			return nil, "", fmt.Errorf("unable to get log info: %w", err)
-		}
-
 		sct, err := ctp.pub.SubmitToSingleCTWithResult(ctx, &pubpb.Request{
-			LogURL:       url,
-			LogPublicKey: key,
+			LogURL:       l.Url,
+			LogPublicKey: base64.StdEncoding.EncodeToString(l.Key),
 			Der:          cert,
 			Kind:         pubpb.SubmissionType_sct,
 		})
 		if err != nil {
-			return nil, url, fmt.Errorf("ct submission to %q (%q) failed: %w", g, url, err)
+			return nil, fmt.Errorf("ct submission to %q (%q) failed: %w", l.Name, l.Url, err)
 		}
 
-		return sct.Sct, url, nil
+		return sct.Sct, nil
 	}
 
-	// Ensure that this channel has a buffer equal to the number of goroutines
-	// we're kicking off, so that they're all guaranteed to be able to write to
-	// it and exit without blocking and leaking.
-	results := make(chan result, len(ctp.sctLogs))
+	// Identify the set of candidate logs whose temporal interval includes this
+	// cert's expiry. Randomize the order of the logs so that we're not always
+	// trying to submit to the same two.
+	logs := ctp.sctLogs.ForTime(expiration).Permute()
 
 	// Kick off a collection of goroutines to try to submit the precert to each
-	// log operator group. Randomize the order of the groups so that we're not
-	// always trying to submit to the same two operators.
-	for i, group := range ctp.sctLogs.Permute() {
-		go func(i int, g string) {
-			sctDER, url, err := getOne(i, g)
-			results <- result{sct: sctDER, url: url, err: err}
-		}(i, group)
+	// log. Ensure that the results channel has a buffer equal to the number of
+	// goroutines we're kicking off, so that they're all guaranteed to be able to
+	// write to it and exit without blocking and leaking.
+	resChan := make(chan result, len(logs))
+	for i, log := range logs {
+		go func(i int, l loglist.Log) {
+			sctDER, err := getOne(i, l)
+			resChan <- result{log: l, sct: sctDER, err: err}
+		}(i, log)
 	}
 
 	go ctp.submitPrecertInformational(cert, expiration)
 
 	// Finally, collect SCTs and/or errors from our results channel. We know that
-	// we will collect len(ctp.sctLogs) results from the channel because every
-	// goroutine is guaranteed to write one result to the channel.
-	scts := make(core.SCTDERs, 0)
+	// we can collect len(logs) results from the channel because every goroutine
+	// is guaranteed to write one result (either sct or error) to the channel.
+	results := make([]result, 0)
 	errs := make([]string, 0)
-	for range len(ctp.sctLogs) {
-		res := <-results
+	for range len(logs) {
+		res := <-resChan
 		if res.err != nil {
 			errs = append(errs, res.err.Error())
-			if res.url != "" {
-				ctp.winnerCounter.WithLabelValues(res.url, failed).Inc()
-			}
+			ctp.winnerCounter.WithLabelValues(res.log.Url, failed).Inc()
 			continue
 		}
-		scts = append(scts, res.sct)
-		ctp.winnerCounter.WithLabelValues(res.url, succeeded).Inc()
-		if len(scts) >= 2 {
+		results = append(results, res)
+		ctp.winnerCounter.WithLabelValues(res.log.Url, succeeded).Inc()
+
+		scts := compliantSet(results)
+		if scts != nil {
 			return scts, nil
 		}
 	}
@@ -196,6 +169,36 @@ func (ctp *CTPolicy) GetSCTs(ctx context.Context, cert core.CertDER, expiration
 	return nil, berrors.MissingSCTsError("failed to get 2 SCTs, got %d error(s): %s", len(errs), strings.Join(errs, "; "))
 }
 
+// compliantSet returns a slice of SCTs which complies with all relevant CT Log
+// Policy requirements, namely that the set of SCTs:
+// - contain at least two SCTs, which
+// - come from logs run by at least two different operators, and
+// - contain at least one RFC6962-compliant (i.e. non-static/tiled) log.
+//
+// If no such set of SCTs exists, returns nil.
+func compliantSet(results []result) core.SCTDERs {
+	for _, first := range results {
+		if first.err != nil {
+			continue
+		}
+		for _, second := range results {
+			if second.err != nil {
+				continue
+			}
+			if first.log.Operator == second.log.Operator {
+				// The two SCTs must come from different operators.
+				continue
+			}
+			if first.log.Tiled && second.log.Tiled {
+				// At least one must come from a non-tiled log.
+				continue
+			}
+			return core.SCTDERs{first.sct, second.sct}
+		}
+	}
+	return nil
+}
+
 // submitAllBestEffort submits the given certificate or precertificate to every
 // log ("informational" for precerts, "final" for certs) configured in the policy.
 // It neither waits for these submission to complete, nor tracks their success.
@@ -205,29 +208,26 @@ func (ctp *CTPolicy) submitAllBestEffort(blob core.CertDER, kind pubpb.Submissio
 		logs = ctp.infoLogs
 	}
 
-	for _, group := range logs {
-		for _, log := range group {
-			if log.StartInclusive.After(expiry) || log.EndExclusive.Equal(expiry) || log.EndExclusive.Before(expiry) {
-				continue
-			}
-
-			go func(log loglist.Log) {
-				_, err := ctp.pub.SubmitToSingleCTWithResult(
-					context.Background(),
-					&pubpb.Request{
-						LogURL:       log.Url,
-						LogPublicKey: log.Key,
-						Der:          blob,
-						Kind:         kind,
-					},
-				)
-				if err != nil {
-					ctp.log.Warningf("ct submission of cert to log %q failed: %s", log.Url, err)
-				}
-			}(log)
+	for _, log := range logs {
+		if log.StartInclusive.After(expiry) || log.EndExclusive.Equal(expiry) || log.EndExclusive.Before(expiry) {
+			continue
 		}
-	}
 
+		go func(log loglist.Log) {
+			_, err := ctp.pub.SubmitToSingleCTWithResult(
+				context.Background(),
+				&pubpb.Request{
+					LogURL:       log.Url,
+					LogPublicKey: base64.StdEncoding.EncodeToString(log.Key),
+					Der:          blob,
+					Kind:         kind,
+				},
+			)
+			if err != nil {
+				ctp.log.Warningf("ct submission of cert to log %q failed: %s", log.Url, err)
+			}
+		}(log)
+	}
 }
 
 // submitPrecertInformational submits precertificates to any configured