Prevent parallel validation attempts

errors/errors.go

@@ -326,6 +326,10 @@ func UnknownSerialError() error {
 	return newf(UnknownSerial, "unknown serial")
 }
 
+func ConflictError(msg string, args ...any) error {
+	return newf(Conflict, msg, args...)
+}
+
 func InvalidProfileError(msg string, args ...any) error {
 	return newf(InvalidProfile, msg, args...)
 }

features/features.go

@@ -68,6 +68,12 @@ type Config struct {
 	// during certificate issuance. This flag must be set to true in the
 	// RA and VA services for full functionality.
 	DNSPersist01Enabled bool
+
+	// SetAuthzProcessing controls whether the RA attempts to mark authorizations
+	// as "processing" before dispatching validation to the VA. This reduces
+	// unnecessary work due to parallel validations, but requires a database
+	// change to work.
+	SetAuthzProcessing bool
 }
 
 var fMu = new(sync.RWMutex)

ra/ra.go

@@ -1447,6 +1447,14 @@ func (ra *RegistrationAuthorityImpl) PerformValidation(
 		return nil, berrors.MalformedError("cannot validate challenge: %s", cErr.Error())
 	}
 
+	// Set the authorization to "processing", to prevent parallel attempts.
+	if features.Get().SetAuthzProcessing {
+		_, err = ra.SA.SetAuthzProcessing(ctx, &sapb.AuthorizationID2{Id: authz.ID})
+		if err != nil {
+			return nil, fmt.Errorf("failed to mark authz as processing: %w", err)
+		}
+	}
+
 	// Dispatch to the VA for service
 	ra.drainWG.Go(func() {
 		ctx := context.WithoutCancel(ctx)

sa/db/01-boulder_sa_next.sql

@@ -251,3 +251,5 @@ ALTER TABLE `revokedCertificates` ADD KEY `serial` (`serial`);
 ALTER TABLE `orders`
   ADD COLUMN `mtcLogID` varchar(255) DEFAULT NULL,
   ADD COLUMN `mtcSerialNumber` bigint(20) unsigned DEFAULT NULL;
+
+ALTER TABLE `authz2` ADD COLUMN `beganProcessing` tinyint(1) NOT NULL DEFAULT 0;

sa/proto/sa.proto

@@ -82,6 +82,7 @@ service StorageAuthority {
   rpc NewOrderAndAuthzs(NewOrderAndAuthzsRequest) returns (core.Order) {}
   rpc NewRegistration(core.Registration) returns (core.Registration) {}
   rpc RevokeCertificate(RevokeCertificateRequest) returns (google.protobuf.Empty) {}
+  rpc SetAuthzProcessing(AuthorizationID2) returns (google.protobuf.Empty) {}
   rpc SetOrderError(SetOrderErrorRequest) returns (google.protobuf.Empty) {}
   rpc SetOrderProcessing(OrderRequest) returns (google.protobuf.Empty) {}
   rpc UpdateRegistrationKey(UpdateRegistrationKeyRequest) returns (core.Registration) {}

sa/sa.go

@@ -597,6 +597,40 @@ func containsDuplicates(ids []int64) bool {
 	return false
 }
 
+// SetAuthzProcessing sets the "beganProcessing" bool for an authorization.
+// This does not affect its public-facing status (unlike orders, authzs do not
+// have an RFC 8555 "processing" state), but does prevent further requests to
+// the challenge endpoint from kicking off parallel validation attempts.
+func (ssa *SQLStorageAuthority) SetAuthzProcessing(ctx context.Context, req *sapb.AuthorizationID2) (*emptypb.Empty, error) {
+	if req.Id == 0 {
+		return nil, errIncompleteRequest
+	}
+	_, overallError := db.WithTransaction(ctx, ssa.dbMap, func(tx db.Executor) (any, error) {
+		result, err := tx.ExecContext(ctx, `
+		UPDATE authz2
+		SET beganProcessing = ?
+		WHERE id = ?
+		AND beganProcessing = ?`,
+			true,
+			req.Id,
+			false)
+		if err != nil {
+			return nil, berrors.InternalServerError("error updating authz to beganProcessing status")
+		}
+
+		n, err := result.RowsAffected()
+		if err != nil || n == 0 {
+			return nil, berrors.ConflictError("Authorization is already being validated. This may indicate your client attempted the same challenge multiple times, possibly due to a client bug.")
+		}
+
+		return nil, nil
+	})
+	if overallError != nil {
+		return nil, overallError
+	}
+	return &emptypb.Empty{}, nil
+}
+
 // SetOrderProcessing updates an order from pending status to processing
 // status by updating the `beganProcessing` field of the corresponding
 // Order table row in the DB.

sa/sa_test.go

@@ -1266,6 +1266,35 @@ func TestNewOrderAndAuthzs_Profile(t *testing.T) {
 	}
 }
 
+func TestSetAuthzProcessing(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("TestSetAuthzProcessing requires config-next")
+	}
+
+	sa, fc := initSA(t)
+
+	reg := createWorkingRegistration(t, sa)
+
+	// Add one valid authz
+	expires := fc.Now().Add(time.Hour)
+	authzID := createPendingAuthorization(t, sa, reg.Id, identifier.NewDNS("example.com"), expires)
+
+	// Set the authz to processing
+	_, err := sa.SetAuthzProcessing(t.Context(), &sapb.AuthorizationID2{Id: authzID})
+	if err != nil {
+		t.Fatalf("SetAuthzProcessing = %q, but want success", err)
+	}
+
+	// Try to set the same authz to be processing again. We should get an error.
+	_, err = sa.SetAuthzProcessing(context.Background(), &sapb.AuthorizationID2{Id: authzID})
+	if err == nil {
+		t.Fatal("SetAuthzProcessing again succeeded, but want error")
+	}
+	if !errors.Is(err, berrors.Conflict) {
+		t.Errorf("SetAuthzProcessing = %T, but want berrors.Conflict", err)
+	}
+}
+
 func TestSetOrderProcessing(t *testing.T) {
 	sa, fc := initSA(t)
 

test/config-next/ra.json

@@ -186,7 +186,8 @@
 			"CAARechecksFailOrder": true,
 			"AutomaticallyPauseZombieClients": true,
 			"DNSAccount01Enabled": true,
-			"DNSPersist01Enabled": true
+			"DNSPersist01Enabled": true,
+			"SetAuthzProcessing": true
 		}
 	},
 	"pa": {

test/integration/authz_test.go

@@ -3,6 +3,9 @@
 package integration
 
 import (
+	"os"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -54,3 +57,70 @@ func TestValidAuthzExpires(t *testing.T) {
 			actualExpires, expectedExpiresMin, expectedExpiresMax)
 	}
 }
