Implements 'check' command in API

.gitignore

@@ -9,3 +9,14 @@ ci/test/output/
 .env
 *.crt
 *.key
+
+# Build artifacts
+build/
+dist/
+*.egg-info/
+
+# Virtual environment
+venv/
+
+# Misc
+.DS_Store

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 - Add support for Role Memberships endpoint
   [conjur-api-python#30](https://github.com/cyberark/conjur-api-python/pull/33)
+- Add support for Check Privilege endpoint
+  [conjur-api-python#34](https://github.com/cyberark/conjur-api-python/pull/34)
 - Add support for Show Role endpoint
   [conjur-api-python#30](https://github.com/cyberark/conjur-api-python/pull/30)
 - Add support for Show Resource endpoint

README.md

@@ -193,6 +193,10 @@ For example: `client.list({'kind': 'user', 'inspect': True})`
 | search           | Search for resources based on specified query                |
 | inspect          | List the metadata for resources                              |
 
+### `check_privilege(kind, resource_id, privilege, role_id)`
+
+Checks for a privilege on a resource based on its kind, resource ID, and an optional role ID. Returns a boolean.
+
 #### `get_resource(kind, resource_id)`
 
 Gets a resource based on its kind and ID. Resource is json data that contains metadata about the resource.

conjur_api/client.py

@@ -105,6 +105,12 @@ async def list(self, list_constraints: dict = None) -> dict:
         """
         return await self._api.resources_list(list_constraints)
 
+    async def check_privilege(self, kind: str, resource_id: str, privilege: str, role_id: str = None) -> bool:
+        """
+        Checks a privilege on a resource based on its kind, ID, role, and privilege.
+        """
+        return await self._api.check_privilege(kind, resource_id, privilege, role_id)
+
     async def get_resource(self, kind: str, resource_id: str) -> json:
         """
         Gets a resource based on its kind and ID

conjur_api/http/api.py

@@ -156,6 +156,34 @@ async def resources_list(self, list_constraints: dict = None) -> dict:
         # ?tocpath=Developer%7CREST%C2%A0APIs%7C_____17
         return resources
 
+    async def check_privilege(self, kind: str, resource_id: str, privilege: str, role_id: str) -> bool:
+        """
+        This method is used to check for a privilege on a resource.
+        """
+        params = {
+            'account': self._account,
+            'kind': kind,
+            'identifier': resource_id,
+            'privilege': privilege,
+            'role': role_id if role_id else "",
+        }
+        params.update(self._default_params)
+
+        try:
+            await invoke_endpoint(HttpVerb.HEAD, ConjurEndpoint.PRIVILEGE,
+                                  params,
+                                  api_token=await self.api_token,
+                                  ssl_verification_metadata=self.ssl_verification_data)
+        except HttpStatusError as err:
+            if err.status == 204:
+                return True
+            if err.status == 404:
+                return False
+
+            raise err
+
+        return True
+
     async def get_resource(self, kind: str, resource_id: str) -> dict:
         """
         This method is used to fetch a specific resource.

conjur_api/http/endpoints.py

@@ -24,6 +24,7 @@ class ConjurEndpoint(Enum):
     POLICIES = "{url}/policies/{account}/policy/{identifier}"
     BATCH_SECRETS = "{url}/secrets"
     SECRETS = "{url}/secrets/{account}/{kind}/{identifier}"
+    PRIVILEGE = "{url}/resources/{account}/{kind}/{identifier}?check=true&role={role}&privilege={privilege}"
     RESOURCES = "{url}/resources/{account}"
     RESOURCE = "{url}/resources/{account}/{kind}/{identifier}"
     ROTATE_API_KEY = "{url}/authn/{account}/api_key"

tests/https/test_unit_client.py

@@ -275,7 +275,7 @@ async def test_client_get_role_invokes_api(self, mock_api_token, mock_invoke_end
         self.assertTrue(exists_in_args('dummy', args))
         mock_invoke_endpoint.assert_called_once()
 
-    @patch.object(Api, '_api_token', new_callable=PropertyMock)  
+    @patch.object(Api, '_api_token', new_callable=PropertyMock)
     async def test_client_role_direct_memberships_invokes_api(self, mock_api_token, mock_invoke_endpoint):
         mock_api_token.return_value = 'test_token'
         await self.client.role_memberships('policy', 'dummy', True)
@@ -287,7 +287,7 @@ async def test_client_role_direct_memberships_invokes_api(self, mock_api_token,
         self.assertTrue(exists_in_args('memberships', args))
         mock_invoke_endpoint.assert_called_once()
 
-    @patch.object(Api, '_api_token', new_callable=PropertyMock)  
+    @patch.object(Api, '_api_token', new_callable=PropertyMock)
     async def test_client_role_memberships_invokes_api(self, mock_api_token, mock_invoke_endpoint):
         mock_api_token.return_value = 'test_token'
         await self.client.role_memberships('policy', 'dummy')
@@ -298,3 +298,16 @@ async def test_client_role_memberships_invokes_api(self, mock_api_token, mock_in
         self.assertTrue(exists_in_args('dummy', args))
         self.assertTrue(exists_in_args('all', args))
         mock_invoke_endpoint.assert_called_once()
+
+    @patch.object(Api, '_api_token', new_callable=PropertyMock)
+    async def test_client_check_privilege_invokes_api(self, mock_api_token, mock_invoke_endpoint):
+        mock_api_token.return_value = 'test_token'
+        await self.client.check_privilege('policy', 'dummy1', 'dummy2', 'dummy3')
+
+        args, kwargs = mock_invoke_endpoint.call_args
+        self.assertEqual('test_token', kwargs.get('api_token'))
+        self.assertTrue(exists_in_args('policy', args))
+        self.assertTrue(exists_in_args('dummy1', args))
+        self.assertTrue(exists_in_args('dummy2', args))
+        self.assertTrue(exists_in_args('dummy3', args))
+        mock_invoke_endpoint.assert_called_once()

tests/https/test_unit_endpoints.py

@@ -67,3 +67,12 @@ def test_endpoint_has_correct_role_memberships_all_template_string(self):
                                                              identifier='roleid',
                                                              membership='all')
         self.assertEqual(endpoint, 'http://host/roles/myacct/rolekind/roleid?all')
+
+    def test_endpoint_has_correct_check_privilege_template_string(self):
+        endpoint = ConjurEndpoint.PRIVILEGE.value.format(url='http://host',
+                                                             account='myacct',
+                                                             kind='resourcekind',
+                                                             identifier='resourceid',
+                                                             role='roleid',
+                                                             privilege='respriv')
+        self.assertEqual(endpoint, 'http://host/resources/myacct/resourcekind/resourceid?check=true&role=roleid&privilege=respriv')

tests/https/test_unit_http_ssl.py

@@ -38,8 +38,9 @@
 ]
 
 valid_badssl_endpoints = [
-    "https://ecc256.badssl.com",
-    "https://ecc384.badssl.com",
+    # [certificate has expired] errors - https://github.com/chromium/badssl.com/issues/510
+    # "https://ecc256.badssl.com",
+    # "https://ecc384.badssl.com",
     "https://extended-validation.badssl.com",
     "https://rsa2048.badssl.com",
     "https://rsa4096.badssl.com",