Added UMA 1 client

Author: Peter Slump

src/keycloak/realm.py

@@ -3,6 +3,7 @@
 from keycloak.client import KeycloakClient
 from keycloak.openid_connect import KeycloakOpenidConnect
 from keycloak.uma import KeycloakUMA
+from keycloak.uma1 import KeycloakUMA1
 
 
 class KeycloakRealm(object):
@@ -70,10 +71,28 @@ def authz(self, client_id):
     def uma(self):
         """
         Get UMA client
+
+        This method is here for backwards compatibility
+
         :return: keycloak.uma.KeycloakUMA
         """
+        return self.uma2
+
+    @property
+    def uma2(self):
+        """
+        Starting from Keycloak 4 UMA2 is supported
+        :rtype: keycloak.uma.KeycloakUMA
+        """
         return KeycloakUMA(realm=self)
 
+    @property
+    def uma1(self):
+        """
+        :rtype: keycloak.uma1.KeycloakUMA1
+        """
+        return KeycloakUMA1(realm=self)
+
     def close(self):
         if self._client is not None:
             self._client.close()

src/keycloak/uma1.py

@@ -0,0 +1,133 @@
+import json
+
+try:
+    from urllib.parse import urlencode  # noqa: F401
+except ImportError:
+    from urllib import urlencode  # noqa: F401
+
+from keycloak.mixins import WellKnownMixin
+
+PATH_WELL_KNOWN = "auth/realms/{}/.well-known/uma-configuration"
+
+
+class KeycloakUMA1(WellKnownMixin, object):
+    DEFAULT_HEADERS = {"Content-type": 'application/json'}
+
+    _realm = None
+    _well_known = None
+    _dumps = staticmethod(json.dumps)
+
+    def __init__(self, realm):
+        """
+        :type realm: keycloak.realm.KeycloakRealm
+        """
+        self._realm = realm
+
+    def get_path_well_known(self):
+        return PATH_WELL_KNOWN
+
+    def resource_set_create(self, token, name, **kwargs):
+        """
+        Create a resource set.
+
+        https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#rfc.section.2.2.1
+
+        :param str token: client access token
+        :param str name:
+        :param str uri: (optional)
+        :param str type: (optional)
+        :param list scopes: (optional)
+        :param str icon_uri: (optional)
+        :rtype: str
+        """
+        return self._realm.client.post(
+            self.well_known['resource_set_registration_endpoint'],
+            data=self._get_data(name=name, **kwargs),
+            headers=self.get_headers(token)
+        )
+
+    def resource_set_update(self, token, id, name, **kwargs):
+        """
+        Update a resource set.
+
+        https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#update-resource-set
+
+        :param str token: client access token
+        :param str id: Identifier of the resource set
+        :param str name:
+        :param str uri: (optional)
+        :param str type: (optional)
+        :param list scopes: (optional)
+        :param str icon_uri: (optional)
+        :rtype: str
+        """
+        return self._realm.client.put(
+            '{}/{}'.format(
+                self.well_known['resource_set_registration_endpoint'], id),
+            data=self._get_data(name=name, **kwargs),
+            headers=self.get_headers(token)
+        )
+
+    def resource_set_read(self, token, id):
+        """
+        Read a resource set.
+
+        https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#read-resource-set
+
+        :param str token: client access token
+        :param str id: Identifier of the resource set
+        :rtype: dict
+        """
+        return self._realm.client.get(
+            '{}/{}'.format(
+                self.well_known['resource_set_registration_endpoint'], id),
+            headers=self.get_headers(token)
+        )
+
+    def resource_set_delete(self, token, id):
+        """
+        Delete a resource set.
+
+        https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#delete-resource-set
+
+        :param str token: client access token
+        :param str id: Identifier of the resource set
+        """
+        return self._realm.client.delete(
+            '{}/{}'.format(
+                self.well_known['resource_set_registration_endpoint'], id),
+            headers=self.get_headers(token)
+        )
+
+    def resource_set_list(self, token, **kwargs):
+        """
+        List a resource set.
+
+        https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#list-resource-sets
+
+        :param str token: client access token
+        :param str name: (optional)
+        :param str uri: (optional)
+        :param str owner: (optional)
+        :param str type: (optional)
+        :param str scope: (optional)
+        :rtype: list
+        """
+        return self._realm.client.get(
+            self.well_known['resource_set_registration_endpoint'],
+            headers=self.get_headers(token),
+            **kwargs
+        )
+
+    @classmethod
+    def get_headers(cls, token):
+        return dict(cls.DEFAULT_HEADERS, **{
+            "Authorization": "Bearer " + token,
+        })
+
+    @staticmethod
+    def get_payload(name, scopes=None, **kwargs):
+        return dict(name=name, scopes=scopes or [], **kwargs)
+
+    def _get_data(self, *args, **kwargs):
+        return self._dumps(self.get_payload(*args, **kwargs))

