- Added test cases for SSH CA public key retrieval

- Added abstract methods for HTTP operations POST and GET on TPP implementations.

Author: rvelaVenafi

examples/ssh_certificates/retrieve_ca_public_key.py

@@ -15,12 +15,9 @@
 # limitations under the License.
 #
 import logging
-import random
-import string
 from os import environ
 
-from vcert import venafi_connection, Authentication, SCOPE_SSH, SSHKeyPair, SSHCertRequest, write_ssh_files, \
-    VenafiPlatform, SSHCATemplateRequest
+from vcert import venafi_connection, Authentication, SCOPE_SSH, VenafiPlatform, SSHCATemplateRequest
 
 logging.basicConfig(level=logging.INFO)
 logging.getLogger("urllib3").setLevel(logging.ERROR)
@@ -37,7 +34,8 @@ def main():
 
     # A Connector can be instantiated with no values by using the platform argument.
     # url argument is always required for TPP.
-    connector = venafi_connection(platform=VenafiPlatform.TPP, url=url, http_request_kwargs={"verify": False})
+    connector = venafi_connection(platform=VenafiPlatform.TPP, url=url,
+                                  http_request_kwargs={"verify": "/tmp/chain.pem"})
     # Optionally, the connector can be instantiated passing the specific arguments:
     # connector = venafi_connection(url=url, user=user, password=password, http_request_kwargs={"verify": False})
 
@@ -72,10 +70,8 @@ def main():
     ssh_config = connector.retrieve_ssh_config(ca_request=request)
     with open("./ca2-pub.key", 'w') as ca_file:
         ca_file.write(pub_key_data)
-
-    print("CA principals:\n")
-    print(ssh_config.ca_principals)
+    print("Certificate Authority principals: %s" % ssh_config.ca_principals)
 
 
 if __name__ == '__main__':
-    main()
+    main()

tests/test_ssh.py

@@ -23,8 +23,8 @@
 from test_env import TPP_TOKEN_URL, TPP_USER, TPP_PASSWORD, TPP_SSH_CADN
 from test_utils import timestamp
 from vcert import CommonConnection, SSHCertRequest, TPPTokenConnection, Authentication, \
-    SCOPE_SSH, write_ssh_files, logger
-from vcert.ssh_utils import SSHRetrieveResponse, SSHKeyPair
+    SCOPE_SSH, write_ssh_files, logger, venafi_connection, VenafiPlatform
+from vcert.ssh_utils import SSHRetrieveResponse, SSHKeyPair, SSHCATemplateRequest
 
 log = logger.get_child("test-ssh")
 
@@ -34,7 +34,7 @@
 
 class TestTPPSSHCertificate(unittest.TestCase):
     def __init__(self, *args, **kwargs):
-        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={"verify": False})
+        self.tpp_conn = TPPTokenConnection(url=TPP_TOKEN_URL, http_request_kwargs={"verify": "/tmp/chain.pem"})
         auth = Authentication(user=TPP_USER, password=TPP_PASSWORD, scope=SCOPE_SSH)
         self.tpp_conn.get_access_token(auth)
         super(TestTPPSSHCertificate, self).__init__(*args, **kwargs)
@@ -65,9 +65,25 @@ def test_enroll_service_generated_keypair(self):
         self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR % ("Public", "", request.key_id))
         self.assertTrue(response.certificate_data, SSH_CERT_DATA_ERROR % request.key_id)
 
