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.

2865 lines
102 KiB

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