User Profile: Add initial search profile API (#83191)

This PR adds an initial API for search user profiles. As discussed, it
is intentially kept minimal and implements in its basic form:
* It's a dedicate API (no query dsl)
* It searches against names (username, full name etc) with the given
  string
* It takes a size parameter to control the response size

It also optionally returns the data section of the profile document
since we agreed that read is not subject to namespacing.

docs/changelog/83191.yaml

@@ -0,0 +1,5 @@
+pr: 83191
+summary: "User Profile: Add initial search profile API"
+area: Security
+type: enhancement
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/profile/GetProfileRequest.java

@@ -18,32 +18,32 @@ import java.util.Set;
 public class GetProfileRequest extends ActionRequest {
 
     private final String uid;
-    private final Set<String> datKeys;
+    private final Set<String> dataKeys;
 
-    public GetProfileRequest(String uid, Set<String> datKeys) {
+    public GetProfileRequest(String uid, Set<String> dataKeys) {
         this.uid = uid;
-        this.datKeys = datKeys;
+        this.dataKeys = dataKeys;
     }
 
     public GetProfileRequest(StreamInput in) throws IOException {
         super(in);
         this.uid = in.readString();
-        this.datKeys = in.readSet(StreamInput::readString);
+        this.dataKeys = in.readSet(StreamInput::readString);
     }
 
     public String getUid() {
         return uid;
     }
 
-    public Set<String> getDatKeys() {
-        return datKeys;
+    public Set<String> getDataKeys() {
+        return dataKeys;
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeString(uid);
-        out.writeStringCollection(datKeys);
+        out.writeStringCollection(dataKeys);
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/profile/Profile.java

@@ -124,15 +124,19 @@ public record Profile(
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
+        innerToXContent(builder, params);
+        versionControl.toXContent(builder, params);
+        builder.endObject();
+        return builder;
+    }
+
+    public void innerToXContent(XContentBuilder builder, Params params) throws IOException {
         builder.field("uid", uid);
         builder.field("enabled", enabled);
         builder.field("last_synchronized", lastSynchronized);
         user.toXContent(builder, params);
         builder.field("access", access);
         builder.field("data", applicationData);
-        versionControl.toXContent(builder, params);
-        builder.endObject();
-        return builder;
     }
 
     @Override

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/profile/SearchProfilesAction.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.action.profile;
+
+import org.elasticsearch.action.ActionType;
+
+public class SearchProfilesAction extends ActionType<SearchProfilesResponse> {
+
+    public static final String NAME = "cluster:admin/xpack/security/profile/search";
+    public static final SearchProfilesAction INSTANCE = new SearchProfilesAction();
+
+    public SearchProfilesAction() {
+        super(NAME, SearchProfilesResponse::new);
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/profile/SearchProfilesRequest.java

@@ -0,0 +1,71 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.action.profile;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
+public class SearchProfilesRequest extends ActionRequest {
+
+    private final Set<String> dataKeys;
+    /**
+     * String to search name related fields of a profile document
+     */
+    private final String name;
+    private final int size;
+
+    public SearchProfilesRequest(Set<String> dataKeys, String name, int size) {
+        this.dataKeys = Objects.requireNonNull(dataKeys, "data parameter must not be null");
+        this.name = Objects.requireNonNull(name, "name must not be null");
+        this.size = size;
+    }
+
+    public SearchProfilesRequest(StreamInput in) throws IOException {
+        super(in);
+        this.dataKeys = in.readSet(StreamInput::readString);
+        this.name = in.readOptionalString();
+        this.size = in.readVInt();
+    }
+
+    public Set<String> getDataKeys() {
+        return dataKeys;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public int getSize() {
+        return size;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringCollection(dataKeys);
+        out.writeOptionalString(name);
+        out.writeVInt(size);
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (size < 0) {
+            validationException = addValidationError("[size] parameter cannot be negative but was [" + size + "]", validationException);
+        }
+        return validationException;
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/profile/SearchProfilesResponse.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.action.profile;
+
+import org.apache.lucene.search.TotalHits;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+public class SearchProfilesResponse extends ActionResponse implements ToXContentObject {
+
+    private final ProfileHit[] profileHits;
+    private final long tookInMillis;
+    private final TotalHits totalHits;
+
+    public SearchProfilesResponse(ProfileHit[] profileHits, long tookInMillis, TotalHits totalHits) {
+        this.profileHits = profileHits;
+        this.tookInMillis = tookInMillis;
+        this.totalHits = totalHits;
+    }
+
+    public SearchProfilesResponse(StreamInput in) throws IOException {
+        super(in);
+        this.profileHits = in.readArray(ProfileHit::new, ProfileHit[]::new);
+        this.tookInMillis = in.readVLong();
+        this.totalHits = Lucene.readTotalHits(in);
+    }
+
+    public ProfileHit[] getProfileHits() {
+        return profileHits;
+    }
+
+    public long getTookInMillis() {
+        return tookInMillis;
+    }
+
+    public TotalHits getTotalHits() {
+        return totalHits;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeArray(profileHits);
+        out.writeVLong(tookInMillis);
+        Lucene.writeTotalHits(out, totalHits);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.field("took", tookInMillis);
+            builder.startObject("total");
+            {
+                builder.field("value", totalHits.value);
+                builder.field("relation", totalHits.relation == TotalHits.Relation.EQUAL_TO ? "eq" : "gte");
+            }
+            builder.endObject();
+            builder.startArray("users");
+            {
+                for (ProfileHit profileHit : profileHits) {
+                    profileHit.toXContent(builder, params);
+                }
+            }
+            builder.endArray();
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public record ProfileHit(Profile profile, float score) implements Writeable, ToXContentObject {
+
+        public ProfileHit(StreamInput in) throws IOException {
+            this(new Profile(in), in.readFloat());
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            profile.writeTo(out);
+            out.writeFloat(score);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field("_score", score);
+                builder.field("uid", profile.uid());
+                profile.user().toXContent(builder, params);
+                builder.field("access", profile.access());
+                builder.field("data", profile.applicationData());
+                // TODO: output a field of sort which is just score plus uid?
+            }
+            builder.endObject();
+            return builder;
+        }
+    }
+}

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -197,6 +197,7 @@ public class Constants {
         "cluster:admin/xpack/security/profile/activate",
         "cluster:admin/xpack/security/profile/get",
         "cluster:admin/xpack/security/profile/put/data",
+        "cluster:admin/xpack/security/profile/search",
         "cluster:admin/xpack/security/realm/cache/clear",
         "cluster:admin/xpack/security/role/delete",
         "cluster:admin/xpack/security/role/get",

x-pack/plugin/security/qa/profile/src/javaRestTest/java/org/elasticsearch/xpack/security/profile/ProfileIT.java

@@ -24,6 +24,8 @@ import java.util.Map;
 import static org.hamcrest.Matchers.anEmptyMap;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.hasSize;
 
 public class ProfileIT extends ESRestTestCase {
 
@@ -137,6 +139,27 @@ public class ProfileIT extends ESRestTestCase {
         assertThat(castToMap(profileMap1.get("data")), equalTo(Map.of("app1", Map.of("theme", "default"))));
     }
 
+    public void testSearchProfile() throws IOException {
+        final Map<String, Object> activateProfileMap = doActivateProfile();
+        final String uid = (String) activateProfileMap.get("uid");
+        final Request searchProfilesRequest1 = new Request(randomFrom("GET", "POST"), "_security/profile/_search");
+        searchProfilesRequest1.setJsonEntity("""
+            {
+              "name": "rac",
+              "size": 10
+            }""");
+        final Response searchProfilesResponse1 = adminClient().performRequest(searchProfilesRequest1);
+        assertOK(searchProfilesResponse1);
+        final Map<String, Object> searchProfileResponseMap1 = responseAsMap(searchProfilesResponse1);
+        assertThat(searchProfileResponseMap1, hasKey("took"));
+        assertThat(searchProfileResponseMap1.get("total"), equalTo(Map.of("value", 1, "relation", "eq")));
+        @SuppressWarnings("unchecked")
+        final List<Map<String, Object>> users = (List<Map<String, Object>>) searchProfileResponseMap1.get("users");
+        assertThat(users, hasSize(1));
+        assertThat(users.get(0), hasKey("_score"));
+        assertThat(users.get(0).get("uid"), equalTo(uid));
+    }
+
     private Map<String, Object> doActivateProfile() throws IOException {
         final Request activateProfileRequest = new Request("POST", "_security/profile/_activate");
         activateProfileRequest.setJsonEntity("""

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/profile/ProfileSingleNodeTests.java

@@ -7,30 +7,42 @@
 
 package org.elasticsearch.xpack.security.profile;
 
+import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.action.admin.indices.get.GetIndexAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
+import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.engine.DocumentMissingException;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.core.security.action.profile.GetProfileAction;
+import org.elasticsearch.xpack.core.security.action.profile.GetProfileRequest;
+import org.elasticsearch.xpack.core.security.action.profile.GetProfilesResponse;
 import org.elasticsearch.xpack.core.security.action.profile.Profile;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesAction;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesRequest;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesResponse;
 import org.elasticsearch.xpack.core.security.action.profile.UpdateProfileDataAction;
 import org.elasticsearch.xpack.core.security.action.profile.UpdateProfileDataRequest;
 import org.elasticsearch.xpack.core.security.action.user.PutUserAction;
 import org.elasticsearch.xpack.core.security.action.user.PutUserRequest;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.test.SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING;
 import static org.elasticsearch.xpack.security.support.SecuritySystemIndices.INTERNAL_SECURITY_PROFILE_INDEX_8;
 import static org.elasticsearch.xpack.security.support.SecuritySystemIndices.SECURITY_PROFILE_ALIAS;
 import static org.hamcrest.Matchers.anEmptyMap;
-import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.emptyArray;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItemInArray;
 import static org.hamcrest.Matchers.hasItems;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.hasSize;
@@ -49,19 +61,17 @@ public class ProfileSingleNodeTests extends AbstractProfileSingleNodeTestCase {
     }
 
     public void testProfileIndexAutoCreation() {
+        // Index does not exist yet
+        assertThat(getProfileIndexResponse().getIndices(), not(hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8)));
+
+        // Trigger index creation by indexing
         var indexResponse = client().prepareIndex(randomFrom(INTERNAL_SECURITY_PROFILE_INDEX_8, SECURITY_PROFILE_ALIAS))
             .setSource(Map.of("user_profile", Map.of("uid", randomAlphaOfLength(22))))
             .get();
-
         assertThat(indexResponse.status().getStatus(), equalTo(201));
 
-        var getIndexRequest = new GetIndexRequest();
-        getIndexRequest.indices(INTERNAL_SECURITY_PROFILE_INDEX_8);
-
-        var getIndexResponse = client().execute(GetIndexAction.INSTANCE, getIndexRequest).actionGet();
-
-        assertThat(getIndexResponse.getIndices(), arrayContaining(INTERNAL_SECURITY_PROFILE_INDEX_8));
-
+        final GetIndexResponse getIndexResponse = getProfileIndexResponse();
+        assertThat(getIndexResponse.getIndices(), hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8));
         var aliases = getIndexResponse.getAliases().get(INTERNAL_SECURITY_PROFILE_INDEX_8);
         assertThat(aliases, hasSize(1));
         assertThat(aliases.get(0).alias(), equalTo(SECURITY_PROFILE_ALIAS));
@@ -222,4 +232,111 @@ public class ProfileSingleNodeTests extends AbstractProfileSingleNodeTestCase {
             () -> client().execute(UpdateProfileDataAction.INSTANCE, updateProfileDataRequest3).actionGet()
         );
     }
+
+    public void testSearchProfiles() {
+        final String nativeRacUserPasswordHash = new String(getFastStoredHashAlgoForTests().hash(NATIVE_RAC_USER_PASSWORD));
+        final Map<String, String> users = Map.of(
+            "user_foo",
+            "Very Curious User Foo",
+            "user_bar",
+            "Super Curious Admin Bar",
+            "user_baz",
+            "Very Anxious User Baz",
+            "user_qux",
+            "Super Anxious Admin Qux"
+        );
+        users.forEach((key, value) -> {
+            final PutUserRequest putUserRequest1 = new PutUserRequest();
+            putUserRequest1.username(key);
+            putUserRequest1.fullName(value);
+            putUserRequest1.roles("rac_role");
+            putUserRequest1.passwordHash(nativeRacUserPasswordHash.toCharArray());
+            assertThat(client().execute(PutUserAction.INSTANCE, putUserRequest1).actionGet().created(), is(true));
+            doActivateProfile(key, NATIVE_RAC_USER_PASSWORD);
+        });
+
+        final SearchProfilesResponse.ProfileHit[] profiles1 = doSearch("");
+        assertThat(extractUsernames(profiles1), equalTo(users.keySet()));
+
+        final SearchProfilesResponse.ProfileHit[] profiles2 = doSearch(randomFrom("super admin", "admin super"));
+        assertThat(extractUsernames(profiles2), equalTo(Set.of("user_bar", "user_qux")));
+
+        // Prefix match on full name
+        final SearchProfilesResponse.ProfileHit[] profiles3 = doSearch("ver");
+        assertThat(extractUsernames(profiles3), equalTo(Set.of("user_foo", "user_baz")));
+
+        // Prefix match on the username
+        final SearchProfilesResponse.ProfileHit[] profiles4 = doSearch("user");
+        assertThat(extractUsernames(profiles4), equalTo(users.keySet()));
+        // Documents scored higher are those with matches in more fields
+        assertThat(extractUsernames(Arrays.copyOfRange(profiles4, 0, 2)), equalTo(Set.of("user_foo", "user_baz")));
+
+        // Match of different terms on different fields
+        final SearchProfilesResponse.ProfileHit[] profiles5 = doSearch(randomFrom("admin very", "very admin"));
+        assertThat(extractUsernames(profiles5), equalTo(users.keySet()));
+    }
+
+    public void testProfileAPIsWhenIndexNotCreated() {
+        // Ensure index does not exist
+        assertThat(getProfileIndexResponse().getIndices(), not(hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8)));
+
+        // Get Profile by ID returns empty result
+        final GetProfilesResponse getProfilesResponse = client().execute(
+            GetProfileAction.INSTANCE,
+            new GetProfileRequest(randomAlphaOfLength(20), Set.of())
+        ).actionGet();
+        assertThat(getProfilesResponse.getProfiles(), arrayWithSize(0));
+
+        // Ensure index does not exist
+        assertThat(getProfileIndexResponse().getIndices(), not(hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8)));
+
+        // Search returns empty result
+        final SearchProfilesResponse.ProfileHit[] profiles1 = doSearch("");
+        assertThat(profiles1, emptyArray());
+
+        // Ensure index does not exist
+        assertThat(getProfileIndexResponse().getIndices(), not(hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8)));
+
+        // Updating profile data results into doc missing exception
+        // But the index is created in the process
+        final DocumentMissingException e1 = expectThrows(
+            DocumentMissingException.class,
+            () -> client().execute(
+                UpdateProfileDataAction.INSTANCE,
+                new UpdateProfileDataRequest(
+                    randomAlphaOfLength(20),
+                    null,
+                    Map.of(randomAlphaOfLengthBetween(3, 8), randomAlphaOfLengthBetween(3, 8)),
+                    -1,
+                    -1,
+                    WriteRequest.RefreshPolicy.WAIT_UNTIL
+                )
+            ).actionGet()
+        );
+
+        // TODO: The index is created after the update call regardless. Should it not do that?
+        assertThat(getProfileIndexResponse().getIndices(), hasItemInArray(INTERNAL_SECURITY_PROFILE_INDEX_8));
+    }
+
+    private SearchProfilesResponse.ProfileHit[] doSearch(String query) {
+        final SearchProfilesRequest searchProfilesRequest = new SearchProfilesRequest(Set.of(), query, 10);
+        final SearchProfilesResponse searchProfilesResponse = client().execute(SearchProfilesAction.INSTANCE, searchProfilesRequest)
+            .actionGet();
+        assertThat(searchProfilesResponse.getTotalHits().relation, is(TotalHits.Relation.EQUAL_TO));
+        return searchProfilesResponse.getProfileHits();
+    }
+
+    private Set<String> extractUsernames(SearchProfilesResponse.ProfileHit[] profileHits) {
+        return Arrays.stream(profileHits)
+            .map(SearchProfilesResponse.ProfileHit::profile)
+            .map(Profile::user)
+            .map(Profile.ProfileUser::username)
+            .collect(Collectors.toUnmodifiableSet());
+    }
+
+    private GetIndexResponse getProfileIndexResponse() {
+        final GetIndexRequest getIndexRequest = new GetIndexRequest();
+        getIndexRequest.indices(".*");
+        return client().execute(GetIndexAction.INSTANCE, getIndexRequest).actionGet();
+    }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java

@@ -110,6 +110,7 @@ import org.elasticsearch.xpack.core.security.action.privilege.GetPrivilegesActio
 import org.elasticsearch.xpack.core.security.action.privilege.PutPrivilegesAction;
 import org.elasticsearch.xpack.core.security.action.profile.ActivateProfileAction;
 import org.elasticsearch.xpack.core.security.action.profile.GetProfileAction;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesAction;
 import org.elasticsearch.xpack.core.security.action.profile.UpdateProfileDataAction;
 import org.elasticsearch.xpack.core.security.action.realm.ClearRealmCacheAction;
 import org.elasticsearch.xpack.core.security.action.role.ClearRolesCacheAction;
@@ -186,6 +187,7 @@ import org.elasticsearch.xpack.security.action.privilege.TransportGetPrivilegesA
 import org.elasticsearch.xpack.security.action.privilege.TransportPutPrivilegesAction;
 import org.elasticsearch.xpack.security.action.profile.TransportActivateProfileAction;
 import org.elasticsearch.xpack.security.action.profile.TransportGetProfileAction;
+import org.elasticsearch.xpack.security.action.profile.TransportSearchProfilesAction;
 import org.elasticsearch.xpack.security.action.profile.TransportUpdateProfileDataAction;
 import org.elasticsearch.xpack.security.action.realm.TransportClearRealmCacheAction;
 import org.elasticsearch.xpack.security.action.role.TransportClearRolesCacheAction;
@@ -280,6 +282,7 @@ import org.elasticsearch.xpack.security.rest.action.privilege.RestGetPrivilegesA
 import org.elasticsearch.xpack.security.rest.action.privilege.RestPutPrivilegesAction;
 import org.elasticsearch.xpack.security.rest.action.profile.RestActivateProfileAction;
 import org.elasticsearch.xpack.security.rest.action.profile.RestGetProfileAction;
+import org.elasticsearch.xpack.security.rest.action.profile.RestSearchProfilesAction;
 import org.elasticsearch.xpack.security.rest.action.profile.RestUpdateProfileDataAction;
 import org.elasticsearch.xpack.security.rest.action.realm.RestClearRealmCacheAction;
 import org.elasticsearch.xpack.security.rest.action.role.RestClearRolesCacheAction;
@@ -1216,7 +1219,8 @@ public class Security extends Plugin
                 Stream.of(
                     new ActionHandler<>(GetProfileAction.INSTANCE, TransportGetProfileAction.class),
                     new ActionHandler<>(ActivateProfileAction.INSTANCE, TransportActivateProfileAction.class),
-                    new ActionHandler<>(UpdateProfileDataAction.INSTANCE, TransportUpdateProfileDataAction.class)
+                    new ActionHandler<>(UpdateProfileDataAction.INSTANCE, TransportUpdateProfileDataAction.class),
+                    new ActionHandler<>(SearchProfilesAction.INSTANCE, TransportSearchProfilesAction.class)
                 )
             ).toList();
         } else {
@@ -1301,7 +1305,8 @@ public class Security extends Plugin
                 Stream.of(
                     new RestGetProfileAction(settings, getLicenseState()),
                     new RestActivateProfileAction(settings, getLicenseState()),
-                    new RestUpdateProfileDataAction(settings, getLicenseState())
+                    new RestUpdateProfileDataAction(settings, getLicenseState()),
+                    new RestSearchProfilesAction(settings, getLicenseState())
                 )
             ).toList();
         } else {

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/profile/TransportGetProfileAction.java

@@ -30,6 +30,6 @@ public class TransportGetProfileAction extends HandledTransportAction<GetProfile
 
     @Override
     protected void doExecute(Task task, GetProfileRequest request, ActionListener<GetProfilesResponse> listener) {
-        profileService.getProfile(request.getUid(), request.getDatKeys(), listener.map(GetProfilesResponse::new));
+        profileService.getProfile(request.getUid(), request.getDataKeys(), listener.map(GetProfilesResponse::new));
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/profile/TransportSearchProfilesAction.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.action.profile;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesAction;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesRequest;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesResponse;
+import org.elasticsearch.xpack.security.profile.ProfileService;
+
+public class TransportSearchProfilesAction extends HandledTransportAction<SearchProfilesRequest, SearchProfilesResponse> {
+
+    private final ProfileService profileService;
+
+    @Inject
+    public TransportSearchProfilesAction(TransportService transportService, ActionFilters actionFilters, ProfileService profileService) {
+        super(SearchProfilesAction.NAME, transportService, actionFilters, SearchProfilesRequest::new);
+        this.profileService = profileService;
+    }
+
+    @Override
+    protected void doExecute(Task task, SearchProfilesRequest request, ActionListener<SearchProfilesResponse> listener) {
+        profileService.searchProfile(request, listener);
+    }
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/profile/ProfileService.java

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.security.profile;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteResponse;
@@ -34,9 +35,11 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.MultiMatchQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -45,6 +48,8 @@ import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.security.action.profile.Profile;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesRequest;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesResponse;
 import org.elasticsearch.xpack.core.security.action.profile.UpdateProfileDataRequest;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationContext;
@@ -161,6 +166,72 @@ public class ProfileService {
         );
     }
 
+    public void searchProfile(SearchProfilesRequest request, ActionListener<SearchProfilesResponse> listener) {
+        tryFreezeAndCheckIndex(listener.map(response -> {
+            assert response == null : "only null response can reach here";
+            return new SearchProfilesResponse(new SearchProfilesResponse.ProfileHit[] {}, 0, new TotalHits(0, TotalHits.Relation.EQUAL_TO));
+        })).ifPresent(frozenProfileIndex -> {
+            final BoolQueryBuilder query = QueryBuilders.boolQuery().filter(QueryBuilders.termQuery("user_profile.enabled", true));
+            if (Strings.hasText(request.getName())) {
+                query.must(
+                    QueryBuilders.multiMatchQuery(
+                        request.getName(),
+                        "user_profile.user.username",
+                        "user_profile.user.username._2gram",
+                        "user_profile.user.username._3gram",
+                        "user_profile.user.full_name",
+                        "user_profile.user.full_name._2gram",
+                        "user_profile.user.full_name._3gram",
+                        "user_profile.user.display_name",
+                        "user_profile.user.display_name._2gram",
+                        "user_profile.user.display_name._3gram"
+                    ).type(MultiMatchQueryBuilder.Type.BOOL_PREFIX)
+                );
+            }
+            final SearchRequest searchRequest = client.prepareSearch(SECURITY_PROFILE_ALIAS)
+                .setQuery(query)
+                .setSize(request.getSize())
+                .addSort("_score", SortOrder.DESC)
+                .addSort("user_profile.last_synchronized", SortOrder.DESC)
+                .request();
+
+            frozenProfileIndex.checkIndexVersionThenExecute(
+                listener::onFailure,
+                () -> executeAsyncWithOrigin(
+                    client,
+                    SECURITY_ORIGIN,
+                    SearchAction.INSTANCE,
+                    searchRequest,
+                    ActionListener.wrap(searchResponse -> {
+                        final SearchHits searchHits = searchResponse.getHits();
+                        final SearchHit[] hits = searchHits.getHits();
+                        final SearchProfilesResponse.ProfileHit[] profileHits;
+                        if (hits.length == 0) {
+                            profileHits = new SearchProfilesResponse.ProfileHit[0];
+                        } else {
+                            profileHits = new SearchProfilesResponse.ProfileHit[hits.length];
+                            for (int i = 0; i < hits.length; i++) {
+                                final SearchHit hit = hits[i];
+                                final VersionedDocument versionedDocument = new VersionedDocument(
+                                    buildProfileDocument(hit.getSourceRef()),
+                                    hit.getPrimaryTerm(),
+                                    hit.getSeqNo()
+                                );
+                                profileHits[i] = new SearchProfilesResponse.ProfileHit(
+                                    versionedDocument.toProfile(request.getDataKeys()),
+                                    hit.getScore()
+                                );
+                            }
+                        }
+                        listener.onResponse(
+                            new SearchProfilesResponse(profileHits, searchResponse.getTook().millis(), searchHits.getTotalHits())
+                        );
+                    }, listener::onFailure)
+                )
+            );
+        });
+    }
+
     private void getVersionedDocument(String uid, ActionListener<VersionedDocument> listener) {
         tryFreezeAndCheckIndex(listener).ifPresent(frozenProfileIndex -> {
             final GetRequest getRequest = new GetRequest(SECURITY_PROFILE_ALIAS, uidToDocId(uid));
@@ -385,7 +456,7 @@ public class ProfileService {
 
     /**
      * Freeze the profile index check its availability and return it if everything is ok.
-     * Otherwise it returns null.
+     * Otherwise it calls the listener with null and returns an empty Optional.
      */
     private <T> Optional<SecurityIndexManager> tryFreezeAndCheckIndex(ActionListener<T> listener) {
         final SecurityIndexManager frozenProfileIndex = profileIndex.freeze();

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/action/profile/RestSearchProfilesAction.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.security.rest.action.profile;
+
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesAction;
+import org.elasticsearch.xpack.core.security.action.profile.SearchProfilesRequest;
+import org.elasticsearch.xpack.security.rest.action.SecurityBaseRestHandler;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+
+import static org.elasticsearch.rest.RestRequest.Method.GET;
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
+
+public class RestSearchProfilesAction extends SecurityBaseRestHandler {
+
+    static final ConstructingObjectParser<Payload, Void> PARSER = new ConstructingObjectParser<>(
+        "search_profile_request_payload",
+        a -> new Payload((String) a[0], (Integer) a[1])
+    );
+
+    static {
+        PARSER.declareString(optionalConstructorArg(), new ParseField("name"));
+        PARSER.declareInt(optionalConstructorArg(), new ParseField("size"));
+    }
+
+    public RestSearchProfilesAction(Settings settings, XPackLicenseState licenseState) {
+        super(settings, licenseState);
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(GET, "/_security/profile/_search"), new Route(POST, "/_security/profile/_search"));
+    }
+
+    @Override
+    public String getName() {
+        return "xpack_security_search_profile";
+    }
+
+    @Override
+    protected RestChannelConsumer innerPrepareRequest(RestRequest request, NodeClient client) throws IOException {
+        final Set<String> dataKeys = Strings.tokenizeByCommaToSet(request.param("data", null));
+        final Payload payload = request.hasContent() ? PARSER.parse(request.contentParser(), null) : new Payload(null, null);
+
+        final SearchProfilesRequest searchProfilesRequest = new SearchProfilesRequest(dataKeys, payload.name(), payload.size());
+        return channel -> client.execute(SearchProfilesAction.INSTANCE, searchProfilesRequest, new RestToXContentListener<>(channel));
+    }
+
+    record Payload(String name, Integer size) {
+
+        public String name() {
+            return name != null ? name : "";
+        }
+
+        public Integer size() {
+            return size != null ? size : 10;
+        }
+    }
+}