Browse Source

feat: add create_client_authz_scope_permission and create_client_authz_policy to keycloak_admin

Signed-off-by: Cainã S. G <p-caina.galante@pd.tec.br>
pull/460/head
Cainã S. G 2 years ago
parent
commit
b11d384446
  1. 79
      src/keycloak/keycloak_admin.py
  2. 3
      src/keycloak/urls_patterns.py
  3. 78
      tests/test_keycloak_admin.py

79
src/keycloak/keycloak_admin.py

@ -1622,6 +1622,43 @@ class KeycloakAdmin:
data_raw, KeycloakPostError, expected_codes=[201], skip_exists=skip_exists
)
def create_client_authz_policy(self, client_id, payload, skip_exists=False):
"""Create an authz policy of client.
Payload example::
payload={
"name": "Policy-time-based",
"type": "time",
"logic": "POSITIVE",
"decisionStrategy": "UNANIMOUS",
"config": {
"hourEnd": "18",
"hour": "9"
}
}
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/18.0/rest-api/index.html#_clientrepresentation
:type client_id: str
:param payload: No Document
:type payload: dict
:param skip_exists: Skip creation in case the object exists
:type skip_exists: bool
:return: Keycloak server response
:rtype: bytes
"""
params_path = {"realm-name": self.connection.realm_name, "id": client_id}
data_raw = self.connection.raw_post(
urls_patterns.URL_ADMIN_CLIENT_AUTHZ_POLICY.format(**params_path),
data=json.dumps(payload),
)
return raise_error_from_response(
data_raw, KeycloakPostError, expected_codes=[201], skip_exists=skip_exists
)
def create_client_authz_resource_based_permission(self, client_id, payload, skip_exists=False):
"""Create resource-based permission of client.
@ -1661,6 +1698,48 @@ class KeycloakAdmin:
data_raw, KeycloakPostError, expected_codes=[201], skip_exists=skip_exists
)
def create_client_authz_scope_based_permission(self, client_id, payload, skip_exists=False):
"""Create scope-based permission of client.
Payload example::
payload={
"type": "resource",
"logic": "POSITIVE",
"decisionStrategy": "UNANIMOUS",
"name": "Permission-Name",
"resources": [
resource_id
],
"policies": [
policy_id
],
"scopes": [
scope_id
]
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/18.0/rest-api/index.html#_clientrepresentation
:type client_id: str
:param payload: PolicyRepresentation
https://www.keycloak.org/docs-api/18.0/rest-api/index.html#_policyrepresentation
:type payload: dict
:param skip_exists: Skip creation in case the object already exists
:type skip_exists: bool
:return: Keycloak server response
:rtype: bytes
"""
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.raw_post(
urls_patterns.URL_ADMIN_CLIENT_AUTHZ_SCOPE_BASED_PERMISSION.format(**params_path),
data=json.dumps(payload),
)
return raise_error_from_response(
data_raw, KeycloakPostError, expected_codes=[201], skip_exists=skip_exists
)
def get_client_authz_scopes(self, client_id):
"""Get scopes from client.

3
src/keycloak/urls_patterns.py

@ -121,6 +121,9 @@ URL_ADMIN_CLIENT_AUTHZ_ROLE_BASED_POLICY = URL_ADMIN_CLIENT_AUTHZ + "/policy/rol
URL_ADMIN_CLIENT_AUTHZ_RESOURCE_BASED_PERMISSION = (
URL_ADMIN_CLIENT_AUTHZ + "/permission/resource?max=-1"
)
URL_ADMIN_CLIENT_AUTHZ_SCOPE_BASED_PERMISSION = (
URL_ADMIN_CLIENT + "/authz/resource-server/permission/scope?max=-1"
)
URL_ADMIN_CLIENT_AUTHZ_POLICY = URL_ADMIN_CLIENT_AUTHZ + "/policy/{policy-id}"
URL_ADMIN_CLIENT_AUTHZ_POLICY_SCOPES = URL_ADMIN_CLIENT_AUTHZ_POLICY + "/scopes"
URL_ADMIN_CLIENT_AUTHZ_POLICY_RESOURCES = URL_ADMIN_CLIENT_AUTHZ_POLICY + "/resources"

78
tests/test_keycloak_admin.py

@ -868,6 +868,49 @@ def test_clients(admin: KeycloakAdmin, realm: str):
admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
assert err.match("404: b''")
res = admin.create_client_authz_policy(
client_id=auth_client_id,
payload={
"name": "test-authz-policy",
"type": "time",
"config": {"hourEnd": "18", "hour": "9"},
},
)
assert res["name"] == "test-authz-policy", res
with pytest.raises(KeycloakPostError) as err:
admin.create_client_authz_policy(
client_id=auth_client_id,
payload={
"name": "test-authz-policy",
"type": "time",
"config": {"hourEnd": "18", "hour": "9"},
},
)
assert err.match('409: b\'{"error":"Policy with name')
assert admin.create_client_authz_policy(
client_id=auth_client_id,
payload={
"name": "test-authz-policy",
"type": "time",
"config": {"hourEnd": "18", "hour": "9"},
},
skip_exists=True,
) == {"msg": "Already exists"}
assert len(admin.get_client_authz_policies(client_id=auth_client_id)) == 2
res = admin.create_client_authz_policy(
client_id=auth_client_id,
payload={
"name": "test-authz-policy",
"type": "time",
"config": {"hourEnd": "18", "hour": "9"},
},
)
res2 = admin.get_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
assert res["id"] == res2["id"]
admin.delete_client_authz_policy(client_id=auth_client_id, policy_id=res["id"])
# Test authz permissions
res = admin.get_client_authz_permissions(client_id=auth_client_id)
assert len(res) == 1, res
@ -910,6 +953,7 @@ def test_clients(admin: KeycloakAdmin, realm: str):
client_id=auth_client_id, payload={"name": "test-authz-scope"}
)
assert res["name"] == "test-authz-scope", res
test_scope_id = res["id"]
with pytest.raises(KeycloakPostError) as err:
admin.create_client_authz_scopes(
@ -924,6 +968,40 @@ def test_clients(admin: KeycloakAdmin, realm: str):
assert len(res) == 1
assert {x["name"] for x in res} == {"test-authz-scope"}
res = admin.create_client_authz_scope_based_permission(
client_id=auth_client_id,
payload={
"name": "test-permission-sb",
"resources": [test_resource_id],
"scope": [test_scope_id],
},
)
assert res, res
assert res["name"] == "test-permission-sb"
assert res["resources"] == [test_resource_id]
assert res["scopes"] == [test_scope_id]
with pytest.raises(KeycloakPostError) as err:
admin.create_client_authz_scope_based_permission(
client_id=auth_client_id,
payload={
"name": "test-permission-sb",
"resources": [test_resource_id],
"scope": [test_scope_id],
},
)
assert err.match('409: b\'{"error":"Policy with name')
assert admin.create_client_authz_scope_based_permission(
client_id=auth_client_id,
payload={
"name": "test-permission-sb",
"resources": [test_resource_id],
"scope": [test_scope_id],
},
skip_exists=True,
) == {"msg": "Already exists"}
assert len(admin.get_client_authz_permissions(client_id=auth_client_id)) == 3
# Test service account user
res = admin.get_client_service_account_user(client_id=auth_client_id)
assert res["username"] == "service-account-authz-client", res

Loading…
Cancel
Save