+
+func TestParallelValidationConflict(t *testing.T) {
+	t.Parallel()
+
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("TestParallelValidationConflict requires config-next")
+	}
+
+	c, err := makeClient()
+	if err != nil {
+		t.Fatalf("making client: %s", err)
+	}
+
+	name := random_domain()
+	order, err := c.NewOrder(c.Account, []acme.Identifier{{Type: "dns", Value: name}})
+	if err != nil {
+		t.Fatalf("making order: %s", err)
+	}
+
+	authz, err := c.FetchAuthorization(c.Account, order.Authorizations[0])
+	if err != nil {
+		t.Fatalf("fetching authz: %s", err)
+	}
+
+	chall, ok := authz.ChallengeMap[acme.ChallengeTypeDNS01]
+	if !ok {
+		t.Fatalf("authz doesn't have dns-01 challenge")
+	}
+
+	// Setting up chall-test-srv to have an actual response isn't strictly
+	// necessary, but it makes the success/failure difference between the attempts
+	// below more obvious.
+	_, err = testSrvClient.AddDNS01Response(name, chall.KeyAuthorization)
+	if err != nil {
+		t.Fatalf("prepping chall-test-srv: %s", err)
+	}
+	t.Cleanup(func() {
+		testSrvClient.RemoveDNS01Response(chall.Token)
+	})
+
+	// Kick off two validations in parallel.
+	var wg sync.WaitGroup
+	errs := make([]error, 2)
+	for i := range 2 {
+		wg.Go(func() {
+			_, err := c.UpdateChallenge(c.Account, chall)
+			errs[i] = err
+		})
+	}
+	wg.Wait()
+
+	// Make sure we got one error and one success.
+	if errs[0] == nil && errs[1] == nil {
+		t.Error("parallel UpdateChallenge both succeeded, but want one failure")
+	} else if errs[0] != nil && errs[1] != nil {
+		t.Errorf("parallel UpdateChallenge both failed (%q and %q), but want one success", errs[0], errs[1])
+	}
+
+	// Make sure the one error is of type "conflict"
+	err = errs[0]
+	if err == nil {
+		err = errs[1]
+	}
+	if !strings.Contains(err.Error(), "urn:ietf:params:acme:error:conflict") {
+		t.Errorf("parallel UpdateChallenge = %q, but want 'conflict'", err)
+	}
+}