User attribute value length extension

Closes #9758

Signed-off-by: vramik <vramik@redhat.com>
Signed-off-by: Alexander Schwartz <aschwart@redhat.com>
Signed-off-by: Michal Hajas <mhajas@redhat.com>
Co-authored-by: Alexander Schwartz <aschwart@redhat.com>
Co-authored-by: Michal Hajas <mhajas@redhat.com>

docs/documentation/release_notes/topics/24_0_0.adoc

@@ -217,6 +217,13 @@ When enabling metrics for {project_name}'s embedded caches, the metrics now use
 For more details, check the
 link:{upgradingguide_link}[{upgradingguide_name}].
 
+= User attribute value length extension
+
+As of this release, {project_name} supports storing and searching by user attribute values longer than 255 characters, which was previously a limitation.
+
+For more details, check the
+link:{upgradingguide_link}[{upgradingguide_name}].
+
 = Authorization Policy
 
 In previous versions of Keycloak when the last member of a User, Group or Client policy was deleted then that policy would also be deleted. Unfortunately this could lead to an escalation of privileges if the policy was used in an aggregate policy. To avoid privilege escalation the effect policies are no longer deleted and an administrator will need to update those policies.

docs/documentation/server_admin/topics.adoc

@@ -67,6 +67,7 @@ include::topics/threat/host.adoc[]
 include::topics/threat/admin.adoc[]
 include::topics/threat/brute-force.adoc[]
 include::topics/threat/read-only-attributes.adoc[]
+include::topics/threat/validate-user-attributes.adoc[]
 include::topics/threat/clickjacking.adoc[]
 include::topics/threat/ssl.adoc[]
 include::topics/threat/csrf.adoc[]

docs/documentation/server_admin/topics/threat/validate-user-attributes.adoc

@@ -0,0 +1,10 @@
+[[validate_user_attributes]]
+=== Validate user attributes
+
+With the functionality in <<user-profile>>, administrators can restrict the data users enter for attributes, for example, in user registration or the account console.
+
+Administrators should not allow unmanaged attributes for users to prevent attackers adding an unlimited number of attributes.
+Attributes should have a validation that restricts the amount of data entered by attackers.
+
+When using regular expressions to validate user attributes, avoid regular expressions that use an excessive amount of memory or CPU.
+See https://owasp.org/www-community/attacks/Regular_expression_Denial_of_Service_-_ReDoS[OWASP's Regular expression Denial of Service] for details.

docs/documentation/server_admin/topics/users/user-profile.adoc

@@ -102,7 +102,7 @@ You can choose to completely disable or only support unmanaged attributes when m
 When unmanaged attributes are enabled (even if partially) you can manage them from the administration console at the `Attributes` tab in the User Details UI.
 If the policy is set to `Disabled` this tab is not available.
 
-As a recommendation, try to adhere to the most strict policy as much as possible (e.g.: `Disabled` or `Admin can edit`) to prevent unexpected
+As a security recommendation, try to adhere to the most strict policy as much as possible (e.g.: `Disabled` or `Admin can edit`) to prevent unexpected
 attributes (and values) set to your users when they are managing their profile through end-user contexts.
 Avoid setting the `Enabled` policy and prefer defining all the attributes that end-users can manage in your user profile configuration, under your control.
 
@@ -114,6 +114,9 @@ behavior when using custom themes and extending the server with their own custom
 
 As you will see in the following sections, you can also restrict the audience for an attribute by choosing if it should be visible or writable by users and/or administrators.
 
+For unmanaged attributes, the maximum length is 2048 characters.
+To specify a different minimum or maximum length, change the unmanaged attribute to a managed attribute and add a `length` validator.
+
 == Managing the User Profile
 
 The user profile configuration is managed on a per-realm basis. For that, click on the
@@ -198,6 +201,9 @@ image:images/user-profile-validation.png[]
 Validation happens at any time when writing to an attribute, and they can throw errors that will be shown in UIs when the value
 fails a validation.
 
+For security reasons, every attribute that is editable by users should have a validation to restrict the size of the values users enter.
+If no `length` validator has been specified, {project_name} defaults to a maximum length of 2048 characters.
+
 === Built-in Validators
 
 {project_name} provides some built-in validators that you can choose from, and you are also able to provide

docs/documentation/upgrading/topics/keycloak/changes-24_0_0.adoc

@@ -239,6 +239,47 @@ To revert the change for an installation, use a custom Infinispan XML configurat
 <metrics names-as-tags="false" />
 ----
 
+= User attribute value length extension
+
+As of this release, {project_name} supports storing and searching by user attribute values longer than 255 characters, which was previously a limitation.
+
+In setups where users are allowed to update attributes, for example, via the account console, prevent denial of service attacks by adding validations.
+Ensure that no unmanaged attributes are allowed and all editable attributes have a validation that limits the input length.
+
+For unmanaged attributes, the maximum length is 2048 characters.
+For managed attributes, the default maximum length is 2048 characters. Administrator can change this by adding a validator of type `length`.
+
+This change adds new indexes on the tables `USER_ATTRIBUTE` and `FED_USER_ATTRIBUTE`.
+If those tables contain more than 300000 entries, Keycloak will skip the index creation by default during the automatic schema migration and instead log the SQL statement on the console during migration to be applied manually after {project_name}'s startup.
+See the link:{upgradingguide_link}[{upgradingguide_name}] for details on how to configure a different limit.
+
+== Additional migration steps for LDAP
+
+This is for installations that match all the following criteria:
+
+* User attributes in the LDAP directory are larger than 2048 characters or binary attributes that are larger than 1500 bytes.
+* The attributes are changed by admins or users via the admin console, the APIs or the account console.
+
+To be able to enable changing those attributes via UI and REST APIs, perform the following steps:
+
+. Declare the attributes identified above as managed attributes in the user profile of the realm.
+. Define a `length` validator for each attribute added in the previous step specifying the desired minimum and maximum length of the attribute value.
+For binary values, add 33% to the expected binary length to count in the overhead for {project_name}'s internal base64 encoding of binary values.
+
+== Additional migration steps for custom user storage providers
+
+This is for installations that match all the following criteria:
+
+* Running MariaDB or MySQL as a database for {project_name}.
+* Entries in table `FED_USER_ATTRIBUTE` with contents in the `VALUE` column that are larger than 2048 characters.
+This table is used for custom user providers which have federation enabled.
+* The long attributes are changed by admins or users via the admin console or the account console.
+
+To be able to enable changing those attributes via UI and REST APIs,  perform the following steps:
+
+. Declare the attributes identified above as managed attributes in the user profile of the realm.
+. Define a `length` validator for each attribute added in the previous step specifying the desired minimum and maximum length of the attribute value.
+
 = The Admin send-verify-email API now uses the same email verification template
 
 ----

