Add ListValidatedMFAChallenges RPC to MFA service for leaf cluster resource replication (#63773)

* Try impl leaf cluster replication using pure watcher API.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Add ValidatedMFAChallenge replication to leaf clusters.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Remove changes unrelated to MFA proto/gRPC updates.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Add optional filtering.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Authorize local proxy identities only. Remove cluster exists check.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Make target cluster primary key in storing resource.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

* Remove unnecessary filter.

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

---------

Signed-off-by: Chris Thach <chris.thach@goteleport.com>

Author: cthach

api/gen/proto/go/teleport/mfa/v1/service.pb.go

@@ -30,6 +30,10 @@ service MFAService {
   // ValidateSessionChallenge validates the MFA challenge response for a user session and stores the validated response
   // in the backend.
   rpc ValidateSessionChallenge(ValidateSessionChallengeRequest) returns (ValidateSessionChallengeResponse);
+  // ListValidatedMFAChallenges lists validated MFA challenges that have been created for all user sessions. This is
+  // intended to be used by the reverse tunnel server to watch for new validated challenges that need to be replicated
+  // for SSH session establishment in leaf clusters.
+  rpc ListValidatedMFAChallenges(ListValidatedMFAChallengesRequest) returns (ListValidatedMFAChallengesResponse);
   // ReplicateValidatedMFAChallenge replicates a validated MFA challenge from root cluster to leaf cluster for
   // verification during SSH session establishment. The reverse tunnel server watches for validated challenges in the
   // root cluster and invokes this RPC on the leaf cluster. This is a NOOP when invoked in the root cluster.
@@ -74,6 +78,32 @@ message ValidateSessionChallengeRequest {
 // ValidateSessionChallengeResponse is the response message for ValidateSessionChallenge.
 message ValidateSessionChallengeResponse {}
 
+// ListValidatedMFAChallengesFilter is used to filter validated MFA challenges in ListValidatedMFAChallengesRequest.
+message ListValidatedMFAChallengesFilter {
+  // If set, only return validated MFA challenges for the specified target cluster.
+  optional string target_cluster = 1;
+}
+
+// ListValidatedMFAChallengesRequest is the request message for ListValidatedMFAChallenges.
+message ListValidatedMFAChallengesRequest {
+  // The maximum number of items to return. The server may impose a different page size at its discretion.
+  int32 page_size = 1;
+  // The next_page_token value returned from a previous List request, if any.
+  string page_token = 2;
+  // Collection of fields to filter challenges by. Challenges must match all filter fields to be included in the
+  // results. If unset, no filtering is performed and all validated MFA challenges are returned (up to the page size
+  // limit).
+  ListValidatedMFAChallengesFilter filter = 3;
+}
+
+// ListValidatedMFAChallengesResponse is the response message for ListValidatedMFAChallenges.
+message ListValidatedMFAChallengesResponse {
+  // List of validated MFA challenges that have been created for user sessions.
+  repeated ValidatedMFAChallenge validated_challenges = 1;
+  // Token to retrieve the next page of results, or empty if there are no more results exist.
+  string next_page_token = 2;
+}
+
 // ReplicateValidatedMFAChallengeRequest is the request message for ReplicateValidatedMFAChallenge.
 message ReplicateValidatedMFAChallengeRequest {
   // Resource name for the issued challenge. Must match the AuthenticateChallenge.name in order to find the correct

api/gen/proto/go/teleport/mfa/v1/service_grpc.pb.go

@@ -42,6 +42,9 @@ message ValidatedMFAChallengeSpec {
   string source_cluster = 2;
   // Name of the cluster where the SSH session is being established and this resource is intended for.
   string target_cluster = 3;
+  // Username of the Teleport user for whom the challenge was issued. This should be the Teleport username (not the SSH
+  // login name) and must correspond to a user in the cluster specified to source_cluster.
+  string username = 4;
 }
 
 // SessionIdentifyingPayload contains a value that uniquely identifies a user's session. It must be computed by the

api/gen/proto/go/teleport/mfa/v1/validated_challenge.pb.go

@@ -145,6 +145,13 @@ type mockMFAService struct {
 
 	createValidatedMFAChallengeError error
 	getValidatedMFAChallengeError    error
+
+	listValidatedMFAChallenges          []*mfav1.ValidatedMFAChallenge
+	listValidatedMFAChallengesToken     string
+	listValidatedMFAChallengesError     error
+	listValidatedMFAChallengesPageSize  int32
+	listValidatedMFAChallengesPageToken string
+	listValidatedMFAChallengesTarget    string
 }
 
 func (m *mockMFAService) CreateValidatedMFAChallenge(
@@ -178,3 +185,22 @@ func (m *mockMFAService) GetValidatedMFAChallenge(
 
 	return m.chal, nil
 }
+
+func (m *mockMFAService) ListValidatedMFAChallenges(
+	_ context.Context,
+	pageSize int32,
+	pageToken string,
+	targetCluster string,
+) ([]*mfav1.ValidatedMFAChallenge, string, error) {
+	if m.listValidatedMFAChallengesError != nil {
+		return nil, "", m.listValidatedMFAChallengesError
+	}
+
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.listValidatedMFAChallengesPageSize = pageSize
+	m.listValidatedMFAChallengesPageToken = pageToken
+	m.listValidatedMFAChallengesTarget = targetCluster
+
+	return m.listValidatedMFAChallenges, m.listValidatedMFAChallengesToken, nil
+}