+    def test_retrieve_ca_public_key(self):
+        tpp_connector = venafi_connection(platform=VenafiPlatform.TPP, url=TPP_TOKEN_URL,
+                                          http_request_kwargs={"verify": "/tmp/chain.pem"})
+        request = SSHCATemplateRequest(ca_template=TPP_SSH_CADN)
+        ssh_config = tpp_connector.retrieve_ssh_config(ca_request=request)
+        self.assertIsNotNone(ssh_config.ca_public_key, "%s Public Key data is empty" % TPP_SSH_CADN)
+        self.assertIsNone(ssh_config.ca_principals, "%s default principals is not empty" % TPP_SSH_CADN)
+        log.debug("%s Public Key data:\n%s" % (TPP_SSH_CADN, ssh_config.ca_public_key))
+
+    def test_retrieve_ca_public_key_and_principals(self):
+        request = SSHCATemplateRequest(ca_template=TPP_SSH_CADN)
+        ssh_config = self.tpp_conn.retrieve_ssh_config(ca_request=request)
+        self.assertIsNotNone(ssh_config.ca_public_key, "%s Public Key data is empty" % TPP_SSH_CADN)
+        self.assertIsNotNone(ssh_config.ca_principals, "%s default principals is empty" % TPP_SSH_CADN)
+        log.debug("%s Public Key data: %s" % (TPP_SSH_CADN, ssh_config.ca_public_key))
+        log.debug("%s default principals: %s" % (TPP_SSH_CADN, ssh_config.ca_principals))
 
-class TestSSHUtils(unittest.TestCase):
 
+class TestSSHUtils(unittest.TestCase):
     def test_write_ssh_files(self):
         key_id = _random_key_id()
         normalized_name = re.sub(r"[^A-Za-z0-9]+", "_", key_id)

vcert/connection_tpp.py

@@ -57,6 +57,28 @@ def __setattr__(self, key, value):
     def __str__(self):
         return "[TPP] %s" % self._base_url
 
