Merge pull request #79 from Venafi/ssh_ca_config

Retrieve SSH CA config

Author: rvelaVenafi

examples/ssh_certificates/retrieve_ca_public_key.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+#
+# Copyright 2021 Venafi, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import logging
+from os import environ
+
+from vcert import venafi_connection, Authentication, SCOPE_SSH, VenafiPlatform, SSHCATemplateRequest
+
+logging.basicConfig(level=logging.INFO)
+logging.getLogger("urllib3").setLevel(logging.ERROR)
+
+
+def main():
+    # Get credentials from environment variables.
+    url = environ.get('TPP_URL')
+    ca_dn = environ.get('TPP_SSH_CADN')
+    ca_guid = environ.get('TPP_SSH_CA_GUID')
+    # Authentication is required for retrieving the CA principals only.
+    user = environ.get("TPP_USER")
+    password = environ.get("TPP_PASSWORD")
+
+    # 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": "/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})
+
+    # If your TPP server certificate is signed with your own CA, or available only via proxy,
+    # you can specify a trust bundle using requests vars:
+    # connector = venafi_connection(url=url, api_key=api_key, access_token=access_token,
+    #                          http_request_kwargs={"verify": "/path-to/bundle.pem"})
+
+    # Create an SSHCATemplateRequest to pass the identifier of the SSH Certificate Authority to retrieve.
+    # Either CADN or Guid can be used as identifiers.
+    request = SSHCATemplateRequest(ca_template=ca_dn)
+    # request = SSHCATemplateRequest(ca_guid=ca_guid)
+
+    # Retrieve the public key.
+    # No Authentication is provided to the Connector so, only the public key is available.
+    ssh_config = connector.retrieve_ssh_config(ca_request=request)
+    pub_key_data = ssh_config.ca_public_key
+    with open("./ca-pub.key", 'w') as ca_file:
+        ca_file.write(pub_key_data)
+
+    # To retrieve the CA principals create an Authentication object with the proper scope to manage SSH certificates.
+    auth = Authentication(user=user, password=password, scope=SCOPE_SSH)
+    # Additionally, you may change the default client id for a custom one.
+    # Make sure this id has been registered on the TPP instance beforehand.
+    # Also, the user (TTP_USER) should be allowed to use this application
+    # and the application should have the ssh permissions enabled.
+    auth.client_id = 'vcert-ssh-ca-pubkey-demo'
+    # Request an access token.
+    # After the request is successful, subsequent api calls will use the same token
+    connector.get_access_token(auth)
+    # Retrieve SSH Certificate Authority public key and principals
+    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("Certificate Authority principals: %s" % ssh_config.ca_principals)
+
+
+if __name__ == '__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/__init__.py

@@ -13,17 +13,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
-from .logger import setup_logger, get_logger, get_child
 from .common import CertificateRequest, CommonConnection, RevocationRequest, ZoneConfig, CertField, KeyType, \
     CustomField, Authentication, SCOPE_CM, SCOPE_PM, SCOPE_SSH, CSR_ORIGIN_LOCAL, CSR_ORIGIN_PROVIDED, \
-    CSR_ORIGIN_SERVICE, CHAIN_OPTION_FIRST, CHAIN_OPTION_IGNORE, CHAIN_OPTION_LAST
+    CSR_ORIGIN_SERVICE, CHAIN_OPTION_FIRST, CHAIN_OPTION_IGNORE, CHAIN_OPTION_LAST, VenafiPlatform
 from .connection_cloud import CloudConnection
 from .connection_tpp import TPPConnection
 from .connection_tpp_token import TPPTokenConnection
 from .connection_fake import FakeConnection
+from .errors import VenafiError
+from .logger import setup_logger, get_logger, get_child
 from .pem import Certificate
-from .ssh_utils import SSHCertRequest, SSHKeyPair, write_ssh_files
+from .ssh_utils import SSHCertRequest, SSHKeyPair, write_ssh_files, SSHCATemplateRequest, SSHConfig
 from .tpp_utils import IssuerHint
 
 setup_logger()
