add a rate limiter per issue

Author: phbnf

cmd/tesseract/aws/main.go

@@ -71,6 +71,7 @@ var (
 	acceptSHA1               = flag.Bool("accept_sha1_signing_algorithms", true, "If true, accept chains that use SHA-1 based signing algorithms. This flag will eventually be removed, and such algorithms will be rejected.")
 	enablePublicationAwaiter = flag.Bool("enable_publication_awaiter", true, "If true then the certificate is integrated into log before returning the response.")
 	notBeforeRL              = flag.String("rate_limit_old_not_before", "", "Optionally rate limits submissions with old notBefore dates. Expects a value of with the format: \"<go duration>:<rate limit>\", e.g. \"30d:50\" would impose a limit of 50 certs/s on submissions whose notBefore date is >= 30days old.")
+	issuerRL                 = flag.Float64("rate_limit_per_issuer", -1, "Optionally rate limits submissions per issuer per second. Disabled when null or negative.")
 
 	// Performance flags
 	httpDeadline              = flag.Duration("http_deadline", time.Second*10, "Deadline for HTTP requests.")
@@ -141,6 +142,7 @@ eventually go away. See /internal/lax509/README.md for more information.`)
 
 	hOpts := tesseract.LogHandlerOpts{
 		NotBeforeRL: notBeforeRLFromFlags(),
+		IssuerRL:    *issuerRL,
 		DedupRL:     dedupRL,
 	}
 	logHandler, err := tesseract.NewLogHandler(ctx, *origin, signer, chainValidationConfig, newAWSStorage, *httpDeadline, *maskInternalErrors, *pathPrefix, hOpts)

cmd/tesseract/gcp/main.go

@@ -72,6 +72,7 @@ var (
 	enablePublicationAwaiter = flag.Bool("enable_publication_awaiter", true, "If true then the certificate is integrated into log before returning the response.")
 	witnessPolicyFile        = flag.String("witness_policy_file", "", "(Optional) Path to the file containing the witness policy in the format described at https://git.glasklar.is/sigsum/core/sigsum-go/-/blob/main/doc/policy.md")
 	notBeforeRL              = flag.String("rate_limit_old_not_before", "", "Optionally rate limits submissions with old notBefore dates. Expects a value of with the format: \"<go duration>:<rate limit>\", e.g. \"30d:50\" would impose a limit of 50 certs/s on submissions whose notBefore date is >= 30days old.")
+	issuerRL                 = flag.Float64("rate_limit_per_issuer", -1, "Optionally rate limits submissions per issuer per second. Disabled when null or negative.")
 
 	// Performance flags
 	httpDeadline              = flag.Duration("http_deadline", time.Second*10, "Deadline for HTTP requests.")
@@ -128,6 +129,7 @@ eventually go away. See /internal/lax509/README.md for more information.`)
 
 	hOpts := tesseract.LogHandlerOpts{
 		NotBeforeRL: notBeforeRLFromFlags(),
+		IssuerRL:    *issuerRL,
 		DedupRL:     dedupRL,
 	}
 	logHandler, err := tesseract.NewLogHandler(ctx, *origin, signer, chainValidationConfig, newGCPStorage, *httpDeadline, *maskInternalErrors, *pathPrefix, hOpts)

cmd/tesseract/posix/main.go

