Add Integration tests in Python

conjur-openapi.yml

@@ -1089,6 +1089,12 @@ paths:
         required: true
         schema:
           $ref: '#/components/schemas/HostFactoryToken'
+      requestBody:
+        required: false
+        content:
+          text/plain:
+            schema:
+              type: string
       responses:
         "204":
           description: "Token was successfully revoked"

test/config/.pylintrc

@@ -2,6 +2,8 @@
 disable=
 
     # redundant, we generally only have one class per module for integration tests
-    missing-module-docstring
+    missing-module-docstring,
+    # pylint catches auto-generated import statements as code duplication
+    R0801
 
 max-line-length=100

test/config/policy.yaml

@@ -1 +1,2 @@
 - !variable one/password
+- !variable testSecret

test/python/api_config.py

@@ -0,0 +1,59 @@
+
+import pathlib
+import os
+import unittest
+
+import openapi_client
+
+CERT_DIR = pathlib.Path('config/https')
+SSL_CERT_FILE = 'ca.crt'
+CONJUR_CERT_FILE = 'conjur.crt'
+CONJUR_KEY_FILE = 'conjur.key'
+
+# Environment Constants
+CONJUR_AUTHN_API_KEY = 'CONJUR_AUTHN_API_KEY'
+CONJUR_AUTHN_LOGIN = 'CONJUR_AUTHN_LOGIN'
+CONJUR_ACCOUNT = 'CONJUR_ACCOUNT'
+
+def get_api_config():
+    """Gets a default API config to be used with the testsing setup"""
+    config = openapi_client.Configuration(
+            host='https://conjur-https',
+        )
+
+    config.ssl_ca_cert = CERT_DIR.joinpath(SSL_CERT_FILE)
+    config.cert_file = CERT_DIR.joinpath(CONJUR_CERT_FILE)
+    config.key_file = CERT_DIR.joinpath(CONJUR_KEY_FILE)
+    config.username = os.environ[CONJUR_AUTHN_LOGIN]
+    return config
+
+def get_api_client():
+    """Gets an authenticated ApiClient to be used with the testing setup"""
+    api_key = os.environ[CONJUR_AUTHN_API_KEY]
+    account = os.environ[CONJUR_ACCOUNT]
+
+    config = get_api_config()
+
+    client = openapi_client.ApiClient(config)
+    auth = openapi_client.api.AuthnApi(client)
+    new_key = auth.authenticate(
+            account,
+            os.environ['CONJUR_AUTHN_LOGIN'],
+            api_key,
+            accept_encoding='base64'
+        )
+    client.configuration.api_key = {'Authorization': f'Token token="{new_key}"'}
+    return client
+
+class ConfiguredTest(unittest.TestCase):
+    """Meant for test classes to inherit. Sets up an authenticated api client
+    for the test to use"""
+    @classmethod
+    def setUpClass(cls):
+        cls.account = os.environ[CONJUR_ACCOUNT]
+
+        cls.client = get_api_client()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.client.close()

test/python/test_authn_api.py

@@ -0,0 +1,110 @@
+from __future__ import absolute_import
+
+import os
+import unittest
+import json
+
+import openapi_client
+
+from . import api_config
+from .api_config import CONJUR_AUTHN_API_KEY, CONJUR_ACCOUNT
+
+class TestAuthnApi(unittest.TestCase):
+    """AuthnApi integration tests. Ensures that authentication with a Conjur server is working"""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.api_key = os.environ[CONJUR_AUTHN_API_KEY]
+        cls.account = os.environ[CONJUR_ACCOUNT]
+
+    @classmethod
+    def tearDownClass(cls):
+        # ensures that the proper API key is set for other test cases after the key rotation test
+        os.environ.update({CONJUR_AUTHN_API_KEY: cls.api_key})
+
+    def setUp(self):
+        # Reset the config password for each run because we change it in some tests
+        self.config = api_config.get_api_config()
+        self.config.password = self.api_key
+
+        self.client = openapi_client.ApiClient(self.config)
+        self.api = openapi_client.api.authn_api.AuthnApi(self.client)
+
+    def tearDown(self):
+        self.client.close()
+
+    def test_authenticate(self):
+        """Test case for authenticate
+
+        Gets a short-lived access token, which can be used to authenticate requests
+        to (most of) the rest of the Conjur API.
+        """
+        login = self.config.username
+        body = self.api_key
+
+        response = self.api.authenticate(self.account, login, body).replace("\'","\"")
+        response_json = json.loads(response)
+        response_keys = response_json.keys()
+
+        self.assertIn("protected", response_keys)
+        self.assertIn("payload", response_keys)
+        self.assertIn("signature", response_keys)
+
+    def test_login(self):
+        """Test case for login
+
+        Gets the API key of a user given the username and password via HTTP Basic Authentication.
+        """
+        # Attempt to login
+        self.api.login(self.account)
+
+        # Ensure we cannot login with a bad password
+        self.config.password = "FakePassword123"
+
+        with self.assertRaises(openapi_client.exceptions.ApiException):
+            self.api.login(self.account)
+
+    def test_rotate_api_key(self):
+        """Test case for rotate_api_key
+
+        Rotates a user’s API key.
+        """
+        # Rotate the key and attempt to login with it
+        new_key = self.api.rotate_api_key(self.account)
+
+        self.config.password = new_key
+
+        self.api.login(self.account)
+
+        # We rotated the API key so we have to update the class variable
+        self.__class__.api_key = new_key
+
+    def test_set_password(self):
+        """Test case for set_password
+
+        Changes a user’s password.
+        """
+        # Set a new password and try to authenticate with it
+        test_password = "PAssword!234"
+
+        self.api.set_password(self.account, body=test_password)
+        self.config.password = test_password
+
+        self.api.login(self.account)
+
+        # Attempt to change password with bad auth info
+        self.config.password = "BadPassword"
+
+        with self.assertRaises(openapi_client.exceptions.ApiException):
+            self.api.login(self.account)
+
+        self.config.password = test_password
+
+        # Attempt to set the users password to something invalid
+        invalid_pass = 'SomethingInvalid'
+
+        with self.assertRaises(openapi_client.exceptions.ApiException):
+            self.api.set_password(self.account, body=invalid_pass)
+
+if __name__ == '__main__':
+    unittest.main()