+    def get(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        url = args[self.ARG_URL] if self.ARG_URL in args else None
+        params = args[self.ARG_PARAMS] if self.ARG_PARAMS in args else None
+
+        return self._get(url=url, params=params)
+
+    def post(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        url = args[self.ARG_URL] if self.ARG_URL in args else None
+        data = args[self.ARG_DATA] if self.ARG_DATA in args else None
+
+        return self._post(url=url, data=data)
+
     def _get(self, url="", params=None):
         if not self._token or self._token[1] < time.time() + 1:
             self.auth()

vcert/connection_tpp_abstract.py

@@ -598,18 +598,27 @@ def retrieve_ssh_config(self, ca_request):
         value = None
         if ca_request.template:
             key = 'DN'
-            value = "%s%s%s" % (CA_ROOT_PATH, PATH_SEPARATOR, ca_request.guid)
+            value = ca_request.template
+            if not value.startswith(PATH_SEPARATOR):
+                value = "%s%s" % (PATH_SEPARATOR, value)
+            if not value.startswith(CA_ROOT_PATH):
+                value = "%s%s" % (CA_ROOT_PATH, value)
         elif ca_request.guid:
             key = 'guid'
             value = ca_request.guid
         else:
             raise ClientBadData("CA Guid or CA template must be provided to retrieve SSH config.")
 
+        value = url_parse.quote(value)
         query = "%s=%s" % (key, value)
-        query = url_parse.quote(query)
         url = "%s?%s" % (URLS.SSH_CA_PUBLIC_KEY, query)
 
-        status, data = self._get(url=url, params=None)
+        args = {
+            self.ARG_URL: url,
+            self.ARG_CHECK_TOKEN: False,
+            self.ARG_INCLUDE_TOKEN_HEADER: False
+        }
+        status, data = self.get(args=args)
         if status == HTTPStatus.OK:
             ssh_config_response = SSHConfig()
             ssh_config_response.ca_public_key = data
@@ -621,6 +630,28 @@ def retrieve_ssh_config(self, ca_request):
             raise ServerUnexptedBehavior("Server returns %d status on requesting SSH CA Public Key Data for %s = %s."
                                          % (status, key, value))
 
+    ARG_URL = 'url'
+    ARG_PARAMS = 'params'
+    ARG_CHECK_TOKEN = 'check_token'
+    ARG_INCLUDE_TOKEN_HEADER = 'include_token_header'
+    ARG_DATA = 'data'
+
+    def get(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        raise NotImplementedError
+
+    def post(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        raise NotImplementedError
+
     # ======================================== API IMPLEMENTATION ENDS ======================================== #
     # ========================================================================================================= #
 
@@ -876,13 +907,22 @@ def _retrieve_ssh_ca_details(self, ca_request):
         """
         json_request = dict()
         if ca_request.template:
-            json_request['DN'] = "%s%s%s" % (CA_ROOT_PATH, PATH_SEPARATOR, ca_request.guid)
+            value = ca_request.template
+            if not value.startswith(PATH_SEPARATOR):
+                value = "%s%s" % (PATH_SEPARATOR, value)
+            if not value.startswith(CA_ROOT_PATH):
+                value = "%s%s" % (CA_ROOT_PATH, value)
+            json_request['DN'] = value
         elif ca_request.guid:
             json_request['Guid'] = ca_request.guid
         else:
             raise ClientBadData("CA Guid or CA template must be provided to retrieve SSH CA details.")
 
-        status, data = self._post(URLS.SSH_CA_DETAILS, json_request)
+        args = {
+            self.ARG_URL: URLS.SSH_CA_DETAILS,
+            self.ARG_DATA: json_request
+        }
+        status, data = self.post(args=args)
         if status == HTTPStatus.OK:
             response_object = SSHResponse(data['Response'])
             if response_object.success:

vcert/connection_tpp_token.py

@@ -63,26 +63,58 @@ def __setattr__(self, key, value):
     def __str__(self):
         return "[TPP] %s" % self._base_url
 
-    def _get(self, url=None, params=None, check_token=True, include_headers=True):
+    def get(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        url = args[self.ARG_URL] if self.ARG_URL in args else None
+        params = args[self.ARG_PARAMS] if self.ARG_PARAMS in args else None
+        check_token = args[self.ARG_CHECK_TOKEN] if self.ARG_CHECK_TOKEN in args else True
+        include_token_header = args[self.ARG_INCLUDE_TOKEN_HEADER] if self.ARG_INCLUDE_TOKEN_HEADER in args else True
+
+        return self._get(url=url, params=params, check_token=check_token, include_token_header=include_token_header)
+
+    def post(self, args):
+        """
+
+        :param dict args:
+        :rtype: tuple[Any, Any]
+        """
+        url = args[self.ARG_URL] if self.ARG_URL in args else None
+        data = args[self.ARG_DATA] if self.ARG_DATA in args else None
+        check_token = args[self.ARG_CHECK_TOKEN] if self.ARG_CHECK_TOKEN in args else True
+        include_token_header = args[self.ARG_INCLUDE_TOKEN_HEADER] if self.ARG_INCLUDE_TOKEN_HEADER in args else True
+
+        return self._post(url=url, data=data, check_token=check_token, include_token_header=include_token_header)
+
+    def _get(self, url=None, params=None, check_token=True, include_token_header=True):
         if check_token:
             self._check_token()
 
-        headers = {}
-        if include_headers:
+        headers = {
+            'content-type': MIME_JSON,
+            'cache-control': "no-cache"
+        }
+        if include_token_header:
             token = self._get_auth_header_value(self._auth.access_token)
-            headers = {HEADER_AUTHORIZATION: token, 'content-type': MIME_JSON, 'cache-control': 'no-cache'}
+            headers[HEADER_AUTHORIZATION] = token
 
         r = requests.get(self._base_url + url, headers=headers, params=params, **self._http_request_kwargs)
         return self.process_server_response(r)
 
-    def _post(self, url=None, data=None, check_token=True, include_headers=True):
+    def _post(self, url=None, data=None, check_token=True, include_token_header=True):
         if check_token:
             self._check_token()
 
-        headers = {}
-        if include_headers:
+        headers = {
+            'content-type': MIME_JSON,
+            'cache-control': "no-cache"
+        }
+        if include_token_header:
             token = self._get_auth_header_value(self._auth.access_token)
-            headers = {HEADER_AUTHORIZATION: token, 'content-type': MIME_JSON, "cache-control": "no-cache"}
+            headers[HEADER_AUTHORIZATION] = token
 
         if isinstance(data, dict):
             log.debug("POST Request\n\tURL: %s\n\tHeaders:%s\n\tBody:%s\n" % (self._base_url+url, headers, data))