tests/keycloak/test_uma1.py

@@ -0,0 +1,133 @@
+from unittest import TestCase
+
+import mock
+
+from keycloak.uma1 import KeycloakUMA1
+from keycloak.realm import KeycloakRealm
+from keycloak.well_known import KeycloakWellKnown
+
+
+class KeycloakUma1TestCase(TestCase):
+
+    def setUp(self):
+        self.realm = mock.MagicMock(spec_set=KeycloakRealm)
+        self.uma_client = KeycloakUMA1(realm=self.realm)
+        self.uma_client.well_known.contents = {
+            'resource_set_registration_endpoint':
+                'https://resource_registration',
+        }
+
+    def test_well_known(self):
+        """
+        Case: .well-known is requested
+        Expected: it's returned and the second time the same is returned
+        """
+        well_known = self.uma_client.well_known
+
+        self.assertIsInstance(well_known, KeycloakWellKnown)
+        self.assertEqual(well_known, self.uma_client.well_known)
+
+    def test_get_headers(self):
+        result = self.uma_client.get_headers(token='test-token')
+        self.assertEqual(result, {
+            "Authorization": "Bearer test-token",
+            "Content-type": 'application/json'
+        })
+
+    def test_get_payload(self):
+        result = self.uma_client.get_payload(
+            name='test-name',
+            scopes=['test-scope'],
+            optional_param='test-optional-param'
+        )
+
+        self.assertEqual(result,  {
+            'name': 'test-name',
+            'scopes': ['test-scope'],
+            'optional_param': 'test-optional-param'
+        })
+
+        result = self.uma_client.get_payload(
+            name='test-name',
+            optional_param='test-optional-param'
+        )
+
+        self.assertEqual(result, {
+            'name': 'test-name',
+            'scopes': [],
+            'optional_param': 'test-optional-param'
+        })
+
+    def test_resource_set_create(self):
+        result = self.uma_client.resource_set_create(
+            token='test-token',
+            name='test-name',
+            optional_param='test-optional-param'
+        )
+
+        self.realm.client.post.assert_called_once_with(
+            'https://resource_registration',
+            data=self.uma_client._get_data(
+                name='test-name',
+                optional_param='test-optional-param'
+            ),
+            headers=self.uma_client.get_headers('test-token')
+        )
+        self.assertEqual(result, self.realm.client.post.return_value)
+
+    def test_resource_set_update(self):
+        result = self.uma_client.resource_set_update(
+            token='test-token',
+            id='test-id',
+            name='test-name',
+            optional_param='test-optional-param'
+        )
+
+        self.realm.client.put.assert_called_once_with(
+            'https://resource_registration/test-id',
+            data=self.uma_client._get_data(
+                name='test-name',
+                optional_param='test-optional-param'
+            ),
+            headers=self.uma_client.get_headers('test-token')
+        )
+        self.assertEqual(result, self.realm.client.put.return_value)
+
+    def test_resource_set_read(self):
+        result = self.uma_client.resource_set_read(
+            token='test-token',
+            id='test-id',
+        )
+
+        self.realm.client.get.assert_called_once_with(
+            'https://resource_registration/test-id',
+            headers=self.uma_client.get_headers('test-token')
+        )
+        self.assertEqual(result, self.realm.client.get.return_value)
+
+    def test_resource_set_delete(self):
+        result = self.uma_client.resource_set_delete(
+            token='test-token',
+            id='test-id',
+        )
+
+        self.realm.client.delete.assert_called_once_with(
+            'https://resource_registration/test-id',
+            headers=self.uma_client.get_headers('test-token')
+        )
+        self.assertEqual(result, self.realm.client.delete.return_value)
+
+    def test_resource_set_list(self):
+        result = self.uma_client.resource_set_list(
+            token='test-token',
+            name='test-name',
+            owner='test-owner'
+        )
+
+        self.realm.client.get.assert_called_once_with(
+            'https://resource_registration',
+            name='test-name',
+            owner='test-owner',
+            headers=self.uma_client.get_headers('test-token')
+        )
+        self.assertEqual(result, self.realm.client.get.return_value)