model/jpa/pom.xml

@@ -30,6 +30,10 @@
     <description/>
 
     <properties>
+        <maven.compiler.release>17</maven.compiler.release>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+
         <keycloak.connectionsJpa.driver>org.h2.Driver</keycloak.connectionsJpa.driver>
         <keycloak.connectionsJpa.database>keycloak</keycloak.connectionsJpa.database>
         <keycloak.connectionsJpa.user>sa</keycloak.connectionsJpa.user>

model/jpa/src/main/java/org/keycloak/connections/jpa/updater/liquibase/custom/FederatedUserAttributeTextColumnMigration.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2018 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.keycloak.connections.jpa.updater.liquibase.custom;
+
+import liquibase.database.core.MySQLDatabase;
+import liquibase.exception.CustomChangeException;
+import liquibase.statement.core.RawParameterizedSqlStatement;
+import org.keycloak.storage.jpa.JpaHashUtils;
+
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+
+/**
+ * The MySQL database is the only database where columns longer than 255 characters are changed to a TEXT column, allowing
+ * for up to 64k characters. See {@link org.keycloak.connections.jpa.updater.liquibase.MySQL8VarcharType} for the implementation.
+ * As the new code expects all information longer than 2024 characters in the new column, this migration copies over the values.
+ *
+ * @author Alexander Schwartz
+ */
+public class FederatedUserAttributeTextColumnMigration extends CustomKeycloakTask {
+
+    @Override
+    protected void generateStatementsImpl() throws CustomChangeException {
+
+        if (database instanceof MySQLDatabase) {
+
+            try (PreparedStatement ps = connection.prepareStatement("SELECT t.ID, t.VALUE" +
+                    "  FROM " + getTableName("FED_USER_ATTRIBUTE") + " t" +
+                    "  WHERE LENGTH(t.VALUE) > 2024");
+                 ResultSet resultSet = ps.executeQuery()
+            ) {
+                while (resultSet.next()) {
+                    String id = resultSet.getString(1);
+                    String value = resultSet.getString(2);
+                    // The SQL LENGTH() will count bytes, where Java's length() will count Unicode characters.
+                    // There's also SQL CHAR_LENGTH() which is probably equivalent to Java's Character.codePointCount(),
+                    // but it is not a fit here as we're not using code points in the JPA entities
+                    if (value.length() > 2024) {
+                        statements.add(new RawParameterizedSqlStatement("UPDATE " + getTableName("FED_USER_ATTRIBUTE") + " SET VALUE = null, LONG_VALUE_HASH = ?, LONG_VALUE_HASH_LOWER_CASE = ?, LONG_VALUE = ? WHERE ID = ?",
+                                JpaHashUtils.hashForAttributeValue(value),
+                                JpaHashUtils.hashForAttributeValueLowerCase(value),
+                                value,
+                                id));
+                    }
+                }
+            } catch (Exception e) {
+                throw new CustomChangeException(getTaskId() + ": Exception when updating data from previous version", e);
+            }
+        }
+
+    }
+
+    @Override
+    protected String getTaskId() {
+        return "Leave only single offline session per user and client";
+    }
+
+}

model/jpa/src/main/java/org/keycloak/models/jpa/JpaUserProvider.java

@@ -61,18 +61,22 @@ import jakarta.persistence.criteria.JoinType;
 import jakarta.persistence.criteria.Predicate;
 import jakarta.persistence.criteria.Root;
 import jakarta.persistence.criteria.Subquery;
+import org.keycloak.storage.jpa.JpaHashUtils;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Stream;
 
 import static org.keycloak.models.jpa.PaginationUtils.paginateQuery;
+import static org.keycloak.storage.jpa.JpaHashUtils.predicateForFilteringUsersByAttributes;
 import static org.keycloak.utils.StreamsUtil.closing;
 
 
@@ -668,7 +672,7 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         Expression<Long> count = qb.count(from);
 
         userQuery = userQuery.select(count);
-        List<Predicate> restrictions = predicates(params, from);
+        List<Predicate> restrictions = predicates(params, from, Map.of());
         restrictions.add(qb.equal(from.get("realmId"), realm.getId()));
 
         userQuery = userQuery.where(restrictions.toArray(new Predicate[0]));
@@ -691,7 +695,7 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         Root<UserEntity> root = countQuery.from(UserEntity.class);
         countQuery.select(cb.count(root));
 
-        List<Predicate> restrictions = predicates(params, root);
+        List<Predicate> restrictions = predicates(params, root, Map.of());
         restrictions.add(cb.equal(root.get("realmId"), realm.getId()));
         
         groupsWithPermissionsSubquery(countQuery, groupIds, root, restrictions);
@@ -735,7 +739,8 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         CriteriaQuery<UserEntity> queryBuilder = builder.createQuery(UserEntity.class);
         Root<UserEntity> root = queryBuilder.from(UserEntity.class);
 
-        List<Predicate> predicates = predicates(attributes, root);
+        Map<String, String> customLongValueSearchAttributes = new HashMap<>();
+        List<Predicate> predicates = predicates(attributes, root, customLongValueSearchAttributes);
 
         predicates.add(builder.equal(root.get("realmId"), realm.getId()));
 
@@ -745,24 +750,35 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
             groupsWithPermissionsSubquery(queryBuilder, userGroups, root, predicates);
         }
 
