|
|
@ -29,6 +29,8 @@ |
|
|
|
import json |
|
|
|
from builtins import isinstance |
|
|
|
from typing import Iterable |
|
|
|
import copy |
|
|
|
from requests_toolbelt import MultipartEncoder |
|
|
|
|
|
|
|
from . import urls_patterns |
|
|
|
from .connection import ConnectionManager |
|
|
@ -2810,3 +2812,52 @@ class KeycloakAdmin: |
|
|
|
data=json.dumps(payload), |
|
|
|
) |
|
|
|
return raise_error_from_response(data_raw, KeycloakPostError, expected_codes=[201]) |
|
|
|
|
|
|
|
def get_composite_client_roles_of_group(self, client_id, group_id): |
|
|
|
params_path = {"realm-name": self.realm_name, "id": group_id, "client-id": client_id} |
|
|
|
data_raw = self.raw_get(urls_patterns.URL_ADMIN_GROUPS_CLIENT_ROLES_COMPOSITE.format(**params_path)) |
|
|
|
return raise_error_from_response(data_raw, KeycloakGetError) |
|
|
|
|
|
|
|
def get_role_client_level_children(self, client_id, role_id): |
|
|
|
params_path = {"realm-name": self.realm_name, "role-id": role_id, "client-id": client_id} |
|
|
|
data_raw = self.raw_get(urls_patterns.URL_ADMIN_CLIENT_ROLE_CHILDREN.format(**params_path)) |
|
|
|
return raise_error_from_response(data_raw, KeycloakGetError) |
|
|
|
|
|
|
|
def get_user_credentials(self, user_id): |
|
|
|
params_path = {"realm-name": self.realm_name, "id": user_id} |
|
|
|
data_raw = self.raw_get(urls_patterns.URL_ADMIN_USER_CREDENTIALS.format(**params_path)) |
|
|
|
return raise_error_from_response(data_raw, KeycloakGetError) |
|
|
|
|
|
|
|
def upload_certificate(self, client_id, certcont): |
|
|
|
params_path = {"realm-name": self.realm_name, "id": client_id, "attr": "jwt.credential"} |
|
|
|
m = MultipartEncoder( |
|
|
|
fields={ |
|
|
|
"keystoreFormat": "Certificate PEM", |
|
|
|
"file": certcont |
|
|
|
} |
|
|
|
) |
|
|
|
new_headers = copy.deepcopy(self.connection.headers) |
|
|
|
new_headers["Content-Type"] = m.content_type |
|
|
|
self.connection.headers = new_headers |
|
|
|
data_raw = self.raw_post(urls_patterns.URL_ADMIN_CLIENT_CERT_UPLOAD.format(**params_path), |
|
|
|
data=m, headers=new_headers) |
|
|
|
return raise_error_from_response(data_raw, KeycloakPostError) |
|
|
|
|
|
|
|
def get_required_action_by_alias(self, action_alias): |
|
|
|
actions = self.get_required_actions() |
|
|
|
for a in actions: |
|
|
|
if a['alias'] == action_alias: |
|
|
|
break |
|
|
|
return a |
|
|
|
|
|
|
|
def get_required_actions(self): |
|
|
|
params_path = {"realm-name": self.realm_name} |
|
|
|
data_raw = self.raw_get(urls_patterns.URL_ADMIN_REQUIRED_ACTIONS.format(**params_path)) |
|
|
|
return raise_error_from_response(data_raw, KeycloakGetError) |
|
|
|
|
|
|
|
def update_required_action(self, action_alias, payload): |
|
|
|
if not isinstance(payload, str): |
|
|
|
payload = json.dumps(payload) |
|
|
|
params_path = {"realm-name": self.realm_name, "action-alias": action_alias} |
|
|
|
data_raw = self.raw_put(urls_patterns.URL_ADMIN_REQUIRED_ACTIONS_ALIAS.format(**params_path), data=payload) |
|
|
|
return raise_error_from_response(data_raw, KeycloakPutError) |