- Added argument "platform" on the venafi_connection method to instantiate a given Connector (TPP, VaaS) with no need to pass other values.

- Added example for retrieving a TPP CA public key and principals.

Author: rvelaVenafi

examples/ssh_certificates/retrieve_ca_public_key.py

@@ -0,0 +1,81 @@
+#!/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
+import random
+import string
+from os import environ
+
+from vcert import venafi_connection, Authentication, SCOPE_SSH, SSHKeyPair, SSHCertRequest, write_ssh_files, \
+    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": False})
+    # 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("CA principals:\n")
+    print(ssh_config.ca_principals)
+
+
+if __name__ == '__main__':
+    main()

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
@@ -729,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/ssh_utils.py

@@ -176,7 +176,7 @@ def __init__(self, response):
 
 
 class SSHCATemplateRequest:
-    def __init__(self, ca_template, ca_guid):
+    def __init__(self, ca_template=None, ca_guid=None):
         """
 
         :param str ca_template: