SA: Implement schema and methods for (account, hostname) pausing (#7490)

Add the storage implementation for our new (account, hostname) pair
pausing feature.

- Add schema and model for for the new paused table
- Add SA service methods for interacting with the paused table

Part of #7406
Part of #7475

mocks/sa.go

@@ -244,6 +244,26 @@ func (sa *StorageAuthority) SerialsForIncident(ctx context.Context, _ *sapb.Seri
 	return &ServerStreamClient[sapb.IncidentSerial]{}, nil
 }
 
+// CheckIdentifiersPaused is a mock
+func (sa *StorageAuthorityReadOnly) CheckIdentifiersPaused(_ context.Context, _ *sapb.PauseRequest, _ ...grpc.CallOption) (*sapb.Identifiers, error) {
+	return nil, nil
+}
+
+// CheckIdentifiersPaused is a mock
+func (sa *StorageAuthority) CheckIdentifiersPaused(_ context.Context, _ *sapb.PauseRequest, _ ...grpc.CallOption) (*sapb.Identifiers, error) {
+	return nil, nil
+}
+
+// GetPausedIdentifiers is a mock
+func (sa *StorageAuthorityReadOnly) GetPausedIdentifiers(_ context.Context, _ *sapb.RegistrationID, _ ...grpc.CallOption) (*sapb.Identifiers, error) {
+	return nil, nil
+}
+
+// GetPausedIdentifiers is a mock
+func (sa *StorageAuthority) GetPausedIdentifiers(_ context.Context, _ *sapb.RegistrationID, _ ...grpc.CallOption) (*sapb.Identifiers, error) {
+	return nil, nil
+}
+
 // GetRevokedCerts is a mock
 func (sa *StorageAuthorityReadOnly) GetRevokedCerts(ctx context.Context, _ *sapb.GetRevokedCertsRequest, _ ...grpc.CallOption) (sapb.StorageAuthorityReadOnly_GetRevokedCertsClient, error) {
 	return &ServerStreamClient[corepb.CRLEntry]{}, nil
@@ -457,7 +477,7 @@ func (sa *StorageAuthorityReadOnly) GetValidAuthorizations2(ctx context.Context,
 			RegistrationID: req.RegistrationID,
 			Expires:        &exp,
 			Identifier: identifier.ACMEIdentifier{
-				Type:  "dns",
+				Type:  identifier.DNS,
 				Value: name,
 			},
 			Challenges: []core.Challenge{

sa/database.go

@@ -290,6 +290,7 @@ func initTables(dbMap *borp.DbMap) {
 	dbMap.AddTableWithName(crlShardModel{}, "crlShards").SetKeys(true, "ID")
 	dbMap.AddTableWithName(revokedCertModel{}, "revokedCertificates").SetKeys(true, "ID")
 	dbMap.AddTableWithName(replacementOrderModel{}, "replacementOrders").SetKeys(true, "ID")
+	dbMap.AddTableWithName(pausedModel{}, "paused")
 
 	// Read-only maps used for selecting subsets of columns.
 	dbMap.AddTableWithName(CertStatusMetadata{}, "certificateStatus")

sa/db-next/boulder_sa/20240514000000_Paused.sql

@@ -0,0 +1,20 @@
+-- +migrate Up
+-- SQL in section 'Up' is executed when this migration is applied
+
+-- This table has no auto-incrementing primary key because we don't plan to
+-- partition it. This table expected to be < 800K rows initially and grow at a
+-- rate of ~18% per year.
+
+CREATE TABLE `paused` (
+  `registrationID` bigint(20) NOT NULL,
+  `identifierType` tinyint(4) NOT NULL,
+  `identifierValue` varchar(255) NOT NULL,
+  `pausedAt` datetime NOT NULL,
+  `unpausedAt` datetime DEFAULT NULL,
+  PRIMARY KEY (`registrationID`, `identifierType`, `identifierValue`)
+);
+
+-- +migrate Down
+-- SQL section 'Down' is executed when this migration is rolled back
+
+DROP TABLE `paused`;

sa/db-users/boulder_sa.sql

@@ -34,6 +34,8 @@ GRANT SELECT ON incidents TO 'sa'@'localhost';
 GRANT SELECT,INSERT,UPDATE ON crlShards TO 'sa'@'localhost';
 GRANT SELECT,INSERT,UPDATE ON revokedCertificates TO 'sa'@'localhost';
 GRANT SELECT,INSERT,UPDATE ON replacementOrders TO 'sa'@'localhost';
+-- Tests need to be able to TRUNCATE this table, so DROP is necessary.
+GRANT SELECT,INSERT,UPDATE,DROP ON paused TO 'sa'@'localhost';
 
 GRANT SELECT ON certificates TO 'sa_ro'@'localhost';
 GRANT SELECT ON certificateStatus TO 'sa_ro'@'localhost';
@@ -54,6 +56,7 @@ GRANT SELECT ON incidents TO 'sa_ro'@'localhost';
 GRANT SELECT ON crlShards TO 'sa_ro'@'localhost';
 GRANT SELECT ON revokedCertificates TO 'sa_ro'@'localhost';
 GRANT SELECT ON replacementOrders TO 'sa_ro'@'localhost';
+GRANT SELECT ON paused TO 'sa_ro'@'localhost';
 
 -- OCSP Responder
 GRANT SELECT ON certificateStatus TO 'ocsp_resp'@'localhost';

sa/model.go

@@ -1294,3 +1294,67 @@ func setReplacementOrderFinalized(ctx context.Context, db db.Execer, orderID int
 	}
 	return nil
 }
+
+type identifierModel struct {
+	Type  uint8  `db:"identifierType"`
+	Value string `db:"identifierValue"`
+}
+
+func newIdentifierModelFromPB(pb *sapb.Identifier) (identifierModel, error) {
+	idType, ok := identifierTypeToUint[pb.Type]
+	if !ok {
+		return identifierModel{}, fmt.Errorf("unsupported identifier type %q", pb.Type)
+	}
+
+	return identifierModel{
+		Type:  idType,
+		Value: pb.Value,
+	}, nil
+}
+
+func newPBFromIdentifierModel(id identifierModel) (*sapb.Identifier, error) {
+	idType, ok := uintToIdentifierType[id.Type]
+	if !ok {
+		return nil, fmt.Errorf("unsupported identifier type %d", id.Type)
+	}
+
+	return &sapb.Identifier{
+		Type:  idType,
+		Value: id.Value,
+	}, nil
+}
+
+func newIdentifierModelsFromPB(pb []*sapb.Identifier) ([]identifierModel, error) {
+	var ids []identifierModel
+	for _, p := range pb {
+		id, err := newIdentifierModelFromPB(p)
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, nil
+}
+
+func newPBFromIdentifierModels(ids []identifierModel) (*sapb.Identifiers, error) {
+	var pb []*sapb.Identifier
+	for _, id := range ids {
+		p, err := newPBFromIdentifierModel(id)
+		if err != nil {
+			return nil, err
+		}
+		pb = append(pb, p)
+	}
+	return &sapb.Identifiers{Identifiers: pb}, nil
+}
+
+// pausedModel represents a row in the paused table. The pausedAt and unpausedAt
+// fields are pointers because they are NULL-able columns. Valid states are:
+//   - Identifier paused: pausedAt is non-NULL, unpausedAt is NULL
+//   - Identifier unpaused: pausedAt is non-NULL, unpausedAt is non-NULL
+type pausedModel struct {
+	identifierModel
+	RegistrationID int64      `db:"registrationID"`
+	PausedAt       *time.Time `db:"pausedAt"`
+	UnpausedAt     *time.Time `db:"unpausedAt"`
+}

sa/proto/sa.proto

@@ -41,6 +41,8 @@ service StorageAuthorityReadOnly {
   rpc KeyBlocked(SPKIHash) returns (Exists) {}
   rpc ReplacementOrderExists(Serial) returns (Exists) {}
   rpc SerialsForIncident (SerialsForIncidentRequest) returns (stream IncidentSerial) {}
+  rpc CheckIdentifiersPaused (PauseRequest) returns (Identifiers) {}
+  rpc GetPausedIdentifiers (RegistrationID) returns (Identifiers) {}
 }
 
 // StorageAuthority provides full read/write access to the database.
@@ -77,6 +79,8 @@ service StorageAuthority {
   rpc KeyBlocked(SPKIHash) returns (Exists) {}
   rpc ReplacementOrderExists(Serial) returns (Exists) {}
   rpc SerialsForIncident (SerialsForIncidentRequest) returns (stream IncidentSerial) {}
+  rpc CheckIdentifiersPaused (PauseRequest) returns (Identifiers) {}
+  rpc GetPausedIdentifiers (RegistrationID) returns (Identifiers) {}
   // Adders
   rpc AddBlockedKey(AddBlockedKeyRequest) returns (google.protobuf.Empty) {}
   rpc AddCertificate(AddCertificateRequest) returns (google.protobuf.Empty) {}
@@ -96,6 +100,8 @@ service StorageAuthority {
   rpc UpdateRevokedCertificate(RevokeCertificateRequest) returns (google.protobuf.Empty) {}
   rpc LeaseCRLShard(LeaseCRLShardRequest) returns (LeaseCRLShardResponse) {}
   rpc UpdateCRLShard(UpdateCRLShardRequest) returns (google.protobuf.Empty) {}
+  rpc PauseIdentifiers(PauseRequest) returns (PauseIdentifiersResponse) {}
+  rpc UnpauseAccount(RegistrationID) returns (google.protobuf.Empty) {}
 }
 
 message RegistrationID {
@@ -414,3 +420,22 @@ message UpdateCRLShardRequest {
   google.protobuf.Timestamp thisUpdate = 3;
   google.protobuf.Timestamp nextUpdate = 4;
 }
+
+message Identifier {
+  string type = 1;
+  string value = 2;
+}
+
+message Identifiers {
+  repeated Identifier identifiers = 1;
+}
+
+message PauseRequest {
+  int64 registrationID = 1;
+  repeated Identifier identifiers = 2;
+}
+
+message PauseIdentifiersResponse {
+  int64 paused = 1;
+  int64 repaused = 2;
+}

sa/proto/sa_grpc.pb.go

@@ -53,6 +53,8 @@ const (
 	StorageAuthorityReadOnly_KeyBlocked_FullMethodName                   = "/sa.StorageAuthorityReadOnly/KeyBlocked"
 	StorageAuthorityReadOnly_ReplacementOrderExists_FullMethodName       = "/sa.StorageAuthorityReadOnly/ReplacementOrderExists"
 	StorageAuthorityReadOnly_SerialsForIncident_FullMethodName           = "/sa.StorageAuthorityReadOnly/SerialsForIncident"
+	StorageAuthorityReadOnly_CheckIdentifiersPaused_FullMethodName       = "/sa.StorageAuthorityReadOnly/CheckIdentifiersPaused"
+	StorageAuthorityReadOnly_GetPausedIdentifiers_FullMethodName         = "/sa.StorageAuthorityReadOnly/GetPausedIdentifiers"
 )
 
 // StorageAuthorityReadOnlyClient is the client API for StorageAuthorityReadOnly service.
@@ -90,6 +92,8 @@ type StorageAuthorityReadOnlyClient interface {
 	KeyBlocked(ctx context.Context, in *SPKIHash, opts ...grpc.CallOption) (*Exists, error)
 	ReplacementOrderExists(ctx context.Context, in *Serial, opts ...grpc.CallOption) (*Exists, error)
 	SerialsForIncident(ctx context.Context, in *SerialsForIncidentRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[IncidentSerial], error)
+	CheckIdentifiersPaused(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*Identifiers, error)
+	GetPausedIdentifiers(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*Identifiers, error)
 }
 
 type storageAuthorityReadOnlyClient struct {
@@ -446,6 +450,26 @@ func (c *storageAuthorityReadOnlyClient) SerialsForIncident(ctx context.Context,
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type StorageAuthorityReadOnly_SerialsForIncidentClient = grpc.ServerStreamingClient[IncidentSerial]
 
+func (c *storageAuthorityReadOnlyClient) CheckIdentifiersPaused(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*Identifiers, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(Identifiers)
+	err := c.cc.Invoke(ctx, StorageAuthorityReadOnly_CheckIdentifiersPaused_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *storageAuthorityReadOnlyClient) GetPausedIdentifiers(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*Identifiers, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(Identifiers)
+	err := c.cc.Invoke(ctx, StorageAuthorityReadOnly_GetPausedIdentifiers_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // StorageAuthorityReadOnlyServer is the server API for StorageAuthorityReadOnly service.
 // All implementations must embed UnimplementedStorageAuthorityReadOnlyServer
 // for forward compatibility
@@ -481,6 +505,8 @@ type StorageAuthorityReadOnlyServer interface {
 	KeyBlocked(context.Context, *SPKIHash) (*Exists, error)
 	ReplacementOrderExists(context.Context, *Serial) (*Exists, error)
 	SerialsForIncident(*SerialsForIncidentRequest, grpc.ServerStreamingServer[IncidentSerial]) error
+	CheckIdentifiersPaused(context.Context, *PauseRequest) (*Identifiers, error)
+	GetPausedIdentifiers(context.Context, *RegistrationID) (*Identifiers, error)
 	mustEmbedUnimplementedStorageAuthorityReadOnlyServer()
 }
 
@@ -581,6 +607,12 @@ func (UnimplementedStorageAuthorityReadOnlyServer) ReplacementOrderExists(contex
 func (UnimplementedStorageAuthorityReadOnlyServer) SerialsForIncident(*SerialsForIncidentRequest, grpc.ServerStreamingServer[IncidentSerial]) error {
 	return status.Errorf(codes.Unimplemented, "method SerialsForIncident not implemented")
 }
+func (UnimplementedStorageAuthorityReadOnlyServer) CheckIdentifiersPaused(context.Context, *PauseRequest) (*Identifiers, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CheckIdentifiersPaused not implemented")
+}
+func (UnimplementedStorageAuthorityReadOnlyServer) GetPausedIdentifiers(context.Context, *RegistrationID) (*Identifiers, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetPausedIdentifiers not implemented")
+}
 func (UnimplementedStorageAuthorityReadOnlyServer) mustEmbedUnimplementedStorageAuthorityReadOnlyServer() {
 }
 
@@ -1125,6 +1157,42 @@ func _StorageAuthorityReadOnly_SerialsForIncident_Handler(srv interface{}, strea
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type StorageAuthorityReadOnly_SerialsForIncidentServer = grpc.ServerStreamingServer[IncidentSerial]
 
+func _StorageAuthorityReadOnly_CheckIdentifiersPaused_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PauseRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityReadOnlyServer).CheckIdentifiersPaused(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthorityReadOnly_CheckIdentifiersPaused_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityReadOnlyServer).CheckIdentifiersPaused(ctx, req.(*PauseRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _StorageAuthorityReadOnly_GetPausedIdentifiers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(RegistrationID)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityReadOnlyServer).GetPausedIdentifiers(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthorityReadOnly_GetPausedIdentifiers_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityReadOnlyServer).GetPausedIdentifiers(ctx, req.(*RegistrationID))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // StorageAuthorityReadOnly_ServiceDesc is the grpc.ServiceDesc for StorageAuthorityReadOnly service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -1240,6 +1308,14 @@ var StorageAuthorityReadOnly_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "ReplacementOrderExists",
 			Handler:    _StorageAuthorityReadOnly_ReplacementOrderExists_Handler,
 		},
+		{
+			MethodName: "CheckIdentifiersPaused",
+			Handler:    _StorageAuthorityReadOnly_CheckIdentifiersPaused_Handler,
+		},
+		{
+			MethodName: "GetPausedIdentifiers",
+			Handler:    _StorageAuthorityReadOnly_GetPausedIdentifiers_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{
@@ -1298,6 +1374,8 @@ const (
 	StorageAuthority_KeyBlocked_FullMethodName                   = "/sa.StorageAuthority/KeyBlocked"
 	StorageAuthority_ReplacementOrderExists_FullMethodName       = "/sa.StorageAuthority/ReplacementOrderExists"
 	StorageAuthority_SerialsForIncident_FullMethodName           = "/sa.StorageAuthority/SerialsForIncident"
+	StorageAuthority_CheckIdentifiersPaused_FullMethodName       = "/sa.StorageAuthority/CheckIdentifiersPaused"
+	StorageAuthority_GetPausedIdentifiers_FullMethodName         = "/sa.StorageAuthority/GetPausedIdentifiers"
 	StorageAuthority_AddBlockedKey_FullMethodName                = "/sa.StorageAuthority/AddBlockedKey"
 	StorageAuthority_AddCertificate_FullMethodName               = "/sa.StorageAuthority/AddCertificate"
 	StorageAuthority_AddPrecertificate_FullMethodName            = "/sa.StorageAuthority/AddPrecertificate"
@@ -1316,6 +1394,8 @@ const (
 	StorageAuthority_UpdateRevokedCertificate_FullMethodName     = "/sa.StorageAuthority/UpdateRevokedCertificate"
 	StorageAuthority_LeaseCRLShard_FullMethodName                = "/sa.StorageAuthority/LeaseCRLShard"
 	StorageAuthority_UpdateCRLShard_FullMethodName               = "/sa.StorageAuthority/UpdateCRLShard"
+	StorageAuthority_PauseIdentifiers_FullMethodName             = "/sa.StorageAuthority/PauseIdentifiers"
+	StorageAuthority_UnpauseAccount_FullMethodName               = "/sa.StorageAuthority/UnpauseAccount"
 )
 
 // StorageAuthorityClient is the client API for StorageAuthority service.
@@ -1354,6 +1434,8 @@ type StorageAuthorityClient interface {
 	KeyBlocked(ctx context.Context, in *SPKIHash, opts ...grpc.CallOption) (*Exists, error)
 	ReplacementOrderExists(ctx context.Context, in *Serial, opts ...grpc.CallOption) (*Exists, error)
 	SerialsForIncident(ctx context.Context, in *SerialsForIncidentRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[IncidentSerial], error)
+	CheckIdentifiersPaused(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*Identifiers, error)
+	GetPausedIdentifiers(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*Identifiers, error)
 	// Adders
 	AddBlockedKey(ctx context.Context, in *AddBlockedKeyRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 	AddCertificate(ctx context.Context, in *AddCertificateRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
@@ -1373,6 +1455,8 @@ type StorageAuthorityClient interface {
 	UpdateRevokedCertificate(ctx context.Context, in *RevokeCertificateRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 	LeaseCRLShard(ctx context.Context, in *LeaseCRLShardRequest, opts ...grpc.CallOption) (*LeaseCRLShardResponse, error)
 	UpdateCRLShard(ctx context.Context, in *UpdateCRLShardRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	PauseIdentifiers(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*PauseIdentifiersResponse, error)
+	UnpauseAccount(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*emptypb.Empty, error)
 }
 
 type storageAuthorityClient struct {
@@ -1729,6 +1813,26 @@ func (c *storageAuthorityClient) SerialsForIncident(ctx context.Context, in *Ser
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type StorageAuthority_SerialsForIncidentClient = grpc.ServerStreamingClient[IncidentSerial]
 
+func (c *storageAuthorityClient) CheckIdentifiersPaused(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*Identifiers, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(Identifiers)
+	err := c.cc.Invoke(ctx, StorageAuthority_CheckIdentifiersPaused_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *storageAuthorityClient) GetPausedIdentifiers(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*Identifiers, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(Identifiers)
+	err := c.cc.Invoke(ctx, StorageAuthority_GetPausedIdentifiers_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *storageAuthorityClient) AddBlockedKey(ctx context.Context, in *AddBlockedKeyRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
 	out := new(emptypb.Empty)
@@ -1909,6 +2013,26 @@ func (c *storageAuthorityClient) UpdateCRLShard(ctx context.Context, in *UpdateC
 	return out, nil
 }
 
+func (c *storageAuthorityClient) PauseIdentifiers(ctx context.Context, in *PauseRequest, opts ...grpc.CallOption) (*PauseIdentifiersResponse, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(PauseIdentifiersResponse)
+	err := c.cc.Invoke(ctx, StorageAuthority_PauseIdentifiers_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *storageAuthorityClient) UnpauseAccount(ctx context.Context, in *RegistrationID, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(emptypb.Empty)
+	err := c.cc.Invoke(ctx, StorageAuthority_UnpauseAccount_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // StorageAuthorityServer is the server API for StorageAuthority service.
 // All implementations must embed UnimplementedStorageAuthorityServer
 // for forward compatibility
@@ -1945,6 +2069,8 @@ type StorageAuthorityServer interface {
 	KeyBlocked(context.Context, *SPKIHash) (*Exists, error)
 	ReplacementOrderExists(context.Context, *Serial) (*Exists, error)
 	SerialsForIncident(*SerialsForIncidentRequest, grpc.ServerStreamingServer[IncidentSerial]) error
+	CheckIdentifiersPaused(context.Context, *PauseRequest) (*Identifiers, error)
+	GetPausedIdentifiers(context.Context, *RegistrationID) (*Identifiers, error)
 	// Adders
 	AddBlockedKey(context.Context, *AddBlockedKeyRequest) (*emptypb.Empty, error)
 	AddCertificate(context.Context, *AddCertificateRequest) (*emptypb.Empty, error)
@@ -1964,6 +2090,8 @@ type StorageAuthorityServer interface {
 	UpdateRevokedCertificate(context.Context, *RevokeCertificateRequest) (*emptypb.Empty, error)
 	LeaseCRLShard(context.Context, *LeaseCRLShardRequest) (*LeaseCRLShardResponse, error)
 	UpdateCRLShard(context.Context, *UpdateCRLShardRequest) (*emptypb.Empty, error)
+	PauseIdentifiers(context.Context, *PauseRequest) (*PauseIdentifiersResponse, error)
+	UnpauseAccount(context.Context, *RegistrationID) (*emptypb.Empty, error)
 	mustEmbedUnimplementedStorageAuthorityServer()
 }
 
@@ -2064,6 +2192,12 @@ func (UnimplementedStorageAuthorityServer) ReplacementOrderExists(context.Contex
 func (UnimplementedStorageAuthorityServer) SerialsForIncident(*SerialsForIncidentRequest, grpc.ServerStreamingServer[IncidentSerial]) error {
 	return status.Errorf(codes.Unimplemented, "method SerialsForIncident not implemented")
 }
+func (UnimplementedStorageAuthorityServer) CheckIdentifiersPaused(context.Context, *PauseRequest) (*Identifiers, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method CheckIdentifiersPaused not implemented")
+}
+func (UnimplementedStorageAuthorityServer) GetPausedIdentifiers(context.Context, *RegistrationID) (*Identifiers, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetPausedIdentifiers not implemented")
+}
 func (UnimplementedStorageAuthorityServer) AddBlockedKey(context.Context, *AddBlockedKeyRequest) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method AddBlockedKey not implemented")
 }
@@ -2118,6 +2252,12 @@ func (UnimplementedStorageAuthorityServer) LeaseCRLShard(context.Context, *Lease
 func (UnimplementedStorageAuthorityServer) UpdateCRLShard(context.Context, *UpdateCRLShardRequest) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method UpdateCRLShard not implemented")
 }
+func (UnimplementedStorageAuthorityServer) PauseIdentifiers(context.Context, *PauseRequest) (*PauseIdentifiersResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method PauseIdentifiers not implemented")
+}
+func (UnimplementedStorageAuthorityServer) UnpauseAccount(context.Context, *RegistrationID) (*emptypb.Empty, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method UnpauseAccount not implemented")
+}
 func (UnimplementedStorageAuthorityServer) mustEmbedUnimplementedStorageAuthorityServer() {}
 
 // UnsafeStorageAuthorityServer may be embedded to opt out of forward compatibility for this service.
@@ -2661,6 +2801,42 @@ func _StorageAuthority_SerialsForIncident_Handler(srv interface{}, stream grpc.S
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type StorageAuthority_SerialsForIncidentServer = grpc.ServerStreamingServer[IncidentSerial]
 
+func _StorageAuthority_CheckIdentifiersPaused_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PauseRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityServer).CheckIdentifiersPaused(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthority_CheckIdentifiersPaused_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityServer).CheckIdentifiersPaused(ctx, req.(*PauseRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _StorageAuthority_GetPausedIdentifiers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(RegistrationID)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityServer).GetPausedIdentifiers(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthority_GetPausedIdentifiers_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityServer).GetPausedIdentifiers(ctx, req.(*RegistrationID))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _StorageAuthority_AddBlockedKey_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(AddBlockedKeyRequest)
 	if err := dec(in); err != nil {
@@ -2985,6 +3161,42 @@ func _StorageAuthority_UpdateCRLShard_Handler(srv interface{}, ctx context.Conte
 	return interceptor(ctx, in, info, handler)
 }
 
+func _StorageAuthority_PauseIdentifiers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PauseRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityServer).PauseIdentifiers(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthority_PauseIdentifiers_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityServer).PauseIdentifiers(ctx, req.(*PauseRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _StorageAuthority_UnpauseAccount_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(RegistrationID)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StorageAuthorityServer).UnpauseAccount(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StorageAuthority_UnpauseAccount_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StorageAuthorityServer).UnpauseAccount(ctx, req.(*RegistrationID))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // StorageAuthority_ServiceDesc is the grpc.ServiceDesc for StorageAuthority service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -3100,6 +3312,14 @@ var StorageAuthority_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "ReplacementOrderExists",
 			Handler:    _StorageAuthority_ReplacementOrderExists_Handler,
 		},
+		{
+			MethodName: "CheckIdentifiersPaused",
+			Handler:    _StorageAuthority_CheckIdentifiersPaused_Handler,
+		},
+		{
+			MethodName: "GetPausedIdentifiers",
+			Handler:    _StorageAuthority_GetPausedIdentifiers_Handler,
+		},
 		{
 			MethodName: "AddBlockedKey",
 			Handler:    _StorageAuthority_AddBlockedKey_Handler,
@@ -3172,6 +3392,14 @@ var StorageAuthority_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "UpdateCRLShard",
 			Handler:    _StorageAuthority_UpdateCRLShard_Handler,
 		},
+		{
+			MethodName: "PauseIdentifiers",
+			Handler:    _StorageAuthority_PauseIdentifiers_Handler,
+		},
+		{
+			MethodName: "UnpauseAccount",
+			Handler:    _StorageAuthority_UnpauseAccount_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{

sa/sa.go

@@ -3,6 +3,7 @@ package sa
 import (
 	"context"
 	"crypto/x509"
+	"database/sql"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -1312,3 +1313,131 @@ func (ssa *SQLStorageAuthority) UpdateCRLShard(ctx context.Context, req *sapb.Up
 
 	return &emptypb.Empty{}, nil
 }
+
+// PauseIdentifiers pauses a set of identifiers for the provided account. If an
+// identifier is currently paused, this is a no-op. If an identifier was
+// previously paused and unpaused, it will be repaused. All work is accomplished
+// in a transaction to limit possible race conditions.
+func (ssa *SQLStorageAuthority) PauseIdentifiers(ctx context.Context, req *sapb.PauseRequest) (*sapb.PauseIdentifiersResponse, error) {
+	if core.IsAnyNilOrZero(req.RegistrationID, req.Identifiers) {
+		return nil, errIncompleteRequest
+	}
+
+	// Marshal the identifier now that we've crossed the RPC boundary.
+	identifiers, err := newIdentifierModelsFromPB(req.Identifiers)
+	if err != nil {
+		return nil, err
+	}
+
+	response := &sapb.PauseIdentifiersResponse{}
+	_, err = db.WithTransaction(ctx, ssa.dbMap, func(tx db.Executor) (interface{}, error) {
+		for _, identifier := range identifiers {
+			pauseError := func(op string, err error) error {
+				return fmt.Errorf("while %s identifier %s for registration ID %d: %w",
+					op, identifier.Value, req.RegistrationID, err,
+				)
+			}
+
+			var entry pausedModel
+			err := tx.SelectOne(ctx, &entry, `
+			SELECT pausedAt, unpausedAt
+			FROM paused
+			WHERE 
+				registrationID = ? AND 
+				identifierType = ? AND 
+				identifierValue = ?`,
+				req.RegistrationID,
+				identifier.Type,
+				identifier.Value,
+			)
+
+			switch {
+			case err != nil && !errors.Is(err, sql.ErrNoRows):
+				// Error querying the database.
+				return nil, pauseError("querying pause status for", err)
+
+			case err != nil && errors.Is(err, sql.ErrNoRows):
+				// Not currently or previously paused, insert a new pause record.
+				pausedAt := ssa.clk.Now().Truncate(time.Second)
+				err = tx.Insert(ctx, &pausedModel{
+					RegistrationID: req.RegistrationID,
+					PausedAt:       &pausedAt,
+					identifierModel: identifierModel{
+						Type:  identifier.Type,
+						Value: identifier.Value,
+					},
+				})
+				if err != nil && !db.IsDuplicate(err) {
+					return nil, pauseError("pausing", err)
+				}
+
+				// Identifier successfully paused.
+				response.Paused++
+				continue
+
+			case entry.UnpausedAt == nil || entry.PausedAt.After(*entry.UnpausedAt):
+				// Identifier is already paused.
+				continue
+
+			case entry.UnpausedAt.After(*entry.PausedAt):
+				// Previously paused (and unpaused), repause the identifier.
+				_, err := tx.ExecContext(ctx, `
+				UPDATE paused
+				SET pausedAt = ?,
+					unpausedAt = NULL
+				WHERE 
+					registrationID = ? AND 
+					identifierType = ? AND 
+					identifierValue = ? AND
+					unpausedAt IS NOT NULL`,
+					ssa.clk.Now().Truncate(time.Second),
+					req.RegistrationID,
+					identifier.Type,
+					identifier.Value,
+				)
+				if err != nil {
+					return nil, pauseError("repausing", err)
+				}
+
+				// Identifier successfully repaused.
+				response.Repaused++
+				continue
+
+			default:
+				// This indicates a database state which should never occur.
+				return nil, fmt.Errorf("impossible database state encountered while pausing identifier %s",
+					identifier.Value,
+				)
+			}
+		}
+		return nil, nil
+	})
+	if err != nil {
+		// Error occurred during transaction.
+		return nil, err
+	}
+	return response, nil
+}
+
+// UnpauseAccount will unpause all paused identifiers for the provided account.
+// If no identifiers are currently paused, this is a no-op.
+func (ssa *SQLStorageAuthority) UnpauseAccount(ctx context.Context, req *sapb.RegistrationID) (*emptypb.Empty, error) {
+	if core.IsAnyNilOrZero(req.Id) {
+		return nil, errIncompleteRequest
+	}
+
+	_, err := ssa.dbMap.ExecContext(ctx, `
+	UPDATE paused
+	SET unpausedAt = ?
+	WHERE 
+		registrationID = ? AND
+		unpausedAt IS NULL`,
+		ssa.clk.Now().Truncate(time.Second),
+		req.Id,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	return nil, nil
+}

sa/sa_test.go

@@ -40,6 +40,7 @@ import (
 	berrors "github.com/letsencrypt/boulder/errors"
 	"github.com/letsencrypt/boulder/features"
 	bgrpc "github.com/letsencrypt/boulder/grpc"
+	"github.com/letsencrypt/boulder/identifier"
 	blog "github.com/letsencrypt/boulder/log"
 	"github.com/letsencrypt/boulder/metrics"
 	"github.com/letsencrypt/boulder/probs"
@@ -4307,3 +4308,583 @@ func TestGetSerialsByAccount(t *testing.T) {
 	test.AssertNotError(t, err, "calling GetSerialsByAccount")
 	test.AssertEquals(t, len(seen), 2)
 }
+
+func TestUnpauseAccount(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("Test requires paused database table")
+	}
+	sa, _, cleanUp := initSA(t)
+	defer cleanUp()
+
+	ptrTime := func(t time.Time) *time.Time {
+		return &t
+	}
+
+	type args struct {
+		state []pausedModel
+		req   *sapb.RegistrationID
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "UnpauseAccount with no paused identifiers",
+			args: args{
+				state: nil,
+				req:   &sapb.RegistrationID{Id: 1},
+			},
+		},
+		{
+			name: "UnpauseAccount with one paused identifier",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+				},
+				req: &sapb.RegistrationID{Id: 1},
+			},
+		},
+		{
+			name: "UnpauseAccount with multiple paused identifiers",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.net",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.org",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+				},
+				req: &sapb.RegistrationID{Id: 1},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup table state.
+			for _, state := range tt.args.state {
+				err := sa.dbMap.Insert(ctx, &state)
+				test.AssertNotError(t, err, "inserting test identifier")
+			}
+
+			_, err := sa.UnpauseAccount(ctx, tt.args.req)
+			test.AssertNotError(t, err, "Unexpected error for UnpauseAccount()")
+
+			// Count the number of paused identifiers.
+			var count int
+			err = sa.dbReadOnlyMap.SelectOne(
+				ctx,
+				&count,
+				"SELECT COUNT(*) FROM paused WHERE registrationID = ? AND unpausedAt IS NULL",
+				tt.args.req.Id,
+			)
+			test.AssertNotError(t, err, "SELECT COUNT(*) failed")
+			test.AssertEquals(t, count, 0)
+
+			// Drop all rows from the paused table.
+			_, err = sa.dbMap.ExecContext(ctx, "TRUNCATE TABLE paused")
+			test.AssertNotError(t, err, "Truncate table paused failed")
+		})
+	}
+}
+
+func TestPauseIdentifiers(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("Test requires paused database table")
+	}
+	sa, _, cleanUp := initSA(t)
+	defer cleanUp()
+
+	ptrTime := func(t time.Time) *time.Time {
+		return &t
+	}
+
+	type args struct {
+		state []pausedModel
+		req   *sapb.PauseRequest
+	}
+	tests := []struct {
+		name string
+		args args
+		want *sapb.PauseIdentifiersResponse
+	}{
+		{
+			name: "An identifier which is not now or previously paused",
+			args: args{
+				state: nil,
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+					},
+				},
+			},
+			want: &sapb.PauseIdentifiersResponse{
+				Paused:   1,
+				Repaused: 0,
+			},
+		},
+		{
+			name: "One unpaused entry which was previously paused",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt:   ptrTime(sa.clk.Now().Add(-time.Hour)),
+						UnpausedAt: ptrTime(sa.clk.Now().Add(-time.Minute)),
+					},
+				},
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+					},
+				},
+			},
+			want: &sapb.PauseIdentifiersResponse{
+				Paused:   0,
+				Repaused: 1,
+			},
+		},
+		{
+			name: "An identifier which is currently paused",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+				},
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+					},
+				},
+			},
+			want: &sapb.PauseIdentifiersResponse{
+				Paused:   0,
+				Repaused: 0,
+			},
+		},
+		{
+			name: "Two previously paused entries and one new entry",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt:   ptrTime(sa.clk.Now().Add(-time.Hour)),
+						UnpausedAt: ptrTime(sa.clk.Now().Add(-time.Minute)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.net",
+						},
+						PausedAt:   ptrTime(sa.clk.Now().Add(-time.Hour)),
+						UnpausedAt: ptrTime(sa.clk.Now().Add(-time.Minute)),
+					},
+				},
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.net",
+						},
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.org",
+						},
+					},
+				},
+			},
+			want: &sapb.PauseIdentifiersResponse{
+				Paused:   1,
+				Repaused: 2,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup table state.
+			for _, state := range tt.args.state {
+				err := sa.dbMap.Insert(ctx, &state)
+				test.AssertNotError(t, err, "inserting test identifier")
+			}
+
+			got, err := sa.PauseIdentifiers(ctx, tt.args.req)
+			test.AssertNotError(t, err, "Unexpected error for PauseIdentifiers()")
+			test.AssertEquals(t, got.Paused, tt.want.Paused)
+			test.AssertEquals(t, got.Repaused, tt.want.Repaused)
+
+			// Drop all rows from the paused table.
+			_, err = sa.dbMap.ExecContext(ctx, "TRUNCATE TABLE paused")
+			test.AssertNotError(t, err, "Truncate table paused failed")
+		})
+	}
+}
+
+func TestCheckIdentifiersPaused(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("Test requires paused database table")
+	}
+	sa, _, cleanUp := initSA(t)
+	defer cleanUp()
+
+	ptrTime := func(t time.Time) *time.Time {
+		return &t
+	}
+
+	type args struct {
+		state []pausedModel
+		req   *sapb.PauseRequest
+	}
+	tests := []struct {
+		name string
+		args args
+		want *sapb.Identifiers
+	}{
+		{
+			name: "No paused identifiers",
+			args: args{
+				state: nil,
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+					},
+				},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: nil,
+			},
+		},
+		{
+			name: "One paused identifier",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+				},
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+					},
+				},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: []*sapb.Identifier{
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.com",
+					},
+				},
+			},
+		},
+		{
+			name: "Two paused identifiers, one unpaused",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.net",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.org",
+						},
+						PausedAt:   ptrTime(sa.clk.Now().Add(-time.Hour)),
+						UnpausedAt: ptrTime(sa.clk.Now().Add(-time.Minute)),
+					},
+				},
+				req: &sapb.PauseRequest{
+					RegistrationID: 1,
+					Identifiers: []*sapb.Identifier{
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.com",
+						},
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.net",
+						},
+						{
+							Type:  string(identifier.DNS),
+							Value: "example.org",
+						},
+					},
+				},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: []*sapb.Identifier{
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.com",
+					},
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.net",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup table state.
+			for _, state := range tt.args.state {
+				err := sa.dbMap.Insert(ctx, &state)
+				test.AssertNotError(t, err, "inserting test identifier")
+			}
+
+			got, err := sa.CheckIdentifiersPaused(ctx, tt.args.req)
+			test.AssertNotError(t, err, "Unexpected error for PauseIdentifiers()")
+			test.AssertDeepEquals(t, got.Identifiers, tt.want.Identifiers)
+
+			// Drop all rows from the paused table.
+			_, err = sa.dbMap.ExecContext(ctx, "TRUNCATE TABLE paused")
+			test.AssertNotError(t, err, "Truncate table paused failed")
+		})
+	}
+}
+
+func TestGetPausedIdentifiers(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("Test requires paused database table")
+	}
+	sa, _, cleanUp := initSA(t)
+	defer cleanUp()
+
+	ptrTime := func(t time.Time) *time.Time {
+		return &t
+	}
+
+	type args struct {
+		state []pausedModel
+		req   *sapb.RegistrationID
+	}
+	tests := []struct {
+		name string
+		args args
+		want *sapb.Identifiers
+	}{
+		{
+			name: "No paused identifiers",
+			args: args{
+				state: nil,
+				req:   &sapb.RegistrationID{Id: 1},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: nil,
+			},
+		},
+		{
+			name: "One paused identifier",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+				},
+				req: &sapb.RegistrationID{Id: 1},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: []*sapb.Identifier{
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.com",
+					},
+				},
+			},
+		},
+		{
+			name: "Two paused identifiers, one unpaused",
+			args: args{
+				state: []pausedModel{
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.com",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.net",
+						},
+						PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+					},
+					{
+						RegistrationID: 1,
+						identifierModel: identifierModel{
+							Type:  identifierTypeToUint[string(identifier.DNS)],
+							Value: "example.org",
+						},
+						PausedAt:   ptrTime(sa.clk.Now().Add(-time.Hour)),
+						UnpausedAt: ptrTime(sa.clk.Now().Add(-time.Minute)),
+					},
+				},
+				req: &sapb.RegistrationID{Id: 1},
+			},
+			want: &sapb.Identifiers{
+				Identifiers: []*sapb.Identifier{
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.com",
+					},
+					{
+						Type:  string(identifier.DNS),
+						Value: "example.net",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup table state.
+			for _, state := range tt.args.state {
+				err := sa.dbMap.Insert(ctx, &state)
+				test.AssertNotError(t, err, "inserting test identifier")
+			}
+
+			got, err := sa.GetPausedIdentifiers(ctx, tt.args.req)
+			test.AssertNotError(t, err, "Unexpected error for PauseIdentifiers()")
+			test.AssertDeepEquals(t, got.Identifiers, tt.want.Identifiers)
+
+			// Drop all rows from the paused table.
+			_, err = sa.dbMap.ExecContext(ctx, "TRUNCATE TABLE paused")
+			test.AssertNotError(t, err, "Truncate table paused failed")
+		})
+	}
+}
+
+func TestGetPausedIdentifiersOnlyUnpausesOneAccount(t *testing.T) {
+	if os.Getenv("BOULDER_CONFIG_DIR") != "test/config-next" {
+		t.Skip("Test requires paused database table")
+	}
+	sa, _, cleanUp := initSA(t)
+	defer cleanUp()
+
+	ptrTime := func(t time.Time) *time.Time {
+		return &t
+	}
+
+	// Insert two paused identifiers for two different accounts.
+	err := sa.dbMap.Insert(ctx, &pausedModel{
+		RegistrationID: 1,
+		identifierModel: identifierModel{
+			Type:  identifierTypeToUint[string(identifier.DNS)],
+			Value: "example.com",
+		},
+		PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+	})
+	test.AssertNotError(t, err, "inserting test identifier")
+
+	err = sa.dbMap.Insert(ctx, &pausedModel{
+		RegistrationID: 2,
+		identifierModel: identifierModel{
+			Type:  identifierTypeToUint[string(identifier.DNS)],
+			Value: "example.net",
+		},
+		PausedAt: ptrTime(sa.clk.Now().Add(-time.Hour)),
+	})
+	test.AssertNotError(t, err, "inserting test identifier")
+
+	// Unpause the first account.
+	_, err = sa.UnpauseAccount(ctx, &sapb.RegistrationID{Id: 1})
+	test.AssertNotError(t, err, "UnpauseAccount failed")
+
+	// Check that the second account's identifier is still paused.
+	identifiers, err := sa.GetPausedIdentifiers(ctx, &sapb.RegistrationID{Id: 2})
+	test.AssertNotError(t, err, "GetPausedIdentifiers failed")
+	test.AssertEquals(t, len(identifiers.Identifiers), 1)
+	test.AssertEquals(t, identifiers.Identifiers[0].Value, "example.net")
+}

sa/saro.go

@@ -1402,3 +1402,96 @@ func (ssa *SQLStorageAuthorityRO) GetSerialsByAccount(req *sapb.RegistrationID,
 		return stream.Send(&sapb.Serial{Serial: row.Serial})
 	})
 }
+
+// CheckIdentifiersPaused takes a slice of identifiers and returns a slice of
+// the first 15 identifier values which are currently paused for the provided
+// account. If no matches are found, an empty slice is returned.
+func (ssa *SQLStorageAuthorityRO) CheckIdentifiersPaused(ctx context.Context, req *sapb.PauseRequest) (*sapb.Identifiers, error) {
+	if core.IsAnyNilOrZero(req.RegistrationID, req.Identifiers) {
+		return nil, errIncompleteRequest
+	}
+
+	identifiers, err := newIdentifierModelsFromPB(req.Identifiers)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(identifiers) == 0 {
+		// No identifier values to check.
+		return nil, nil
+	}
+
+	identifiersByType := map[uint8][]string{}
+	for _, id := range identifiers {
+		identifiersByType[id.Type] = append(identifiersByType[id.Type], id.Value)
+	}
+
+	// Build a query to retrieve up to 15 paused identifiers using OR clauses
+	// for conditions specific to each type. This approach handles mixed
+	// identifier types in a single query. Assuming 3 DNS identifiers and 1 IP
+	// identifier, the resulting query would look like:
+	//
+	// SELECT identifierType, identifierValue
+	// FROM paused WHERE registrationID = ? AND
+	// unpausedAt IS NULL AND
+	//     ((identifierType = ? AND identifierValue IN (?, ?, ?)) OR
+	//      (identifierType = ? AND identifierValue IN (?)))
+	// LIMIT 15
+	//
+	// Corresponding args array for placeholders: [<regID>, 0, "example.com",
+	// "example.net", "example.org", 1, "1.2.3.4"]
+
+	var conditions []string
+	args := []interface{}{req.RegistrationID}
+	for idType, values := range identifiersByType {
+		conditions = append(conditions,
+			fmt.Sprintf("identifierType = ? AND identifierValue IN (%s)",
+				db.QuestionMarks(len(values)),
+			),
+		)
+		args = append(args, idType)
+		for _, value := range values {
+			args = append(args, value)
+		}
+	}
+
+	query := fmt.Sprintf(`
+        SELECT identifierType, identifierValue
+        FROM paused
+        WHERE registrationID = ? AND unpausedAt IS NULL AND (%s) LIMIT 15`,
+		strings.Join(conditions, " OR "))
+
+	var matches []identifierModel
+	_, err = ssa.dbReadOnlyMap.Select(ctx, &matches, query, args...)
+	if err != nil && !db.IsNoRows(err) {
+		// Error querying the database.
+		return nil, err
+	}
+
+	return newPBFromIdentifierModels(matches)
+}
+
+// GetPausedIdentifiers returns a slice of paused identifiers for the provided
+// account. If no paused identifiers are found, an empty slice is returned. The
+// results are limited to the first 15 paused identifiers.
+func (ssa *SQLStorageAuthorityRO) GetPausedIdentifiers(ctx context.Context, req *sapb.RegistrationID) (*sapb.Identifiers, error) {
+	if core.IsAnyNilOrZero(req.Id) {
+		return nil, errIncompleteRequest
+	}
+
+	var matches []identifierModel
+	_, err := ssa.dbReadOnlyMap.Select(ctx, &matches, `
+		SELECT identifierType, identifierValue
+		FROM paused
+		WHERE 
+			registrationID = ? AND
+			unpausedAt IS NULL
+		LIMIT 15`,
+		req.Id,
+	)
+	if err != nil && !db.IsNoRows(err) {
+		return nil, err
+	}
+
+	return newPBFromIdentifierModels(matches)
+}