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.

2878 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 len(groups[0]["subGroups"]) == 2, groups["subGroups"]
  536. assert groups[0]["id"] == group_id
  537. assert {x["id"] for x in groups[0]["subGroups"]} == {subgroup_id_1, subgroup_id_2}
  538. # Test get groups query
  539. groups = admin.get_groups(query={"max": 10})
  540. assert len(groups) == 1, groups
  541. assert len(groups[0]["subGroups"]) == 2, groups["subGroups"]
  542. assert groups[0]["id"] == group_id
  543. assert {x["id"] for x in groups[0]["subGroups"]} == {subgroup_id_1, subgroup_id_2}
  544. # Test get group
  545. res = admin.get_group(group_id=subgroup_id_1)
  546. assert res["id"] == subgroup_id_1, res
  547. assert res["name"] == "subgroup-1"
  548. assert res["path"] == "/main-group/subgroup-1"
  549. # Test get group fail
  550. with pytest.raises(KeycloakGetError) as err:
  551. admin.get_group(group_id="does-not-exist")
  552. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  553. # Create 1 more subgroup
  554. subsubgroup_id_1 = admin.create_group(payload={"name": "subsubgroup-1"}, parent=subgroup_id_2)
  555. main_group = admin.get_group(group_id=group_id)
  556. # Test nested searches
  557. res = admin.get_subgroups(group=main_group, path="/main-group/subgroup-2/subsubgroup-1")
  558. assert res is not None, res
  559. assert res["id"] == subsubgroup_id_1
  560. # Test empty search
  561. res = admin.get_subgroups(group=main_group, path="/none")
  562. assert res is None, res
  563. # Test get group by path
  564. res = admin.get_group_by_path(path="/main-group/subgroup-1")
  565. assert res is not None, res
  566. assert res["id"] == subgroup_id_1, res
  567. with pytest.raises(KeycloakGetError) as err:
  568. admin.get_group_by_path(path="/main-group/subgroup-2/subsubgroup-1/test")
  569. assert err.match('404: b\'{"error":"Group path does not exist"}\'')
  570. res = admin.get_group_by_path(path="/main-group/subgroup-2/subsubgroup-1")
  571. assert res is not None, res
  572. assert res["id"] == subsubgroup_id_1
  573. res = admin.get_group_by_path(path="/main-group")
  574. assert res is not None, res
  575. assert res["id"] == group_id, res
  576. # Test group members
  577. res = admin.get_group_members(group_id=subgroup_id_2)
  578. assert len(res) == 0, res
  579. # Test fail group members
  580. with pytest.raises(KeycloakGetError) as err:
  581. admin.get_group_members(group_id="does-not-exist")
  582. assert err.match('404: b\'{"error":"Could not find group by id"}\'')
  583. res = admin.group_user_add(user_id=user, group_id=subgroup_id_2)
  584. assert res == dict(), res
  585. res = admin.get_group_members(group_id=subgroup_id_2)
  586. assert len(res) == 1, res
  587. assert res[0]["id"] == user
  588. # Test get group members query
  589. res = admin.get_group_members(group_id=subgroup_id_2, query={"max": 10})
  590. assert len(res) == 1, res
  591. assert res[0]["id"] == user
  592. with pytest.raises(KeycloakDeleteError) as err:
  593. admin.group_user_remove(user_id="does-not-exist", group_id=subgroup_id_2)
  594. assert err.match('404: b\'{"error":"User not found"}\''), err
  595. res = admin.group_user_remove(user_id=user, group_id=subgroup_id_2)
  596. assert res == dict(), res
  597. # Test set permissions
  598. res = admin.group_set_permissions(group_id=subgroup_id_2, enabled=True)
  599. assert res["enabled"], res
  600. res = admin.group_set_permissions(group_id=subgroup_id_2, enabled=False)
  601. assert not res["enabled"], res
  602. with pytest.raises(KeycloakPutError) as err:
  603. admin.group_set_permissions(group_id=subgroup_id_2, enabled="blah")
  604. assert err.match('b\'{"error":"unknown_error"}\''), err
  605. # Test update group
  606. res = admin.update_group(group_id=subgroup_id_2, payload={"name": "new-subgroup-2"})
  607. assert res == dict(), res
  608. assert admin.get_group(group_id=subgroup_id_2)["name"] == "new-subgroup-2"
  609. # test update fail
  610. with pytest.raises(KeycloakPutError) as err:
  611. admin.update_group(group_id="does-not-exist", payload=dict())
  612. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  613. # Test delete
  614. res = admin.delete_group(group_id=group_id)
  615. assert res == dict(), res
  616. assert len(admin.get_groups()) == 0
  617. # Test delete fail
  618. with pytest.raises(KeycloakDeleteError) as err:
  619. admin.delete_group(group_id="does-not-exist")
  620. assert err.match('404: b\'{"error":"Could not find group by id"}\''), err
  621. def test_clients(admin: KeycloakAdmin, realm: str):
  622. """Test clients.
  623. :param admin: Keycloak Admin client
  624. :type admin: KeycloakAdmin
  625. :param realm: Keycloak realm
  626. :type realm: str
  627. """
  628. admin.change_current_realm(realm)
  629. # Test get clients
  630. clients = admin.get_clients()
  631. assert len(clients) == 6, clients
  632. assert {x["name"] for x in clients} == set(
  633. [
  634. "${client_admin-cli}",
  635. "${client_security-admin-console}",
  636. "${client_account-console}",
  637. "${client_broker}",
  638. "${client_account}",
  639. "${client_realm-management}",
  640. ]
  641. ), clients
  642. # Test create client
  643. client_id = admin.create_client(payload={"name": "test-client", "clientId": "test-client"})
  644. assert client_id, client_id
  645. with pytest.raises(KeycloakPostError) as err:
  646. admin.create_client(payload={"name": "test-client", "clientId": "test-client"})
  647. assert err.match('409: b\'{"errorMessage":"Client test-client already exists"}\''), err
  648. client_id_2 = admin.create_client(
  649. payload={"name": "test-client", "clientId": "test-client"}, skip_exists=True
  650. )
  651. assert client_id == client_id_2, client_id_2
  652. # Test get client
  653. res = admin.get_client(client_id=client_id)
  654. assert res["clientId"] == "test-client", res
  655. assert res["name"] == "test-client", res
  656. assert res["id"] == client_id, res
  657. with pytest.raises(KeycloakGetError) as err:
  658. admin.get_client(client_id="does-not-exist")
  659. assert err.match('404: b\'{"error":"Could not find client"}\'')
  660. assert len(admin.get_clients()) == 7
  661. # Test get client id
  662. assert admin.get_client_id(client_id="test-client") == client_id
  663. assert admin.get_client_id(client_id="does-not-exist") is None
  664. # Test update client
  665. res = admin.update_client(client_id=client_id, payload={"name": "test-client-change"})
  666. assert res == dict(), res
  667. with pytest.raises(KeycloakPutError) as err:
  668. admin.update_client(client_id="does-not-exist", payload={"name": "test-client-change"})
  669. assert err.match('404: b\'{"error":"Could not find client"}\'')
  670. # Test client mappers
  671. res = admin.get_mappers_from_client(client_id=client_id)
  672. assert len(res) == 0
  673. with pytest.raises(KeycloakPostError) as err:
  674. admin.add_mapper_to_client(client_id="does-not-exist", payload=dict())
  675. assert err.match('404: b\'{"error":"Could not find client"}\'')
  676. res = admin.add_mapper_to_client(
  677. client_id=client_id,
  678. payload={
  679. "name": "test-mapper",
  680. "protocol": "openid-connect",
  681. "protocolMapper": "oidc-usermodel-attribute-mapper",
  682. },
  683. )
  684. assert res == b""
  685. assert len(admin.get_mappers_from_client(client_id=client_id)) == 1
  686. mapper = admin.get_mappers_from_client(client_id=client_id)[0]
  687. with pytest.raises(KeycloakPutError) as err:
  688. admin.update_client_mapper(client_id=client_id, mapper_id="does-not-exist", payload=dict())
  689. assert err.match('404: b\'{"error":"Model not found"}\'')
  690. mapper["config"]["user.attribute"] = "test"
  691. res = admin.update_client_mapper(client_id=client_id, mapper_id=mapper["id"], payload=mapper)
  692. assert res == dict()
  693. res = admin.remove_client_mapper(client_id=client_id, client_mapper_id=mapper["id"])
  694. assert res == dict()
  695. with pytest.raises(KeycloakDeleteError) as err:
  696. admin.remove_client_mapper(client_id=client_id, client_mapper_id=mapper["id"])
  697. assert err.match('404: b\'{"error":"Model not found"}\'')
  698. # Test client sessions
  699. with pytest.raises(KeycloakGetError) as err:
  700. admin.get_client_all_sessions(client_id="does-not-exist")
  701. assert err.match('404: b\'{"error":"Could not find client"}\'')
  702. assert admin.get_client_all_sessions(client_id=client_id) == list()
  703. assert admin.get_client_sessions_stats() == list()
  704. # Test authz
  705. auth_client_id = admin.create_client(
  706. payload={
  707. "name": "authz-client",
  708. "clientId": "authz-client",
  709. "authorizationServicesEnabled": True,
  710. "serviceAccountsEnabled": True,
  711. }
  712. )
  713. res = admin.get_client_authz_settings(client_id=auth_client_id)
  714. assert res["allowRemoteResourceManagement"]
  715. assert res["decisionStrategy"] == "UNANIMOUS"
  716. assert len(res["policies"]) >= 0
  717. with pytest.raises(KeycloakGetError) as err:
  718. admin.get_client_authz_settings(client_id=client_id)
  719. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  720. # Authz resources
  721. res = admin.get_client_authz_resources(client_id=auth_client_id)
  722. assert len(res) == 1
  723. assert res[0]["name"] == "Default Resource"
  724. with pytest.raises(KeycloakGetError) as err:
  725. admin.get_client_authz_resources(client_id=client_id)
  726. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  727. res = admin.create_client_authz_resource(
  728. client_id=auth_client_id, payload={"name": "test-resource"}
  729. )
  730. assert res["name"] == "test-resource", res
  731. test_resource_id = res["_id"]
  732. res = admin.get_client_authz_resource(client_id=auth_client_id, resource_id=test_resource_id)
  733. assert res["_id"] == test_resource_id, res
  734. assert res["name"] == "test-resource", res
  735. with pytest.raises(KeycloakPostError) as err:
  736. admin.create_client_authz_resource(
  737. client_id=auth_client_id, payload={"name": "test-resource"}
  738. )
  739. assert err.match('409: b\'{"error":"invalid_request"')
  740. assert admin.create_client_authz_resource(
  741. client_id=auth_client_id, payload={"name": "test-resource"}, skip_exists=True
  742. ) == {"msg": "Already exists"}
  743. res = admin.get_client_authz_resources(client_id=auth_client_id)
  744. assert len(res) == 2
  745. assert {x["name"] for x in res} == {"Default Resource", "test-resource"}
  746. res = admin.create_client_authz_resource(
  747. client_id=auth_client_id, payload={"name": "temp-resource"}
  748. )
  749. assert res["name"] == "temp-resource", res
  750. temp_resource_id: str = res["_id"]
  751. # Test update authz resources
  752. admin.update_client_authz_resource(
  753. client_id=auth_client_id,
  754. resource_id=temp_resource_id,
  755. payload={"name": "temp-updated-resource"},
  756. )
  757. res = admin.get_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  758. assert res["name"] == "temp-updated-resource", res
  759. with pytest.raises(KeycloakPutError) as err:
  760. admin.update_client_authz_resource(
  761. client_id=auth_client_id,
  762. resource_id="invalid_resource_id",
  763. payload={"name": "temp-updated-resource"},
  764. )
  765. assert err.match("404: b''"), err
  766. admin.delete_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  767. with pytest.raises(KeycloakGetError) as err:
  768. admin.get_client_authz_resource(client_id=auth_client_id, resource_id=temp_resource_id)
  769. assert err.match("404: b''")
  770. # Authz policies
  771. res = admin.get_client_authz_policies(client_id=auth_client_id)
  772. assert len(res) == 1, res
  773. assert res[0]["name"] == "Default Policy"
  774. with pytest.raises(KeycloakGetError) as err:
  775. admin.get_client_authz_policies(client_id="does-not-exist")
  776. assert err.match('404: b\'{"error":"Could not find client"}\'')
  777. role_id = admin.get_realm_role(role_name="offline_access")["id"]
  778. res = admin.create_client_authz_role_based_policy(
  779. client_id=auth_client_id,
  780. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  781. )
  782. assert res["name"] == "test-authz-rb-policy", res
  783. with pytest.raises(KeycloakPostError) as err:
  784. admin.create_client_authz_role_based_policy(
  785. client_id=auth_client_id,
  786. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  787. )
  788. assert err.match('409: b\'{"error":"Policy with name')
  789. assert admin.create_client_authz_role_based_policy(
  790. client_id=auth_client_id,
  791. payload={"name": "test-authz-rb-policy", "roles": [{"id": role_id}]},
  792. skip_exists=True,
  793. ) == {"msg": "Already exists"}
  794. assert len(admin.get_client_authz_policies(client_id=auth_client_id)) == 2
  795. res = admin.create_client_authz_role_based_policy(
  796. client_id=auth_client_id,
  797. payload={"name": "test-authz-rb-policy-delete", "roles": [{"id": role_id}]},
  798. )
  799. res2 = admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  800. assert res["id"] == res2["id"]
  801. admin.delete_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  802. with pytest.raises(KeycloakGetError) as err:
  803. admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
  804. assert err.match("404: b''")
  805. res = admin.create_client_authz_policy(
  806. client_id=auth_client_id,
  807. payload={
  808. "name": "test-authz-policy",
  809. "type": "time",
  810. "config": {"hourEnd": "18", "hour": "9"},
  811. },
  812. )
  813. assert res["name"] == "test-authz-policy", res
  814. with pytest.raises(KeycloakPostError) as err:
  815. admin.create_client_authz_policy(
  816. client_id=auth_client_id,
  817. payload={
  818. "name": "test-authz-policy",
  819. "type": "time",
  820. "config": {"hourEnd": "18", "hour": "9"},
  821. },
  822. )
  823. assert err.match('409: b\'{"error":"Policy with name')
  824. assert admin.create_client_authz_policy(
  825. client_id=auth_client_id,
  826. payload={
  827. "name": "test-authz-policy",
  828. "type": "time",
  829. "config": {"hourEnd": "18", "hour": "9"},
  830. },
  831. skip_exists=True,
  832. ) == {"msg": "Already exists"}
  833. assert len(admin.get_client_authz_policies(client_id=auth_client_id)) == 3
  834. # Test authz permissions
  835. res = admin.get_client_authz_permissions(client_id=auth_client_id)
  836. assert len(res) == 1, res
  837. assert res[0]["name"] == "Default Permission"
  838. with pytest.raises(KeycloakGetError) as err:
  839. admin.get_client_authz_permissions(client_id="does-not-exist")
  840. assert err.match('404: b\'{"error":"Could not find client"}\'')
  841. res = admin.create_client_authz_resource_based_permission(
  842. client_id=auth_client_id,
  843. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  844. )
  845. assert res, res
  846. assert res["name"] == "test-permission-rb"
  847. assert res["resources"] == [test_resource_id]
  848. with pytest.raises(KeycloakPostError) as err:
  849. admin.create_client_authz_resource_based_permission(
  850. client_id=auth_client_id,
  851. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  852. )
  853. assert err.match('409: b\'{"error":"Policy with name')
  854. assert admin.create_client_authz_resource_based_permission(
  855. client_id=auth_client_id,
  856. payload={"name": "test-permission-rb", "resources": [test_resource_id]},
  857. skip_exists=True,
  858. ) == {"msg": "Already exists"}
  859. assert len(admin.get_client_authz_permissions(client_id=auth_client_id)) == 2
  860. # Test authz scopes
  861. res = admin.get_client_authz_scopes(client_id=auth_client_id)
  862. assert len(res) == 0, res
  863. with pytest.raises(KeycloakGetError) as err:
  864. admin.get_client_authz_scopes(client_id=client_id)
  865. assert err.match('404: b\'{"error":"HTTP 404 Not Found"}\'')
  866. res = admin.create_client_authz_scopes(
  867. client_id=auth_client_id, payload={"name": "test-authz-scope"}
  868. )
  869. assert res["name"] == "test-authz-scope", res
  870. with pytest.raises(KeycloakPostError) as err:
  871. admin.create_client_authz_scopes(
  872. client_id="invalid_client_id", payload={"name": "test-authz-scope"}
  873. )
  874. assert err.match('404: b\'{"error":"Could not find client"')
  875. assert admin.create_client_authz_scopes(
  876. client_id=auth_client_id, payload={"name": "test-authz-scope"}
  877. )
  878. res = admin.get_client_authz_scopes(client_id=auth_client_id)
  879. assert len(res) == 1
  880. assert {x["name"] for x in res} == {"test-authz-scope"}
  881. # Test service account user
  882. res = admin.get_client_service_account_user(client_id=auth_client_id)
  883. assert res["username"] == "service-account-authz-client", res
  884. with pytest.raises(KeycloakGetError) as err:
  885. admin.get_client_service_account_user(client_id=client_id)
  886. assert err.match('400: b\'{"error":"unknown_error"}\'')
  887. # Test delete client
  888. res = admin.delete_client(client_id=auth_client_id)
  889. assert res == dict(), res
  890. with pytest.raises(KeycloakDeleteError) as err:
  891. admin.delete_client(client_id=auth_client_id)
  892. assert err.match('404: b\'{"error":"Could not find client"}\'')
  893. # Test client credentials
  894. admin.create_client(
  895. payload={
  896. "name": "test-confidential",
  897. "enabled": True,
  898. "protocol": "openid-connect",
  899. "publicClient": False,
  900. "redirectUris": ["http://localhost/*"],
  901. "webOrigins": ["+"],
  902. "clientId": "test-confidential",
  903. "secret": "test-secret",
  904. "clientAuthenticatorType": "client-secret",
  905. }
  906. )
  907. with pytest.raises(KeycloakGetError) as err:
  908. admin.get_client_secrets(client_id="does-not-exist")
  909. assert err.match('404: b\'{"error":"Could not find client"}\'')
  910. secrets = admin.get_client_secrets(
  911. client_id=admin.get_client_id(client_id="test-confidential")
  912. )
  913. assert secrets == {"type": "secret", "value": "test-secret"}
  914. with pytest.raises(KeycloakPostError) as err:
  915. admin.generate_client_secrets(client_id="does-not-exist")
  916. assert err.match('404: b\'{"error":"Could not find client"}\'')
  917. res = admin.generate_client_secrets(
  918. client_id=admin.get_client_id(client_id="test-confidential")
  919. )
  920. assert res
  921. assert (
  922. admin.get_client_secrets(client_id=admin.get_client_id(client_id="test-confidential"))
  923. == res
  924. )
  925. def test_realm_roles(admin: KeycloakAdmin, realm: str):
  926. """Test realm roles.
  927. :param admin: Keycloak Admin client
  928. :type admin: KeycloakAdmin
  929. :param realm: Keycloak realm
  930. :type realm: str
  931. """
  932. admin.change_current_realm(realm)
  933. # Test get realm roles
  934. roles = admin.get_realm_roles()
  935. assert len(roles) == 3, roles
  936. role_names = [x["name"] for x in roles]
  937. assert "uma_authorization" in role_names, role_names
  938. assert "offline_access" in role_names, role_names
  939. # Test get realm roles with search text
  940. searched_roles = admin.get_realm_roles(search_text="uma_a")
  941. searched_role_names = [x["name"] for x in searched_roles]
  942. assert "uma_authorization" in searched_role_names, searched_role_names
  943. assert "offline_access" not in searched_role_names, searched_role_names
  944. # Test empty members
  945. with pytest.raises(KeycloakGetError) as err:
  946. admin.get_realm_role_members(role_name="does-not-exist")
  947. assert err.match('404: b\'{"error":"Could not find role"}\'')
  948. members = admin.get_realm_role_members(role_name="offline_access")
  949. assert members == list(), members
  950. # Test create realm role
  951. role_id = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  952. assert role_id, role_id
  953. with pytest.raises(KeycloakPostError) as err:
  954. admin.create_realm_role(payload={"name": "test-realm-role"})
  955. assert err.match('409: b\'{"errorMessage":"Role with name test-realm-role already exists"}\'')
  956. role_id_2 = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  957. assert role_id == role_id_2
  958. # Test get realm role by its id
  959. role_id = admin.get_realm_role(role_name="test-realm-role")["id"]
  960. res = admin.get_realm_role_by_id(role_id)
  961. assert res["name"] == "test-realm-role"
  962. # Test update realm role
  963. res = admin.update_realm_role(
  964. role_name="test-realm-role", payload={"name": "test-realm-role-update"}
  965. )
  966. assert res == dict(), res
  967. with pytest.raises(KeycloakPutError) as err:
  968. admin.update_realm_role(
  969. role_name="test-realm-role", payload={"name": "test-realm-role-update"}
  970. )
  971. assert err.match('404: b\'{"error":"Could not find role"}\''), err
  972. # Test realm role user assignment
  973. user_id = admin.create_user(payload={"username": "role-testing", "email": "test@test.test"})
  974. with pytest.raises(KeycloakPostError) as err:
  975. admin.assign_realm_roles(user_id=user_id, roles=["bad"])
  976. assert err.match('b\'{"error":"unknown_error"}\''), err
  977. res = admin.assign_realm_roles(
  978. user_id=user_id,
  979. roles=[
  980. admin.get_realm_role(role_name="offline_access"),
  981. admin.get_realm_role(role_name="test-realm-role-update"),
  982. ],
  983. )
  984. assert res == dict(), res
  985. assert admin.get_user(user_id=user_id)["username"] in [
  986. x["username"] for x in admin.get_realm_role_members(role_name="offline_access")
  987. ]
  988. assert admin.get_user(user_id=user_id)["username"] in [
  989. x["username"] for x in admin.get_realm_role_members(role_name="test-realm-role-update")
  990. ]
  991. roles = admin.get_realm_roles_of_user(user_id=user_id)
  992. assert len(roles) == 3
  993. assert "offline_access" in [x["name"] for x in roles]
  994. assert "test-realm-role-update" in [x["name"] for x in roles]
  995. with pytest.raises(KeycloakDeleteError) as err:
  996. admin.delete_realm_roles_of_user(user_id=user_id, roles=["bad"])
  997. assert err.match('b\'{"error":"unknown_error"}\''), err
  998. res = admin.delete_realm_roles_of_user(
  999. user_id=user_id, roles=[admin.get_realm_role(role_name="offline_access")]
  1000. )
  1001. assert res == dict(), res
  1002. assert admin.get_realm_role_members(role_name="offline_access") == list()
  1003. roles = admin.get_realm_roles_of_user(user_id=user_id)
  1004. assert len(roles) == 2
  1005. assert "offline_access" not in [x["name"] for x in roles]
  1006. assert "test-realm-role-update" in [x["name"] for x in roles]
  1007. roles = admin.get_available_realm_roles_of_user(user_id=user_id)
  1008. assert len(roles) == 2
  1009. assert "offline_access" in [x["name"] for x in roles]
  1010. assert "uma_authorization" in [x["name"] for x in roles]
  1011. # Test realm role group assignment
  1012. group_id = admin.create_group(payload={"name": "test-group"})
  1013. with pytest.raises(KeycloakPostError) as err:
  1014. admin.assign_group_realm_roles(group_id=group_id, roles=["bad"])
  1015. assert err.match('b\'{"error":"unknown_error"}\''), err
  1016. res = admin.assign_group_realm_roles(
  1017. group_id=group_id,
  1018. roles=[
  1019. admin.get_realm_role(role_name="offline_access"),
  1020. admin.get_realm_role(role_name="test-realm-role-update"),
  1021. ],
  1022. )
  1023. assert res == dict(), res
  1024. roles = admin.get_group_realm_roles(group_id=group_id)
  1025. assert len(roles) == 2
  1026. assert "offline_access" in [x["name"] for x in roles]
  1027. assert "test-realm-role-update" in [x["name"] for x in roles]
  1028. with pytest.raises(KeycloakDeleteError) as err:
  1029. admin.delete_group_realm_roles(group_id=group_id, roles=["bad"])
  1030. assert err.match('b\'{"error":"unknown_error"}\''), err
  1031. res = admin.delete_group_realm_roles(
  1032. group_id=group_id, roles=[admin.get_realm_role(role_name="offline_access")]
  1033. )
  1034. assert res == dict(), res
  1035. roles = admin.get_group_realm_roles(group_id=group_id)
  1036. assert len(roles) == 1
  1037. assert "test-realm-role-update" in [x["name"] for x in roles]
  1038. # Test composite realm roles
  1039. composite_role = admin.create_realm_role(payload={"name": "test-composite-role"})
  1040. with pytest.raises(KeycloakPostError) as err:
  1041. admin.add_composite_realm_roles_to_role(role_name=composite_role, roles=["bad"])
  1042. assert err.match('b\'{"error":"unknown_error"}\''), err
  1043. res = admin.add_composite_realm_roles_to_role(
  1044. role_name=composite_role, roles=[admin.get_realm_role(role_name="test-realm-role-update")]
  1045. )
  1046. assert res == dict(), res
  1047. res = admin.get_composite_realm_roles_of_role(role_name=composite_role)
  1048. assert len(res) == 1
  1049. assert "test-realm-role-update" in res[0]["name"]
  1050. with pytest.raises(KeycloakGetError) as err:
  1051. admin.get_composite_realm_roles_of_role(role_name="bad")
  1052. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1053. res = admin.get_composite_realm_roles_of_user(user_id=user_id)
  1054. assert len(res) == 4
  1055. assert "offline_access" in {x["name"] for x in res}
  1056. assert "test-realm-role-update" in {x["name"] for x in res}
  1057. assert "uma_authorization" in {x["name"] for x in res}
  1058. with pytest.raises(KeycloakGetError) as err:
  1059. admin.get_composite_realm_roles_of_user(user_id="bad")
  1060. assert err.match('b\'{"error":"User not found"}\''), err
  1061. with pytest.raises(KeycloakDeleteError) as err:
  1062. admin.remove_composite_realm_roles_to_role(role_name=composite_role, roles=["bad"])
  1063. assert err.match('b\'{"error":"unknown_error"}\''), err
  1064. res = admin.remove_composite_realm_roles_to_role(
  1065. role_name=composite_role, roles=[admin.get_realm_role(role_name="test-realm-role-update")]
  1066. )
  1067. assert res == dict(), res
  1068. res = admin.get_composite_realm_roles_of_role(role_name=composite_role)
  1069. assert len(res) == 0
  1070. # Test realm role group list
  1071. res = admin.get_realm_role_groups(role_name="test-realm-role-update")
  1072. assert len(res) == 1
  1073. assert res[0]["id"] == group_id
  1074. with pytest.raises(KeycloakGetError) as err:
  1075. admin.get_realm_role_groups(role_name="non-existent-role")
  1076. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1077. # Test delete realm role
  1078. res = admin.delete_realm_role(role_name=composite_role)
  1079. assert res == dict(), res
  1080. with pytest.raises(KeycloakDeleteError) as err:
  1081. admin.delete_realm_role(role_name=composite_role)
  1082. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1083. @pytest.mark.parametrize(
  1084. "testcase, arg_brief_repr, includes_attributes",
  1085. [
  1086. ("brief True", {"brief_representation": True}, False),
  1087. ("brief False", {"brief_representation": False}, True),
  1088. ("default", {}, False),
  1089. ],
  1090. )
  1091. def test_role_attributes(
  1092. admin: KeycloakAdmin,
  1093. realm: str,
  1094. client: str,
  1095. arg_brief_repr: dict,
  1096. includes_attributes: bool,
  1097. testcase: str,
  1098. ):
  1099. """Test getting role attributes for bulk calls.
  1100. :param admin: Keycloak admin
  1101. :type admin: KeycloakAdmin
  1102. :param realm: Keycloak realm
  1103. :type realm: str
  1104. :param client: Keycloak client
  1105. :type client: str
  1106. :param arg_brief_repr: Brief representation
  1107. :type arg_brief_repr: dict
  1108. :param includes_attributes: Indicator whether to include attributes
  1109. :type includes_attributes: bool
  1110. :param testcase: Test case
  1111. :type testcase: str
  1112. """
  1113. # setup
  1114. attribute_role = "test-realm-role-w-attr"
  1115. test_attrs = {"attr1": ["val1"], "attr2": ["val2-1", "val2-2"]}
  1116. role_id = admin.create_realm_role(
  1117. payload={"name": attribute_role, "attributes": test_attrs}, skip_exists=True
  1118. )
  1119. assert role_id, role_id
  1120. cli_role_id = admin.create_client_role(
  1121. client, payload={"name": attribute_role, "attributes": test_attrs}, skip_exists=True
  1122. )
  1123. assert cli_role_id, cli_role_id
  1124. if not includes_attributes:
  1125. test_attrs = None
  1126. # tests
  1127. roles = admin.get_realm_roles(**arg_brief_repr)
  1128. roles_filtered = [role for role in roles if role["name"] == role_id]
  1129. assert roles_filtered, roles_filtered
  1130. role = roles_filtered[0]
  1131. assert role.get("attributes") == test_attrs, testcase
  1132. roles = admin.get_client_roles(client, **arg_brief_repr)
  1133. roles_filtered = [role for role in roles if role["name"] == cli_role_id]
  1134. assert roles_filtered, roles_filtered
  1135. role = roles_filtered[0]
  1136. assert role.get("attributes") == test_attrs, testcase
  1137. # cleanup
  1138. res = admin.delete_realm_role(role_name=attribute_role)
  1139. assert res == dict(), res
  1140. res = admin.delete_client_role(client, role_name=attribute_role)
  1141. assert res == dict(), res
  1142. def test_client_scope_realm_roles(admin: KeycloakAdmin, realm: str):
  1143. """Test client realm roles.
  1144. :param admin: Keycloak admin
  1145. :type admin: KeycloakAdmin
  1146. :param realm: Keycloak realm
  1147. :type realm: str
  1148. """
  1149. admin.change_current_realm(realm)
  1150. # Test get realm roles
  1151. roles = admin.get_realm_roles()
  1152. assert len(roles) == 3, roles
  1153. role_names = [x["name"] for x in roles]
  1154. assert "uma_authorization" in role_names, role_names
  1155. assert "offline_access" in role_names, role_names
  1156. # create realm role for test
  1157. role_id = admin.create_realm_role(payload={"name": "test-realm-role"}, skip_exists=True)
  1158. assert role_id, role_id
  1159. # Test realm role client assignment
  1160. client_id = admin.create_client(
  1161. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1162. )
  1163. with pytest.raises(KeycloakPostError) as err:
  1164. admin.assign_realm_roles_to_client_scope(client_id=client_id, roles=["bad"])
  1165. assert err.match('b\'{"error":"unknown_error"}\''), err
  1166. res = admin.assign_realm_roles_to_client_scope(
  1167. client_id=client_id,
  1168. roles=[
  1169. admin.get_realm_role(role_name="offline_access"),
  1170. admin.get_realm_role(role_name="test-realm-role"),
  1171. ],
  1172. )
  1173. assert res == dict(), res
  1174. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1175. assert len(roles) == 2
  1176. client_role_names = [x["name"] for x in roles]
  1177. assert "offline_access" in client_role_names, client_role_names
  1178. assert "test-realm-role" in client_role_names, client_role_names
  1179. assert "uma_authorization" not in client_role_names, client_role_names
  1180. # Test remove realm role of client
  1181. with pytest.raises(KeycloakDeleteError) as err:
  1182. admin.delete_realm_roles_of_client_scope(client_id=client_id, roles=["bad"])
  1183. assert err.match('b\'{"error":"unknown_error"}\''), err
  1184. res = admin.delete_realm_roles_of_client_scope(
  1185. client_id=client_id, roles=[admin.get_realm_role(role_name="offline_access")]
  1186. )
  1187. assert res == dict(), res
  1188. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1189. assert len(roles) == 1
  1190. assert "test-realm-role" in [x["name"] for x in roles]
  1191. res = admin.delete_realm_roles_of_client_scope(
  1192. client_id=client_id, roles=[admin.get_realm_role(role_name="test-realm-role")]
  1193. )
  1194. assert res == dict(), res
  1195. roles = admin.get_realm_roles_of_client_scope(client_id=client_id)
  1196. assert len(roles) == 0
  1197. def test_client_scope_client_roles(admin: KeycloakAdmin, realm: str, client: str):
  1198. """Test client assignment of other client roles.
  1199. :param admin: Keycloak admin
  1200. :type admin: KeycloakAdmin
  1201. :param realm: Keycloak realm
  1202. :type realm: str
  1203. :param client: Keycloak client
  1204. :type client: str
  1205. """
  1206. admin.change_current_realm(realm)
  1207. client_id = admin.create_client(
  1208. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1209. )
  1210. # Test get client roles
  1211. roles = admin.get_client_roles_of_client_scope(client_id, client)
  1212. assert len(roles) == 0, roles
  1213. # create client role for test
  1214. client_role_id = admin.create_client_role(
  1215. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1216. )
  1217. assert client_role_id, client_role_id
  1218. # Test client role assignment to other client
  1219. with pytest.raises(KeycloakPostError) as err:
  1220. admin.assign_client_roles_to_client_scope(
  1221. client_id=client_id, client_roles_owner_id=client, roles=["bad"]
  1222. )
  1223. assert err.match('b\'{"error":"unknown_error"}\''), err
  1224. res = admin.assign_client_roles_to_client_scope(
  1225. client_id=client_id,
  1226. client_roles_owner_id=client,
  1227. roles=[admin.get_client_role(client_id=client, role_name="client-role-test")],
  1228. )
  1229. assert res == dict(), res
  1230. roles = admin.get_client_roles_of_client_scope(
  1231. client_id=client_id, client_roles_owner_id=client
  1232. )
  1233. assert len(roles) == 1
  1234. client_role_names = [x["name"] for x in roles]
  1235. assert "client-role-test" in client_role_names, client_role_names
  1236. # Test remove realm role of client
  1237. with pytest.raises(KeycloakDeleteError) as err:
  1238. admin.delete_client_roles_of_client_scope(
  1239. client_id=client_id, client_roles_owner_id=client, roles=["bad"]
  1240. )
  1241. assert err.match('b\'{"error":"unknown_error"}\''), err
  1242. res = admin.delete_client_roles_of_client_scope(
  1243. client_id=client_id,
  1244. client_roles_owner_id=client,
  1245. roles=[admin.get_client_role(client_id=client, role_name="client-role-test")],
  1246. )
  1247. assert res == dict(), res
  1248. roles = admin.get_client_roles_of_client_scope(
  1249. client_id=client_id, client_roles_owner_id=client
  1250. )
  1251. assert len(roles) == 0
  1252. def test_client_default_client_scopes(admin: KeycloakAdmin, realm: str, client: str):
  1253. """Test client assignment of default client scopes.
  1254. :param admin: Keycloak admin
  1255. :type admin: KeycloakAdmin
  1256. :param realm: Keycloak realm
  1257. :type realm: str
  1258. :param client: Keycloak client
  1259. :type client: str
  1260. """
  1261. admin.change_current_realm(realm)
  1262. client_id = admin.create_client(
  1263. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1264. )
  1265. # Test get client default scopes
  1266. # keycloak default roles: web-origins, acr, profile, roles, email
  1267. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1268. assert len(default_client_scopes) == 5, default_client_scopes
  1269. # Test add a client scope to client default scopes
  1270. default_client_scope = "test-client-default-scope"
  1271. new_client_scope = {
  1272. "name": default_client_scope,
  1273. "description": f"Test Client Scope: {default_client_scope}",
  1274. "protocol": "openid-connect",
  1275. "attributes": {},
  1276. }
  1277. new_client_scope_id = admin.create_client_scope(new_client_scope, skip_exists=False)
  1278. new_default_client_scope_data = {
  1279. "realm": realm,
  1280. "client": client_id,
  1281. "clientScopeId": new_client_scope_id,
  1282. }
  1283. admin.add_client_default_client_scope(
  1284. client_id, new_client_scope_id, new_default_client_scope_data
  1285. )
  1286. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1287. assert len(default_client_scopes) == 6, default_client_scopes
  1288. # Test remove a client default scope
  1289. admin.delete_client_default_client_scope(client_id, new_client_scope_id)
  1290. default_client_scopes = admin.get_client_default_client_scopes(client_id)
  1291. assert len(default_client_scopes) == 5, default_client_scopes
  1292. def test_client_optional_client_scopes(admin: KeycloakAdmin, realm: str, client: str):
  1293. """Test client assignment of optional client scopes.
  1294. :param admin: Keycloak admin
  1295. :type admin: KeycloakAdmin
  1296. :param realm: Keycloak realm
  1297. :type realm: str
  1298. :param client: Keycloak client
  1299. :type client: str
  1300. """
  1301. admin.change_current_realm(realm)
  1302. client_id = admin.create_client(
  1303. payload={"name": "role-testing-client", "clientId": "role-testing-client"}
  1304. )
  1305. # Test get client optional scopes
  1306. # keycloak optional roles: microprofile-jwt, offline_access, address, phone
  1307. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1308. assert len(optional_client_scopes) == 4, optional_client_scopes
  1309. # Test add a client scope to client optional scopes
  1310. optional_client_scope = "test-client-optional-scope"
  1311. new_client_scope = {
  1312. "name": optional_client_scope,
  1313. "description": f"Test Client Scope: {optional_client_scope}",
  1314. "protocol": "openid-connect",
  1315. "attributes": {},
  1316. }
  1317. new_client_scope_id = admin.create_client_scope(new_client_scope, skip_exists=False)
  1318. new_optional_client_scope_data = {
  1319. "realm": realm,
  1320. "client": client_id,
  1321. "clientScopeId": new_client_scope_id,
  1322. }
  1323. admin.add_client_optional_client_scope(
  1324. client_id, new_client_scope_id, new_optional_client_scope_data
  1325. )
  1326. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1327. assert len(optional_client_scopes) == 5, optional_client_scopes
  1328. # Test remove a client optional scope
  1329. admin.delete_client_optional_client_scope(client_id, new_client_scope_id)
  1330. optional_client_scopes = admin.get_client_optional_client_scopes(client_id)
  1331. assert len(optional_client_scopes) == 4, optional_client_scopes
  1332. def test_client_roles(admin: KeycloakAdmin, client: str):
  1333. """Test client roles.
  1334. :param admin: Keycloak Admin client
  1335. :type admin: KeycloakAdmin
  1336. :param client: Keycloak client
  1337. :type client: str
  1338. """
  1339. # Test get client roles
  1340. res = admin.get_client_roles(client_id=client)
  1341. assert len(res) == 0
  1342. with pytest.raises(KeycloakGetError) as err:
  1343. admin.get_client_roles(client_id="bad")
  1344. assert err.match('404: b\'{"error":"Could not find client"}\'')
  1345. # Test create client role
  1346. client_role_id = admin.create_client_role(
  1347. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1348. )
  1349. with pytest.raises(KeycloakPostError) as err:
  1350. admin.create_client_role(client_role_id=client, payload={"name": "client-role-test"})
  1351. assert err.match('409: b\'{"errorMessage":"Role with name client-role-test already exists"}\'')
  1352. client_role_id_2 = admin.create_client_role(
  1353. client_role_id=client, payload={"name": "client-role-test"}, skip_exists=True
  1354. )
  1355. assert client_role_id == client_role_id_2
  1356. # Test get client role
  1357. res = admin.get_client_role(client_id=client, role_name="client-role-test")
  1358. assert res["name"] == client_role_id
  1359. with pytest.raises(KeycloakGetError) as err:
  1360. admin.get_client_role(client_id=client, role_name="bad")
  1361. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1362. res_ = admin.get_client_role_id(client_id=client, role_name="client-role-test")
  1363. assert res_ == res["id"]
  1364. with pytest.raises(KeycloakGetError) as err:
  1365. admin.get_client_role_id(client_id=client, role_name="bad")
  1366. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1367. assert len(admin.get_client_roles(client_id=client)) == 1
  1368. # Test update client role
  1369. res = admin.update_client_role(
  1370. client_role_id=client,
  1371. role_name="client-role-test",
  1372. payload={"name": "client-role-test-update"},
  1373. )
  1374. assert res == dict()
  1375. with pytest.raises(KeycloakPutError) as err:
  1376. res = admin.update_client_role(
  1377. client_role_id=client,
  1378. role_name="client-role-test",
  1379. payload={"name": "client-role-test-update"},
  1380. )
  1381. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1382. # Test user with client role
  1383. res = admin.get_client_role_members(client_id=client, role_name="client-role-test-update")
  1384. assert len(res) == 0
  1385. with pytest.raises(KeycloakGetError) as err:
  1386. admin.get_client_role_members(client_id=client, role_name="bad")
  1387. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1388. user_id = admin.create_user(payload={"username": "test", "email": "test@test.test"})
  1389. with pytest.raises(KeycloakPostError) as err:
  1390. admin.assign_client_role(user_id=user_id, client_id=client, roles=["bad"])
  1391. assert err.match('b\'{"error":"unknown_error"}\''), err
  1392. res = admin.assign_client_role(
  1393. user_id=user_id,
  1394. client_id=client,
  1395. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1396. )
  1397. assert res == dict()
  1398. assert (
  1399. len(admin.get_client_role_members(client_id=client, role_name="client-role-test-update"))
  1400. == 1
  1401. )
  1402. roles = admin.get_client_roles_of_user(user_id=user_id, client_id=client)
  1403. assert len(roles) == 1, roles
  1404. with pytest.raises(KeycloakGetError) as err:
  1405. admin.get_client_roles_of_user(user_id=user_id, client_id="bad")
  1406. assert err.match('404: b\'{"error":"Client not found"}\'')
  1407. roles = admin.get_composite_client_roles_of_user(user_id=user_id, client_id=client)
  1408. assert len(roles) == 1, roles
  1409. with pytest.raises(KeycloakGetError) as err:
  1410. admin.get_composite_client_roles_of_user(user_id=user_id, client_id="bad")
  1411. assert err.match('404: b\'{"error":"Client not found"}\'')
  1412. roles = admin.get_available_client_roles_of_user(user_id=user_id, client_id=client)
  1413. assert len(roles) == 0, roles
  1414. with pytest.raises(KeycloakGetError) as err:
  1415. admin.get_composite_client_roles_of_user(user_id=user_id, client_id="bad")
  1416. assert err.match('404: b\'{"error":"Client not found"}\'')
  1417. with pytest.raises(KeycloakDeleteError) as err:
  1418. admin.delete_client_roles_of_user(user_id=user_id, client_id=client, roles=["bad"])
  1419. assert err.match('b\'{"error":"unknown_error"}\''), err
  1420. admin.delete_client_roles_of_user(
  1421. user_id=user_id,
  1422. client_id=client,
  1423. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1424. )
  1425. assert len(admin.get_client_roles_of_user(user_id=user_id, client_id=client)) == 0
  1426. # Test groups and client roles
  1427. res = admin.get_client_role_groups(client_id=client, role_name="client-role-test-update")
  1428. assert len(res) == 0
  1429. with pytest.raises(KeycloakGetError) as err:
  1430. admin.get_client_role_groups(client_id=client, role_name="bad")
  1431. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1432. group_id = admin.create_group(payload={"name": "test-group"})
  1433. res = admin.get_group_client_roles(group_id=group_id, client_id=client)
  1434. assert len(res) == 0
  1435. with pytest.raises(KeycloakGetError) as err:
  1436. admin.get_group_client_roles(group_id=group_id, client_id="bad")
  1437. assert err.match('404: b\'{"error":"Client not found"}\'')
  1438. with pytest.raises(KeycloakPostError) as err:
  1439. admin.assign_group_client_roles(group_id=group_id, client_id=client, roles=["bad"])
  1440. assert err.match('b\'{"error":"unknown_error"}\''), err
  1441. res = admin.assign_group_client_roles(
  1442. group_id=group_id,
  1443. client_id=client,
  1444. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1445. )
  1446. assert res == dict()
  1447. assert (
  1448. len(admin.get_client_role_groups(client_id=client, role_name="client-role-test-update"))
  1449. == 1
  1450. )
  1451. assert len(admin.get_group_client_roles(group_id=group_id, client_id=client)) == 1
  1452. with pytest.raises(KeycloakDeleteError) as err:
  1453. admin.delete_group_client_roles(group_id=group_id, client_id=client, roles=["bad"])
  1454. assert err.match('b\'{"error":"unknown_error"}\''), err
  1455. res = admin.delete_group_client_roles(
  1456. group_id=group_id,
  1457. client_id=client,
  1458. roles=[admin.get_client_role(client_id=client, role_name="client-role-test-update")],
  1459. )
  1460. assert res == dict()
  1461. # Test composite client roles
  1462. with pytest.raises(KeycloakPostError) as err:
  1463. admin.add_composite_client_roles_to_role(
  1464. client_role_id=client, role_name="client-role-test-update", roles=["bad"]
  1465. )
  1466. assert err.match('b\'{"error":"unknown_error"}\''), err
  1467. res = admin.add_composite_client_roles_to_role(
  1468. client_role_id=client,
  1469. role_name="client-role-test-update",
  1470. roles=[admin.get_realm_role(role_name="offline_access")],
  1471. )
  1472. assert res == dict()
  1473. assert admin.get_client_role(client_id=client, role_name="client-role-test-update")[
  1474. "composite"
  1475. ]
  1476. # Test delete of client role
  1477. res = admin.delete_client_role(client_role_id=client, role_name="client-role-test-update")
  1478. assert res == dict()
  1479. with pytest.raises(KeycloakDeleteError) as err:
  1480. admin.delete_client_role(client_role_id=client, role_name="client-role-test-update")
  1481. assert err.match('404: b\'{"error":"Could not find role"}\'')
  1482. def test_enable_token_exchange(admin: KeycloakAdmin, realm: str):
  1483. """Test enable token exchange.
  1484. :param admin: Keycloak Admin client
  1485. :type admin: KeycloakAdmin
  1486. :param realm: Keycloak realm
  1487. :type realm: str
  1488. :raises AssertionError: In case of bad configuration
  1489. """
  1490. # Test enabling token exchange between two confidential clients
  1491. admin.change_current_realm(realm)
  1492. # Create test clients
  1493. source_client_id = admin.create_client(
  1494. payload={"name": "Source Client", "clientId": "source-client"}
  1495. )
  1496. target_client_id = admin.create_client(
  1497. payload={"name": "Target Client", "clientId": "target-client"}
  1498. )
  1499. for c in admin.get_clients():
  1500. if c["clientId"] == "realm-management":
  1501. realm_management_id = c["id"]
  1502. break
  1503. else:
  1504. raise AssertionError("Missing realm management client")
  1505. # Enable permissions on the Superset client
  1506. admin.update_client_management_permissions(
  1507. payload={"enabled": True}, client_id=target_client_id
  1508. )
  1509. # Fetch various IDs and strings needed when creating the permission
  1510. token_exchange_permission_id = admin.get_client_management_permissions(
  1511. client_id=target_client_id
  1512. )["scopePermissions"]["token-exchange"]
  1513. scopes = admin.get_client_authz_policy_scopes(
  1514. client_id=realm_management_id, policy_id=token_exchange_permission_id
  1515. )
  1516. for s in scopes:
  1517. if s["name"] == "token-exchange":
  1518. token_exchange_scope_id = s["id"]
  1519. break
  1520. else:
  1521. raise AssertionError("Missing token-exchange scope")
  1522. resources = admin.get_client_authz_policy_resources(
  1523. client_id=realm_management_id, policy_id=token_exchange_permission_id
  1524. )
  1525. for r in resources:
  1526. if r["name"] == f"client.resource.{target_client_id}":
  1527. token_exchange_resource_id = r["_id"]
  1528. break
  1529. else:
  1530. raise AssertionError("Missing client resource")
  1531. # Create a client policy for source client
  1532. policy_name = "Exchange source client token with target client token"
  1533. client_policy_id = admin.create_client_authz_client_policy(
  1534. payload={
  1535. "type": "client",
  1536. "logic": "POSITIVE",
  1537. "decisionStrategy": "UNANIMOUS",
  1538. "name": policy_name,
  1539. "clients": [source_client_id],
  1540. },
  1541. client_id=realm_management_id,
  1542. )["id"]
  1543. policies = admin.get_client_authz_client_policies(client_id=realm_management_id)
  1544. for policy in policies:
  1545. if policy["name"] == policy_name:
  1546. assert policy["clients"] == [source_client_id]
  1547. break
  1548. else:
  1549. raise AssertionError("Missing client policy")
  1550. # Update permissions on the target client to reference this policy
  1551. permission_name = admin.get_client_authz_scope_permission(
  1552. client_id=realm_management_id, scope_id=token_exchange_permission_id
  1553. )["name"]
  1554. admin.update_client_authz_scope_permission(
  1555. payload={
  1556. "id": token_exchange_permission_id,
  1557. "name": permission_name,
  1558. "type": "scope",
  1559. "logic": "POSITIVE",
  1560. "decisionStrategy": "UNANIMOUS",
  1561. "resources": [token_exchange_resource_id],
  1562. "scopes": [token_exchange_scope_id],
  1563. "policies": [client_policy_id],
  1564. },
  1565. client_id=realm_management_id,
  1566. scope_id=token_exchange_permission_id,
  1567. )
  1568. # Create permissions on the target client to reference this policy
  1569. admin.create_client_authz_scope_permission(
  1570. payload={
  1571. "id": "some-id",
  1572. "name": "test-permission",
  1573. "type": "scope",
  1574. "logic": "POSITIVE",
  1575. "decisionStrategy": "UNANIMOUS",
  1576. "resources": [token_exchange_resource_id],
  1577. "scopes": [token_exchange_scope_id],
  1578. "policies": [client_policy_id],
  1579. },
  1580. client_id=realm_management_id,
  1581. )
  1582. permission_name = admin.get_client_authz_scope_permission(
  1583. client_id=realm_management_id, scope_id=token_exchange_permission_id
  1584. )["name"]
  1585. assert permission_name.startswith("token-exchange.permission.client.")
  1586. with pytest.raises(KeycloakPostError) as err:
  1587. admin.create_client_authz_scope_permission(
  1588. payload={"name": "test-permission", "scopes": [token_exchange_scope_id]},
  1589. client_id="realm_management_id",
  1590. )
  1591. assert err.match('404: b\'{"error":"Could not find client"}\'')
  1592. def test_email(admin: KeycloakAdmin, user: str):
  1593. """Test email.
  1594. :param admin: Keycloak Admin client
  1595. :type admin: KeycloakAdmin
  1596. :param user: Keycloak user
  1597. :type user: str
  1598. """
  1599. # Emails will fail as we don't have SMTP test setup
  1600. with pytest.raises(KeycloakPutError) as err:
  1601. admin.send_update_account(user_id=user, payload=dict())
  1602. assert err.match('b\'{"error":"unknown_error"}\''), err
  1603. admin.update_user(user_id=user, payload={"enabled": True})
  1604. with pytest.raises(KeycloakPutError) as err:
  1605. admin.send_verify_email(user_id=user)
  1606. assert err.match('500: b\'{"errorMessage":"Failed to send execute actions email"}\'')
  1607. def test_get_sessions(admin: KeycloakAdmin):
  1608. """Test get sessions.
  1609. :param admin: Keycloak Admin client
  1610. :type admin: KeycloakAdmin
  1611. """
  1612. sessions = admin.get_sessions(user_id=admin.get_user_id(username=admin.username))
  1613. assert len(sessions) >= 1
  1614. with pytest.raises(KeycloakGetError) as err:
  1615. admin.get_sessions(user_id="bad")
  1616. assert err.match('404: b\'{"error":"User not found"}\'')
  1617. def test_get_client_installation_provider(admin: KeycloakAdmin, client: str):
  1618. """Test get client installation provider.
  1619. :param admin: Keycloak Admin client
  1620. :type admin: KeycloakAdmin
  1621. :param client: Keycloak client
  1622. :type client: str
  1623. """
  1624. with pytest.raises(KeycloakGetError) as err:
  1625. admin.get_client_installation_provider(client_id=client, provider_id="bad")
  1626. assert err.match('404: b\'{"error":"Unknown Provider"}\'')
  1627. installation = admin.get_client_installation_provider(
  1628. client_id=client, provider_id="keycloak-oidc-keycloak-json"
  1629. )
  1630. assert set(installation.keys()) == {
  1631. "auth-server-url",
  1632. "confidential-port",
  1633. "credentials",
  1634. "realm",
  1635. "resource",
  1636. "ssl-required",
  1637. }
  1638. def test_auth_flows(admin: KeycloakAdmin, realm: str):
  1639. """Test auth flows.
  1640. :param admin: Keycloak Admin client
  1641. :type admin: KeycloakAdmin
  1642. :param realm: Keycloak realm
  1643. :type realm: str
  1644. """
  1645. admin.change_current_realm(realm)
  1646. res = admin.get_authentication_flows()
  1647. assert len(res) <= 8, res
  1648. default_flows = len(res)
  1649. assert {x["alias"] for x in res}.issubset(
  1650. {
  1651. "reset credentials",
  1652. "browser",
  1653. "registration",
  1654. "http challenge",
  1655. "docker auth",
  1656. "direct grant",
  1657. "first broker login",
  1658. "clients",
  1659. }
  1660. )
  1661. assert set(res[0].keys()) == {
  1662. "alias",
  1663. "authenticationExecutions",
  1664. "builtIn",
  1665. "description",
  1666. "id",
  1667. "providerId",
  1668. "topLevel",
  1669. }
  1670. assert {x["alias"] for x in res}.issubset(
  1671. {
  1672. "reset credentials",
  1673. "browser",
  1674. "registration",
  1675. "docker auth",
  1676. "direct grant",
  1677. "first broker login",
  1678. "clients",
  1679. "http challenge",
  1680. }
  1681. )
  1682. with pytest.raises(KeycloakGetError) as err:
  1683. admin.get_authentication_flow_for_id(flow_id="bad")
  1684. assert err.match('404: b\'{"error":"Could not find flow with id"}\'')
  1685. browser_flow_id = [x for x in res if x["alias"] == "browser"][0]["id"]
  1686. res = admin.get_authentication_flow_for_id(flow_id=browser_flow_id)
  1687. assert res["alias"] == "browser"
  1688. # Test copying
  1689. with pytest.raises(KeycloakPostError) as err:
  1690. admin.copy_authentication_flow(payload=dict(), flow_alias="bad")
  1691. assert err.match("404: b''")
  1692. res = admin.copy_authentication_flow(payload={"newName": "test-browser"}, flow_alias="browser")
  1693. assert res == b"", res
  1694. assert len(admin.get_authentication_flows()) == (default_flows + 1)
  1695. # Test create
  1696. res = admin.create_authentication_flow(
  1697. payload={"alias": "test-create", "providerId": "basic-flow"}
  1698. )
  1699. assert res == b""
  1700. with pytest.raises(KeycloakPostError) as err:
  1701. admin.create_authentication_flow(payload={"alias": "test-create", "builtIn": False})
  1702. assert err.match('409: b\'{"errorMessage":"Flow test-create already exists"}\'')
  1703. assert admin.create_authentication_flow(
  1704. payload={"alias": "test-create"}, skip_exists=True
  1705. ) == {"msg": "Already exists"}
  1706. # Test flow executions
  1707. res = admin.get_authentication_flow_executions(flow_alias="browser")
  1708. assert len(res) == 8, res
  1709. with pytest.raises(KeycloakGetError) as err:
  1710. admin.get_authentication_flow_executions(flow_alias="bad")
  1711. assert err.match("404: b''")
  1712. exec_id = res[0]["id"]
  1713. res = admin.get_authentication_flow_execution(execution_id=exec_id)
  1714. assert set(res.keys()) == {
  1715. "alternative",
  1716. "authenticator",
  1717. "authenticatorFlow",
  1718. "conditional",
  1719. "disabled",
  1720. "enabled",
  1721. "id",
  1722. "parentFlow",
  1723. "priority",
  1724. "required",
  1725. "requirement",
  1726. }, res
  1727. with pytest.raises(KeycloakGetError) as err:
  1728. admin.get_authentication_flow_execution(execution_id="bad")
  1729. assert err.match('404: b\'{"error":"Illegal execution"}\'')
  1730. with pytest.raises(KeycloakPostError) as err:
  1731. admin.create_authentication_flow_execution(payload=dict(), flow_alias="browser")
  1732. assert err.match('400: b\'{"error":"It is illegal to add execution to a built in flow"}\'')
  1733. res = admin.create_authentication_flow_execution(
  1734. payload={"provider": "auth-cookie"}, flow_alias="test-create"
  1735. )
  1736. assert res == b""
  1737. assert len(admin.get_authentication_flow_executions(flow_alias="test-create")) == 1
  1738. with pytest.raises(KeycloakPutError) as err:
  1739. admin.update_authentication_flow_executions(
  1740. payload={"required": "yes"}, flow_alias="test-create"
  1741. )
  1742. assert err.match('400: b\'{"error":"Unrecognized field')
  1743. payload = admin.get_authentication_flow_executions(flow_alias="test-create")[0]
  1744. payload["displayName"] = "test"
  1745. res = admin.update_authentication_flow_executions(payload=payload, flow_alias="test-create")
  1746. assert res
  1747. exec_id = admin.get_authentication_flow_executions(flow_alias="test-create")[0]["id"]
  1748. res = admin.delete_authentication_flow_execution(execution_id=exec_id)
  1749. assert res == dict()
  1750. with pytest.raises(KeycloakDeleteError) as err:
  1751. admin.delete_authentication_flow_execution(execution_id=exec_id)
  1752. assert err.match('404: b\'{"error":"Illegal execution"}\'')
  1753. # Test subflows
  1754. res = admin.create_authentication_flow_subflow(
  1755. payload={
  1756. "alias": "test-subflow",
  1757. "provider": "basic-flow",
  1758. "type": "something",
  1759. "description": "something",
  1760. },
  1761. flow_alias="test-browser",
  1762. )
  1763. assert res == b""
  1764. with pytest.raises(KeycloakPostError) as err:
  1765. admin.create_authentication_flow_subflow(
  1766. payload={"alias": "test-subflow", "providerId": "basic-flow"},
  1767. flow_alias="test-browser",
  1768. )
  1769. assert err.match('409: b\'{"errorMessage":"New flow alias name already exists"}\'')
  1770. res = admin.create_authentication_flow_subflow(
  1771. payload={
  1772. "alias": "test-subflow",
  1773. "provider": "basic-flow",
  1774. "type": "something",
  1775. "description": "something",
  1776. },
  1777. flow_alias="test-create",
  1778. skip_exists=True,
  1779. )
  1780. assert res == {"msg": "Already exists"}
  1781. # Test delete auth flow
  1782. flow_id = [x for x in admin.get_authentication_flows() if x["alias"] == "test-browser"][0][
  1783. "id"
  1784. ]
  1785. res = admin.delete_authentication_flow(flow_id=flow_id)
  1786. assert res == dict()
  1787. with pytest.raises(KeycloakDeleteError) as err:
  1788. admin.delete_authentication_flow(flow_id=flow_id)
  1789. assert err.match('404: b\'{"error":"Could not find flow with id"}\'')
  1790. def test_authentication_configs(admin: KeycloakAdmin, realm: str):
  1791. """Test authentication configs.
  1792. :param admin: Keycloak Admin client
  1793. :type admin: KeycloakAdmin
  1794. :param realm: Keycloak realm
  1795. :type realm: str
  1796. """
  1797. admin.change_current_realm(realm)
  1798. # Test list of auth providers
  1799. res = admin.get_authenticator_providers()
  1800. assert len(res) <= 38
  1801. res = admin.get_authenticator_provider_config_description(provider_id="auth-cookie")
  1802. assert res == {
  1803. "helpText": "Validates the SSO cookie set by the auth server.",
  1804. "name": "Cookie",
  1805. "properties": [],
  1806. "providerId": "auth-cookie",
  1807. }
  1808. # Test authenticator config
  1809. # Currently unable to find a sustainable way to fetch the config id,
  1810. # therefore testing only failures
  1811. with pytest.raises(KeycloakGetError) as err:
  1812. admin.get_authenticator_config(config_id="bad")
  1813. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1814. with pytest.raises(KeycloakPutError) as err:
  1815. admin.update_authenticator_config(payload=dict(), config_id="bad")
  1816. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1817. with pytest.raises(KeycloakDeleteError) as err:
  1818. admin.delete_authenticator_config(config_id="bad")
  1819. assert err.match('404: b\'{"error":"Could not find authenticator config"}\'')
  1820. def test_sync_users(admin: KeycloakAdmin, realm: str):
  1821. """Test sync users.
  1822. :param admin: Keycloak Admin client
  1823. :type admin: KeycloakAdmin
  1824. :param realm: Keycloak realm
  1825. :type realm: str
  1826. """
  1827. admin.change_current_realm(realm)
  1828. # Only testing the error message
  1829. with pytest.raises(KeycloakPostError) as err:
  1830. admin.sync_users(storage_id="does-not-exist", action="triggerFullSync")
  1831. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1832. def test_client_scopes(admin: KeycloakAdmin, realm: str):
  1833. """Test client scopes.
  1834. :param admin: Keycloak Admin client
  1835. :type admin: KeycloakAdmin
  1836. :param realm: Keycloak realm
  1837. :type realm: str
  1838. """
  1839. admin.change_current_realm(realm)
  1840. # Test get client scopes
  1841. res = admin.get_client_scopes()
  1842. scope_names = {x["name"] for x in res}
  1843. assert len(res) == 10
  1844. assert "email" in scope_names
  1845. assert "profile" in scope_names
  1846. assert "offline_access" in scope_names
  1847. with pytest.raises(KeycloakGetError) as err:
  1848. admin.get_client_scope(client_scope_id="does-not-exist")
  1849. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1850. scope = admin.get_client_scope(client_scope_id=res[0]["id"])
  1851. assert res[0] == scope
  1852. scope = admin.get_client_scope_by_name(client_scope_name=res[0]["name"])
  1853. assert res[0] == scope
  1854. # Test create client scope
  1855. res = admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=True)
  1856. assert res
  1857. res2 = admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=True)
  1858. assert res == res2
  1859. with pytest.raises(KeycloakPostError) as err:
  1860. admin.create_client_scope(payload={"name": "test-scope"}, skip_exists=False)
  1861. assert err.match('409: b\'{"errorMessage":"Client Scope test-scope already exists"}\'')
  1862. # Test update client scope
  1863. with pytest.raises(KeycloakPutError) as err:
  1864. admin.update_client_scope(client_scope_id="does-not-exist", payload=dict())
  1865. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1866. res_update = admin.update_client_scope(
  1867. client_scope_id=res, payload={"name": "test-scope-update"}
  1868. )
  1869. assert res_update == dict()
  1870. admin.get_client_scope(client_scope_id=res)["name"] == "test-scope-update"
  1871. # Test get mappers
  1872. mappers = admin.get_mappers_from_client_scope(client_scope_id=res)
  1873. assert mappers == list()
  1874. # Test add mapper
  1875. with pytest.raises(KeycloakPostError) as err:
  1876. admin.add_mapper_to_client_scope(client_scope_id=res, payload=dict())
  1877. assert err.match('404: b\'{"error":"ProtocolMapper provider not found"}\'')
  1878. res_add = admin.add_mapper_to_client_scope(
  1879. client_scope_id=res,
  1880. payload={
  1881. "name": "test-mapper",
  1882. "protocol": "openid-connect",
  1883. "protocolMapper": "oidc-usermodel-attribute-mapper",
  1884. },
  1885. )
  1886. assert res_add == b""
  1887. assert len(admin.get_mappers_from_client_scope(client_scope_id=res)) == 1
  1888. # Test update mapper
  1889. test_mapper = admin.get_mappers_from_client_scope(client_scope_id=res)[0]
  1890. with pytest.raises(KeycloakPutError) as err:
  1891. admin.update_mapper_in_client_scope(
  1892. client_scope_id="does-not-exist", protocol_mapper_id=test_mapper["id"], payload=dict()
  1893. )
  1894. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1895. test_mapper["config"]["user.attribute"] = "test"
  1896. res_update = admin.update_mapper_in_client_scope(
  1897. client_scope_id=res, protocol_mapper_id=test_mapper["id"], payload=test_mapper
  1898. )
  1899. assert res_update == dict()
  1900. assert (
  1901. admin.get_mappers_from_client_scope(client_scope_id=res)[0]["config"]["user.attribute"]
  1902. == "test"
  1903. )
  1904. # Test delete mapper
  1905. res_del = admin.delete_mapper_from_client_scope(
  1906. client_scope_id=res, protocol_mapper_id=test_mapper["id"]
  1907. )
  1908. assert res_del == dict()
  1909. with pytest.raises(KeycloakDeleteError) as err:
  1910. admin.delete_mapper_from_client_scope(
  1911. client_scope_id=res, protocol_mapper_id=test_mapper["id"]
  1912. )
  1913. assert err.match('404: b\'{"error":"Model not found"}\'')
  1914. # Test default default scopes
  1915. res_defaults = admin.get_default_default_client_scopes()
  1916. assert len(res_defaults) == 6
  1917. with pytest.raises(KeycloakPutError) as err:
  1918. admin.add_default_default_client_scope(scope_id="does-not-exist")
  1919. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1920. res_add = admin.add_default_default_client_scope(scope_id=res)
  1921. assert res_add == dict()
  1922. assert len(admin.get_default_default_client_scopes()) == 7
  1923. with pytest.raises(KeycloakDeleteError) as err:
  1924. admin.delete_default_default_client_scope(scope_id="does-not-exist")
  1925. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1926. res_del = admin.delete_default_default_client_scope(scope_id=res)
  1927. assert res_del == dict()
  1928. assert len(admin.get_default_default_client_scopes()) == 6
  1929. # Test default optional scopes
  1930. res_defaults = admin.get_default_optional_client_scopes()
  1931. assert len(res_defaults) == 4
  1932. with pytest.raises(KeycloakPutError) as err:
  1933. admin.add_default_optional_client_scope(scope_id="does-not-exist")
  1934. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1935. res_add = admin.add_default_optional_client_scope(scope_id=res)
  1936. assert res_add == dict()
  1937. assert len(admin.get_default_optional_client_scopes()) == 5
  1938. with pytest.raises(KeycloakDeleteError) as err:
  1939. admin.delete_default_optional_client_scope(scope_id="does-not-exist")
  1940. assert err.match('404: b\'{"error":"Client scope not found"}\'')
  1941. res_del = admin.delete_default_optional_client_scope(scope_id=res)
  1942. assert res_del == dict()
  1943. assert len(admin.get_default_optional_client_scopes()) == 4
  1944. # Test client scope delete
  1945. res_del = admin.delete_client_scope(client_scope_id=res)
  1946. assert res_del == dict()
  1947. with pytest.raises(KeycloakDeleteError) as err:
  1948. admin.delete_client_scope(client_scope_id=res)
  1949. assert err.match('404: b\'{"error":"Could not find client scope"}\'')
  1950. def test_components(admin: KeycloakAdmin, realm: str):
  1951. """Test components.
  1952. :param admin: Keycloak Admin client
  1953. :type admin: KeycloakAdmin
  1954. :param realm: Keycloak realm
  1955. :type realm: str
  1956. """
  1957. admin.change_current_realm(realm)
  1958. # Test get components
  1959. res = admin.get_components()
  1960. assert len(res) == 12
  1961. with pytest.raises(KeycloakGetError) as err:
  1962. admin.get_component(component_id="does-not-exist")
  1963. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1964. res_get = admin.get_component(component_id=res[0]["id"])
  1965. assert res_get == res[0]
  1966. # Test create component
  1967. with pytest.raises(KeycloakPostError) as err:
  1968. admin.create_component(payload={"bad": "dict"})
  1969. assert err.match('400: b\'{"error":"Unrecognized field')
  1970. res = admin.create_component(
  1971. payload={
  1972. "name": "Test Component",
  1973. "providerId": "max-clients",
  1974. "providerType": "org.keycloak.services.clientregistration."
  1975. + "policy.ClientRegistrationPolicy",
  1976. "config": {"max-clients": ["1000"]},
  1977. }
  1978. )
  1979. assert res
  1980. assert admin.get_component(component_id=res)["name"] == "Test Component"
  1981. # Test update component
  1982. component = admin.get_component(component_id=res)
  1983. component["name"] = "Test Component Update"
  1984. with pytest.raises(KeycloakPutError) as err:
  1985. admin.update_component(component_id="does-not-exist", payload=dict())
  1986. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1987. res_upd = admin.update_component(component_id=res, payload=component)
  1988. assert res_upd == dict()
  1989. assert admin.get_component(component_id=res)["name"] == "Test Component Update"
  1990. # Test delete component
  1991. res_del = admin.delete_component(component_id=res)
  1992. assert res_del == dict()
  1993. with pytest.raises(KeycloakDeleteError) as err:
  1994. admin.delete_component(component_id=res)
  1995. assert err.match('404: b\'{"error":"Could not find component"}\'')
  1996. def test_keys(admin: KeycloakAdmin, realm: str):
  1997. """Test keys.
  1998. :param admin: Keycloak Admin client
  1999. :type admin: KeycloakAdmin
  2000. :param realm: Keycloak realm
  2001. :type realm: str
  2002. """
  2003. admin.change_current_realm(realm)
  2004. assert set(admin.get_keys()["active"].keys()) == {"AES", "HS256", "RS256", "RSA-OAEP"}
  2005. assert {k["algorithm"] for k in admin.get_keys()["keys"]} == {
  2006. "HS256",
  2007. "RSA-OAEP",
  2008. "AES",
  2009. "RS256",
  2010. }
  2011. def test_admin_events(admin: KeycloakAdmin, realm: str):
  2012. """Test events.
  2013. :param admin: Keycloak Admin client
  2014. :type admin: KeycloakAdmin
  2015. :param realm: Keycloak realm
  2016. :type realm: str
  2017. """
  2018. admin.change_current_realm(realm)
  2019. admin.create_client(payload={"name": "test", "clientId": "test"})
  2020. events = admin.get_admin_events()
  2021. assert events == list()
  2022. def test_user_events(admin: KeycloakAdmin, realm: str):
  2023. """Test events.
  2024. :param admin: Keycloak Admin client
  2025. :type admin: KeycloakAdmin
  2026. :param realm: Keycloak realm
  2027. :type realm: str
  2028. """
  2029. admin.change_current_realm(realm)
  2030. events = admin.get_events()
  2031. assert events == list()
  2032. with pytest.raises(KeycloakPutError) as err:
  2033. admin.set_events(payload={"bad": "conf"})
  2034. assert err.match('400: b\'{"error":"Unrecognized field')
  2035. res = admin.set_events(payload={"adminEventsDetailsEnabled": True, "adminEventsEnabled": True})
  2036. assert res == dict()
  2037. admin.create_client(payload={"name": "test", "clientId": "test"})
  2038. events = admin.get_events()
  2039. assert events == list()
  2040. @freezegun.freeze_time("2023-02-25 10:00:00")
  2041. def test_auto_refresh(admin_frozen: KeycloakAdmin, realm: str):
  2042. """Test auto refresh token.
  2043. :param admin_frozen: Keycloak Admin client with time frozen in place
  2044. :type admin_frozen: KeycloakAdmin
  2045. :param realm: Keycloak realm
  2046. :type realm: str
  2047. """
  2048. admin = admin_frozen
  2049. # Test get refresh
  2050. admin.connection.custom_headers = {
  2051. "Authorization": "Bearer bad",
  2052. "Content-Type": "application/json",
  2053. }
  2054. with pytest.raises(KeycloakAuthenticationError) as err:
  2055. admin.get_realm(realm_name=realm)
  2056. assert err.match('401: b\'{"error":"HTTP 401 Unauthorized"}\'')
  2057. # Freeze time to simulate the access token expiring
  2058. with freezegun.freeze_time("2023-02-25 10:05:00"):
  2059. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:05:00")
  2060. assert admin.get_realm(realm_name=realm)
  2061. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:05:00")
  2062. # Test bad refresh token, but first make sure access token has expired again
  2063. with freezegun.freeze_time("2023-02-25 10:10:00"):
  2064. admin.connection.custom_headers = {"Content-Type": "application/json"}
  2065. admin.connection.token["refresh_token"] = "bad"
  2066. with pytest.raises(KeycloakPostError) as err:
  2067. admin.get_realm(realm_name="test-refresh")
  2068. assert err.match(
  2069. '400: b\'{"error":"invalid_grant","error_description":"Invalid refresh token"}\''
  2070. )
  2071. admin.connection.get_token()
  2072. # Test post refresh
  2073. with freezegun.freeze_time("2023-02-25 10:15:00"):
  2074. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:15:00")
  2075. admin.connection.token = None
  2076. assert admin.create_realm(payload={"realm": "test-refresh"}) == b""
  2077. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:15:00")
  2078. # Test update refresh
  2079. with freezegun.freeze_time("2023-02-25 10:25:00"):
  2080. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:25:00")
  2081. admin.connection.token = None
  2082. assert (
  2083. admin.update_realm(realm_name="test-refresh", payload={"accountTheme": "test"})
  2084. == dict()
  2085. )
  2086. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:25:00")
  2087. # Test delete refresh
  2088. with freezegun.freeze_time("2023-02-25 10:35:00"):
  2089. assert admin.connection.expires_at < datetime_parser.parse("2023-02-25 10:35:00")
  2090. admin.connection.token = None
  2091. assert admin.delete_realm(realm_name="test-refresh") == dict()
  2092. assert admin.connection.expires_at > datetime_parser.parse("2023-02-25 10:35:00")
  2093. def test_get_required_actions(admin: KeycloakAdmin, realm: str):
  2094. """Test required actions.
  2095. :param admin: Keycloak Admin client
  2096. :type admin: KeycloakAdmin
  2097. :param realm: Keycloak realm
  2098. :type realm: str
  2099. """
  2100. admin.change_current_realm(realm)
  2101. ractions = admin.get_required_actions()
  2102. assert isinstance(ractions, list)
  2103. for ra in ractions:
  2104. for key in [
  2105. "alias",
  2106. "name",
  2107. "providerId",
  2108. "enabled",
  2109. "defaultAction",
  2110. "priority",
  2111. "config",
  2112. ]:
  2113. assert key in ra
  2114. def test_get_required_action_by_alias(admin: KeycloakAdmin, realm: str):
  2115. """Test get required action by alias.
  2116. :param admin: Keycloak Admin client
  2117. :type admin: KeycloakAdmin
  2118. :param realm: Keycloak realm
  2119. :type realm: str
  2120. """
  2121. admin.change_current_realm(realm)
  2122. ractions = admin.get_required_actions()
  2123. ra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2124. assert ra in ractions
  2125. assert ra["alias"] == "UPDATE_PASSWORD"
  2126. assert admin.get_required_action_by_alias("does-not-exist") is None
  2127. def test_update_required_action(admin: KeycloakAdmin, realm: str):
  2128. """Test update required action.
  2129. :param admin: Keycloak Admin client
  2130. :type admin: KeycloakAdmin
  2131. :param realm: Keycloak realm
  2132. :type realm: str
  2133. """
  2134. admin.change_current_realm(realm)
  2135. ra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2136. old = copy.deepcopy(ra)
  2137. ra["enabled"] = False
  2138. admin.update_required_action("UPDATE_PASSWORD", ra)
  2139. newra = admin.get_required_action_by_alias("UPDATE_PASSWORD")
  2140. assert old != newra
  2141. assert newra["enabled"] is False
  2142. def test_get_composite_client_roles_of_group(
  2143. admin: KeycloakAdmin, realm: str, client: str, group: str, composite_client_role: str
  2144. ):
  2145. """Test get composite client roles of group.
  2146. :param admin: Keycloak Admin client
  2147. :type admin: KeycloakAdmin
  2148. :param realm: Keycloak realm
  2149. :type realm: str
  2150. :param client: Keycloak client
  2151. :type client: str
  2152. :param group: Keycloak group
  2153. :type group: str
  2154. :param composite_client_role: Composite client role
  2155. :type composite_client_role: str
  2156. """
  2157. admin.change_current_realm(realm)
  2158. role = admin.get_client_role(client, composite_client_role)
  2159. admin.assign_group_client_roles(group_id=group, client_id=client, roles=[role])
  2160. result = admin.get_composite_client_roles_of_group(client, group)
  2161. assert role["id"] in [x["id"] for x in result]
  2162. def test_get_role_client_level_children(
  2163. admin: KeycloakAdmin, realm: str, client: str, composite_client_role: str, client_role: str
  2164. ):
  2165. """Test get children of composite client role.
  2166. :param admin: Keycloak Admin client
  2167. :type admin: KeycloakAdmin
  2168. :param realm: Keycloak realm
  2169. :type realm: str
  2170. :param client: Keycloak client
  2171. :type client: str
  2172. :param composite_client_role: Composite client role
  2173. :type composite_client_role: str
  2174. :param client_role: Client role
  2175. :type client_role: str
  2176. """
  2177. admin.change_current_realm(realm)
  2178. child = admin.get_client_role(client, client_role)
  2179. parent = admin.get_client_role(client, composite_client_role)
  2180. res = admin.get_role_client_level_children(client, parent["id"])
  2181. assert child["id"] in [x["id"] for x in res]
  2182. def test_upload_certificate(admin: KeycloakAdmin, realm: str, client: str, selfsigned_cert: tuple):
  2183. """Test upload certificate.
  2184. :param admin: Keycloak Admin client
  2185. :type admin: KeycloakAdmin
  2186. :param realm: Keycloak realm
  2187. :type realm: str
  2188. :param client: Keycloak client
  2189. :type client: str
  2190. :param selfsigned_cert: Selfsigned certificates
  2191. :type selfsigned_cert: tuple
  2192. """
  2193. admin.change_current_realm(realm)
  2194. cert, _ = selfsigned_cert
  2195. cert = cert.decode("utf-8").strip()
  2196. admin.upload_certificate(client, cert)
  2197. cl = admin.get_client(client)
  2198. assert cl["attributes"]["jwt.credential.certificate"] == "".join(cert.splitlines()[1:-1])
  2199. def test_get_bruteforce_status_for_user(
  2200. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2201. ):
  2202. """Test users.
  2203. :param admin: Keycloak Admin client
  2204. :type admin: KeycloakAdmin
  2205. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2206. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2207. :param realm: Keycloak realm
  2208. :type realm: str
  2209. """
  2210. oid, username, password = oid_with_credentials
  2211. admin.change_current_realm(realm)
  2212. # Turn on bruteforce protection
  2213. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2214. res = admin.get_realm(realm_name=realm)
  2215. assert res["bruteForceProtected"] is True
  2216. # Test login user with wrong credentials
  2217. try:
  2218. oid.token(username=username, password="wrongpassword")
  2219. except KeycloakAuthenticationError:
  2220. pass
  2221. user_id = admin.get_user_id(username)
  2222. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2223. assert bruteforce_status["numFailures"] == 1
  2224. # Cleanup
  2225. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2226. res = admin.get_realm(realm_name=realm)
  2227. assert res["bruteForceProtected"] is False
  2228. def test_clear_bruteforce_attempts_for_user(
  2229. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2230. ):
  2231. """Test users.
  2232. :param admin: Keycloak Admin client
  2233. :type admin: KeycloakAdmin
  2234. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2235. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2236. :param realm: Keycloak realm
  2237. :type realm: str
  2238. """
  2239. oid, username, password = oid_with_credentials
  2240. admin.change_current_realm(realm)
  2241. # Turn on bruteforce protection
  2242. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2243. res = admin.get_realm(realm_name=realm)
  2244. assert res["bruteForceProtected"] is True
  2245. # Test login user with wrong credentials
  2246. try:
  2247. oid.token(username=username, password="wrongpassword")
  2248. except KeycloakAuthenticationError:
  2249. pass
  2250. user_id = admin.get_user_id(username)
  2251. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2252. assert bruteforce_status["numFailures"] == 1
  2253. res = admin.clear_bruteforce_attempts_for_user(user_id)
  2254. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2255. assert bruteforce_status["numFailures"] == 0
  2256. # Cleanup
  2257. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2258. res = admin.get_realm(realm_name=realm)
  2259. assert res["bruteForceProtected"] is False
  2260. def test_clear_bruteforce_attempts_for_all_users(
  2261. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str], realm: str
  2262. ):
  2263. """Test users.
  2264. :param admin: Keycloak Admin client
  2265. :type admin: KeycloakAdmin
  2266. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2267. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2268. :param realm: Keycloak realm
  2269. :type realm: str
  2270. """
  2271. oid, username, password = oid_with_credentials
  2272. admin.change_current_realm(realm)
  2273. # Turn on bruteforce protection
  2274. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": True})
  2275. res = admin.get_realm(realm_name=realm)
  2276. assert res["bruteForceProtected"] is True
  2277. # Test login user with wrong credentials
  2278. try:
  2279. oid.token(username=username, password="wrongpassword")
  2280. except KeycloakAuthenticationError:
  2281. pass
  2282. user_id = admin.get_user_id(username)
  2283. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2284. assert bruteforce_status["numFailures"] == 1
  2285. res = admin.clear_all_bruteforce_attempts()
  2286. bruteforce_status = admin.get_bruteforce_detection_status(user_id)
  2287. assert bruteforce_status["numFailures"] == 0
  2288. # Cleanup
  2289. res = admin.update_realm(realm_name=realm, payload={"bruteForceProtected": False})
  2290. res = admin.get_realm(realm_name=realm)
  2291. assert res["bruteForceProtected"] is False
  2292. def test_default_realm_role_present(realm: str, admin: KeycloakAdmin) -> None:
  2293. """Test that the default realm role is present in a brand new realm.
  2294. :param realm: Realm name
  2295. :type realm: str
  2296. :param admin: Keycloak admin
  2297. :type admin: KeycloakAdmin
  2298. """
  2299. admin.change_current_realm(realm)
  2300. assert f"default-roles-{realm}" in [x["name"] for x in admin.get_realm_roles()]
  2301. assert (
  2302. len([x["name"] for x in admin.get_realm_roles() if x["name"] == f"default-roles-{realm}"])
  2303. == 1
  2304. )
  2305. def test_get_default_realm_role_id(realm: str, admin: KeycloakAdmin) -> None:
  2306. """Test getter for the ID of the default realm role.
  2307. :param realm: Realm name
  2308. :type realm: str
  2309. :param admin: Keycloak admin
  2310. :type admin: KeycloakAdmin
  2311. """
  2312. admin.change_current_realm(realm)
  2313. assert (
  2314. admin.get_default_realm_role_id()
  2315. == [x["id"] for x in admin.get_realm_roles() if x["name"] == f"default-roles-{realm}"][0]
  2316. )
  2317. def test_realm_default_roles(admin: KeycloakAdmin, realm: str) -> None:
  2318. """Test getting, adding and deleting default realm roles.
  2319. :param realm: Realm name
  2320. :type realm: str
  2321. :param admin: Keycloak admin
  2322. :type admin: KeycloakAdmin
  2323. """
  2324. admin.change_current_realm(realm)
  2325. # Test listing all default realm roles
  2326. roles = admin.get_realm_default_roles()
  2327. assert len(roles) == 2
  2328. assert {x["name"] for x in roles} == {"offline_access", "uma_authorization"}
  2329. with pytest.raises(KeycloakGetError) as err:
  2330. admin.realm_name = "doesnotexist"
  2331. admin.get_realm_default_roles()
  2332. assert err.match('404: b\'{"error":"Realm not found."}\'')
  2333. admin.change_current_realm(realm)
  2334. # Test removing a default realm role
  2335. res = admin.remove_realm_default_roles(payload=[roles[0]])
  2336. assert res == {}
  2337. assert roles[0] not in admin.get_realm_default_roles()
  2338. assert len(admin.get_realm_default_roles()) == 1
  2339. with pytest.raises(KeycloakDeleteError) as err:
  2340. admin.remove_realm_default_roles(payload=[{"id": "bad id"}])
  2341. assert err.match('404: b\'{"error":"Could not find composite role"}\'')
  2342. # Test adding a default realm role
  2343. res = admin.add_realm_default_roles(payload=[roles[0]])
  2344. assert res == {}
  2345. assert roles[0] in admin.get_realm_default_roles()
  2346. assert len(admin.get_realm_default_roles()) == 2
  2347. with pytest.raises(KeycloakPostError) as err:
  2348. admin.add_realm_default_roles(payload=[{"id": "bad id"}])
  2349. assert err.match('404: b\'{"error":"Could not find composite role"}\'')
  2350. def test_clear_keys_cache(realm: str, admin: KeycloakAdmin) -> None:
  2351. """Test clearing the keys cache.
  2352. :param realm: Realm name
  2353. :type realm: str
  2354. :param admin: Keycloak admin
  2355. :type admin: KeycloakAdmin
  2356. """
  2357. admin.change_current_realm(realm)
  2358. res = admin.clear_keys_cache()
  2359. assert res == {}
  2360. def test_clear_realm_cache(realm: str, admin: KeycloakAdmin) -> None:
  2361. """Test clearing the realm cache.
  2362. :param realm: Realm name
  2363. :type realm: str
  2364. :param admin: Keycloak admin
  2365. :type admin: KeycloakAdmin
  2366. """
  2367. admin.change_current_realm(realm)
  2368. res = admin.clear_realm_cache()
  2369. assert res == {}
  2370. def test_clear_user_cache(realm: str, admin: KeycloakAdmin) -> None:
  2371. """Test clearing the user cache.
  2372. :param realm: Realm name
  2373. :type realm: str
  2374. :param admin: Keycloak admin
  2375. :type admin: KeycloakAdmin
  2376. """
  2377. admin.change_current_realm(realm)
  2378. res = admin.clear_user_cache()
  2379. assert res == {}
  2380. def test_initial_access_token(
  2381. admin: KeycloakAdmin, oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2382. ) -> None:
  2383. """Test initial access token and client creation.
  2384. :param admin: Keycloak admin
  2385. :type admin: KeycloakAdmin
  2386. :param oid_with_credentials: Keycloak OpenID client with pre-configured user credentials
  2387. :type oid_with_credentials: Tuple[KeycloakOpenID, str, str]
  2388. """
  2389. res = admin.create_initial_access_token(2, 3)
  2390. assert "token" in res
  2391. assert res["count"] == 2
  2392. assert res["expiration"] == 3
  2393. oid, username, password = oid_with_credentials
  2394. client = str(uuid.uuid4())
  2395. secret = str(uuid.uuid4())
  2396. res = oid.register_client(
  2397. token=res["token"],
  2398. payload={
  2399. "name": "DynamicRegisteredClient",
  2400. "clientId": client,
  2401. "enabled": True,
  2402. "publicClient": False,
  2403. "protocol": "openid-connect",
  2404. "secret": secret,
  2405. "clientAuthenticatorType": "client-secret",
  2406. },
  2407. )
  2408. assert res["clientId"] == client
  2409. new_secret = str(uuid.uuid4())
  2410. res = oid.update_client(res["registrationAccessToken"], client, payload={"secret": new_secret})
  2411. assert res["secret"] == new_secret