-        queryBuilder.where(predicates.toArray(new Predicate[0])).orderBy(builder.asc(root.get(UserModel.USERNAME)));
+        queryBuilder.where(predicates.toArray(Predicate[]::new)).orderBy(builder.asc(root.get(UserModel.USERNAME)));
 
         TypedQuery<UserEntity> query = em.createQuery(queryBuilder);
 
         UserProvider users = session.users();
         return closing(paginateQuery(query, firstResult, maxResults).getResultStream())
+                // following check verifies that there are no collisions with hashes
+                .filter(predicateForFilteringUsersByAttributes(customLongValueSearchAttributes, JpaHashUtils::compareSourceValueLowerCase))
                 .map(userEntity -> users.getUserById(realm, userEntity.getId()))
                 .filter(Objects::nonNull);
     }
 
     @Override
     public Stream<UserModel> searchForUserByUserAttributeStream(RealmModel realm, String attrName, String attrValue) {
-        TypedQuery<UserEntity> query = em.createNamedQuery("getRealmUsersByAttributeNameAndValue", UserEntity.class);
+        boolean longAttribute = attrValue != null && attrValue.length() > 255;
-        query.setParameter("name", attrName);
+        TypedQuery<UserEntity> query = longAttribute ?
-        query.setParameter("value", attrValue);
+                em.createNamedQuery("getRealmUsersByAttributeNameAndLongValue", UserEntity.class)
-        query.setParameter("realmId", realm.getId());
+                        .setParameter("realmId", realm.getId())
+                        .setParameter("name", attrName)
+                        .setParameter("longValueHash", JpaHashUtils.hashForAttributeValue(attrValue)):
+                em.createNamedQuery("getRealmUsersByAttributeNameAndValue", UserEntity.class)
+                        .setParameter("realmId", realm.getId())
+                        .setParameter("name", attrName)
+                        .setParameter("value", attrValue);
 
-        return closing(query.getResultStream().map(userEntity -> new UserAdapter(session, realm, em, userEntity)));
+        return closing(query.getResultStream()
+                // following check verifies that there are no collisions with hashes
+                .filter(longAttribute ? predicateForFilteringUsersByAttributes(Map.of(attrName, attrValue), JpaHashUtils::compareSourceValue) : u -> true)
+                .map(userEntity -> new UserAdapter(session, realm, em, userEntity)));
     }
 
     private FederatedIdentityEntity findFederatedIdentity(UserModel user, String identityProvider, LockModeType lockMode) {
@@ -929,7 +945,7 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         return em.contains(user) ? user : null;
     }
 
