You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

2877 lines
102 KiB

  1. """Test the keycloak admin object."""
  2. import copy
  3. import uuid
  4. from typing import Tuple
  5. import freezegun
  6. import pytest
  7. from dateutil import parser as datetime_parser
  8. import keycloak
  9. from keycloak import KeycloakAdmin, KeycloakOpenID, KeycloakOpenIDConnection
  10. from keycloak.connection import ConnectionManager
  11. from keycloak.exceptions import (
  12. KeycloakAuthenticationError,
  13. KeycloakDeleteError,
  14. KeycloakGetError,
  15. KeycloakPostError,
  16. KeycloakPutError,
  17. )
  18. def test_keycloak_version():
  19. """Test version."""
  20. assert keycloak.__version__, keycloak.__version__
  21. def test_keycloak_admin_init(env):
  22. """Test keycloak admin init.
  23. :param env: Environment fixture
  24. :type env: KeycloakTestEnv
  25. """
  26. admin = KeycloakAdmin(
  27. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  28. username=env.KEYCLOAK_ADMIN,
  29. password=env.KEYCLOAK_ADMIN_PASSWORD,
  30. )
  31. assert admin.server_url == f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}", admin.server_url
  32. assert admin.realm_name == "master", admin.realm_name
  33. assert isinstance(admin.connection, ConnectionManager), type(admin.connection)
  34. assert admin.client_id == "admin-cli", admin.client_id
  35. assert admin.client_secret_key is None, admin.client_secret_key
  36. assert admin.verify, admin.verify
  37. assert admin.username == env.KEYCLOAK_ADMIN, admin.username
  38. assert admin.password == env.KEYCLOAK_ADMIN_PASSWORD, admin.password
  39. assert admin.totp is None, admin.totp
  40. assert admin.token is not None, admin.token
  41. assert admin.user_realm_name is None, admin.user_realm_name
  42. assert admin.custom_headers is None, admin.custom_headers
  43. assert admin.token
  44. admin = KeycloakAdmin(
  45. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  46. username=env.KEYCLOAK_ADMIN,
  47. password=env.KEYCLOAK_ADMIN_PASSWORD,
  48. realm_name=None,
  49. user_realm_name="master",
  50. )
  51. assert admin.token
  52. admin = KeycloakAdmin(
  53. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  54. username=env.KEYCLOAK_ADMIN,
  55. password=env.KEYCLOAK_ADMIN_PASSWORD,
  56. realm_name=None,
  57. user_realm_name=None,
  58. )
  59. assert admin.token
  60. token = admin.token
  61. admin = KeycloakAdmin(
  62. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  63. token=token,
  64. realm_name=None,
  65. user_realm_name=None,
  66. )
  67. assert admin.token == token
  68. admin.create_realm(payload={"realm": "authz", "enabled": True})
  69. admin.realm_name = "authz"
  70. admin.create_client(
  71. payload={
  72. "name": "authz-client",
  73. "clientId": "authz-client",
  74. "authorizationServicesEnabled": True,
  75. "serviceAccountsEnabled": True,
  76. "clientAuthenticatorType": "client-secret",
  77. "directAccessGrantsEnabled": False,
  78. "enabled": True,
  79. "implicitFlowEnabled": False,
  80. "publicClient": False,
  81. }
  82. )
  83. secret = admin.generate_client_secrets(client_id=admin.get_client_id("authz-client"))
  84. assert KeycloakAdmin(
  85. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  86. user_realm_name="authz",
  87. client_id="authz-client",
  88. client_secret_key=secret["value"],
  89. ).token
  90. admin.delete_realm(realm_name="authz")
  91. assert (
  92. KeycloakAdmin(
  93. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  94. username=None,
  95. password=None,
  96. client_secret_key=None,
  97. custom_headers={"custom": "header"},
  98. ).token
  99. is None
  100. )
  101. keycloak_connection = KeycloakOpenIDConnection(
  102. server_url=f"http://{env.KEYCLOAK_HOST}:{env.KEYCLOAK_PORT}",
  103. username=env.KEYCLOAK_ADMIN,
  104. password=env.KEYCLOAK_ADMIN_PASSWORD,
  105. realm_name="master",
  106. client_id="admin-cli",
  107. verify=True,
  108. )
  109. keycloak_admin = KeycloakAdmin(connection=keycloak_connection)
  110. assert keycloak_admin.token
  111. def test_realms(admin: KeycloakAdmin):
  112. """Test realms.
  113. :param admin: Keycloak Admin client
  114. :type admin: KeycloakAdmin
  115. """
  116. # Get realms
  117. realms = admin.get_realms()
  118. assert len(realms) == 1, realms
  119. assert "master" == realms[0]["realm"]
  120. # Create a test realm
  121. res = admin.create_realm(payload={"realm": "test"})
  122. assert res == b"", res
  123. # Create the same realm, should fail
  124. with pytest.raises(KeycloakPostError) as err:
  125. res = admin.create_realm(payload={"realm": "test"})
  126. assert err.match('409: b\'{"errorMessage":"Conflict detected. See logs for details"}\'')
  127. # Create the same realm, skip_exists true
  128. res = admin.create_realm(payload={"realm": "test"}, skip_exists=True)
  129. assert res == {"msg": "Already exists"}, res
  130. # Get a single realm
  131. res = admin.get_realm(realm_name="test")
  132. assert res["realm"] == "test"
  133. # Get non-existing realm
  134. with pytest.raises(KeycloakGetError) as err:
  135. admin.get_realm(realm_name="non-existent")
  136. assert err.match('404: b\'{"error":"Realm not found."}\'')
  137. # Update realm
  138. res = admin.update_realm(realm_name="test", payload={"accountTheme": "test"})
  139. assert res == dict(), res
  140. # Check that the update worked
  141. res = admin.get_realm(realm_name="test")
  142. assert res["realm"] == "test"
  143. assert res["accountTheme"] == "test"
  144. # Update wrong payload
  145. with pytest.raises(KeycloakPutError) as err:
  146. admin.update_realm(realm_name="test", payload={"wrong": "payload"})
  147. assert err.match('400: b\'{"error":"Unrecognized field')
  148. # Check that get realms returns both realms
  149. realms = admin.get_realms()
  150. realm_names = [x["realm"] for x in realms]
  151. assert len(realms) == 2, realms
  152. assert "master" in realm_names, realm_names
  153. assert "test" in realm_names, realm_names
  154. # Delete the realm
  155. res = admin.delete_realm(realm_name="test")
  156. assert res == dict(), res
  157. # Check that the realm does not exist anymore
  158. with pytest.raises(KeycloakGetError) as err:
  159. admin.get_realm(realm_name="test")
  160. assert err.match('404: b\'{"error":"Realm not found."}\'')
  161. # Delete non-existing realm
  162. with pytest.raises(KeycloakDeleteError) as err:
  163. admin.delete_realm(realm_name="non-existent")
  164. assert err.match('404: b\'{"error":"Realm not found."}\'')
  165. def test_changing_of_realms(admin: KeycloakAdmin, realm: str):
  166. """Test changing of realms.
  167. :param admin: Keycloak Admin client
  168. :type admin: KeycloakAdmin
  169. :param realm: Keycloak realm
  170. :type realm: str
  171. """
  172. assert admin.get_current_realm() == "master"
  173. admin.change_current_realm(realm)
  174. assert admin.get_current_realm() == realm
  175. def test_import_export_realms(admin: KeycloakAdmin, realm: str):
  176. """Test import and export of realms.
  177. :param admin: Keycloak Admin client
  178. :type admin: KeycloakAdmin
  179. :param realm: Keycloak realm
  180. :type realm: str
  181. """
  182. admin.change_current_realm(realm)
  183. realm_export = admin.export_realm(export_clients=True, export_groups_and_role=True)
  184. assert realm_export != dict(), realm_export
  185. admin.delete_realm(realm_name=realm)
  186. admin.realm_name = "master"
  187. res = admin.import_realm(payload=realm_export)
  188. assert res == b"", res
  189. # Test bad import
  190. with pytest.raises(KeycloakPostError) as err:
  191. admin.import_realm(payload=dict())
  192. assert err.match('500: b\'{"error":"unknown_error"}\'')
  193. def test_partial_import_realm(admin: KeycloakAdmin, realm: str):
  194. """Test partial import of realm configuration.
  195. :param admin: Keycloak Admin client
  196. :type admin: KeycloakAdmin
  197. :param realm: Keycloak realm
  198. :type realm: str
  199. """
  200. test_realm_role = str(uuid.uuid4())
  201. test_user = str(uuid.uuid4())
  202. test_client = str(uuid.uuid4())
  203. admin.change_current_realm(realm)
  204. client_id = admin.create_client(payload={"name": test_client, "clientId": test_client})
  205. realm_export = admin.export_realm(export_clients=True, export_groups_and_role=False)
  206. client_config = [
  207. client_entry for client_entry in realm_export["clients"] if client_entry["id"] == client_id
  208. ][0]
  209. # delete before partial import
  210. admin.delete_client(client_id)
  211. payload = {
  212. "ifResourceExists": "SKIP",
  213. "id": realm_export["id"],
  214. "realm": realm,
  215. "clients": [client_config],
  216. "roles": {"realm": [{"name": test_realm_role}]},
  217. "users": [{"username": test_user, "email": f"{test_user}@test.test"}],
  218. }
  219. # check add
  220. res = admin.partial_import_realm(realm_name=realm, payload=payload)
  221. assert res["added"] == 3
  222. # check skip
  223. res = admin.partial_import_realm(realm_name=realm, payload=payload)
  224. assert res["skipped"] == 3
  225. # check overwrite
  226. payload["ifResourceExists"] = "OVERWRITE"
  227. res = admin.partial_import_realm(realm_name=realm, payload=payload)
  228. assert res["overwritten"] == 3
  229. def test_users(admin: KeycloakAdmin, realm: str):
  230. """Test users.
  231. :param admin: Keycloak Admin client
  232. :type admin: KeycloakAdmin
  233. :param realm: Keycloak realm
  234. :type realm: str
  235. """
  236. admin.change_current_realm(realm)
  237. # Check no users present
  238. users = admin.get_users()
  239. assert users == list(), users
  240. # Test create user
  241. user_id = admin.create_user(payload={"username": "test", "email": "test@test.test"})
  242. assert user_id is not None, user_id
  243. # Test create the same user
  244. with pytest.raises(KeycloakPostError) as err:
  245. admin.create_user(payload={"username": "test", "email": "test@test.test"})
  246. assert err.match('409: b\'{"errorMessage":"User exists with same username"}\'')
  247. # Test create the same user, exists_ok true
  248. user_id_2 = admin.create_user(
  249. payload={"username": "test", "email": "test@test.test"}, exist_ok=True
  250. )
  251. assert user_id == user_id_2
  252. # Test get user
  253. user = admin.get_user(user_id=user_id)
  254. assert user["username"] == "test", user["username"]
  255. assert user["email"] == "test@test.test", user["email"]
  256. # Test update user
  257. res = admin.update_user(user_id=user_id, payload={"firstName": "Test"})
  258. assert res == dict(), res
  259. user = admin.get_user(user_id=user_id)
  260. assert user["firstName"] == "Test"
  261. # Test update user fail
  262. with pytest.raises(KeycloakPutError) as err:
  263. admin.update_user(user_id=user_id, payload={"wrong": "payload"})
  264. assert err.match('400: b\'{"error":"Unrecognized field')
  265. # Test get users again
  266. users = admin.get_users()
  267. usernames = [x["username"] for x in users]
  268. assert "test" in usernames
  269. # Test users counts
  270. count = admin.users_count()
  271. assert count == 1, count
  272. # Test users count with query
  273. count = admin.users_count(query={"username": "notpresent"})
  274. assert count == 0
  275. # Test user groups
  276. groups = admin.get_user_groups(user_id=user["id"])
  277. assert len(groups) == 0
  278. # Test user groups bad id
  279. with pytest.raises(KeycloakGetError) as err:
  280. admin.get_user_groups(user_id="does-not-exist")
  281. assert err.match('404: b\'{"error":"User not found"}\'')
  282. # Test logout
  283. res = admin.user_logout(user_id=user["id"])
  284. assert res == dict(), res
  285. # Test logout fail
  286. with pytest.raises(KeycloakPostError) as err:
  287. admin.user_logout(user_id="non-existent-id")
  288. assert err.match('404: b\'{"error":"User not found"}\'')
  289. # Test consents
  290. res = admin.user_consents(user_id=user["id"])
  291. assert len(res) == 0, res
  292. # Test consents fail
  293. with pytest.raises(KeycloakGetError) as err:
  294. admin.user_consents(user_id="non-existent-id")
  295. assert err.match('404: b\'{"error":"User not found"}\'')
  296. # Test delete user
  297. res = admin.delete_user(user_id=user_id)
  298. assert res == dict(), res
  299. with pytest.raises(KeycloakGetError) as err:
  300. admin.get_user(user_id=user_id)
  301. err.match('404: b\'{"error":"User not found"}\'')
  302. # Test delete fail
  303. with pytest.raises(KeycloakDeleteError) as err:
  304. admin.delete_user(user_id="non-existent-id")
  305. assert err.match('404: b\'{"error":"User not found"}\'')
  306. def test_users_pagination(admin: KeycloakAdmin, realm: str):
  307. """Test user pagination.
  308. :param admin: Keycloak Admin client
  309. :type admin: KeycloakAdmin
  310. :param realm: Keycloak realm
  311. :type realm: str
  312. """
  313. admin.change_current_realm(realm)
  314. for ind in range(admin.PAGE_SIZE + 50):
  315. username = f"user_{ind}"
  316. admin.create_user(payload={"username": username, "email": f"{username}@test.test"})
  317. users = admin.get_users()
  318. assert len(users) == admin.PAGE_SIZE + 50, len(users)
  319. users = admin.get_users(query={"first": 100})
  320. assert len(users) == 50, len(users)
  321. users = admin.get_users(query={"max": 20})
  322. assert len(users) == 20, len(users)
  323. def test_user_groups_pagination(admin: KeycloakAdmin, realm: str):
  324. """Test user groups pagination.
  325. :param admin: Keycloak Admin client
  326. :type admin: KeycloakAdmin
  327. :param realm: Keycloak realm
  328. :type realm: str
  329. """
  330. admin.change_current_realm(realm)
  331. user_id = admin.create_user(
  332. payload={"username": "username_1", "email": "username_1@test.test"}
  333. )
  334. for ind in range(admin.PAGE_SIZE + 50):
  335. group_name = f"group_{ind}"
  336. group_id = admin.create_group(payload={"name": group_name})
  337. admin.group_user_add(user_id=user_id, group_id=group_id)
  338. groups = admin.get_user_groups(user_id=user_id)
  339. assert len(groups) == admin.PAGE_SIZE + 50, len(groups)
  340. groups = admin.get_user_groups(user_id=user_id, query={"first": 100, "max": -1, "search": ""})
  341. assert len(groups) == 50, len(groups)
  342. groups = admin.get_user_groups(user_id=user_id, query={"max": 20, "first": -1, "search": ""})
  343. assert len(groups) == 20, len(groups)
  344. def test_idps(admin: KeycloakAdmin, realm: str):
  345. """Test IDPs.
  346. :param admin: Keycloak Admin client
  347. :type admin: KeycloakAdmin
  348. :param realm: Keycloak realm
  349. :type realm: str
  350. """
  351. admin.change_current_realm(realm)
  352. # Create IDP
  353. res = admin.create_idp(
  354. payload=dict(
  355. providerId="github", alias="github", config=dict(clientId="test", clientSecret="test")
  356. )
  357. )
  358. assert res == b"", res
  359. # Test create idp fail
  360. with pytest.raises(KeycloakPostError) as err:
  361. admin.create_idp(payload={"providerId": "does-not-exist", "alias": "something"})
  362. assert err.match("Invalid identity provider id"), err
  363. # Test listing
  364. idps = admin.get_idps()
  365. assert len(idps) == 1
  366. assert "github" == idps[0]["alias"]
  367. # Test get idp
  368. idp = admin.get_idp("github")
  369. assert "github" == idp["alias"]
  370. assert idp.get("config")
  371. assert "test" == idp["config"]["clientId"]
  372. assert "**********" == idp["config"]["clientSecret"]
  373. # Test get idp fail
  374. with pytest.raises(KeycloakGetError) as err:
  375. admin.get_idp("does-not-exist")
  376. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  377. # Test IdP update
  378. res = admin.update_idp(idp_alias="github", payload=idps[0])
  379. assert res == {}, res
  380. # Test adding a mapper
  381. res = admin.add_mapper_to_idp(
  382. idp_alias="github",
  383. payload={
  384. "identityProviderAlias": "github",
  385. "identityProviderMapper": "github-user-attribute-mapper",
  386. "name": "test",
  387. },
  388. )
  389. assert res == b"", res
  390. # Test mapper fail
  391. with pytest.raises(KeycloakPostError) as err:
  392. admin.add_mapper_to_idp(idp_alias="does-no-texist", payload=dict())
  393. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  394. # Test IdP mappers listing
  395. idp_mappers = admin.get_idp_mappers(idp_alias="github")
  396. assert len(idp_mappers) == 1
  397. # Test IdP mapper update
  398. res = admin.update_mapper_in_idp(
  399. idp_alias="github",
  400. mapper_id=idp_mappers[0]["id"],
  401. # For an obscure reason, keycloak expect all fields
  402. payload={
  403. "id": idp_mappers[0]["id"],
  404. "identityProviderAlias": "github-alias",
  405. "identityProviderMapper": "github-user-attribute-mapper",
  406. "name": "test",
  407. "config": idp_mappers[0]["config"],
  408. },
  409. )
  410. assert res == dict(), res
  411. # Test delete
  412. res = admin.delete_idp(idp_alias="github")
  413. assert res == dict(), res
  414. # Test delete fail
  415. with pytest.raises(KeycloakDeleteError) as err:
  416. admin.delete_idp(idp_alias="does-not-exist")
  417. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  418. def test_user_credentials(admin: KeycloakAdmin, user: str):
  419. """Test user credentials.
  420. :param admin: Keycloak Admin client
  421. :type admin: KeycloakAdmin
  422. :param user: Keycloak user
  423. :type user: str
  424. """
  425. res = admin.set_user_password(user_id=user, password="booya", temporary=True)
  426. assert res == dict(), res
  427. # Test user password set fail
  428. with pytest.raises(KeycloakPutError) as err:
  429. admin.set_user_password(user_id="does-not-exist", password="")
  430. assert err.match('404: b\'{"error":"User not found"}\'')
  431. credentials = admin.get_credentials(user_id=user)
  432. assert len(credentials) == 1
  433. assert credentials[0]["type"] == "password", credentials
  434. # Test get credentials fail
  435. with pytest.raises(KeycloakGetError) as err:
  436. admin.get_credentials(user_id="does-not-exist")
  437. assert err.match('404: b\'{"error":"User not found"}\'')
  438. res = admin.delete_credential(user_id=user, credential_id=credentials[0]["id"])
  439. assert res == dict(), res
  440. # Test delete fail
  441. with pytest.raises(KeycloakDeleteError) as err:
  442. admin.delete_credential(user_id=user, credential_id="does-not-exist")
  443. assert err.match('404: b\'{"error":"Credential not found"}\'')
  444. def test_social_logins(admin: KeycloakAdmin, user: str):
  445. """Test social logins.
  446. :param admin: Keycloak Admin client
  447. :type admin: KeycloakAdmin
  448. :param user: Keycloak user
  449. :type user: str
  450. """
  451. res = admin.add_user_social_login(
  452. user_id=user, provider_id="gitlab", provider_userid="test", provider_username="test"
  453. )
  454. assert res == dict(), res
  455. admin.add_user_social_login(
  456. user_id=user, provider_id="github", provider_userid="test", provider_username="test"
  457. )
  458. assert res == dict(), res
  459. # Test add social login fail
  460. with pytest.raises(KeycloakPostError) as err:
  461. admin.add_user_social_login(
  462. user_id="does-not-exist",
  463. provider_id="does-not-exist",
  464. provider_userid="test",
  465. provider_username="test",
  466. )
  467. assert err.match('404: b\'{"error":"User not found"}\'')
  468. res = admin.get_user_social_logins(user_id=user)
  469. assert res == list(), res
  470. # Test get social logins fail
  471. with pytest.raises(KeycloakGetError) as err:
  472. admin.get_user_social_logins(user_id="does-not-exist")
  473. assert err.match('404: b\'{"error":"User not found"}\'')
  474. res = admin.delete_user_social_login(user_id=user, provider_id="gitlab")
  475. assert res == {}, res
  476. res = admin.delete_user_social_login(user_id=user, provider_id="github")
  477. assert res == {}, res
  478. with pytest.raises(KeycloakDeleteError) as err:
  479. admin.delete_user_social_login(user_id=user, provider_id="instagram")
  480. assert err.match('404: b\'{"error":"Link not found"}\''), err
  481. def test_server_info(admin: KeycloakAdmin):
  482. """Test server info.
  483. :param admin: Keycloak Admin client
  484. :type admin: KeycloakAdmin
  485. """
  486. info = admin.get_server_info()
  487. assert set(info.keys()).issubset(
  488. {
  489. "systemInfo",
  490. "memoryInfo",
  491. "profileInfo",
  492. "features",
  493. "themes",
  494. "socialProviders",
  495. "identityProviders",
  496. "providers",
  497. "protocolMapperTypes",
  498. "builtinProtocolMappers",
  499. "clientInstallations",
  500. "componentTypes",
  501. "passwordPolicies",
  502. "enums",
  503. "cryptoInfo",
  504. "features",
  505. }
  506. ), info.keys()
  507. def test_groups(admin: KeycloakAdmin, user: str):
  508. """Test groups.
  509. :param admin: Keycloak Admin client
  510. :type admin: KeycloakAdmin
  511. :param user: Keycloak user
  512. :type user: str
  513. """
  514. # Test get groups
  515. groups = admin.get_groups()
  516. assert len(groups) == 0
  517. # Test create group
  518. group_id = admin.create_group(payload={"name": "main-group"})
  519. assert group_id is not None, group_id
  520. # Test create subgroups
  521. subgroup_id_1 = admin.create_group(payload={"name": "subgroup-1"}, parent=group_id)
  522. subgroup_id_2 = admin.create_group(payload={"name": "subgroup-2"}, parent=group_id)
  523. # Test create group fail
  524. with pytest.raises(KeycloakPostError) as err:
  525. admin.create_group(payload={"name": "subgroup-1"}, parent=group_id)
  526. assert err.match("409"), err
  527. # Test skip exists OK
  528. subgroup_id_1_eq = admin.create_group(
  529. payload={"name": "subgroup-1"}, parent=group_id, skip_exists=True
  530. )
  531. assert subgroup_id_1_eq is None
  532. # Test get groups again
  533. groups = admin.get_groups()
  534. assert len(groups) == 1, groups
  535. assert groups[0]["subGroupCount"] == 2
  536. assert groups[0]["id"] == group_id
  537. # Test get groups query
  538. groups = admin.get_groups(query={"max": 10})
  539. assert len(groups) == 1, groups
  540. assert groups[0]["subGroupCount"] == 2
  541. assert groups[0]["id"] == group_id
  542. # Test get group
  543. res = admin.get_group(group_id=subgroup_id_1)
  544. assert res["id"] == subgroup_id_1, res
  545. assert res["name"] == "subgroup-1"
  546. assert res["path"] == "/main-group/subgroup-1"
  547. # Test get group fail
  548. with pytest.raises(KeycloakGetError) as err:
  549. admin.get_group(group_id="does-not-exist")
  550. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  551. # Create 1 more subgroup
  552. subsubgroup_id_1 = admin.create_group(payload={"name": "subsubgroup-1"}, parent=subgroup_id_2)
  553. main_group = admin.get_group(group_id=group_id)
  554. subgroup_2 = admin.get_group(group_id=subgroup_id_2)
  555. # Test nested searches
  556. res = admin.get_subgroups(group=subgroup_2, path="/main-group/subgroup-2/subsubgroup-1")
  557. assert res is not None, res
  558. assert res["id"] == subsubgroup_id_1
  559. # Test empty search
  560. res = admin.get_subgroups(group=main_group, path="/none")
  561. assert res is None, res
  562. # Test get group by path
  563. res = admin.get_group_by_path(path="/main-group/subgroup-1")
  564. assert res is not None, res
  565. assert res["id"] == subgroup_id_1, res
  566. with pytest.raises(KeycloakGetError) as err:
  567. admin.get_group_by_path(path="/main-group/subgroup-2/subsubgroup-1/test")
  568. assert err.match('404: b\'{"error":"Group path does not exist"}\'')
  569. res = admin.get_group_by_path(path="/main-group/subgroup-2/subsubgroup-1")
  570. assert res is not None, res
  571. assert res["id"] == subsubgroup_id_1
  572. res = admin.get_group_by_path(path="/main-group")
  573. assert res is not None, res
  574. assert res["id"] == group_id, res
  575. # Test group members
  576. res = admin.get_group_members(group_id=subgroup_id_2)
  577. assert len(res) == 0, res
  578. # Test fail group members
  579. with pytest.raises(KeycloakGetError) as err:
  580. admin.get_group_members(group_id="does-not-exist")
  581. assert err.match('404: b\'{"error":"Could not find group by id"}\'')
  582. res = admin.group_user_add(user_id=user, group_id=subgroup_id_2)
  583. assert res == dict(), res
  584. res = admin.get_group_members(group_id=subgroup_id_2)
  585. assert len(res) == 1, res
  586. assert res[0]["id"] == user
  587. # Test get group members query
  588. res = admin.get_group_members(group_id=subgroup_id_2, query={"max": 10})
  589. assert len(res) == 1, res
  590. assert res[0]["id"] == user
  591. with pytest.raises(KeycloakDeleteError) as err:
  592. admin.group_user_remove(user_id="does-not-exist", group_id=subgroup_id_2)
  593. assert err.match('404: b\'{"error":"User not found"}\''), err
  594. res = admin.group_user_remove(user_id=user, group_id=subgroup_id_2)
  595. assert res == dict(), res
  596. # Test set permissions
  597. res = admin.group_set_permissions(group_id=subgroup_id_2, enabled=True)
  598. assert res["enabled"], res
  599. res = admin.group_set_permissions(group_id=subgroup_id_2, enabled=False)
  600. assert not res["enabled"], res
  601. with pytest.raises(KeycloakPutError) as err:
  602. admin.group_set_permissions(group_id=subgroup_id_2, enabled="blah")
  603. assert err.match('b\'{"error":"unknown_error"}\''), err
  604. # Test update group
  605. res = admin.update_group(group_id=subgroup_id_2, payload={"name": "new-subgroup-2"})
  606. assert res == dict(), res
  607. assert admin.get_group(group_id=subgroup_id_2)["name"] == "new-subgroup-2"
  608. # test update fail
  609. with pytest.raises(KeycloakPutError) as err:
  610. admin.update_group(group_id="does-not-exist", payload=dict())
  611. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  612. # Test delete
  613. res = admin.delete_group(group_id=group_id)
  614. assert res == dict(), res
  615. assert len(admin.get_groups()) == 0
  616. # Test delete fail
  617. with pytest.raises(KeycloakDeleteError) as err:
  618. admin.delete_group(group_id="does-not-exist")
  619. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  620. def test_clients(admin: KeycloakAdmin, realm: str):
  621. """Test clients.
  622. :param admin: Keycloak Admin client
  623. :type admin: KeycloakAdmin
  624. :param realm: Keycloak realm
  625. :type realm: str
  626. """
  627. admin.change_current_realm(realm)
  628. # Test get clients
  629. clients = admin.get_clients()
  630. assert len(clients) == 6, clients
  631. assert {x["name"] for x in clients} == set(
  632. [
  633. "${client_admin-cli}",
  634. "${client_security-admin-console}",
  635. "${client_account-console}",
  636. "${client_broker}",
  637. "${client_account}",
  638. "${client_realm-management}",
  639. ]
  640. ), clients
  641. # Test create client
  642. client_id = admin.create_client(payload={"name": "test-client", "clientId": "test-client"})
  643. assert client_id, client_id
  644. with pytest.raises(KeycloakPostError) as err:
  645. admin.create_client(payload={"name": "test-client", "clientId": "test-client"})
  646. assert err.match('409: b\'{"errorMessage":"Client test-client already exists"}\''), err
  647. client_id_2 = admin.create_client(
  648. payload={"name": "test-client", "clientId": "test-client"}, skip_exists=True
  649. )
  650. assert client_id == client_id_2, client_id_2
  651. # Test get client
  652. res = admin.get_client(client_id=client_id)
  653. assert res["clientId"] == "test-client", res
  654. assert res["name"] == "test-client", res
  655. assert res["id"] == client_id, res
  656. with pytest.raises(KeycloakGetError) as err:
  657. admin.get_client(client_id="does-not-exist")
  658. assert err.match('404: b\'{"error":"Could not find client"}\'')
  659. assert len(admin.get_clients()) == 7
  660. # Test get client id
  661. assert admin.get_client_id(client_id="test-client") == client_id
  662. assert admin.get_client_id(client_id="does-not-exist") is None
  663. # Test update client
  664. res = admin.update_client(client_id=client_id, payload={"name": "test-client-change"})
  665. assert res == dict(), res
  666. with pytest.raises(KeycloakPutError) as err:
  667. admin.update_client(client_id="does-not-exist", payload={"name": "test-client-change"})
  668. assert err.match('404: b\'{"error":"Could not find client"}\'')
  669. # Test client mappers
  670. res = admin.get_mappers_from_client(client_id=client_id)
  671. assert len(res) == 0
  672. with pytest.raises(KeycloakPostError) as err:
  673. admin.add_mapper_to_client(client_id="does-not-exist", payload=dict())
  674. assert err.match('404: b\'{"error":"Could not find client"}\'')
  675. res = admin.add_mapper_to_client(
  676. client_id=client_id,
  677. payload={
  678. "name": "test-mapper",
  679. "protocol": "openid-connect",
  680. "protocolMapper": "oidc-usermodel-attribute-mapper",
  681. },
  682. )
  683. assert res == b""
  684. assert len(admin.get_mappers_from_client(client_id=client_id)) == 1
  685. mapper = admin.get_mappers_from_client(client_id=client_id)[0]
  686. with pytest.raises(KeycloakPutError) as err:
  687. admin.update_client_mapper(client_id=client_id, mapper_id="does-not-exist", payload=dict())
  688. assert err.match('404: b\'{"error":"Model not found"}\'')
  689. mapper["config"]["user.attribute"] = "test"
  690. res = admin.update_client_mapper(client_id=client_id, mapper_id=mapper["id"], payload=mapper)
  691. assert res == dict()
  692. res = admin.remove_client_mapper(client_id=client_id, client_mapper_id=mapper["id"])
  693. assert res == dict()
  694. with pytest.raises(KeycloakDeleteError) as err:
  695. admin.remove_client_mapper(client_id=client_id, client_mapper_id=mapper["id"])
  696. assert err.match('404: b\'{"error":"Model not found"}\'')
  697. # Test client sessions
  698. with pytest.raises(KeycloakGetError) as err:
  699. admin.get_client_all_sessions(client_id="does-not-exist")
  700. assert err.match('404: b\'{"error":"Could not find client"}\'')
  701. assert admin.get_client_all_sessions(client_id=client_id) == list()
  702. assert admin.get_client_sessions_stats() == list()
  703. # Test authz
  704. auth_client_id = admin.create_client(
  705. payload={
  706. "name": "authz-client",
  707. "clientId": "authz-client",
  708. "authorizationServicesEnabled": True,
  709. "serviceAccountsEnabled": True,
  710. }
  711. )
  712. res = admin.get_client_authz_settings(client_id=auth_client_id)
  713. assert res["allowRemoteResourceManagement"]
  714. assert res["decisionStrategy"] == "UNANIMOUS"
  715. assert len(res["policies"]) >= 0
  716. with pytest.raises(KeycloakGetError) as err:
  717. admin.get_client_authz_settings(client_id=client_id)
  718. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  719. # Authz resources
  720. res = admin.get_client_authz_resources(client_id=auth_client_id)
  721. assert len(res) == 1
  722. assert res[0]["name"] == "Default Resource"
  723. with pytest.raises(KeycloakGetError) as err:
  724. admin.get_client_authz_resources(client_id=client_id)
  725. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  726. res = admin.create_client_authz_resource(
  727. client_id=auth_client_id, payload={"name": "test-resource"}
  728. )
  729. assert res["name"] == "test-resource", res
  730. test_resource_id = res["_id"]
  731. res = admin.get_client_authz_resource(client_id=auth_client_id, resource_id=test_resource_id)
  732. assert res["_id"] == test_resource_id, res
  733. assert res["name"] == "test-resource", res
  734. with pytest.raises(KeycloakPostError) as err:
  735. admin.create_client_authz_resource(
  736. client_id=auth_client_id, payload={"name": "test-resource"}
  737. )
  738. assert err.match('409: b\'{"error":"invalid_request"')
  739. assert admin.create_client_authz_resource(
  740. client_id=auth_client_id, payload={"name": "test-resource"}, skip_exists=True
  741. ) == {"msg": "Already exists"}
  742. res = admin.get_client_authz_resources(client_id=auth_client_id)
  743. assert len(res) == 2
  744. assert {x["name"] for x in res} == {"Default Resource", "test-resource"}
  745. res = admin.create_client_authz_resource(
  746. client_id=auth_client_id, payload={"name": "temp-resource"}
  747. )
  748. assert res["name"] == "temp-resource", res
  749. temp_resource_id: str = res["_id"]
  750. # Test update authz resources
  751. admin.update_client_authz_resource(
  752. client_id=auth_client_id,
  753. resource_id=temp_resource_id,
  754. payload={"name": "temp-updated-resource"},
  755. )
  756. res = admin.get_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  757. assert res["name"] == "temp-updated-resource", res
  758. with pytest.raises(KeycloakPutError) as err:
  759. admin.update_client_authz_resource(
  760. client_id=auth_client_id,
  761. resource_id="invalid_resource_id",
  762. payload={"name": "temp-updated-resource"},
  763. )
  764. assert err.match("404: b''"), err
  765. admin.delete_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  766. with pytest.raises(KeycloakGetError) as err:
  767. admin.get_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  768. assert err.match("404: b''")
  769. # Authz policies
  770. res = admin.get_client_authz_policies(client_id=auth_client_id)
  771. assert len(res) == 1, res
  772. assert res[0]["name"] == "Default Policy"
  773. with pytest.raises(KeycloakGetError) as err:
  774. admin.get_client_authz_policies(client_id="does-not-exist")
  775. assert err.match('404: b\'{"error":"Could not find client"}\'')
  776. role_id = admin.get_realm_role(role_name="offline_access")["id"]
  777. res = admin.create_client_authz_role_based_policy(
  778. client_id=auth_client_id,
  779. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  780. )
  781. assert res["name"] == "test-authz-rb-policy", res
  782. with pytest.raises(KeycloakPostError) as err:
  783. admin.create_client_authz_role_based_policy(
  784. client_id=auth_client_id,
  785. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  786. )
  787. assert err.match('409: b\'{"error":"Policy with name')
  788. assert admin.create_client_authz_role_based_policy(
  789. client_id=auth_client_id,
  790. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  791. skip_exists=True,
  792. ) == {"msg": "Already exists"}
  793. assert len(admin.get_client_authz_policies(client_id=auth_client_id)) == 2
  794. res = admin.create_client_authz_role_based_policy(
  795. client_id=auth_client_id,
  796. payload={"name": "test-authz-rb-policy-delete", "roles": [{"id": role_id}]},
  797. )
  798. res2 = admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  799. assert res["id"] == res2["id"]
  800. admin.delete_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  801. with pytest.raises(KeycloakGetError) as err:
  802. admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  803. assert err.match("404: b''")
  804. res = admin.create_client_authz_policy(
  805. client_id=auth_client_id,
  806. payload={
  807. "name": "test-authz-policy",
  808. "type": "time",
  809. "config": {"hourEnd": "18", "hour": "9"},
  810. },
  811. )
  812. assert res["name"] == "test-authz-policy", res
  813. with pytest.raises(KeycloakPostError) as err:
  814. admin.create_client_authz_policy(
  815. client_id=auth_client_id,
  816. payload={
  817. "name": "test-authz-policy",
  818. "type": "time",
  819. "config": {"hourEnd": "18", "hour": "9"},
  820. },
  821. )
  822. assert err.match('409: b\'{"error":"Policy with name')
  823. assert admin.create_client_authz_policy(
  824. client_id=auth_client_id,
  825. payload={
  826. "name": "test-authz-policy",
  827. "type": "time",
  828. "config": {"hourEnd": "18", "hour": "9"},
  829. },
  830. skip_exists=True,
  831. ) == {"msg": "Already exists"}
  832. assert len(admin.get_client_authz_policies(client_id=auth_client_id)) == 3
  833. # Test authz permissions
  834. res = admin.get_client_authz_permissions(client_id=auth_client_id)
  835. assert len(res) == 1, res
  836. assert res[0]["name"] == "Default Permission"
  837. with pytest.raises(KeycloakGetError) as err:
  838. admin.get_client_authz_permissions(client_id="does-not-exist")
  839. assert err.match('404: b\'{"error":"Could not find client"}\'')
  840. res = admin.create_client_authz_resource_based_permission(
  841. client_id=auth_client_id,
  842. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  843. )
  844. assert res, res
  845. assert res["name"] == "test-permission-rb"
  846. assert res["resources"] == [test_resource_id]
  847. with pytest.raises(KeycloakPostError) as err:
  848. admin.create_client_authz_resource_based_permission(
  849. client_id=auth_client_id,
  850. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  851. )
  852. assert err.match('409: b\'{"error":"Policy with name')
  853. assert admin.create_client_authz_resource_based_permission(
  854. client_id=auth_client_id,
  855. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  856. skip_exists=True,
  857. ) == {"msg": "Already exists"}
  858. assert len(admin.get_client_authz_permissions(client_id=auth_client_id)) == 2
  859. # Test authz scopes
  860. res = admin.get_client_authz_scopes(client_id=auth_client_id)
  861. assert len(res) == 0, res
  862. with pytest.raises(KeycloakGetError) as err:
  863. admin.get_client_authz_scopes(client_id=client_id)
  864. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  865. res = admin.create_client_authz_scopes(
  866. client_id=auth_client_id, payload={"name": "test-authz-scope"}
  867. )
  868. assert res["name"] == "test-authz-scope", res
  869. with pytest.raises(KeycloakPostError) as err:
  870. admin.create_client_authz_scopes(
  871. client_id="invalid_client_id", payload={"name": "test-authz-scope"}
  872. )
  873. assert err.match('404: b\'{"error":"Could not find client"')
  874. assert admin.create_client_authz_scopes(
  875. client_id=auth_client_id, payload={"name": "test-authz-scope"}
  876. )
  877. res = admin.get_client_authz_scopes(client_id=auth_client_id)
  878. assert len(res) == 1
  879. assert {x["name"] for x in res} == {"test-authz-scope"}
  880. # Test service account user
  881. res = admin.get_client_service_account_user(client_id=auth_client_id)
  882. assert res["username"] == "service-account-authz-client", res
  883. with pytest.raises(KeycloakGetError) as err:
  884. admin.get_client_service_account_user(client_id=client_id)
  885. assert err.match('400: b\'{"error":"unknown_error"}\'')
  886. # Test delete client
  887. res = admin.delete_client(client_id=auth_client_id)
  888. assert res == dict(), res
  889. with pytest.raises(KeycloakDeleteError) as err:
  890. admin.delete_client(client_id=auth_client_id)
  891. assert err.match('404: b\'{"error":"Could not find client"}\'')
  892. # Test client credentials
  893. admin.create_client(
  894. payload={
  895. "name": "test-confidential",
  896. "enabled": True,
  897. "protocol": "openid-connect",
  898. "publicClient": False,
  899. "redirectUris": ["http://localhost/*"],
  900. "webOrigins": ["+"],
  901. "clientId": "test-confidential",
  902. "secret": "test-secret",
  903. "clientAuthenticatorType": "client-secret",
  904. }
  905. )
  906. with pytest.raises(KeycloakGetError) as err:
  907. admin.get_client_secrets(client_id="does-not-exist")
  908. assert err.match('404: b\'{"error":"Could not find client"}\'')
  909. secrets = admin.get_client_secrets(
  910. client_id=admin.get_client_id(client_id="test-confidential")
  911. )
  912. assert secrets == {"type": "secret", "value": "test-secret"}
  913. with pytest.raises(KeycloakPostError) as err:
  914. admin.generate_client_secrets(client_id="does-not-exist")
  915. assert err.match('404: b\'{"error":"Could not find client"}\'')
  916. res = admin.generate_client_secrets(
  917. client_id=admin.get_client_id(client_id="test-confidential")
  918. )
  919. assert res
  920. assert (
  921. admin.get_client_secrets(client_id=admin.get_client_id(client_id="test-confidential"))
  922. == res
  923. )
  924. def test_realm_roles(admin: KeycloakAdmin, realm: str):
  925. """Test realm roles.
  926. :param admin: Keycloak Admin client
  927. :type admin: KeycloakAdmin
  928. :param realm: Keycloak realm
  929. :type realm: str
  930. """
  931. admin.change_current_realm(realm)
  932. # Test get realm roles
  933. roles = admin.get_realm_roles()
  934. assert len(roles) == 3, roles
  935. role_names = [x["name"] for x in roles]
  936. assert "uma_authorization" in role_names, role_names
  937. assert "offline_access" in role_names, role_names
  938. # Test get realm roles with search text
  939. searched_roles = admin.get_realm_roles(search_text="uma_a")
  940. searched_role_names = [x["name"] for x in searched_roles]
  941. assert "uma_authorization" in searched_role_names, searched_role_names
  942. assert "offline_access" not in searched_role_names, searched_role_names
  943. # Test empty members
  944. with pytest.raises(KeycloakGetError) as err:
  945. admin.get_realm_role_members(role_name="does-not-exist")
  946. assert err.match('404: b\'{"error":"Could not find role"}\'')
  947. members = admin.get_realm_role_members(role_name="offline_access")
  948. assert members == list(), members
  949. # Test create realm role
  950. role_id = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  951. assert role_id, role_id
  952. with pytest.raises(KeycloakPostError) as err:
  953. admin.create_realm_role(payload={"name": "test-realm-role"})
  954. assert err.match('409: b\'{"errorMessage":"Role with name test-realm-role already exists"}\'')
  955. role_id_2 = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  956. assert role_id == role_id_2
  957. # Test get realm role by its id
  958. role_id = admin.get_realm_role(role_name="test-realm-role")["id"]
  959. res = admin.get_realm_role_by_id(role_id)
  960. assert res["name"] == "test-realm-role"
  961. # Test update realm role
  962. res = admin.update_realm_role(
  963. role_name="test-realm-role", payload={"name": "test-realm-role-update"}
  964. )
  965. assert res == dict(), res
  966. with pytest.raises(KeycloakPutError) as err:
  967. admin.update_realm_role(
  968. role_name="test-realm-role", payload={"name": "test-realm-role-update"}
  969. )
  970. assert err.match('404: b\'{"error":"Could not find role"}\''), err
  971. # Test realm role user assignment
  972. user_id = admin.create_user(payload={"username": "role-testing", "email": "test@test.test"})
  973. with pytest.raises(KeycloakPostError) as err:
  974. admin.assign_realm_roles(user_id=user_id, roles=["bad"])
  975. assert err.match('b\'{"error":"unknown_error"}\''), err
  976. res = admin.assign_realm_roles(
  977. user_id=user_id,
  978. roles=[
  979. admin.get_realm_role(role_name="offline_access"),
  980. admin.get_realm_role(role_name="test-realm-role-update"),
  981. ],
  982. )
  983. assert res == dict(), res
  984. assert admin.get_user(user_id=user_id)["username"] in [
  985. x["username"] for x in admin.get_realm_role_members(role_name="offline_access")
  986. ]
  987. assert admin.get_user(user_id=user_id)["username"] in [
  988. x["username"] for x in admin.get_realm_role_members(role_name="test-realm-role-update")
  989. ]
  990. roles = admin.get_realm_roles_of_user(user_id=user_id)
  991. assert len(roles) == 3
  992. assert "offline_access" in [x["name"] for x in roles]
  993. assert "test-realm-role-update" in [x["name"] for x in roles]
  994. with pytest.raises(KeycloakDeleteError) as err:
  995. admin.delete_realm_roles_of_user(user_id=user_id, roles=["bad"])
  996. assert err.match('b\'{"error":"unknown_error"}\''), err
  997. res = admin.delete_realm_roles_of_user(
  998. user_id=user_id, roles=[admin.get_realm_role(role_name="offline_access")]
  999. )
  1000. assert res == dict(), res
  1001. assert admin.get_realm_role_members(role_name="offline_access") == list()
  1002. roles = admin.get_realm_roles_of_user(user_id=user_id)
  1003. assert len(roles) == 2
  1004. assert "offline_access" not in [x["name"] for x in roles]
  1005. assert "test-realm-role-update" in [x["name"] for x in roles]
  1006. roles = admin.get_available_realm_roles_of_user(user_id=user_id)
  1007. assert len(roles) == 2
  1008. assert "offline_access" in [x["name"] for x in roles]
  1009. assert "uma_authorization" in [x["name"] for x in roles]
  1010. # Test realm role group assignment
  1011. group_id = admin.create_group(payload={"name": "test-group"})
  1012. with pytest.raises(KeycloakPostError) as err:
  1013. admin.assign_group_realm_roles(group_id=group_id, roles=["bad"])
  1014. assert err.match('b\'{"error":"unknown_error"}\''), err
  1015. res = admin.assign_group_realm_roles(
  1016. group_id=group_id,
  1017. roles=[
  1018. admin.get_realm_role(role_name="offline_access"),
  1019. admin.get_realm_role(role_name="test-realm-role-update"),
  1020. ],
  1021. )
  1022. assert res == dict(), res
  1023. roles = admin.get_group_realm_roles(group_id=group_id)
  1024. assert len(roles) == 2
  1025. assert "offline_access" in [x["name"] for x in roles]
  1026. assert "test-realm-role-update" in [x["name"] for x in roles]
  1027. with pytest.raises(KeycloakDeleteError) as err:
  1028. admin.delete_group_realm_roles(group_id=group_id, roles=["bad"])
  1029. assert err.match('b\'{"error":"unknown_error"}\''), err
  1030. res = admin.delete_group_realm_roles(
  1031. group_id=group_id, roles=[admin.get_realm_role(role_name="offline_access")]
  1032. )
  1033. assert res == dict(), res
  1034. roles = admin.get_group_realm_roles(group_id=group_id)
  1035. assert len(roles) == 1
  1036. assert "test-realm-role-update" in [x["name"] for x in roles]
  1037. # Test composite realm roles
  1038. composite_role = admin.create_realm_role(payload={"name": "test-composite-role"})
  1039. with pytest.raises(KeycloakPostError) as err:
  1040. admin.add_composite_realm_roles_to_role(role_name=composite_role, roles=["bad"])
  1041. assert err.match('b\'{"error":"unknown_error"}\''), err
  1042. res = admin.add_composite_realm_roles_to_role(
  1043. role_name=composite_role, roles=[admin.get_realm_role(role_name="test-realm-role-update")]
  1044. )
  1045. assert res == dict(), res
  1046. res = admin.get_composite_realm_roles_of_role(role_name=composite_role)
  1047. assert len(res) == 1
  1048. assert "test-realm-role-update" in res[0]["name"]
  1049. with pytest.raises(KeycloakGetError) as err:
  1050. admin.get_composite_realm_roles_of_role(role_name="bad")
  1051. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1052. res = admin.get_composite_realm_roles_of_user(user_id=user_id)
  1053. assert len(res) == 4
  1054. assert "offline_access" in {x["name"] for x in res}
  1055. assert "test-realm-role-update" in {x["name"] for x in res}
  1056. assert "uma_authorization" in {x["name"] for x in res}
  1057. with pytest.raises(KeycloakGetError) as err:
  1058. admin.get_composite_realm_roles_of_user(user_id="bad")
  1059. assert err.match('b\'{"error":"User not found"}\''), err
  1060. with pytest.raises(KeycloakDeleteError) as err:
  1061. admin.remove_composite_realm_roles_to_role(role_name=composite_role, roles=["bad"])
  1062. assert err.match('b\'{"error":"unknown_error"}\''), err
  1063. res = admin.remove_composite_realm_roles_to_role(
  1064. role_name=composite_role, roles=[admin.get_realm_role(role_name="test-realm-role-update")]
  1065. )
  1066. assert res == dict(), res
  1067. res = admin.get_composite_realm_roles_of_role(role_name=composite_role)
  1068. assert len(res) == 0
  1069. # Test realm role group list
  1070. res = admin.get_realm_role_groups(role_name="test-realm-role-update")
  1071. assert len(res) == 1
  1072. assert res[0]["id"] == group_id
  1073. with pytest.raises(KeycloakGetError) as err:
  1074. admin.get_realm_role_groups(role_name="non-existent-role")
  1075. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1076. # Test delete realm role
  1077. res = admin.delete_realm_role(role_name=composite_role)
  1078. assert res == dict(), res
  1079. with pytest.raises(KeycloakDeleteError) as err:
  1080. admin.delete_realm_role(role_name=composite_role)
  1081. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1082. @pytest.mark.parametrize(
  1083. "testcase, arg_brief_repr, includes_attributes",
  1084. [
  1085. ("brief True", {"brief_representation": True}, False),
  1086. ("brief False", {"brief_representation": False}, True),
  1087. ("default", {}, False),
  1088. ],
  1089. )
  1090. def test_role_attributes(
  1091. admin: KeycloakAdmin,
  1092. realm: str,
  1093. client: str,
  1094. arg_brief_repr: dict,
  1095. includes_attributes: bool,
  1096. testcase: str,
  1097. ):
  1098. """Test getting role attributes for bulk calls.
  1099. :param admin: Keycloak admin
  1100. :type admin: KeycloakAdmin
  1101. :param realm: Keycloak realm
  1102. :type realm: str
  1103. :param client: Keycloak client
  1104. :type client: str
  1105. :param arg_brief_repr: Brief representation
  1106. :type arg_brief_repr: dict
  1107. :param includes_attributes: Indicator whether to include attributes
  1108. :type includes_attributes: bool
  1109. :param testcase: Test case
  1110. :type testcase: str
  1111. """
  1112. # setup
  1113. attribute_role = "test-realm-role-w-attr"
  1114. test_attrs = {"attr1": ["val1"], "attr2": ["val2-1", "val2-2"]}
  1115. role_id = admin.create_realm_role(
  1116. payload={"name": attribute_role, "attributes": test_attrs}, skip_exists=True
  1117. )
  1118. assert role_id, role_id
  1119. cli_role_id = admin.create_client_role(
  1120. client, payload={"name": attribute_role, "attributes": test_attrs}, skip_exists=True
  1121. )
  1122. assert cli_role_id, cli_role_id
  1123. if not includes_attributes:
  1124. test_attrs = None
  1125. # tests
  1126. roles = admin.get_realm_roles(**arg_brief_repr)
  1127. roles_filtered = [role for role in roles if role["name"] == role_id]
  1128. assert roles_filtered, roles_filtered
  1129. role = roles_filtered[0]
  1130. assert role.get("attributes") == test_attrs, testcase
  1131. roles = admin.get_client_roles(client, **arg_brief_repr)
  1132. roles_filtered = [role for role in roles if role["name"] == cli_role_id]
  1133. assert roles_filtered, roles_filtered
  1134. role = roles_filtered[0]
  1135. assert role.get("attributes") == test_attrs, testcase
  1136. # cleanup
  1137. res = admin.delete_realm_role(role_name=attribute_role)
  1138. assert res == dict(), res
  1139. res = admin.delete_client_role(client, role_name=attribute_role)
  1140. assert res == dict(), res
  1141. def test_client_scope_realm_roles(admin: KeycloakAdmin, realm: str):
  1142. """Test client realm roles.
  1143. :param admin: Keycloak admin
  1144. :type admin: KeycloakAdmin
  1145. :param realm: Keycloak realm
  1146. :type realm: str
  1147. """
  1148. admin.change_current_realm(realm)
  1149. # Test get realm roles
  1150. roles = admin.get_realm_roles()
  1151. assert len(roles) == 3, roles
  1152. role_names = [x["name"] for x in roles]
  1153. assert "uma_authorization" in role_names, role_names
  1154. assert "offline_access" in role_names, role_names
  1155. # create realm role for test
  1156. role_id = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  1157. assert role_id, role_id
  1158. # Test realm role client assignment
  1159. client_id = admin.create_client(
  1160. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1161. )
  1162. with pytest.raises(KeycloakPostError) as err:
  1163. admin.assign_realm_roles_to_client_scope(client_id=client_id, roles=["bad"])
  1164. assert err.match('b\'{"error":"unknown_error"}\''), err
  1165. res = admin.assign_realm_roles_to_client_scope(
  1166. client_id=client_id,
  1167. roles=[
  1168. admin.get_realm_role(role_name="offline_access"),
  1169. admin.get_realm_role(role_name="test-realm-role"),
  1170. ],
  1171. )
  1172. assert res == dict(), res
  1173. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1174. assert len(roles) == 2
  1175. client_role_names = [x["name"] for x in roles]
  1176. assert "offline_access" in client_role_names, client_role_names
  1177. assert "test-realm-role" in client_role_names, client_role_names
  1178. assert "uma_authorization" not in client_role_names, client_role_names
  1179. # Test remove realm role of client
  1180. with pytest.raises(KeycloakDeleteError) as err:
  1181. admin.delete_realm_roles_of_client_scope(client_id=client_id, roles=["bad"])
  1182. assert err.match('b\'{"error":"unknown_error"}\''), err
  1183. res = admin.delete_realm_roles_of_client_scope(
  1184. client_id=client_id, roles=[admin.get_realm_role(role_name="offline_access")]
  1185. )
  1186. assert res == dict(), res
  1187. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1188. assert len(roles) == 1
  1189. assert "test-realm-role" in [x["name"] for x in roles]
  1190. res = admin.delete_realm_roles_of_client_scope(
  1191. client_id=client_id, roles=[admin.get_realm_role(role_name="test-realm-role")]
  1192. )
  1193. assert res == dict(), res
  1194. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1195. assert len(roles) == 0
  1196. def test_client_scope_client_roles(admin: KeycloakAdmin, realm: str, client: str):
  1197. """Test client assignment of other client roles.
  1198. :param admin: Keycloak admin
  1199. :type admin: KeycloakAdmin
  1200. :param realm: Keycloak realm
  1201. :type realm: str
  1202. :param client: Keycloak client
  1203. :type client: str
  1204. """
  1205. admin.change_current_realm(realm)
  1206. client_id = admin.create_client(
  1207. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1208. )
  1209. # Test get client roles
  1210. roles = admin.get_client_roles_of_client_scope(client_id, client)
  1211. assert len(roles) == 0, roles
  1212. # create client role for test
  1213. client_role_id = admin.create_client_role(
  1214. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1215. )
  1216. assert client_role_id, client_role_id
  1217. # Test client role assignment to other client
  1218. with pytest.raises(KeycloakPostError) as err:
  1219. admin.assign_client_roles_to_client_scope(
  1220. client_id=client_id, client_roles_owner_id=client, roles=["bad"]
  1221. )
  1222. assert err.match('b\'{"error":"unknown_error"}\''), err
  1223. res = admin.assign_client_roles_to_client_scope(
  1224. client_id=client_id,
  1225. client_roles_owner_id=client,
  1226. roles=[admin.get_client_role(client_id=client, role_name="client-role-test")],
  1227. )
  1228. assert res == dict(), res
  1229. roles = admin.get_client_roles_of_client_scope(
  1230. client_id=client_id, client_roles_owner_id=client
  1231. )
  1232. assert len(roles) == 1
  1233. client_role_names = [x["name"] for x in roles]
  1234. assert "client-role-test" in client_role_names, client_role_names
  1235. # Test remove realm role of client
  1236. with pytest.raises(KeycloakDeleteError) as err:
  1237. admin.delete_client_roles_of_client_scope(
  1238. client_id=client_id, client_roles_owner_id=client, roles=["bad"]
  1239. )
  1240. assert err.match('b\'{"error":"unknown_error"}\''), err
  1241. res = admin.delete_client_roles_of_client_scope(
  1242. client_id=client_id,
  1243. client_roles_owner_id=client,
  1244. roles=[admin.get_client_role(client_id=client, role_name="client-role-test")],
  1245. )
  1246. assert res == dict(), res
  1247. roles = admin.get_client_roles_of_client_scope(
  1248. client_id=client_id, client_roles_owner_id=client
  1249. )
  1250. assert len(roles) == 0
  1251. def test_client_default_client_scopes(admin: KeycloakAdmin, realm: str, client: str):
  1252. """Test client assignment of default client scopes.
  1253. :param admin: Keycloak admin
  1254. :type admin: KeycloakAdmin
  1255. :param realm: Keycloak realm
  1256. :type realm: str
  1257. :param client: Keycloak client
  1258. :type client: str
  1259. """
  1260. admin.change_current_realm(realm)
  1261. client_id = admin.create_client(
  1262. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1263. )
  1264. # Test get client default scopes
  1265. # keycloak default roles: web-origins, acr, profile, roles, email
  1266. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1267. assert len(default_client_scopes) == 5, default_client_scopes
  1268. # Test add a client scope to client default scopes
  1269. default_client_scope = "test-client-default-scope"
  1270. new_client_scope = {
  1271. "name": default_client_scope,
  1272. "description": f"Test Client Scope: {default_client_scope}",
  1273. "protocol": "openid-connect",
  1274. "attributes": {},
  1275. }
  1276. new_client_scope_id = admin.create_client_scope(new_client_scope, skip_exists=False)
  1277. new_default_client_scope_data = {
  1278. "realm": realm,
  1279. "client": client_id,
  1280. "clientScopeId": new_client_scope_id,
  1281. }
  1282. admin.add_client_default_client_scope(
  1283. client_id, new_client_scope_id, new_default_client_scope_data
  1284. )
  1285. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1286. assert len(default_client_scopes) == 6, default_client_scopes
  1287. # Test remove a client default scope
  1288. admin.delete_client_default_client_scope(client_id, new_client_scope_id)
  1289. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1290. assert len(default_client_scopes) == 5, default_client_scopes
  1291. def test_client_optional_client_scopes(admin: KeycloakAdmin, realm: str, client: str):
  1292. """Test client assignment of optional client scopes.
  1293. :param admin: Keycloak admin
  1294. :type admin: KeycloakAdmin
  1295. :param realm: Keycloak realm
  1296. :type realm: str
  1297. :param client: Keycloak client
  1298. :type client: str
  1299. """
  1300. admin.change_current_realm(realm)
  1301. client_id = admin.create_client(
  1302. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1303. )
  1304. # Test get client optional scopes
  1305. # keycloak optional roles: microprofile-jwt, offline_access, address, phone
  1306. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1307. assert len(optional_client_scopes) == 4, optional_client_scopes
  1308. # Test add a client scope to client optional scopes
  1309. optional_client_scope = "test-client-optional-scope"
  1310. new_client_scope = {
  1311. "name": optional_client_scope,
  1312. "description": f"Test Client Scope: {optional_client_scope}",
  1313. "protocol": "openid-connect",
  1314. "attributes": {},
  1315. }
  1316. new_client_scope_id = admin.create_client_scope(new_client_scope, skip_exists=False)
  1317. new_optional_client_scope_data = {
  1318. "realm": realm,
  1319. "client": client_id,
  1320. "clientScopeId": new_client_scope_id,
  1321. }
  1322. admin.add_client_optional_client_scope(
  1323. client_id, new_client_scope_id, new_optional_client_scope_data
  1324. )
  1325. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1326. assert len(optional_client_scopes) == 5, optional_client_scopes
  1327. # Test remove a client optional scope
  1328. admin.delete_client_optional_client_scope(client_id, new_client_scope_id)
  1329. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1330. assert len(optional_client_scopes) == 4, optional_client_scopes
  1331. def test_client_roles(admin: KeycloakAdmin, client: str):
  1332. """Test client roles.
  1333. :param admin: Keycloak Admin client
  1334. :type admin: KeycloakAdmin
  1335. :param client: Keycloak client
  1336. :type client: str
  1337. """
  1338. # Test get client roles
  1339. res = admin.get_client_roles(client_id=client)
  1340. assert len(res) == 0
  1341. with pytest.raises(KeycloakGetError) as err:
  1342. admin.get_client_roles(client_id="bad")
  1343. assert err.match('404: b\'{"error":"Could not find client"}\'')
  1344. # Test create client role
  1345. client_role_id = admin.create_client_role(
  1346. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1347. )
  1348. with pytest.raises(KeycloakPostError) as err:
  1349. admin.create_client_role(client_role_id=client, payload={"name": "client-role-test"})
  1350. assert err.match('409: b\'{"errorMessage":"Role with name client-role-test already exists"}\'')
  1351. client_role_id_2 = admin.create_client_role(
  1352. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1353. )
  1354. assert client_role_id == client_role_id_2
  1355. # Test get client role
  1356. res = admin.get_client_role(client_id=client, role_name="client-role-test")
  1357. assert res["name"] == client_role_id
  1358. with pytest.raises(KeycloakGetError) as err:
  1359. admin.get_client_role(client_id=client, role_name="bad")
  1360. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1361. res_ = admin.get_client_role_id(client_id=client, role_name="client-role-test")
  1362. assert res_ == res["id"]
  1363. with pytest.raises(KeycloakGetError) as err:
  1364. admin.get_client_role_id(client_id=client, role_name="bad")
  1365. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1366. assert len(admin.get_client_roles(client_id=client)) == 1
  1367. # Test update client role
  1368. res = admin.update_client_role(
  1369. client_role_id=client,
  1370. role_name="client-role-test",
  1371. payload={"name": "client-role-test-update"},
  1372. )
  1373. assert res == dict()
  1374. with pytest.raises(KeycloakPutError) as err:
  1375. res = admin.update_client_role(
  1376. client_role_id=client,
  1377. role_name="client-role-test",
  1378. payload={"name": "client-role-test-update"},
  1379. )
  1380. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1381. # Test user with client role
  1382. res = admin.get_client_role_members(client_id=client, role_name="client-role-test-update")
  1383. assert len(res) == 0
  1384. with pytest.raises(KeycloakGetError) as err:
  1385. admin.get_client_role_members(client_id=client, role_name="bad")
  1386. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1387. user_id = admin.create_user(payload={"username": "test", "email": "test@test.test"})
  1388. with pytest.raises(KeycloakPostError) as err:
  1389. admin.assign_client_role(user_id=user_id, client_id=client, roles=["bad"])
  1390. assert err.match('b\'{"error":"unknown_error"}\''), err
  1391. res = admin.assign_client_role(
  1392. user_id=user_id,
  1393. client_id=client,
  1394. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1395. )
  1396. assert res == dict()
  1397. assert (
  1398. len(admin.get_client_role_members(client_id=client, role_name="client-role-test-update"))
  1399. == 1
  1400. )
  1401. roles = admin.get_client_roles_of_user(user_id=user_id, client_id=client)
  1402. assert len(roles) == 1, roles
  1403. with pytest.raises(KeycloakGetError) as err:
  1404. admin.get_client_roles_of_user(user_id=user_id, client_id="bad")
  1405. assert err.match('404: b\'{"error":"Client not found"}\'')
  1406. roles = admin.get_composite_client_roles_of_user(user_id=user_id, client_id=client)
  1407. assert len(roles) == 1, roles
  1408. with pytest.raises(KeycloakGetError) as err:
  1409. admin.get_composite_client_roles_of_user(user_id=user_id, client_id="bad")
  1410. assert err.match('404: b\'{"error":"Client not found"}\'')
  1411. roles = admin.get_available_client_roles_of_user(user_id=user_id, client_id=client)
  1412. assert len(roles) == 0, roles
  1413. with pytest.raises(KeycloakGetError) as err:
  1414. admin.get_composite_client_roles_of_user(user_id=user_id, client_id="bad")
  1415. assert err.match('404: b\'{"error":"Client not found"}\'')
  1416. with pytest.raises(KeycloakDeleteError) as err:
  1417. admin.delete_client_roles_of_user(user_id=user_id, client_id=client, roles=["bad"])
  1418. assert err.match('b\'{"error":"unknown_error"}\''), err
  1419. admin.delete_client_roles_of_user(
  1420. user_id=user_id,
  1421. client_id=client,
  1422. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1423. )
  1424. assert len(admin.get_client_roles_of_user(user_id=user_id, client_id=client)) == 0
  1425. # Test groups and client roles
  1426. res = admin.get_client_role_groups(client_id=client, role_name="client-role-test-update")
  1427. assert len(res) == 0
  1428. with pytest.raises(KeycloakGetError) as err:
  1429. admin.get_client_role_groups(client_id=client, role_name="bad")
  1430. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1431. group_id = admin.create_group(payload={"name": "test-group"})
  1432. res = admin.get_group_client_roles(group_id=group_id, client_id=client)
  1433. assert len(res) == 0
  1434. with pytest.raises(KeycloakGetError) as err:
  1435. admin.get_group_client_roles(group_id=group_id, client_id="bad")
  1436. assert err.match('404: b\'{"error":"Client not found"}\'')
  1437. with pytest.raises(KeycloakPostError) as err:
  1438. admin.assign_group_client_roles(group_id=group_id, client_id=client, roles=["bad"])
  1439. assert err.match('b\'{"error":"unknown_error"}\''), err
  1440. res = admin.assign_group_client_roles(
  1441. group_id=group_id,
  1442. client_id=client,
  1443. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1444. )
  1445. assert res == dict()
  1446. assert (
  1447. len(admin.get_client_role_groups(client_id=client, role_name="client-role-test-update"))
  1448. == 1
  1449. )
  1450. assert len(admin.get_group_client_roles(group_id=group_id, client_id=client)) == 1
  1451. with pytest.raises(KeycloakDeleteError) as err:
  1452. admin.delete_group_client_roles(group_id=group_id, client_id=client, roles=["bad"])
  1453. assert err.match('b\'{"error":"unknown_error"}\''), err
  1454. res = admin.delete_group_client_roles(
  1455. group_id=group_id,
  1456. client_id=client,
  1457. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1458. )
  1459. assert res == dict()
  1460. # Test composite client roles
  1461. with pytest.raises(KeycloakPostError) as err:
  1462. admin.add_composite_client_roles_to_role(
  1463. client_role_id=client, role_name="client-role-test-update", roles=["bad"]
  1464. )
  1465. assert err.match('b\'{"error":"unknown_error"}\''), err
  1466. res = admin.add_composite_client_roles_to_role(
  1467. client_role_id=client,
  1468. role_name="client-role-test-update",
  1469. roles=[admin.get_realm_role(role_name="offline_access")],
  1470. )
  1471. assert res == dict()
  1472. assert admin.get_client_role(client_id=client, role_name="client-role-test-update")[
  1473. "composite"
  1474. ]
  1475. # Test delete of client role
  1476. res = admin.delete_client_role(client_role_id=client, role_name="client-role-test-update")
  1477. assert res == dict()
  1478. with pytest.raises(KeycloakDeleteError) as err:
  1479. admin.delete_client_role(client_role_id=client, role_name="client-role-test-update")
  1480. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1481. def test_enable_token_exchange(admin: KeycloakAdmin, realm: str):
  1482. """Test enable token exchange.
  1483. :param admin: Keycloak Admin client
  1484. :type admin: KeycloakAdmin
  1485. :param realm: Keycloak realm
  1486. :type realm: str
  1487. :raises AssertionError: In case of bad configuration
  1488. """
  1489. # Test enabling token exchange between two confidential clients
  1490. admin.change_current_realm(realm)
  1491. # Create test clients
  1492. source_client_id = admin.create_client(
  1493. payload={"name": "Source Client", "clientId": "source-client"}
  1494. )
  1495. target_client_id = admin.create_client(
  1496. payload={"name": "Target Client", "clientId": "target-client"}
  1497. )
  1498. for c in admin.get_clients():
  1499. if c["clientId"] == "realm-management":
  1500. realm_management_id = c["id"]
  1501. break
  1502. else:
  1503. raise AssertionError("Missing realm management client")
  1504. # Enable permissions on the Superset client
  1505. admin.update_client_management_permissions(
  1506. payload={"enabled": True}, client_id=target_client_id
  1507. )
  1508. # Fetch various IDs and strings needed when creating the permission
  1509. token_exchange_permission_id = admin.get_client_management_permissions(
  1510. client_id=target_client_id
  1511. )["scopePermissions"]["token-exchange"]
  1512. scopes = admin.get_client_authz_policy_scopes(
  1513. client_id=realm_management_id, policy_id=token_exchange_permission_id
  1514. )
  1515. for s in scopes:
  1516. if s["name"] == "token-exchange":
  1517. token_exchange_scope_id = s["id"]
  1518. break
  1519. else:
  1520. raise AssertionError("Missing token-exchange scope")
  1521. resources = admin.get_client_authz_policy_resources(
  1522. client_id=realm_management_id, policy_id=token_exchange_permission_id
  1523. )
  1524. for r in resources:
  1525. if r["name"] == f"client.resource.{target_client_id}":
  1526. token_exchange_resource_id = r["_id"]
  1527. break
  1528. else:
  1529. raise AssertionError("Missing client resource")
  1530. # Create a client policy for source client
  1531. policy_name = "Exchange source client token with target client token"
  1532. client_policy_id = admin.create_client_authz_client_policy(
  1533. payload={
  1534. "type": "client",
  1535. "logic": "POSITIVE",
  1536. "decisionStrategy": "UNANIMOUS",
  1537. "name": policy_name,
  1538. "clients": [source_client_id],
  1539. },
  1540. client_id=realm_management_id,
  1541. )["id"]
  1542. policies = admin.get_client_authz_client_policies(client_id=realm_management_id)
  1543. for policy in policies:
  1544. if policy["name"] == policy_name:
  1545. assert policy["clients"] == [source_client_id]
  1546. break
  1547. else:
  1548. raise AssertionError("Missing client policy")
  1549. # Update permissions on the target client to reference this policy
  1550. permission_name = admin.get_client_authz_scope_permission(
  1551. client_id=realm_management_id, scope_id=token_exchange_permission_id
  1552. )["name"]
  1553. admin.update_client_authz_scope_permission(
  1554. payload={
  1555. "id": token_exchange_permission_id,
  1556. "name": permission_name,
  1557. "type": "scope",
  1558. "logic": "POSITIVE",
  1559. "decisionStrategy": "UNANIMOUS",
  1560. "resources": [token_exchange_resource_id],
  1561. "scopes": [token_exchange_scope_id],
  1562. "policies": [client_policy_id],
  1563. },
  1564. client_id=realm_management_id,
  1565. scope_id=token_exchange_permission_id,
  1566. )
  1567. # Create permissions on the target client to reference this policy
  1568. admin.create_client_authz_scope_permission(
  1569. payload={
  1570. "id": "some-id",
  1571. "name": "test-permission",
  1572. "type": "scope",
  1573. "logic": "POSITIVE",
  1574. "decisionStrategy": "UNANIMOUS",
  1575. "resources": [token_exchange_resource_id],
  1576. "scopes": [token_exchange_scope_id],
  1577. "policies": [client_policy_id],
  1578. },
  1579. client_id=realm_management_id,
  1580. )
  1581. permission_name = admin.get_client_authz_scope_permission(
  1582. client_id=realm_management_id, scope_id=token_exchange_permission_id
  1583. )["name"]
  1584. assert permission_name.startswith("token-exchange.permission.client.")
  1585. with pytest.raises(KeycloakPostError) as err:
  1586. admin.create_client_authz_scope_permission(
  1587. payload={"name": "test-permission", "scopes": [token_exchange_scope_id]},
  1588. client_id="realm_management_id",
  1589. )
  1590. assert err.match('404: b\'{"error":"Could not find client"}\'')
  1591. def test_email(admin: KeycloakAdmin, user: str):
  1592. """Test email.
  1593. :param admin: Keycloak Admin client
  1594. :type admin: KeycloakAdmin
  1595. :param user: Keycloak user
  1596. :type user: str
  1597. """
  1598. # Emails will fail as we don't have SMTP test setup
  1599. with pytest.raises(KeycloakPutError) as err:
  1600. admin.send_update_account(user_id=user, payload=dict())
  1601. assert err.match('b\'{"error":"unknown_error"}\''), err
  1602. admin.update_user(user_id=user, payload={"enabled": True})
  1603. with pytest.raises(KeycloakPutError) as err:
  1604. admin.send_verify_email(user_id=user)
  1605. assert err.match('500: b\'{"errorMessage":"Failed to send execute actions email"}\'')
  1606. def test_get_sessions(admin: KeycloakAdmin):
  1607. """Test get sessions.
  1608. :param admin: Keycloak Admin client
  1609. :type admin: KeycloakAdmin
  1610. """
  1611. sessions = admin.get_sessions(user_id=admin.get_user_id(username=admin.username))
  1612. assert len(sessions) >= 1
  1613. with pytest.raises(KeycloakGetError) as err:
  1614. admin.get_sessions(user_id="bad")
  1615. assert err.match('404: b\'{"error":"User not found"}\'')
  1616. def test_get_client_installation_provider(admin: KeycloakAdmin, client: str):
  1617. """Test get client installation provider.
  1618. :param admin: Keycloak Admin client
  1619. :type admin: KeycloakAdmin
  1620. :param client: Keycloak client
  1621. :type client: str
  1622. """
  1623. with pytest.raises(KeycloakGetError) as err:
  1624. admin.get_client_installation_provider(client_id=client, provider_id="bad")
  1625. assert err.match('404: b\'{"error":"Unknown Provider"}\'')
  1626. installation = admin.get_client_installation_provider(
  1627. client_id=client, provider_id="keycloak-oidc-keycloak-json"
  1628. )
  1629. assert set(installation.keys()) == {
  1630. "auth-server-url",
  1631. "confidential-port",
  1632. "credentials",
  1633. "realm",
  1634. "resource",
  1635. "ssl-required",
  1636. }
  1637. def test_auth_flows(admin: KeycloakAdmin, realm: str):
  1638. """Test auth flows.
  1639. :param admin: Keycloak Admin client
  1640. :type admin: KeycloakAdmin
  1641. :param realm: Keycloak realm
  1642. :type realm: str
  1643. """
  1644. admin.change_current_realm(realm)
  1645. res = admin.get_authentication_flows()
  1646. assert len(res) <= 8, res
  1647. default_flows = len(res)
  1648. assert {x["alias"] for x in res}.issubset(
  1649. {
  1650. "reset credentials",
  1651. "browser",
  1652. "registration",
  1653. "http challenge",
  1654. "docker auth",
  1655. "direct grant",
  1656. "first broker login",
  1657. "clients",
  1658. }
  1659. )
  1660. assert set(res[0].keys()) == {
  1661. "alias",
  1662. "authenticationExecutions",
  1663. "builtIn",
  1664. "description",
  1665. "id",
  1666. "providerId",
  1667. "topLevel",
  1668. }
  1669. assert {x["alias"] for x in res}.issubset(
  1670. {
  1671. "reset credentials",
  1672. "browser",
  1673. "registration",
  1674. "docker auth",
  1675. "direct grant",
  1676. "first broker login",
  1677. "clients",
  1678. "http challenge",
  1679. }
  1680. )
  1681. with pytest.raises(KeycloakGetError) as err:
  1682. admin.get_authentication_flow_for_id(flow_id="bad")
  1683. assert err.match('404: b\'{"error":"Could not find flow with id"}\'')
  1684. browser_flow_id = [x for x in res if x["alias"] == "browser"][0]["id"]
  1685. res = admin.get_authentication_flow_for_id(flow_id=browser_flow_id)
  1686. assert res["alias"] == "browser"
  1687. # Test copying
  1688. with pytest.raises(KeycloakPostError) as err:
  1689. admin.copy_authentication_flow(payload=dict(), flow_alias="bad")
  1690. assert err.match("404: b''")
  1691. res = admin.copy_authentication_flow(payload={"newName": "test-browser"}, flow_alias="browser")
  1692. assert res == b"", res
  1693. assert len(admin.get_authentication_flows()) == (default_flows + 1)
  1694. # Test create
  1695. res = admin.create_authentication_flow(
  1696. payload={"alias": "test-create", "providerId": "basic-flow"}
  1697. )
  1698. assert res == b""
  1699. with pytest.raises(KeycloakPostError) as err:
  1700. admin.create_authentication_flow(payload={"alias": "test-create", "builtIn": False})
  1701. assert err.match('409: b\'{"errorMessage":"Flow test-create already exists"}\'')
  1702. assert admin.create_authentication_flow(
  1703. payload={"alias": "test-create"}, skip_exists=True
  1704. ) == {"msg": "Already exists"}
  1705. # Test flow executions
  1706. res = admin.get_authentication_flow_executions(flow_alias="browser")
  1707. assert len(res) == 8, res
  1708. with pytest.raises(KeycloakGetError) as err:
  1709. admin.get_authentication_flow_executions(flow_alias="bad")
  1710. assert err.match("404: b''")
  1711. exec_id = res[0]["id"]
  1712. res = admin.get_authentication_flow_execution(execution_id=exec_id)
  1713. assert set(res.keys()) == {
  1714. "alternative",
  1715. "authenticator",
  1716. "authenticatorFlow",
  1717. "conditional",
  1718. "disabled",
  1719. "enabled",
  1720. "id",
  1721. "parentFlow",
  1722. "priority",
  1723. "required",
  1724. "requirement",
  1725. }, res
  1726. with pytest.raises(KeycloakGetError) as err:
  1727. admin.get_authentication_flow_execution(execution_id="bad")
  1728. assert err.match('404: b\'{"error":"Illegal execution"}\'')
  1729. with pytest.raises(KeycloakPostError) as err:
  1730. admin.create_authentication_flow_execution(payload=dict(), flow_alias="browser")
  1731. assert err.match('400: b\'{"error":"It is illegal to add execution to a built in flow"}\'')
  1732. res = admin.create_authentication_flow_execution(
  1733. payload={"provider": "auth-cookie"}, flow_alias="test-create"
  1734. )
  1735. assert res == b""
  1736. assert len(admin.get_authentication_flow_executions(flow_alias="test-create")) == 1
  1737. with pytest.raises(KeycloakPutError) as err:
  1738. admin.update_authentication_flow_executions(
  1739. payload={"required": "yes"}, flow_alias="test-create"
  1740. )
  1741. assert err.match('400: b\'{"error":"Unrecognized field')
  1742. payload = admin.get_authentication_flow_executions(flow_alias="test-create")[0]
  1743. payload["displayName"] = "test"
  1744. res = admin.update_authentication_flow_executions(payload=payload, flow_alias="test-create")
  1745. assert res
  1746. exec_id = admin.get_authentication_flow_executions(flow_alias="test-create")[0]["id"]
  1747. res = admin.delete_authentication_flow_execution(execution_id=exec_id)
  1748. assert res == dict()
  1749. with pytest.raises(KeycloakDeleteError) as err:
  1750. admin.delete_authentication_flow_execution(execution_id=exec_id)
  1751. assert err.match('404: b\'{"error":"Illegal execution"}\'')
  1752. # Test subflows
  1753. res = admin.create_authentication_flow_subflow(
  1754. payload={
  1755. "alias": "test-subflow",
  1756. "provider": "basic-flow",
  1757. "type": "something",
  1758. "description": "something",
  1759. },
  1760. flow_alias="test-browser",
  1761. )
  1762. assert res == b""
  1763. with pytest.raises(KeycloakPostError) as err:
  1764. admin.create_authentication_flow_subflow(
  1765. payload={"alias": "test-subflow", "providerId": "basic-flow"},
  1766. flow_alias="test-browser",
  1767. )
  1768. assert err.match('409: b\'{"errorMessage":"New flow alias name already exists"}\'')
  1769. res = admin.create_authentication_flow_subflow(
  1770. payload={
  1771. "alias": "test-subflow",
  1772. "provider": "basic-flow",
  1773. "type": "something",
  1774. "description": "something",
  1775. },
  1776. flow_alias="test-create",
  1777. skip_exists=True,
  1778. )
  1779. assert res == {"msg": "Already exists"}
  1780. # Test delete auth flow
  1781. flow_id = [x for x in admin.get_authentication_flows() if x["alias"] == "test-browser"][0][
  1782. "id"
  1783. ]
  1784. res = admin.delete_authentication_flow(flow_id=flow_id)
  1785. assert res == dict()
  1786. with pytest.raises(KeycloakDeleteError) as err:
  1787. admin.delete_authentication_flow(flow_id=flow_id)
  1788. assert err.match('404: b\'{"error":"Could not find flow with id"}\'')
  1789. def test_authentication_configs(admin: KeycloakAdmin, realm: str):
  1790. """Test authentication configs.
  1791. :param admin: Keycloak Admin client
  1792. :type admin: KeycloakAdmin
  1793. :param realm: Keycloak realm
  1794. :type realm: str
  1795. """
  1796. admin.change_current_realm(realm)
  1797. # Test list of auth providers
  1798. res = admin.get_authenticator_providers()
  1799. assert len(res) <= 38
  1800. res = admin.get_authenticator_provider_config_description(provider_id="auth-cookie")
  1801. assert res == {
  1802. "helpText": "Validates the SSO cookie set by the auth server.",
  1803. "name": "Cookie",
  1804. "properties": [],
  1805. "providerId": "auth-cookie",
  1806. }
  1807. # Test authenticator config
  1808. # Currently unable to find a sustainable way to fetch the config id,
  1809. # therefore testing only failures
  1810. with pytest.raises(KeycloakGetError) as err:
  1811. admin.get_authenticator_config(config_id="bad")
  1812. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1813. with pytest.raises(KeycloakPutError) as err:
  1814. admin.update_authenticator_config(payload=dict(), config_id="bad")
  1815. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1816. with pytest.raises(KeycloakDeleteError) as err:
  1817. admin.delete_authenticator_config(config_id="bad")
  1818. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1819. def test_sync_users(admin: KeycloakAdmin, realm: str):
  1820. """Test sync users.
  1821. :param admin: Keycloak Admin client
  1822. :type admin: KeycloakAdmin
  1823. :param realm: Keycloak realm
  1824. :type realm: str
  1825. """
  1826. admin.change_current_realm(realm)
  1827. # Only testing the error message
  1828. with pytest.raises(KeycloakPostError) as err:
  1829. admin.sync_users(storage_id="does-not-exist", action="triggerFullSync")
  1830. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1831. def test_client_scopes(admin: KeycloakAdmin, realm: str):
  1832. """Test client scopes.
  1833. :param admin: Keycloak Admin client
  1834. :type admin: KeycloakAdmin
  1835. :param realm: Keycloak realm
  1836. :type realm: str
  1837. """
  1838. admin.change_current_realm(realm)
  1839. # Test get client scopes
  1840. res = admin.get_client_scopes()
  1841. scope_names = {x["name"] for x in res}
  1842. assert len(res) == 10
  1843. assert "email" in scope_names
  1844. assert "profile" in scope_names
  1845. assert "offline_access" in scope_names
  1846. with pytest.raises(KeycloakGetError) as err:
  1847. admin.get_client_scope(client_scope_id="does-not-exist")
  1848. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1849. scope = admin.get_client_scope(client_scope_id=res[0]["id"])
  1850. assert res[0] == scope
  1851. scope = admin.get_client_scope_by_name(client_scope_name=res[0]["name"])
  1852. assert res[0] == scope
  1853. # Test create client scope
  1854. res = admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=True)
  1855. assert res
  1856. res2 = admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=True)
  1857. assert res == res2
  1858. with pytest.raises(KeycloakPostError) as err:
  1859. admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=False)
  1860. assert err.match('409: b\'{"errorMessage":"Client Scope test-scope already exists"}\'')
  1861. # Test update client scope
  1862. with pytest.raises(KeycloakPutError) as err:
  1863. admin.update_client_scope(client_scope_id="does-not-exist", payload=dict())
  1864. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1865. res_update = admin.update_client_scope(
  1866. client_scope_id=res, payload={"name": "test-scope-update"}
  1867. )
  1868. assert res_update == dict()
  1869. admin.get_client_scope(client_scope_id=res)["name"] == "test-scope-update"
  1870. # Test get mappers
  1871. mappers = admin.get_mappers_from_client_scope(client_scope_id=res)
  1872. assert mappers == list()
  1873. # Test add mapper
  1874. with pytest.raises(KeycloakPostError) as err:
  1875. admin.add_mapper_to_client_scope(client_scope_id=res, payload=dict())
  1876. assert err.match('404: b\'{"error":"ProtocolMapper provider not found"}\'')
  1877. res_add = admin.add_mapper_to_client_scope(
  1878. client_scope_id=res,
  1879. payload={
  1880. "name": "test-mapper",
  1881. "protocol": "openid-connect",
  1882. "protocolMapper": "oidc-usermodel-attribute-mapper",
  1883. },
  1884. )
  1885. assert res_add == b""
  1886. assert len(admin.get_mappers_from_client_scope(client_scope_id=res)) == 1
  1887. # Test update mapper
  1888. test_mapper = admin.get_mappers_from_client_scope(client_scope_id=res)[0]
  1889. with pytest.raises(KeycloakPutError) as err:
  1890. admin.update_mapper_in_client_scope(
  1891. client_scope_id="does-not-exist", protocol_mapper_id=test_mapper["id"], payload=dict()
  1892. )
  1893. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1894. test_mapper["config"]["user.attribute"] = "test"
  1895. res_update = admin.update_mapper_in_client_scope(
  1896. client_scope_id=res, protocol_mapper_id=test_mapper["id"], payload=test_mapper
  1897. )
  1898. assert res_update == dict()
  1899. assert (
  1900. admin.get_mappers_from_client_scope(client_scope_id=res)[0]["config"]["user.attribute"]
  1901. == "test"
  1902. )
  1903. # Test delete mapper
  1904. res_del = admin.delete_mapper_from_client_scope(
  1905. client_scope_id=res, protocol_mapper_id=test_mapper["id"]
  1906. )
  1907. assert res_del == dict()
  1908. with pytest.raises(KeycloakDeleteError) as err:
  1909. admin.delete_mapper_from_client_scope(
  1910. client_scope_id=res, protocol_mapper_id=test_mapper["id"]
  1911. )
  1912. assert err.match('404: b\'{"error":"Model not found"}\'')
  1913. # Test default default scopes
  1914. res_defaults = admin.get_default_default_client_scopes()
  1915. assert len(res_defaults) == 6
  1916. with pytest.raises(KeycloakPutError) as err:
  1917. admin.add_default_default_client_scope(scope_id="does-not-exist")
  1918. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1919. res_add = admin.add_default_default_client_scope(scope_id=res)
  1920. assert res_add == dict()
  1921. assert len(admin.get_default_default_client_scopes()) == 7
  1922. with pytest.raises(KeycloakDeleteError) as err:
  1923. admin.delete_default_default_client_scope(scope_id="does-not-exist")
  1924. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1925. res_del = admin.delete_default_default_client_scope(scope_id=res)
  1926. assert res_del == dict()
  1927. assert len(admin.get_default_default_client_scopes()) == 6
  1928. # Test default optional scopes
  1929. res_defaults = admin.get_default_optional_client_scopes()
  1930. assert len(res_defaults) == 4
  1931. with pytest.raises(KeycloakPutError) as err:
  1932. admin.add_default_optional_client_scope(scope_id="does-not-exist")
  1933. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1934. res_add = admin.add_default_optional_client_scope(scope_id=res)
  1935. assert res_add == dict()
  1936. assert len(admin.get_default_optional_client_scopes()) == 5
  1937. with pytest.raises(KeycloakDeleteError) as err:
  1938. admin.delete_default_optional_client_scope(scope_id="does-not-exist")
  1939. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1940. res_del = admin.delete_default_optional_client_scope(scope_id=res)
  1941. assert res_del == dict()
  1942. assert len(admin.get_default_optional_client_scopes()) == 4
  1943. # Test client scope delete
  1944. res_del = admin.delete_client_scope(client_scope_id=res)
  1945. assert res_del == dict()
  1946. with pytest.raises(KeycloakDeleteError) as err:
  1947. admin.delete_client_scope(client_scope_id=res)
  1948. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1949. def test_components(admin: KeycloakAdmin, realm: str):
  1950. """Test components.
  1951. :param admin: Keycloak Admin client
  1952. :type admin: KeycloakAdmin
  1953. :param realm: Keycloak realm
  1954. :type realm: str
  1955. """
  1956. admin.change_current_realm(realm)
  1957. # Test get components
  1958. res = admin.get_components()
  1959. assert len(res) == 12
  1960. with pytest.raises(KeycloakGetError) as err:
  1961. admin.get_component(component_id="does-not-exist")
  1962. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1963. res_get = admin.get_component(component_id=res[0]["id"])
  1964. assert res_get == res[0]
  1965. # Test create component
  1966. with pytest.raises(KeycloakPostError) as err:
  1967. admin.create_component(payload={"bad": "dict"})
  1968. assert err.match('400: b\'{"error":"Unrecognized field')
  1969. res = admin.create_component(
  1970. payload={
  1971. "name": "Test Component",
  1972. "providerId": "max-clients",
  1973. "providerType": "org.keycloak.services.clientregistration."
  1974. + "policy.ClientRegistrationPolicy",
  1975. "config": {"max-clients": ["1000"]},
  1976. }
  1977. )
  1978. assert res
  1979. assert admin.get_component(component_id=res)["name"] == "Test Component"
  1980. # Test update component
  1981. component = admin.get_component(component_id=res)
  1982. component["name"] = "Test Component Update"
  1983. with pytest.raises(KeycloakPutError) as err:
  1984. admin.update_component(component_id="does-not-exist", payload=dict())
  1985. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1986. res_upd = admin.update_component(component_id=res, payload=component)
  1987. assert res_upd == dict()
  1988. assert admin.get_component(component_id=res)["name"] == "Test Component Update"
  1989. # Test delete component
  1990. res_del = admin.delete_component(component_id=res)
  1991. assert res_del == dict()
  1992. with pytest.raises(KeycloakDeleteError) as err:
  1993. admin.delete_component(component_id=res)
  1994. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1995. def test_keys(admin: KeycloakAdmin, realm: str):
  1996. """Test keys.
  1997. :param admin: Keycloak Admin client
  1998. :type admin: KeycloakAdmin
  1999. :param realm: Keycloak realm
  2000. :type realm: str
  2001. """
  2002. admin.change_current_realm(realm)
  2003. assert set(admin.get_keys()["active"].keys()) == {"AES", "HS256", "RS256", "RSA-OAEP"}
  2004. assert {k["algorithm"] for k in admin.get_keys()["keys"]} == {
  2005. "HS256",
  2006. "RSA-OAEP",
  2007. "AES",
  2008. "RS256",
  2009. }
  2010. def test_admin_events(admin: KeycloakAdmin, realm: str):
  2011. """Test events.
  2012. :param admin: Keycloak Admin client
  2013. :type admin: KeycloakAdmin
  2014. :param realm: Keycloak realm
  2015. :type realm: str
  2016. """
  2017. admin.change_current_realm(realm)
  2018. admin.create_client(payload={"name": "test", "clientId": "test"})
  2019. events = admin.get_admin_events()
  2020. assert events == list()
  2021. def test_user_events(admin: KeycloakAdmin, realm: str):
  2022. """Test events.
  2023. :param admin: Keycloak Admin client
  2024. :type admin: KeycloakAdmin
  2025. :param realm: Keycloak realm
  2026. :type realm: str
  2027. """
  2028. admin.change_current_realm(realm)
  2029. events = admin.get_events()
  2030. assert events == list()
  2031. with pytest.raises(KeycloakPutError) as err:
  2032. admin.set_events(payload={"bad": "conf"})
  2033. assert err.match('400: b\'{"error":"Unrecognized field')
  2034. res = admin.set_events(payload={"adminEventsDetailsEnabled": True, "adminEventsEnabled": True})
  2035. assert res == dict()
  2036. admin.create_client(payload={"name": "test", "clientId": "test"})
  2037. events = admin.get_events()
  2038. assert events == list()
  2039. @freezegun.freeze_time("2023-02-25 10:00:00")
  2040. def test_auto_refresh(admin_frozen: KeycloakAdmin, realm: str):
  2041. """Test auto refresh token.
  2042. :param admin_frozen: Keycloak Admin client with time frozen in place
  2043. :type admin_frozen: KeycloakAdmin
  2044. :param realm: Keycloak realm
  2045. :type realm: str
  2046. """
  2047. admin = admin_frozen
  2048. # Test get refresh
  2049. admin.connection.custom_headers = {
  2050. "Authorization": "Bearer bad",
  2051. "Content-Type": "application/json",
  2052. }
  2053. with pytest.raises(KeycloakAuthenticationError) as err:
  2054. admin.get_realm(realm_name=realm)
  2055. assert err.match('401: b\'{"error":"HTTP 401 Unauthorized"}\'')
  2056. # Freeze time to simulate the access token expiring
  2057. with freezegun.freeze_time("2023-02-25 10:02:00"):
  2058. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:01:00")
  2059. assert admin.get_realm(realm_name=realm)
  2060. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:01:00")
  2061. # Test bad refresh token, but first make sure access token has expired again
  2062. with freezegun.freeze_time("2023-02-25 10:10:00"):
  2063. admin.connection.custom_headers = {"Content-Type": "application/json"}
  2064. admin.connection.token["refresh_token"] = "bad"
  2065. with pytest.raises(KeycloakPostError) as err:
  2066. admin.get_realm(realm_name="test-refresh")
  2067. assert err.match(
  2068. '400: b\'{"error":"invalid_grant","error_description":"Invalid refresh token"}\''
  2069. )
  2070. admin.connection.get_token()
  2071. # Test post refresh
  2072. with freezegun.freeze_time("2023-02-25 10:15:00"):
  2073. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:15:00")
  2074. admin.connection.token = None
  2075. assert admin.create_realm(payload={"realm": "test-refresh"}) == b""
  2076. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:15:00")
  2077. # Test update refresh
  2078. with freezegun.freeze_time("2023-02-25 10:25:00"):
  2079. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:25:00")
  2080. admin.connection.token = None
  2081. assert (
  2082. admin.update_realm(realm_name="test-refresh", payload={"accountTheme": "test"})
  2083. == dict()
  2084. )
  2085. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:25:00")
  2086. # Test delete refresh
  2087. with freezegun.freeze_time("2023-02-25 10:35:00"):
  2088. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:35:00")
  2089. admin.connection.token = None
  2090. assert admin.delete_realm(realm_name="test-refresh") == dict()
  2091. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:35:00")
  2092. def test_get_required_actions(admin: KeycloakAdmin, realm: str):
  2093. """Test required actions.
  2094. :param admin: Keycloak Admin client
  2095. :type admin: KeycloakAdmin
  2096. :param realm: Keycloak realm
  2097. :type realm: str
  2098. """
  2099. admin.change_current_realm(realm)
  2100. ractions = admin.get_required_actions()
  2101. assert isinstance(ractions, list)
  2102. for ra in ractions:
  2103. for key in [
  2104. "alias",
  2105. "name",
  2106. "providerId",
  2107. "enabled",
  2108. "defaultAction",
  2109. "priority",
  2110. "config",
  2111. ]:
  2112. assert key in ra
  2113. def test_get_required_action_by_alias(admin: KeycloakAdmin, realm: str):
  2114. """Test get required action by alias.
  2115. :param admin: Keycloak Admin client
  2116. :type admin: KeycloakAdmin
  2117. :param realm: Keycloak realm
  2118. :type realm: str
  2119. """
  2120. admin.change_current_realm(realm)
  2121. ractions = admin.get_required_actions()
  2122. ra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2123. assert ra in ractions
  2124. assert ra["alias"] == "UPDATE_PASSWORD"
  2125. assert admin.get_required_action_by_alias("does-not-exist") is None
  2126. def test_update_required_action(admin: KeycloakAdmin, realm: str):
  2127. """Test update required action.
  2128. :param admin: Keycloak Admin client
  2129. :type admin: KeycloakAdmin
  2130. :param realm: Keycloak realm
  2131. :type realm: str
  2132. """
  2133. admin.change_current_realm(realm)
  2134. ra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2135. old = copy.deepcopy(ra)
  2136. ra["enabled"] = False
  2137. admin.update_required_action("UPDATE_PASSWORD", ra)
  2138. newra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2139. assert old != newra
  2140. assert newra["enabled"] is False
  2141. def test_get_composite_client_roles_of_group(
  2142. admin: KeycloakAdmin, realm: str, client: str, group: str, composite_client_role: str
  2143. ):
  2144. """Test get composite client roles of group.
  2145. :param admin: Keycloak Admin client
  2146. :type admin: KeycloakAdmin
  2147. :param realm: Keycloak realm
  2148. :type realm: str
  2149. :param client: Keycloak client
  2150. :type client: str
  2151. :param group: Keycloak group
  2152. :type group: str
  2153. :param composite_client_role: Composite client role
  2154. :type composite_client_role: str
  2155. """
  2156. admin.change_current_realm(realm)
  2157. role = admin.get_client_role(client, composite_client_role)
  2158. admin.assign_group_client_roles(group_id=group, client_id=client, roles=[role])
  2159. result = admin.get_composite_client_roles_of_group(client, group)
  2160. assert role["id"] in [x["id"] for x in result]
  2161. def test_get_role_client_level_children(
  2162. admin: KeycloakAdmin, realm: str, client: str, composite_client_role: str, client_role: str
  2163. ):
  2164. """Test get children of composite client role.
  2165. :param admin: Keycloak Admin client
  2166. :type admin: KeycloakAdmin
  2167. :param realm: Keycloak realm
  2168. :type realm: str
  2169. :param client: Keycloak client
  2170. :type client: str
  2171. :param composite_client_role: Composite client role
  2172. :type composite_client_role: str
  2173. :param client_role: Client role
  2174. :type client_role: str
  2175. """
  2176. admin.change_current_realm(realm)
  2177. child = admin.get_client_role(client, client_role)
  2178. parent = admin.get_client_role(client, composite_client_role)
  2179. res = admin.get_role_client_level_children(client, parent["id"])
  2180. assert child["id"] in [x["id"] for x in res]
  2181. def test_upload_certificate(admin: KeycloakAdmin, realm: str, client: str, selfsigned_cert: tuple):
  2182. """Test upload certificate.
  2183. :param admin: Keycloak Admin client
  2184. :type admin: KeycloakAdmin
  2185. :param realm: Keycloak realm
  2186. :type realm: str
  2187. :param client: Keycloak client
  2188. :type client: str
  2189. :param selfsigned_cert: Selfsigned certificates
  2190. :type selfsigned_cert: tuple
  2191. """
  2192. admin.change_current_realm(realm)
  2193. cert, _ = selfsigned_cert
  2194. cert = cert.decode("utf-8").strip()
  2195. admin.upload_certificate(client, cert)
  2196. cl = admin.get_client(client)
  2197. assert cl["attributes"]["jwt.credential.certificate"] == "".join(cert.splitlines()[1:-1])
  2198. def test_get_bruteforce_status_for_user(
  2199. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2200. ):
  2201. """Test users.
  2202. :param admin: Keycloak Admin client
  2203. :type admin: KeycloakAdmin
  2204. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2205. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2206. :param realm: Keycloak realm
  2207. :type realm: str
  2208. """
  2209. oid, username, password = oid_with_credentials
  2210. admin.change_current_realm(realm)
  2211. # Turn on bruteforce protection
  2212. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2213. res = admin.get_realm(realm_name=realm)
  2214. assert res["bruteForceProtected"] is True
  2215. # Test login user with wrong credentials
  2216. try:
  2217. oid.token(username=username, password="wrongpassword")
  2218. except KeycloakAuthenticationError:
  2219. pass
  2220. user_id = admin.get_user_id(username)
  2221. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2222. assert bruteforce_status["numFailures"] == 1
  2223. # Cleanup
  2224. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2225. res = admin.get_realm(realm_name=realm)
  2226. assert res["bruteForceProtected"] is False
  2227. def test_clear_bruteforce_attempts_for_user(
  2228. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2229. ):
  2230. """Test users.
  2231. :param admin: Keycloak Admin client
  2232. :type admin: KeycloakAdmin
  2233. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2234. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2235. :param realm: Keycloak realm
  2236. :type realm: str
  2237. """
  2238. oid, username, password = oid_with_credentials
  2239. admin.change_current_realm(realm)
  2240. # Turn on bruteforce protection
  2241. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2242. res = admin.get_realm(realm_name=realm)
  2243. assert res["bruteForceProtected"] is True
  2244. # Test login user with wrong credentials
  2245. try:
  2246. oid.token(username=username, password="wrongpassword")
  2247. except KeycloakAuthenticationError:
  2248. pass
  2249. user_id = admin.get_user_id(username)
  2250. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2251. assert bruteforce_status["numFailures"] == 1
  2252. res = admin.clear_bruteforce_attempts_for_user(user_id)
  2253. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2254. assert bruteforce_status["numFailures"] == 0
  2255. # Cleanup
  2256. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2257. res = admin.get_realm(realm_name=realm)
  2258. assert res["bruteForceProtected"] is False
  2259. def test_clear_bruteforce_attempts_for_all_users(
  2260. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2261. ):
  2262. """Test users.
  2263. :param admin: Keycloak Admin client
  2264. :type admin: KeycloakAdmin
  2265. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2266. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2267. :param realm: Keycloak realm
  2268. :type realm: str
  2269. """
  2270. oid, username, password = oid_with_credentials
  2271. admin.change_current_realm(realm)
  2272. # Turn on bruteforce protection
  2273. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2274. res = admin.get_realm(realm_name=realm)
  2275. assert res["bruteForceProtected"] is True
  2276. # Test login user with wrong credentials
  2277. try:
  2278. oid.token(username=username, password="wrongpassword")
  2279. except KeycloakAuthenticationError:
  2280. pass
  2281. user_id = admin.get_user_id(username)
  2282. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2283. assert bruteforce_status["numFailures"] == 1
  2284. res = admin.clear_all_bruteforce_attempts()
  2285. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2286. assert bruteforce_status["numFailures"] == 0
  2287. # Cleanup
  2288. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2289. res = admin.get_realm(realm_name=realm)
  2290. assert res["bruteForceProtected"] is False
  2291. def test_default_realm_role_present(realm: str, admin: KeycloakAdmin) -> None:
  2292. """Test that the default realm role is present in a brand new realm.
  2293. :param realm: Realm name
  2294. :type realm: str
  2295. :param admin: Keycloak admin
  2296. :type admin: KeycloakAdmin
  2297. """
  2298. admin.change_current_realm(realm)
  2299. assert f"default-roles-{realm}" in [x["name"] for x in admin.get_realm_roles()]
  2300. assert (
  2301. len([x["name"] for x in admin.get_realm_roles() if x["name"] == f"default-roles-{realm}"])
  2302. == 1
  2303. )
  2304. def test_get_default_realm_role_id(realm: str, admin: KeycloakAdmin) -> None:
  2305. """Test getter for the ID of the default realm role.
  2306. :param realm: Realm name
  2307. :type realm: str
  2308. :param admin: Keycloak admin
  2309. :type admin: KeycloakAdmin
  2310. """
  2311. admin.change_current_realm(realm)
  2312. assert (
  2313. admin.get_default_realm_role_id()
  2314. == [x["id"] for x in admin.get_realm_roles() if x["name"] == f"default-roles-{realm}"][0]
  2315. )
  2316. def test_realm_default_roles(admin: KeycloakAdmin, realm: str) -> None:
  2317. """Test getting, adding and deleting default realm roles.
  2318. :param realm: Realm name
  2319. :type realm: str
  2320. :param admin: Keycloak admin
  2321. :type admin: KeycloakAdmin
  2322. """
  2323. admin.change_current_realm(realm)
  2324. # Test listing all default realm roles
  2325. roles = admin.get_realm_default_roles()
  2326. assert len(roles) == 2
  2327. assert {x["name"] for x in roles} == {"offline_access", "uma_authorization"}
  2328. with pytest.raises(KeycloakGetError) as err:
  2329. admin.realm_name = "doesnotexist"
  2330. admin.get_realm_default_roles()
  2331. assert err.match('404: b\'{"error":"Realm not found."}\'')
  2332. admin.change_current_realm(realm)
  2333. # Test removing a default realm role
  2334. res = admin.remove_realm_default_roles(payload=[roles[0]])
  2335. assert res == {}
  2336. assert roles[0] not in admin.get_realm_default_roles()
  2337. assert len(admin.get_realm_default_roles()) == 1
  2338. with pytest.raises(KeycloakDeleteError) as err:
  2339. admin.remove_realm_default_roles(payload=[{"id": "bad id"}])
  2340. assert err.match('404: b\'{"error":"Could not find composite role"}\'')
  2341. # Test adding a default realm role
  2342. res = admin.add_realm_default_roles(payload=[roles[0]])
  2343. assert res == {}
  2344. assert roles[0] in admin.get_realm_default_roles()
  2345. assert len(admin.get_realm_default_roles()) == 2
  2346. with pytest.raises(KeycloakPostError) as err:
  2347. admin.add_realm_default_roles(payload=[{"id": "bad id"}])
  2348. assert err.match('404: b\'{"error":"Could not find composite role"}\'')
  2349. def test_clear_keys_cache(realm: str, admin: KeycloakAdmin) -> None:
  2350. """Test clearing the keys cache.
  2351. :param realm: Realm name
  2352. :type realm: str
  2353. :param admin: Keycloak admin
  2354. :type admin: KeycloakAdmin
  2355. """
  2356. admin.change_current_realm(realm)
  2357. res = admin.clear_keys_cache()
  2358. assert res == {}
  2359. def test_clear_realm_cache(realm: str, admin: KeycloakAdmin) -> None:
  2360. """Test clearing the realm cache.
  2361. :param realm: Realm name
  2362. :type realm: str
  2363. :param admin: Keycloak admin
  2364. :type admin: KeycloakAdmin
  2365. """
  2366. admin.change_current_realm(realm)
  2367. res = admin.clear_realm_cache()
  2368. assert res == {}
  2369. def test_clear_user_cache(realm: str, admin: KeycloakAdmin) -> None:
  2370. """Test clearing the user cache.
  2371. :param realm: Realm name
  2372. :type realm: str
  2373. :param admin: Keycloak admin
  2374. :type admin: KeycloakAdmin
  2375. """
  2376. admin.change_current_realm(realm)
  2377. res = admin.clear_user_cache()
  2378. assert res == {}
  2379. def test_initial_access_token(
  2380. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2381. ) -> None:
  2382. """Test initial access token and client creation.
  2383. :param admin: Keycloak admin
  2384. :type admin: KeycloakAdmin
  2385. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2386. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2387. """
  2388. res = admin.create_initial_access_token(2, 3)
  2389. assert "token" in res
  2390. assert res["count"] == 2
  2391. assert res["expiration"] == 3
  2392. oid, username, password = oid_with_credentials
  2393. client = str(uuid.uuid4())
  2394. secret = str(uuid.uuid4())
  2395. res = oid.register_client(
  2396. token=res["token"],
  2397. payload={
  2398. "name": "DynamicRegisteredClient",
  2399. "clientId": client,
  2400. "enabled": True,
  2401. "publicClient": False,
  2402. "protocol": "openid-connect",
  2403. "secret": secret,
  2404. "clientAuthenticatorType": "client-secret",
  2405. },
  2406. )
  2407. assert res["clientId"] == client
  2408. new_secret = str(uuid.uuid4())
  2409. res = oid.update_client(res["registrationAccessToken"], client, payload={"secret": new_secret})
  2410. assert res["secret"] == new_secret