test/python/test_host_factory_api.py

@@ -0,0 +1,93 @@
+from __future__ import absolute_import
+
+import unittest
+import datetime
+
+import openapi_client
+
+from . import api_config
+
+TEST_HOST = "testHost"
+HOST_FACTORY = "dev:host_factory:testFactory"
+FACTORY_POLICY = '''
+- !layer testLayer
+- !host_factory
+    id: testFactory
+    annotations:
+        description: Testing factory
+    layers: [ !layer testLayer ]
+'''
+
+HOST_TOKEN_MEMBERS = ['expiration', 'cidr', 'token']
+NEW_HOST_MEMBERS = ['created_at',  'id', 'owner', 'api_key']
+
+class TestHostFactoryApi(api_config.ConfiguredTest):
+    """HostFactoryApi unit test stubs"""
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        policy_api = openapi_client.api.policies_api.PoliciesApi(cls.client)
+        policy_api.update_policy(cls.account, 'root', FACTORY_POLICY)
+
+    def setUp(self):
+        self.api = openapi_client.api.host_factory_api.HostFactoryApi(self.client)
+
+    def get_host_token(self):
+        """Gets a token used for creating new hosts"""
+        tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+        token = self.api.create_host_token(tomorrow, HOST_FACTORY)
+        return token[0]
+
+    def test_create_host(self):
+        """Test case for create_host
+
+        Creates a Host using the Host Factory.
+        """
+        token = self.get_host_token()['token']
+
+        old_key = dict(self.client.configuration.api_key)
+        self.client.configuration.api_key = {'Authorization': f'Token token="{token}"'}
+        new_host = self.api.create_host(TEST_HOST)
+        self.client.configuration.api_key = old_key
+
+        # Ensure the return object has the correct members
+        self.assertIsInstance(new_host, dict)
+        for member in NEW_HOST_MEMBERS:
+            self.assertIn(member, new_host)
+
+        # Make sure the new host can authenticate
+        authn = openapi_client.api.authn_api.AuthnApi(self.client)
+        authn.authenticate(self.account, f'host/{TEST_HOST}', new_host['api_key'])
+
+    def test_create_host_token(self):
+        """Test case for create_host_token
+
+        Creates one or more host identity tokens.
+        """
+        token = self.get_host_token()
+
+        self.assertIsInstance(token, dict)
+        for i in HOST_TOKEN_MEMBERS:
+            self.assertIn(i, token)
+
+    def test_revoke_host_token(self):
+        """Test case for revoke_host_token
+
+        Revokes a token, immediately disabling it.
+        """
+        token = self.get_host_token()['token']
+
+        old_key = dict(self.client.configuration.api_key)
+
+        self.api.revoke_host_token(token, body='')
+
+        self.client.configuration.api_key = {'Authorization': f'Token token="{token}"'}
+
+        with self.assertRaises(openapi_client.exceptions.ApiException):
+            self.api.create_host(TEST_HOST)
+
+        self.client.configuration.api_key = old_key
+
+if __name__ == '__main__':
+    unittest.main()