-    private List<Predicate> predicates(Map<String, String> attributes, Root<UserEntity> root) {
+    private List<Predicate> predicates(Map<String, String> attributes, Root<UserEntity> root, Map<String, String> customLongValueSearchAttributes) {
         CriteriaBuilder builder = em.getCriteriaBuilder();
 
         List<Predicate> predicates = new ArrayList<>();
@@ -991,10 +1007,16 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
                 default:
                     Join<UserEntity, UserAttributeEntity> attributesJoin = root.join("attributes", JoinType.LEFT);
 
+                    if (value.length() > 255) {
+                        customLongValueSearchAttributes.put(key, value);
+                        attributePredicates.add(builder.and(
+                                builder.equal(builder.lower(attributesJoin.get("name")), key.toLowerCase()),
+                                builder.equal(attributesJoin.get("longValueHashLowerCase"), JpaHashUtils.hashForAttributeValueLowerCase(value))));
+                    } else {
                         attributePredicates.add(builder.and(
                                 builder.equal(builder.lower(attributesJoin.get("name")), key.toLowerCase()),
                                 builder.equal(builder.lower(attributesJoin.get("value")), value.toLowerCase())));
-
+                    }
                     break;
                 case UserModel.INCLUDE_SERVICE_ACCOUNT: {
                     if (!attributes.containsKey(UserModel.INCLUDE_SERVICE_ACCOUNT)

model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserAttributeEntity.java

@@ -30,6 +30,7 @@ import jakarta.persistence.ManyToOne;
 import jakarta.persistence.NamedQueries;
 import jakarta.persistence.NamedQuery;
 import jakarta.persistence.Table;
+import org.keycloak.storage.jpa.JpaHashUtils;
 
 /**
  * @author <a href="mailto:bill@burkecentral.com">Bill Burke</a>
@@ -60,6 +61,14 @@ public class UserAttributeEntity {
     @Column(name = "VALUE")
     protected String value;
 
+    @Column(name = "LONG_VALUE_HASH")
+    private byte[] longValueHash;
+    @Column(name = "LONG_VALUE_HASH_LOWER_CASE")
+    private byte[] longValueHashLowerCase;
+    @Nationalized
+    @Column(name = "LONG_VALUE")
+    private String longValue;
+
     public String getId() {
         return id;
     }
@@ -77,11 +86,29 @@ public class UserAttributeEntity {
     }
 
     public String getValue() {
-        return value;
+        if (value != null && longValue != null) {
+            throw new IllegalStateException(String.format("User with id %s should not have set both `value` and `longValue` for attribute %s.", user.getId(), name));
+        }
+        return value != null ? value : longValue;
     }
 
     public void setValue(String value) {
+        if (value == null) {
+            this.value = null;
+            this.longValue = null;
+            this.longValueHash = null;
+            this.longValueHashLowerCase = null;
+        } else if (value.length() > 255) {
+            this.value = null;
+            this.longValue = value;
+            this.longValueHash = JpaHashUtils.hashForAttributeValue(value);
+            this.longValueHashLowerCase = JpaHashUtils.hashForAttributeValueLowerCase(value);
+        } else {
             this.value = value;
+            this.longValue = null;
+            this.longValueHash = null;
+            this.longValueHashLowerCase = null;
+        }
     }
 
     public UserEntity getUser() {

model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserEntity.java

@@ -51,6 +51,8 @@ import java.util.LinkedList;
         @NamedQuery(name="getRealmUserCountExcludeServiceAccount", query="select count(u) from UserEntity u where u.realmId = :realmId and (u.serviceAccountClientLink is null)"),
         @NamedQuery(name="getRealmUsersByAttributeNameAndValue", query="select u from UserEntity u join u.attributes attr " +
                 "where u.realmId = :realmId and attr.name = :name and attr.value = :value"),
+        @NamedQuery(name="getRealmUsersByAttributeNameAndLongValue", query="select u from UserEntity u join u.attributes attr " +
+                "where u.realmId = :realmId and attr.name = :name and attr.longValueHash = :longValueHash"),
         @NamedQuery(name="deleteUsersByRealm", query="delete from UserEntity u where u.realmId = :realmId"),
         @NamedQuery(name="deleteUsersByRealmAndLink", query="delete from UserEntity u where u.realmId = :realmId and u.federationLink=:link"),
         @NamedQuery(name="unlinkUsers", query="update UserEntity u set u.federationLink = null where u.realmId = :realmId and u.federationLink=:link")

model/jpa/src/main/java/org/keycloak/storage/jpa/JpaHashUtils.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.storage.jpa;
+
+import org.keycloak.crypto.HashException;
+import org.keycloak.crypto.JavaAlgorithm;
+import org.keycloak.models.jpa.entities.UserEntity;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiPredicate;
+
+/**
+ * Create hashes for long values stored in the database. Offers different variants for exact and lowercase search.
+ * Keycloak uses lowercase search to approximate a case-insensitive search.
+ * <p>
+ * The lowercase function always uses the English locale to avoid changing hashes due to changing locales which can be surprising
+ * and would be expensive to fix as all hashes would need to be re-calculated.
+ *
+ * @author Alexander Schwartz
+ */
+public class JpaHashUtils {
+
+    private static byte[] hash(byte[] inputBytes) {
+        try {
+            MessageDigest md = MessageDigest.getInstance(JavaAlgorithm.SHA512);
+            md.update(inputBytes);
+            return md.digest();
+        } catch (Exception e) {
+            throw new HashException("Error when creating token hash", e);
+        }
+    }
+
+    public static byte[] hashForAttributeValue(String value) {
+        return JpaHashUtils.hash(value.getBytes(StandardCharsets.UTF_8));
+    }
+
+    public static byte[] hashForAttributeValueLowerCase(String value) {
+        return JpaHashUtils.hash(value.toLowerCase(Locale.ENGLISH).getBytes(StandardCharsets.UTF_8));
+    }
+
+    public static boolean compareSourceValueLowerCase(String value1, String value2) {
+        return Objects.equals(value1.toLowerCase(Locale.ENGLISH), value2.toLowerCase(Locale.ENGLISH));
+    }
+
+    public static boolean compareSourceValue(String value1, String value2) {
+        return Objects.equals(value1, value2);
+    }
+
+    /**
+     * This method returns a predicate that returns true when user has all attributes specified in {@code customLongValueSearchAttributes} map
+     * <p />
+     * The check is performed by exact comparison on attribute name the value
+     * <p />
+     * This is necessary because database can return users without the searched attribute when a hash collision on long user attribute value occurs
+     *
+     * @param customLongValueSearchAttributes required attributes
+     * @param valueComparator                 comparator for comparing attribute values
+     * @return predicate for filtering users based on attributes map
+     */
+    public static java.util.function.Predicate<UserEntity> predicateForFilteringUsersByAttributes(Map<String, String> customLongValueSearchAttributes, BiPredicate<String, String> valueComparator) {
+        return userEntity -> customLongValueSearchAttributes.isEmpty() || // are there some long attribute values
+                customLongValueSearchAttributes
+                        .entrySet()
+                        .stream()
+                        .allMatch(longAttrEntry -> //for all long search attributes
+                                userEntity
+                                        .getAttributes()
+                                        .stream()
+                                        .anyMatch(userAttribute -> //check whether the user indeed has the attribute
+                                                Objects.equals(longAttrEntry.getKey().toLowerCase(), userAttribute.getName().toLowerCase())
+                                                        && valueComparator.test(longAttrEntry.getValue(), userAttribute.getValue())
+                                        )
+                        );
+    }
+
+}

model/jpa/src/main/java/org/keycloak/storage/jpa/JpaUserFederatedStorageProvider.java

@@ -165,12 +165,21 @@ public class JpaUserFederatedStorageProvider implements
 
     @Override
     public Stream<String> getUsersByUserAttributeStream(RealmModel realm, String name, String value) {
+        boolean longAttribute = value != null && value.length() > 2024;
+        if (longAttribute) {
+            TypedQuery<Object[]> query = em.createNamedQuery("getFederatedAttributesByNameAndLongValue", Object[].class)
+                    .setParameter("realmId", realm.getId())
+                    .setParameter("name", name)
+                    .setParameter("longValueHash", JpaHashUtils.hashForAttributeValue(value));
+            return closing(query.getResultStream().filter(objects -> JpaHashUtils.compareSourceValue((String) objects[1], value)).map(objects -> (String) objects[0]));
+        } else {
             TypedQuery<String> query = em.createNamedQuery("getFederatedAttributesByNameAndValue", String.class)
                     .setParameter("realmId", realm.getId())
                     .setParameter("name", name)
                     .setParameter("value", value);
             return closing(query.getResultStream());
         }
+    }
 
     @Override
     public String getUserByFederatedIdentity(FederatedIdentityModel link, RealmModel realm) {

model/jpa/src/main/java/org/keycloak/storage/jpa/entity/FederatedUserAttributeEntity.java

@@ -25,6 +25,8 @@ import jakarta.persistence.Id;
 import jakarta.persistence.NamedQueries;
 import jakarta.persistence.NamedQuery;
 import jakarta.persistence.Table;
+import org.hibernate.annotations.Nationalized;
+import org.keycloak.storage.jpa.JpaHashUtils;
 
 /**
  * @author <a href="mailto:bill@burkecentral.com">Bill Burke</a>
@@ -32,6 +34,7 @@ import jakarta.persistence.Table;
  */
 @NamedQueries({
         @NamedQuery(name="getFederatedAttributesByNameAndValue", query="select attr.userId from FederatedUserAttributeEntity attr where attr.name = :name and attr.value = :value and attr.realmId=:realmId"),
+        @NamedQuery(name="getFederatedAttributesByNameAndLongValue", query="select attr.userId, attr.longValue from FederatedUserAttributeEntity attr where attr.name = :name and attr.longValueHash = :longValueHash and attr.realmId=:realmId"),
         @NamedQuery(name="getFederatedAttributesByUser", query="select attr from FederatedUserAttributeEntity attr where attr.userId = :userId and attr.realmId=:realmId"),
         @NamedQuery(name="deleteUserFederatedAttributesByUser", query="delete from  FederatedUserAttributeEntity attr where attr.userId = :userId and attr.realmId=:realmId"),
         @NamedQuery(name="deleteUserFederatedAttributesByUserAndName", query="delete from  FederatedUserAttributeEntity attr where attr.userId = :userId and attr.name=:name and attr.realmId=:realmId"),
@@ -62,6 +65,14 @@ public class FederatedUserAttributeEntity {
     @Column(name = "VALUE")
     protected String value;
 
+    @Column(name = "LONG_VALUE_HASH")
+    private byte[] longValueHash;
+    @Column(name = "LONG_VALUE_HASH_LOWER_CASE")
+    private byte[] longValueHashLowerCase;
+    @Nationalized
+    @Column(name = "LONG_VALUE")
+    private String longValue;
+
     public String getId() {
         return id;
     }
@@ -79,11 +90,29 @@ public class FederatedUserAttributeEntity {
     }
 
     public String getValue() {
-        return value;
+        if (value != null && longValue != null) {
+            throw new IllegalStateException(String.format("Federated user with id %s should not have set both `value` and `longValue` for attribute %s.", userId, name));
+        }
+        return value != null ? value : longValue;
     }
 
     public void setValue(String value) {
+        if (value == null) {
+            this.value = null;
+            this.longValue = null;
+            this.longValueHash = null;
+            this.longValueHashLowerCase = null;
+        } else if (value.length() > 2024) { // https://github.com/keycloak/keycloak/blob/2785bbd29bcc1b39d9abe90724333dd42af34b10/model/jpa/src/main/resources/META-INF/jpa-changelog-2.1.0.xml#L58
+            this.value = null;
+            this.longValue = value;
+            this.longValueHash = JpaHashUtils.hashForAttributeValue(value);
+            this.longValueHashLowerCase = JpaHashUtils.hashForAttributeValueLowerCase(value);
+        } else {
             this.value = value;
+            this.longValue = null;
+            this.longValueHash = null;
+            this.longValueHashLowerCase = null;
+        }
     }
 
     public String getUserId() {

model/jpa/src/main/resources/META-INF/jpa-changelog-24.0.0.xml

@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!--
+  ~ * Copyright 2024 Red Hat, Inc. and/or its affiliates
+  ~ * and other contributors as indicated by the @author tags.
+  ~ *
+  ~ * Licensed under the Apache License, Version 2.0 (the "License");
+  ~ * you may not use this file except in compliance with the License.
+  ~ * You may obtain a copy of the License at
+  ~ *
+  ~ * http://www.apache.org/licenses/LICENSE-2.0
+  ~ *
+  ~ * Unless required by applicable law or agreed to in writing, software
+  ~ * distributed under the License is distributed on an "AS IS" BASIS,
+  ~ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ * See the License for the specific language governing permissions and
+  ~ * limitations under the License.
+  -->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.1.xsd">
+
+    <changeSet author="keycloak" id="24.0.0-9758">
+        <addColumn tableName="USER_ATTRIBUTE">
+            <column name="LONG_VALUE_HASH" type="BINARY(64)" />
+            <column name="LONG_VALUE_HASH_LOWER_CASE" type="BINARY(64)" />
+            <column name="LONG_VALUE" type="NCLOB" />
+        </addColumn>
+        <addColumn tableName="FED_USER_ATTRIBUTE">
+            <column name="LONG_VALUE_HASH" type="BINARY(64)" />
+            <column name="LONG_VALUE_HASH_LOWER_CASE" type="BINARY(64)" />
+            <column name="LONG_VALUE" type="NCLOB" />
+        </addColumn>
+        <createIndex tableName="USER_ATTRIBUTE" indexName="USER_ATTR_LONG_VALUES">
+            <column name="LONG_VALUE_HASH" />
+            <column name="NAME" />
+        </createIndex>
+        <createIndex tableName="FED_USER_ATTRIBUTE" indexName="FED_USER_ATTR_LONG_VALUES">
+            <column name="LONG_VALUE_HASH" />
+            <column name="NAME" />
+        </createIndex>
+        <createIndex tableName="USER_ATTRIBUTE" indexName="USER_ATTR_LONG_VALUES_LOWER_CASE">
+            <column name="LONG_VALUE_HASH_LOWER_CASE" />
+            <column name="NAME" />
+        </createIndex>
+        <createIndex tableName="FED_USER_ATTRIBUTE" indexName="FED_USER_ATTR_LONG_VALUES_LOWER_CASE">
+            <column name="LONG_VALUE_HASH_LOWER_CASE" />
+            <column name="NAME" />
+        </createIndex>
+    </changeSet>
+
+    <changeSet author="keycloak" id="24.0.0-9758-2">
+        <customChange class="org.keycloak.connections.jpa.updater.liquibase.custom.FederatedUserAttributeTextColumnMigration" />
+    </changeSet>
+
+</databaseChangeLog>

model/jpa/src/main/resources/META-INF/jpa-changelog-master.xml

@@ -79,5 +79,6 @@
     <include file="META-INF/jpa-changelog-21.1.0.xml"/>
     <include file="META-INF/jpa-changelog-22.0.0.xml"/>
     <include file="META-INF/jpa-changelog-23.0.0.xml"/>
+    <include file="META-INF/jpa-changelog-24.0.0.xml"/>
 
 </databaseChangeLog>

model/jpa/src/test/java/org/keycloak/storage/jpa/JpaHashUtilsTest.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright 2024 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.keycloak.storage.jpa;
+
+import org.junit.Test;
+import org.keycloak.models.jpa.entities.UserAttributeEntity;
+import org.keycloak.models.jpa.entities.UserEntity;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+/**
+ * Unit tests for {@link JpaHashUtils}.
+ * @author Alexander Schwartz
+ */
+public class JpaHashUtilsTest {
+
+    @Test
+    public void differentCaseShouldCreateDifferentHash() {
+        byte[] hash1 = JpaHashUtils.hashForAttributeValue("a");
+        byte[] hash2 = JpaHashUtils.hashForAttributeValue("A");
+        assertThat(hash1, not(equalTo(hash2)));
+    }
+
+    @Test
+    public void differentCaseShouldCreateSameHashForLowercase() {
+        byte[] hash1 = JpaHashUtils.hashForAttributeValueLowerCase("a");
+        byte[] hash2 = JpaHashUtils.hashForAttributeValueLowerCase("A");
+        assertThat(hash1, equalTo(hash2));
+    }
+
+    /**
+     * This shows that the default english locale used here correctly handles German umlauts as expected.
+     */
+    @Test
+    public void differentCaseShouldCreateSameHashForLowercaseGermanCharacters() {
+        byte[] hash1 = JpaHashUtils.hashForAttributeValueLowerCase("\u00E4");
+        byte[] hash2 = JpaHashUtils.hashForAttributeValueLowerCase("\u00C4");
+        assertThat(hash1, equalTo(hash2));
+    }
+
+    /**
+     * Although a caller in a turkish context might expect this to work, it won't as we enforce a hard-coded locale
+     * to avoid the need to re-hash on changes in the runtime locale.
+     */
+    @Test
+    public void differentCaseShouldCreateSameHashForLowercaseTurkishI() {
+        byte[] hash1 = JpaHashUtils.hashForAttributeValueLowerCase("I");
+        byte[] hash2 = JpaHashUtils.hashForAttributeValueLowerCase("I".toLowerCase(new Locale("tr")));
+        assertThat(hash1, not(equalTo(hash2)));
+    }
+
+    @Test
+    public void testPredicateForFilteringUsersByAttributes() {
+        UserEntity user = new UserEntity();
+        user.setAttributes(List.of(createAttribute("key1", "value1"), createAttribute("key2", "Value2")));
+
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1", "key2", "Value2"), JpaHashUtils::compareSourceValue).test(user), is(true));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1", "key2", "value2"), JpaHashUtils::compareSourceValue).test(user), is(false));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1"), JpaHashUtils::compareSourceValue).test(user), is(true));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value2"), JpaHashUtils::compareSourceValue).test(user), is(false));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key2", "value1"), JpaHashUtils::compareSourceValue).test(user), is(false));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "v1"), JpaHashUtils::compareSourceValue).test(user), is(false));
+
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1", "key2", "Value2"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(true));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1", "key2", "value2"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(true));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value1"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(true));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "value2"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(false));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key2", "value1"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(false));
+        assertThat(JpaHashUtils.predicateForFilteringUsersByAttributes(Map.of("key1", "v1"), JpaHashUtils::compareSourceValueLowerCase).test(user), is(false));
+    }
+
+    private UserAttributeEntity createAttribute(String key, String value) {
+        UserAttributeEntity attribute = new UserAttributeEntity();
+        attribute.setName(key);
+        attribute.setValue(value);
+        return attribute;
+    }
+}

model/storage/src/main/java/org/keycloak/storage/federated/UserAttributeFederatedStorage.java

@@ -38,7 +38,7 @@ public interface UserAttributeFederatedStorage {
      * @param realm a reference to the realm.
      * @param name the attribute name.
      * @param value the attribute value.
-     * @return a non-null {@link Stream} of users that match the search criteria.
+     * @return a non-null {@link Stream} of user IDs that match the search criteria.
      */
     Stream<String> getUsersByUserAttributeStream(RealmModel realm, String name, String value);
 

server-spi-private/src/main/java/org/keycloak/userprofile/DefaultAttributes.java

@@ -45,6 +45,8 @@ import org.keycloak.representations.userprofile.config.UPConfig.UnmanagedAttribu
 import org.keycloak.utils.StringUtil;
 import org.keycloak.validate.ValidationContext;
 import org.keycloak.validate.ValidationError;
+import org.keycloak.validate.ValidatorConfig;
+import org.keycloak.validate.validators.LengthValidator;
 
 /**
  * <p>The default implementation for {@link Attributes}. Should be reused as much as possible by the different implementations
@@ -67,6 +69,7 @@ public class DefaultAttributes extends HashMap<String, List<String>> implements
      * We should probably remove that once we remove the legacy provider, because this will come from the configuration.
      */
     public static final String READ_ONLY_ATTRIBUTE_KEY = "kc.read.only";
+    public static final String DEFAULT_MAX_LENGTH_ATTRIBUTES = "2048";
 
     protected final UserProfileContext context;
     protected final KeycloakSession session;
@@ -164,6 +167,7 @@ public class DefaultAttributes extends HashMap<String, List<String>> implements
                 .map(Collections::singletonList).orElse(emptyList()));
         metadatas.addAll(Optional.ofNullable(this.metadataByAttribute.get(READ_ONLY_ATTRIBUTE_KEY))
                 .map(Collections::singletonList).orElse(emptyList()));
+        limitLengthOnAttributesWithNoLengthRestriction(name, metadatas);
 
         Boolean result = null;
 
@@ -205,6 +209,27 @@ public class DefaultAttributes extends HashMap<String, List<String>> implements
         return result == null;
     }
 