@@ -75,6 +75,7 @@ var (
 	enablePublicationAwaiter = flag.Bool("enable_publication_awaiter", true, "If true then the certificate is integrated into log before returning the response.")
 	witnessPolicyFile        = flag.String("witness_policy_file", "", "(Optional) Path to the file containing the witness policy in the format describe at https://git.glasklar.is/sigsum/core/sigsum-go/-/blob/main/doc/policy.md")
 	notBeforeRL              = flag.String("rate_limit_old_not_before", "", "Optionally rate limits submissions with old notBefore dates. Expects a value of with the format: \"<go duration>:<rate limit>\", e.g. \"30d:50\" would impose a limit of 50 certs/s on submissions whose notBefore date is >= 30days old.")
+	issuerRL                 = flag.Float64("rate_limit_per_issuer", -1, "Optionally rate limits submissions per issuer per second. Disabled when null or negative.")
 
 	// Performance flags
 	httpDeadline              = flag.Duration("http_deadline", time.Second*10, "Deadline for HTTP requests.")
@@ -121,6 +122,7 @@ eventually go away. See /internal/lax509/README.md for more information.`)
 
 	hOpts := tesseract.LogHandlerOpts{
 		NotBeforeRL: notBeforeRLFromFlags(),
+		IssuerRL:    *issuerRL,
 		DedupRL:     dedupRL,
 	}
 	logHandler, err := tesseract.NewLogHandler(ctx, *origin, signer, chainValidationConfig, newStorage, *httpDeadline, *maskInternalErrors, *pathPrefix, hOpts)

ctlog.go

@@ -131,6 +131,7 @@ type NotBeforeRL struct {
 
 type LogHandlerOpts struct {
 	NotBeforeRL *NotBeforeRL
+	IssuerRL    float64
 	DedupRL     float64
 }
 
@@ -163,6 +164,9 @@ func NewLogHandler(ctx context.Context, origin string, signer crypto.Signer, cfg
 	if opts.NotBeforeRL != nil {
 		ctOpts.RateLimits.NotBefore(opts.NotBeforeRL.AgeThreshold, opts.NotBeforeRL.RateLimit)
 	}
+	if opts.IssuerRL > 0 {
+		ctOpts.RateLimits.Issuer(opts.IssuerRL)
+	}
 	if opts.DedupRL >= 0 {
 		ctOpts.RateLimits.Dedup(opts.DedupRL)
 	}

deployment/modules/gcp/gce/tesseract/main.tf

@@ -41,6 +41,7 @@ module "gce_container_tesseract" {
       "--enable_publication_awaiter=${var.enable_publication_awaiter}",
       "--accept_sha1_signing_algorithms=true",
       "--rate_limit_old_not_before=${var.rate_limit_old_not_before}",
+      "--rate_limit_per_issuer=${var.rate_limit_per_issuer}",
       "--rate_limit_dedup=${var.rate_limit_dedup}"
     ]
     tty : true # maybe remove this

deployment/modules/gcp/gce/tesseract/variables.tf

@@ -106,6 +106,12 @@ variable "rate_limit_old_not_before" {
   default     = ""
 }
 
+variable "rate_limit_per_issuer" {
+  description = "Set to rate limit submissions per issuer per second."
+  type        = number
+  default     = -1 
+}
+
 variable "rate_limit_dedup" {
   description = "Set to rate limit duplicate submissions per second."
   type        = number

deployment/modules/gcp/tesseract/gce/main.tf

@@ -42,6 +42,7 @@ module "gce" {
   batch_max_size                 = var.batch_max_size
   enable_publication_awaiter     = var.enable_publication_awaiter
   rate_limit_old_not_before      = var.rate_limit_old_not_before
+  rate_limit_per_issuer          = var.rate_limit_per_issuer
   rate_limit_dedup               = var.rate_limit_dedup
 
   depends_on = [

deployment/modules/gcp/tesseract/gce/variables.tf

@@ -105,6 +105,12 @@ variable "rate_limit_old_not_before" {
   default     = ""
 }
 
+variable "rate_limit_per_issuer" {
+  description = "Set to rate limit submissions per issuer per second."
+  type        = number
+  default     = -1 
+}
+
 variable "rate_limit_dedup" {
   description = "Set to rate limit duplicate submissions per second."
   type        = number

internal/ct/handlers.go

@@ -16,6 +16,7 @@ package ct
 
 import (
 	"context"
+	"crypto/sha256"
 	"crypto/x509"
 	"encoding/base64"
 	"encoding/json"
@@ -195,6 +196,8 @@ type RateLimits struct {
 	notBeforeLimit time.Duration
 	notBefore      *rate.Limiter
 	dedup          *rate.Limiter
+	issuerLimit    float64
+	issuer         *sync.Map
 }
 
 // NotBefore configures a rate limit on old certs.
@@ -206,6 +209,15 @@ func (r *RateLimits) NotBefore(age time.Duration, limit float64) {
 	klog.Infof("Configured OldSubmission limiter with %0.2f qps for certs aged >= %s", limit, age)
 }
 
+// Issuer configures a rate limit on the first issuer of each chain.
+//
+// Submissions will be subject to the specified number of entries per second.
+func (r *RateLimits) Issuer(limit float64) {
+	r.issuerLimit = limit
+	r.issuer = &sync.Map{}
+	klog.Infof("Configured Issuer limiter with %0.2f qps per issuer", limit)
+}
+
 // Dedup configures a rate limit on entries being deduplicated.
 //
 // Submissions will be subject to the specified number of entries per second.
@@ -231,6 +243,24 @@ func (r *RateLimits) AcceptNotBefore(ctx context.Context, chain []*x509.Certific
 	return true
 }
 
+// AcceptIssuer returns true if a chain should be accepted based on its first issuer, and false othwerwise.
+func (r *RateLimits) AcceptIssuer(ctx context.Context, chain []*x509.Certificate) bool {
+	if len(chain) == 0 {
+		return false
+	}
+	if r.issuer != nil && r.issuerLimit >= 0 {
+		issuer := sha256.Sum256(chain[0].RawIssuer)
+		v, _ := r.issuer.LoadOrStore(issuer, rate.NewLimiter(rate.Limit(r.issuerLimit), int(math.Ceil(r.issuerLimit))))
+		rl := v.(*rate.Limiter)
+		if rl.Allow() {
+			return true
+		}
+		rateLimitedRequests.Add(ctx, 1, metric.WithAttributes(rateLimitReasonKey.String("issuer")))
+		return false
+	}
+	return true
+}
+
 // AcceptDedup returns true if a duplicate entry is permitted to be resolved.
 func (r *RateLimits) AcceptDedup(ctx context.Context) bool {
 	if r.dedup != nil {
@@ -354,6 +384,11 @@ func addChainInternal(ctx context.Context, opts *HandlerOptions, log *log, w htt
 		opts.RequestLog.addCertToChain(ctx, cert)
 	}
 
+	if ok := opts.RateLimits.AcceptIssuer(ctx, chain); !ok {
+		w.Header().Add("Retry-After", strconv.Itoa(rand.IntN(5)+1)) // random retry within [1,6) seconds
+		return http.StatusTooManyRequests, []attribute.KeyValue{tooManyRequestsReasonKey.String("rate_limit_issuer")}, errors.New(http.StatusText(http.StatusTooManyRequests))
+	}
+
 	// Get the current time in the form used throughout RFC6962, namely milliseconds since Unix
 	// epoch, and use this throughout.
 	nanosPerMilli := int64(time.Millisecond / time.Nanosecond)

internal/ct/handlers_test.go

@@ -823,6 +823,71 @@ func TestMaxDedupInFlight(t *testing.T) {
 	}
 }
 
+func TestLimitIssuer(t *testing.T) {
+	var tests = []struct {
+		descr   string
+		chains  [][]string
+		wants   []int
+		maxRate float64
+	}{
+		{
+			descr: "no-limit",
+			chains: [][]string{
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+			},
+			wants:   []int{http.StatusOK, http.StatusOK},
+			maxRate: -1,
+		},
+		{
+			descr: "success",
+			chains: [][]string{
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+			},
+			maxRate: 100,
+			wants:   []int{http.StatusOK, http.StatusOK},
+		},
+		{
+			descr: "issuer-limited",
+			chains: [][]string{
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+				{testdata.CertFromIntermediate, testdata.IntermediateFromRoot, testdata.CACertPEM},
+			},
+			maxRate: 1,
+			wants:   []int{http.StatusOK, http.StatusTooManyRequests},
+		},
+	}
+
+	for _, test := range tests {
+		log, _ := setupTestLog(t)
+		hhOpts := hOpts()
+		if test.maxRate > 0 {
+			hhOpts.RateLimits.Issuer(test.maxRate)
+		}
+		server := setupTestServer(t, log, path.Join(prefix, rfc6962.AddChainPath), hhOpts)
+		defer server.Close()
+		defer timeSource.Reset()
+
+		// Increment time to make it unique for each test case.
+		t.Run(test.descr, func(t *testing.T) {
+			for i, chain := range test.chains {
+				pool := loadCertsIntoPoolOrDie(t, chain)
+				chain := createJSONChain(t, *pool)
+
+				resp, err := http.Post(server.URL+rfc6962.AddChainPath, "application/json", chain)
+
+				if err != nil {
+					t.Fatalf("http.Post(%s)=(_,%q); want (_,nil)", rfc6962.AddChainPath, err)
+				}
+				if got, want := resp.StatusCode, test.wants[i]; got != want {
+					t.Errorf("http.Post(%s)=(%d,nil); want (%d,nil)", rfc6962.AddChainPath, got, want)
+				}
+			}
+		})
+	}
+}
+
 func TestRateLimiter(t *testing.T) {
 	certAge := time.Minute
 	cert := &x509.Certificate{