You can search a user by user attribute via Admin REST API.
Endpoint. It depends on your Keycloak version
{keycloak URL}/auth/admin/realms/{my-realm}/users/?q=userOrg:{uuid}
or
{keycloak URL}/admin/realms/{my-realm}/users/?q=userOrg:{uuid}
In, Get users Returns a stream of users, filtered according to query parameters. section
I demoed it in local Keycloak v18.0.2 and Postman
1 I made 3 users by UI
2 Each users has own userOrg
3 Get master token
4 search user by UUID
http://localhost:8180/auth/admin/realms/my-realm/users/?q=userOrg:3db51e81-7569-4a05-aaf3-91058450c63e
Keycloak response matched user by attribute search API
I tested with 10K users. it take 10~12 msec by my laptop.
Here is my user created program by python
import json
import admin
import random
import uuid
admin = admin.Admin()
token = admin.get_master_token()
first_names = [ \
"AARON","ADAM","ALAN","ALBERT","ANDREW","ANTHONY" \
,"ANTONIO","ARTHUR","BENJAMIN","BILLY","BOBBY" \
,"BRANDON","BRIAN","BRUCE","CARL","CARLOS" \
,"CHARLES","CHRIS","CHRISTOPHER","CLARENCE","CRAIG" \
,"DANIEL","DAVID","DENNIS","DONALD","DOUGLAS" \
,"EARL","EDWARD","ERIC","ERNEST","EUGENE" \
,"FRANK","FRED","GARY","GEORGE","GERALD" \
,"GREGORY","HAROLD","HARRY","HENRY","HOWARD" \
,"JACK","JAMES","JASON","JEFFREY","JEREMY" \
,"JERRY","JESSE","JIMMY","JOE","JOHN" \
,"JOHNNY","JONATHAN","JOSE","JOSEPH","JOSHUA" \
,"JUAN","JUSTIN","KEITH","KENNETH","KEVIN" \
,"LARRY","LAWRENCE","LOUIS","MARK","MARTIN" \
,"MATTHEW","MICHAEL","NICHOLAS","PATRICK","PAUL" \
,"PETER","PHILIP","PHILLIP","RALPH","RANDY" \
,"RAYMOND","RICHARD","ROBERT","ROGER","RONALD" \
,"ROY","RUSSELL","RYAN","SAMUEL","SCOTT" \
,"SEAN","SHAWN","STEPHEN","STEVE","STEVEN" \
,"TERRY","THOMAS","TIMOTHY","TODD","VICTOR" \
,"WALTER","WAYNE","WILLIAM","WILLIE"]
last_names = [ \
"Adams","Allen","Alvarez","Anderson","Bailey",\
"Baker","Bennet","Brooks","Brown","Campbell",\
"Carter","Castillo","Chavez","Clark","Collins",\
"Cook","Cooper","Cox","Cruz","Davis",\
"Diaz","Edwards","Evans","Flores","Foster",\
"Garcia","Gomez","Gonzales","Gray","Green",\
"Gutierrez","Hall","Harris","Hernandez","Hill",\
"Howard","Hughes","Jackson","James","Jimenez",\
"Johnson","Jones","Kelly","Kim","King"\
"Lee","Lewis","Long","Lopez","Martin",\
"Martinez","Mendoza","Miller","Mitchell","Moore",\
"Morales","Morgan","Morris","Murphy","Myers",\
"Nelson","Nguyen","Ortiz","Parker","Patel"\
"Perez","Peterson","Phillips","Price","Ramirez",\
"Ramos","Reed","Reyes","Richardson","Rivera",\
"Roberts","Robinson","Rodriguez","Rogers","Ross",\
"Ruiz","Sanchez","Sanders","Scott","Smith",\
"Stewart","Taylor","Thomas","Thompson","Torres",\
"Turner","Walker","Ward","Watson","White",\
"Williams","Wilson","Wood","Wright","Young",]
user_list = []
index = 1
user_data = {}
access_item = {}
user_list = []
max_user_count = 10000
size_first_name = len(first_names)
size_last_name = len(last_names)
for index in range(1,max_user_count+1):
first_name_index = random.randint(0, size_first_name)
last_name_index = random.randint(0, size_last_name)
user_data['enabled'] = True
user_data['groups'] = []
user_data['emailVerified'] = ''
user_data['firstName'] = first_names[first_name_index-1].capitalize()
user_data['lastName'] = last_names[last_name_index-1]
user_data['username'] = 'user'+str(index)
user_data['email'] = 'user'+str(index)+'@test.com'
user_data['attributes'] = { 'userOrg' : [ str(uuid.uuid4())]}
user_list.append(user_data)
print(json.dumps(user_data))
user_data = {}
# add user if not exist
for user in user_list:
if (not admin.is_user_exist(token, 'test', user['username'])):
admin.add_user(token, user, 'test')
print('Add User', user['username'])
Get token code, name should be admin.py
from urllib.error import HTTPError
import requests
import ast
import json
class Admin:
# Keycloak master realm URL
url = 'http://localhost:8180/auth/realms/master/protocol/openid-connect/token'
# Keycloak master credential
params = {
'client_id': 'admin-cli',
'grant_type': 'password',
'username' : 'admin',
'password': 'admin'
}
def get_master_token(self):
try:
response = requests.post(self.url, self.params, verify=False).content.decode('utf-8')
except HTTPError as http_err:
print(f'HTTP error occurred: {http_err}') # Python 3.6
except Exception as err:
print(f'Other error occurred: {err}') # Python 3.6
print('Keycloak container is not running, Please check your docker container!')
raise SystemExit
else:
return ast.literal_eval(response)['access_token']
def add_user(self, token, user, realm_name):
url ='http://localhost:8180/auth/admin/realms/'+realm_name+'/users'
headers = {
'content-type': 'application/json',
'Authorization' : 'Bearer '+ str(token)
}
params = {
'username': user['username'],
'enabled': True,
'totp': False,
'emailVerified': True,
'firstName': user['firstName'],
'lastName': user['lastName'],
'email': user['email'],
'attributes': user['attributes'],
'disableableCredentialTypes': [],
'requiredActions': [],
'notBefore': 0,
'access': {
'manageGroupMembership': True,
'view': True,
'mapRoles': True,
'impersonate': True,
'manage': True
},
'realmRoles': [ realm_name ]
}
x = requests.post(url, headers=headers, json=params)
return x.content
def is_user_exist(self, token, realm_name, user_name):
url ='http://localhost:8180/auth/admin/realms/'+realm_name+'/users/?username='+user_name.replace(" ", "%20")
headers = {
'content-type': 'application/json',
'Authorization' : 'Bearer '+ str(token)
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
except HTTPError as http_err:
print(f'HTTP error occurred: {http_err}') # Python 3.6
except Exception as err:
print(f'Other error occurred: {err}') # Python 3.6
else:
if len(response.content) == 2: # []
return False
if (json.loads(response.text)[0]['username'] == user_name.lower()):
return True
else:
return False