@@ -54,7 +54,7 @@ def Connection(url=None, token=None, user=None, password=None, fake=False, http_
 
 
 def venafi_connection(url=None, api_key=None, user=None, password=None, access_token=None, refresh_token=None,
-                      fake=False, http_request_kwargs=None):
+                      fake=False, http_request_kwargs=None, platform=None):
     """
     Return connection based on credentials list.
     Venafi Platform (TPP) requires URL and access_token (or user and password for getting a new access_token)
@@ -68,14 +68,26 @@ def venafi_connection(url=None, api_key=None, user=None, password=None, access_t
     :param str refresh_token: TPP refresh token (optional)
     :param bool fake: Use fake connection
     :param dict[str, Any] http_request_kwargs: Option for specifying trust bundle or to operate insecurely.
+    :param VenafiPlatform platform: The platform to be used with the Connector
     :rtype CommonConnection:
     """
-    if fake:
-        return FakeConnection()
-    if url and (access_token or refresh_token or (user and password)):
-        return TPPTokenConnection(url=url, user=user, password=password, access_token=access_token,
-                                  refresh_token=refresh_token, http_request_kwargs=http_request_kwargs)
-    if api_key:
-        return CloudConnection(token=api_key, url=url, http_request_kwargs=http_request_kwargs)
+    if platform:
+        if platform == VenafiPlatform.FAKE:
+            return FakeConnection()
+        elif platform == VenafiPlatform.TPP:
+            return TPPTokenConnection(url=url, user=user, password=password, access_token=access_token,
+                                      refresh_token=refresh_token, http_request_kwargs=http_request_kwargs)
+        elif platform == VenafiPlatform.VAAS:
+            return CloudConnection(token=api_key, url=url, http_request_kwargs=http_request_kwargs)
+        else:
+            raise VenafiError("Invalid Platform: %s. Cannot instantiate a Connector." % platform)
     else:
-        raise Exception("Bad credentials list")
+        if fake:
+            return FakeConnection()
+        if url and (access_token or refresh_token or (user and password)):
+            return TPPTokenConnection(url=url, user=user, password=password, access_token=access_token,
+                                      refresh_token=refresh_token, http_request_kwargs=http_request_kwargs)
+        if api_key:
+            return CloudConnection(token=api_key, url=url, http_request_kwargs=http_request_kwargs)
+        else:
+            raise VenafiError("Bad credentials list")

vcert/common.py

@@ -22,6 +22,7 @@
 import socket
 import sys
 
+import enum
 import ipaddress
 from builtins import bytes
 from cryptography import x509
@@ -36,7 +37,7 @@
 from .errors import VenafiConnectionError, ServerUnexptedBehavior, BadData, ClientBadData
 from .http import HTTPStatus
 from .policy import PolicySpecification
-from .ssh_utils import SSHCertRequest, SSHRetrieveResponse
+from .ssh_utils import SSHCertRequest, SSHRetrieveResponse, SSHCATemplateRequest, SSHConfig
 from .tpp_utils import IssuerHint
 
 MIME_JSON = "application/json"
@@ -691,6 +692,14 @@ def retrieve_ssh_cert(self, request):
         """
         raise NotImplementedError
 
+    def retrieve_ssh_config(self, ca_request):
+        """
+
+        :param SSHCATemplateRequest ca_request:
+        :rtype: SSHConfig
+        """
+        raise NotImplementedError
+
     @staticmethod
     def process_server_response(r):
         if r.status_code not in (HTTPStatus.OK, HTTPStatus.ACCEPTED, HTTPStatus.CREATED, HTTPStatus.CONFLICT):
@@ -721,3 +730,16 @@ def process_server_response(r):
         else:
             log.error("Unexpected content type: %s for request %s" % (content_type, r.request.url))
             raise ServerUnexptedBehavior
+
+
+class VenafiPlatform(enum.IntEnum):
+    def __new__(cls, value, description):
+        obj = int.__new__(cls, value)
+        obj._value_ = value
+        obj.description = description
+
+        return obj
+
+    FAKE = 100, "Connector for testing purposes"
+    TPP = 200, "Trust Protection Platfom"
+    VAAS = 400, "Venafi as a Service"

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()
@@ -116,3 +138,8 @@ def _read_config_dn(self, dn, attribute_name):
         if status != HTTPStatus.OK:
             raise ServerUnexptedBehavior("")
         return data
+
+    def _is_valid_auth(self):
+        if self._token:
+            return True
+        return False