+    /**
+     * In case there are unmanaged attributes or attributes that don't have a length restrictions,
+     * add a default length restriction to avoid a denial of service by a caller.
+     */
+    private static void limitLengthOnAttributesWithNoLengthRestriction(String name, List<AttributeMetadata> metadatas) {
+        for (AttributeMetadata metadata : metadatas) {
+            for (AttributeValidatorMetadata validator : metadata.getValidators()) {
+                if (validator.getValidatorId().equals(LengthValidator.ID)) {
+                    return;
+                }
+            }
+        }
+
+        AttributeMetadata am = new AttributeMetadata(name, -1);
+        Map<String, Object> vc = new HashMap<>();
+        vc.put(LengthValidator.KEY_MIN, "0");
+        vc.put(LengthValidator.KEY_MAX, DEFAULT_MAX_LENGTH_ATTRIBUTES);
+        am.addValidators(Collections.singletonList(new AttributeValidatorMetadata(LengthValidator.ID, new ValidatorConfig(vc))));
+        metadatas.add(am);
+    }
+
     @Override
     public List<String> get(String name) {
         return getOrDefault(name, EMPTY_VALUE);

server-spi/src/main/java/org/keycloak/storage/user/UserQueryMethodsProvider.java

@@ -94,6 +94,8 @@ public interface UserQueryMethodsProvider {
      *     the given userId (case sensitive string)</li>
      * </ul>
      * <p>
+     * Any other parameters will be treated as custom user attributes.
+     * <p>
      * This method is used by the REST API when querying users.
      *
      * @param realm  a reference to the realm.

testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/UserTest.java

@@ -17,7 +17,7 @@
 
 package org.keycloak.testsuite.admin;
 
-import jakarta.ws.rs.WebApplicationException;
+import org.apache.commons.lang3.RandomStringUtils;
 import org.hamcrest.Matchers;
 import org.jboss.arquillian.drone.api.annotation.Drone;
 import org.jboss.arquillian.graphene.page.Page;
@@ -93,6 +93,7 @@ import org.keycloak.testsuite.util.OAuthClient;
 import org.keycloak.testsuite.util.RealmBuilder;
 import org.keycloak.testsuite.util.RoleBuilder;
 import org.keycloak.testsuite.util.UserBuilder;
+import org.keycloak.userprofile.DefaultAttributes;
 import org.keycloak.userprofile.validator.UsernameProhibitedCharactersValidator;
 import org.keycloak.util.JsonSerialization;
 import org.openqa.selenium.By;
@@ -102,6 +103,7 @@ import jakarta.mail.internet.MimeMessage;
 import jakarta.ws.rs.BadRequestException;
 import jakarta.ws.rs.ClientErrorException;
 import jakarta.ws.rs.NotFoundException;
+import jakarta.ws.rs.WebApplicationException;
 import jakarta.ws.rs.core.Response;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -111,6 +113,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -122,6 +125,8 @@ import java.util.stream.Collectors;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -132,6 +137,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.keycloak.storage.UserStorageProviderModel.IMPORT_ENABLED;
@@ -896,6 +902,47 @@ public class UserTest extends AbstractAdminTest {
         }
     }
 
+    @Test
+    public void storeAndReadUserWithLongAttributeValue() {
+        String longValue = RandomStringUtils.random(Integer.parseInt(DefaultAttributes.DEFAULT_MAX_LENGTH_ATTRIBUTES), true, true);
+
+        getCleanup().addUserId(createUser(REALM_NAME, "user1", "password", "user1FirstName", "user1LastName", "user1@example.com",
+                user -> user.setAttributes(Map.of("attr", List.of(longValue)))));
+
+        List<UserRepresentation> users = realm.users().search("user1", true);
+
+        assertThat(users, hasSize(1));
+        assertThat(users.get(0).getAttributes().get("attr").get(0), equalTo(longValue));
+
+        WebApplicationException ex = assertThrows(WebApplicationException.class, () -> getCleanup().addUserId(createUser(REALM_NAME, "user2", "password", "user2FirstName", "user2LastName", "user2@example.com",
+                user -> user.setAttributes(Map.of("attr", List.of(longValue + "a"))))));
+        assertThat(ex.getResponse().getStatusInfo().getStatusCode(), equalTo(400));
+        assertThat(ex.getResponse().readEntity(ErrorRepresentation.class).getErrorMessage(), equalTo("error-invalid-length"));
+    }
+
+    @Test
+    public void searchByLongAttributes() {
+        // random string with suffix that makes it case-sensitive and distinct
+        String longValue = RandomStringUtils.random(Integer.parseInt(DefaultAttributes.DEFAULT_MAX_LENGTH_ATTRIBUTES) - 1, true, true) + "u";
+        String longValue2 = RandomStringUtils.random(Integer.parseInt(DefaultAttributes.DEFAULT_MAX_LENGTH_ATTRIBUTES) - 1, true, true) + "v";
+
+        getCleanup().addUserId(createUser(REALM_NAME, "user1", "password", "user1FirstName", "user1LastName", "user1@example.com",
+                user -> user.setAttributes(Map.of("test1", List.of(longValue, "v2"), "test2", List.of("v2")))));
+        getCleanup().addUserId(createUser(REALM_NAME, "user2", "password", "user2FirstName", "user2LastName", "user2@example.com", 
+                user -> user.setAttributes(Map.of("test1", List.of(longValue, "v2"), "test2", List.of(longValue2)))));
+        getCleanup().addUserId(createUser(REALM_NAME, "user3", "password", "user3FirstName", "user3LastName", "user3@example.com", 
+                user -> user.setAttributes(Map.of("test2", List.of(longValue, "v3"), "test4", List.of("v4")))));
+
+        assertThat(realm.users().searchByAttributes(mapToSearchQuery(Map.of("test1", longValue))).stream().map(UserRepresentation::getUsername).collect(Collectors.toList()), 
+                containsInAnyOrder("user1", "user2"));
+        assertThat(realm.users().searchByAttributes(mapToSearchQuery(Map.of("test1", longValue, "test2", longValue2))).stream().map(UserRepresentation::getUsername).collect(Collectors.toList()), 
+                contains("user2"));
+
+        //case-insensitive search
+        assertThat(realm.users().searchByAttributes(mapToSearchQuery(Map.of("test1", longValue, "test2", longValue2.toLowerCase(Locale.ENGLISH)))).stream().map(UserRepresentation::getUsername).collect(Collectors.toList()),
+                contains("user2"));
+    }
+
     @Test
     public void searchByUsernameExactMatch() {
         createUsers();

testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/federation/ldap/LDAPBinaryAttributesTest.java

@@ -32,6 +32,9 @@ import org.keycloak.models.UserModel;
 import org.keycloak.models.utils.UserModelDelegate;
 import org.keycloak.representations.idm.ComponentRepresentation;
 import org.keycloak.representations.idm.UserRepresentation;
+import org.keycloak.representations.userprofile.config.UPAttribute;
+import org.keycloak.representations.userprofile.config.UPAttributePermissions;
+import org.keycloak.representations.userprofile.config.UPConfig;
 import org.keycloak.storage.UserStoragePrivateUtil;
 import org.keycloak.storage.ldap.LDAPStorageProvider;
 import org.keycloak.storage.ldap.LDAPUtils;
@@ -44,11 +47,15 @@ import org.keycloak.testsuite.util.LDAPRule;
 import org.keycloak.testsuite.util.LDAPTestUtils;
 
 import jakarta.ws.rs.core.Response;
-import java.util.Arrays;
+import org.keycloak.validate.validators.LengthValidator;
+
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Stream;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -152,14 +159,23 @@ public class LDAPBinaryAttributesTest extends AbstractLDAPTest {
     public void test03WritableMapper() {
         String mapperId = addPhotoMapper(testingClient);
 
-        // Create user joe with jpegPHoto
+        // Allow long attribute for photos
+        UPConfig upc = adminClient.realm("test").users().userProfile().getConfiguration();
+        UPAttribute upa = new UPAttribute("jpegPhoto");
+        upa.setValidations(Map.of(LengthValidator.ID, Map.of(LengthValidator.KEY_MIN, "0", LengthValidator.KEY_MAX, JPEG_PHOTO_BASE64.length() * 2)));
+        upa.setPermissions(new UPAttributePermissions(Set.of("user", "admin"), Set.of("user", "admin")));
+        upc.getAttributes().add(upa);
+        adminClient.realm("test").users().userProfile().update(upc);
+
+        // Create user joe with jpegPhoto
         UserRepresentation joe = new UserRepresentation();
         joe.setUsername("joephoto");
         joe.setEmail("joe@photo.org");
-        joe.setAttributes(Collections.singletonMap(LDAPConstants.JPEG_PHOTO, Arrays.asList(JPEG_PHOTO_BASE64)));
+        joe.setAttributes(Collections.singletonMap(LDAPConstants.JPEG_PHOTO, List.of(JPEG_PHOTO_BASE64)));
-        Response response = adminClient.realm("test").users().create(joe);
-        response.close();
 
+        try (Response response = adminClient.realm("test").users().create(joe)) {
+            assertThat(response.getStatusInfo().getStatusCode(), equalTo(201));
+        }
 
         // Assert he is found including jpegPhoto
         joe = getUserAndAssertPhoto("joephoto", true);

testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/federation/storage/UserStorageTest.java

@@ -1,6 +1,7 @@
 package org.keycloak.testsuite.federation.storage;
 
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.RandomStringUtils;
 import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
 import org.jboss.arquillian.graphene.page.Page;
@@ -55,6 +56,7 @@ import org.keycloak.testsuite.updaters.RealmAttributeUpdater;
 import org.keycloak.testsuite.util.AccountHelper;
 import org.keycloak.testsuite.util.GreenMailRule;
 import org.keycloak.testsuite.util.TestCleanup;
+import org.keycloak.userprofile.DefaultAttributes;
 import org.openqa.selenium.Cookie;
 
 import jakarta.mail.internet.MimeMessage;
@@ -69,6 +71,7 @@ import java.util.Calendar;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
@@ -78,12 +81,14 @@ import java.util.stream.Stream;
 import static java.util.Calendar.DAY_OF_WEEK;
 import static java.util.Calendar.HOUR_OF_DAY;
 import static java.util.Calendar.MINUTE;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.keycloak.models.UserModel.RequiredAction.UPDATE_PROFILE;
@@ -515,6 +520,40 @@ public class UserStorageTest extends AbstractAuthTest {
         });
     }
 
+    @Test
+    public void storeAndReadUserWithLongAttributeValue() {
+        testingClient.server().run(session -> {
+            String longValue = RandomStringUtils.random(Integer.parseInt(DefaultAttributes.DEFAULT_MAX_LENGTH_ATTRIBUTES), true, true);
+            RealmModel realm = session.realms().getRealmByName("test");
+            UserModel userModel = session.users().getUserByUsername(realm, "thor");
+            userModel.setSingleAttribute("weapon", longValue);
+
+            List<UserModel> userModels = session.users().searchForUserStream(realm, Map.of(UserModel.USERNAME, "thor"))
+                    .collect(Collectors.toList());
+            assertThat(userModels, hasSize(1));
+            assertThat(userModels.get(0).getAttributes().get("weapon").get(0), equalTo(longValue));
+        });
+    }
+
+    @Test
+    public void searchByLongAttributeValue() {
+        testingClient.server().run(session -> {
+            // random string with suffix that makes it case-sensitive
+            String longValue = RandomStringUtils.random(2999, true, true) + "v";
+            RealmModel realm = session.realms().getRealmByName("test");
+            UserModel userModel = session.users().getUserByUsername(realm, "thor");
+            userModel.setSingleAttribute("weapon", longValue);
+
+            assertThat(session.users().searchForUserByUserAttributeStream(realm, "weapon", longValue).map(UserModel::getUsername).collect(Collectors.toList()), 
+                    containsInAnyOrder("thor"));
+
+            // searching here is always case sensitive
+            assertThat(session.users().searchForUserByUserAttributeStream(realm, "weapon", longValue.toUpperCase(Locale.ENGLISH)).map(UserModel::getUsername).collect(Collectors.toList()),
+                    empty());
+
+        });
+    }
+
     @Test
     public void testQueryExactMatch() {
         assertThat(testRealmResource().users().search("a", true), hasSize(0));