Added log messages to POST operations for better tracing.

Minor fixes.

Author: rvelaVenafi

tests/test_env.py

@@ -21,27 +21,27 @@
 from future.backports.datetime import datetime
 from six import text_type
 
-FAKE = environ.get('FAKE')
+FAKE = environ.get("FAKE")
 RANDOM_DOMAIN = environ.get("RANDOM_DOMAIN")
-TPP_URL = environ.get('TPP_URL')
-TPP_USER = environ.get('TPP_USER')
-TPP_PASSWORD = environ.get('TPP_PASSWORD')
-TPP_ZONE = environ.get('TPP_ZONE')
-TPP_ZONE_ECDSA = environ.get('TPP_ZONE_ECDSA')
+TPP_URL = environ.get("TPP_URL")
+TPP_USER = environ.get("TPP_USER")
+TPP_PASSWORD = environ.get("TPP_PASSWORD")
+TPP_ZONE = environ.get("TPP_ZONE")
+TPP_ZONE_ECDSA = environ.get("TPP_ZONE_ECDSA")
 TPP_TOKEN_URL = environ.get("TPP_TOKEN_URL")
 TPP_ACCESS_TOKEN = environ.get("TPP_ACCESS_TOKEN")
-CLOUD_URL = environ.get('CLOUD_URL')
-CLOUD_APIKEY = environ.get('CLOUD_APIKEY')
-CLOUD_ZONE = environ.get('CLOUD_ZONE')
+CLOUD_URL = environ.get("CLOUD_URL")
+CLOUD_APIKEY = environ.get("CLOUD_APIKEY")
+CLOUD_ZONE = environ.get("CLOUD_ZONE")
 
-TPP_PM_ROOT = environ.get('TPP_PM_ROOT')
-TPP_CA_NAME = environ.get('TPP_CA_NAME')
-CLOUD_ENTRUST_CA_NAME = environ.get('CLOUD_ENTRUST_CA_NAME')
-CLOUD_DIGICERT_CA_NAME = environ.get('CLOUD_DIGICERT_CA_NAME')
+TPP_PM_ROOT = environ.get("TPP_PM_ROOT")
+TPP_CA_NAME = environ.get("TPP_CA_NAME")
+CLOUD_ENTRUST_CA_NAME = environ.get("CLOUD_ENTRUST_CA_NAME")
+CLOUD_DIGICERT_CA_NAME = environ.get("CLOUD_DIGICERT_CA_NAME")
 
-SSH_CADN = "\\VED\\Certificate Authority\\SSH\\Templates\\vCert-team"
+SSH_CADN = environ.get("SSH_CADN")
 
-if not isinstance(RANDOM_DOMAIN, text_type):
+if RANDOM_DOMAIN and not isinstance(RANDOM_DOMAIN, text_type):
     RANDOM_DOMAIN = RANDOM_DOMAIN.decode()
 
 

tests/test_ssh.py

@@ -18,8 +18,9 @@
 import unittest
 
 from test_env import timestamp, TPP_TOKEN_URL, TPP_USER, TPP_PASSWORD, SSH_CADN
-from vcert import CommonConnection, SSHCertRequest, SSHRetrieveResponse, TPPTokenConnection, Authentication, \
+from vcert import CommonConnection, SSHCertRequest, TPPTokenConnection, Authentication, \
     SCOPE_SSH, generate_ssh_keypair
+from vcert.ssh_utils import SSHRetrieveResponse
 
 logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger('vcert-test')
@@ -37,19 +38,19 @@ def __init__(self, *args, **kwargs):
         super(TestTPPSSHCertificate, self).__init__(*args, **kwargs)
 
     def test_enroll_local_generated_keypair(self):
-        pub_key, priv_key = generate_ssh_keypair(key_size=4096, passphrase="foobar")
+        keypair = generate_ssh_keypair(key_size=4096, passphrase="foobar")
 
         request = SSHCertRequest(cadn=SSH_CADN, key_id=_random_key_id())
         request.validity_period = "4h"
         request.source_addresses = ["test.com"]
-        request.public_key_data = pub_key
+        request.set_public_key_data(keypair.public_key)
         response = _enroll_ssh_cert(self.tpp_conn, request)
         self.assertTrue(response.private_key_data is None,
                         SERVICE_GENERATED_NO_KEY_ERROR % ("Private", "not", request.key_id))
         self.assertTrue(response.public_key_data, SERVICE_GENERATED_NO_KEY_ERROR % ("Public", "", request.key_id))
-        self.assertTrue(response.public_key_data == request.public_key_data,
+        self.assertTrue(response.public_key_data == request.get_public_key_data(),
                         "Public key on response does not match request.\nExpected: %s\nGot: %s"
-                        % (request.public_key_data, response.public_key_data))
+                        % (request.get_public_key_data(), response.public_key_data))
         self.assertTrue(response.cert_data, SSH_CERT_DATA_ERROR % request.key_id)
 
     def test_enroll_service_generated_keypair(self):
@@ -69,7 +70,7 @@ def _enroll_ssh_cert(connector, request):
     :rtype: SSHRetrieveResponse
     """
     success = connector.request_ssh_cert(request)
-    assert success  # self.assertTrue(success, "SSH certificate request failed. Key id: %s CADN: %s" % (request.key_id, request.cadn))
+    assert success
     response = connector.retrieve_ssh_cert(request)
     assert isinstance(response, SSHRetrieveResponse)
     return response

vcert/__init__.py

@@ -21,7 +21,7 @@
 from .connection_tpp_token import TPPTokenConnection
 from .connection_fake import FakeConnection
 from .pem import Certificate
-from .ssh_utils import SSHCertRequest, SSHRetrieveResponse, SSHCertDetails, generate_ssh_keypair
+from .ssh_utils import SSHCertRequest, generate_ssh_keypair
 
 
 def Connection(url=None, token=None, user=None, password=None, fake=False, http_request_kwargs=None):

vcert/common.py

@@ -697,7 +697,8 @@ def process_server_response(r):
                 log_errors(r.json())
             except:
                 log_errors(r.content)
-            raise VenafiConnectionError("Server status: %s\nResponse: %s" % (r.status_code, r.request.url))
+            raise VenafiConnectionError("\n\tServer status: %s\n\tURL: %s\n\tResponse: %s"
+                                        % (r.status_code, r.request.url, r.content))
 
         content_type = r.headers.get("content-type")
         # Content-type not present, return status and reason (if any)

vcert/connection_tpp_token.py

@@ -91,6 +91,7 @@ def _post(self, url=None, data=None, check_token=True, include_headers=True):
             headers = {HEADER_AUTHORIZATION: token, 'content-type': MIME_JSON, "cache-control": "no-cache"}
 
         if isinstance(data, dict):
+            log.debug("POST Request\n\tURL: %s\n\tHeaders:%s\n\tBody:%s\n" % (self._base_url+url, headers, data))
             r = requests.post(self._base_url + url, headers=headers, json=data,  **self._http_request_kwargs)
         else:
             log.error("Unexpected client data type: %s for %s" % (type(data), url))

vcert/ssh_utils.py

@@ -60,7 +60,7 @@ def __init__(self, cadn, key_id, validity_period=None, policy_dn=None, object_na
         self.object_name = object_name
         self.destination_addresses = destination_addresses
         self.principals = principals
-        self.public_key_data = public_key_data
+        self._public_key_data = public_key_data
         self.extensions = extensions
         self.force_command = force_command
         self.source_addresses = source_addresses
@@ -74,24 +74,25 @@ def __init__(self, cadn, key_id, validity_period=None, policy_dn=None, object_na
         self.include_cert_details = include_cert_details
         self.timeout = timeout
 
-    @property
-    def public_key_data(self):
+    def get_public_key_data(self):
         """
         :rtype: str
         """
+        if not self._public_key_data:
+            return None
         temp = self._public_key_data.rstrip("\r\n")
         if self.key_id:
             return "%s %s" % (temp, self.key_id)
         else:
             return temp
 
-    @public_key_data.setter
-    def public_key_data(self, public_key):
+    def set_public_key_data(self, key):
         """
-        :param str public_key:
+
+        :param str key:
         :rtype: None
         """
-        self._public_key_data = public_key
+        self._public_key_data = key
 
 
 class SSHCertResponse:
@@ -160,6 +161,17 @@ def __init__(self, response):
         self.error_msg = response["ErrorMessage"] if "ErrorMessage" in response else None
 
 
+class SSHKeyPair:
+    def __init__(self, private, public):
+        """
+
+        :param str private:
+        :param str public:
+        """
+        self.private_key = private
+        self.public_key = public
+
+
 def build_tpp_request(request):
     """
     :param SSHCertRequest request:
@@ -189,8 +201,8 @@ def build_tpp_request(request):
         data["Principals"] = request.principals
     if request.validity_period:
         data["ValidityPeriod"] = request.validity_period
-    if request.public_key_data:
-        data["PublicKeyData"] = request.public_key_data
+    if request.get_public_key_data():
+        data["PublicKeyData"] = request.get_public_key_data()
     if request.extensions:
         data["Extensions"] = request.extensions
     if request.force_command:
@@ -229,7 +241,7 @@ def generate_ssh_keypair(key_size=DEFAULT_SSH_KEY_SIZE, passphrase=None):
     Generates a key pair (private, public) for use with SSH
     :param int key_size:
     :param str passphrase:
-    :rtype: tuple
+    :rtype: SSHKeyPair
     """
     if passphrase:
         encryption = serialization.BestAvailableEncryption(passphrase)
@@ -250,4 +262,4 @@ def generate_ssh_keypair(key_size=DEFAULT_SSH_KEY_SIZE, passphrase=None):
         format=serialization.PublicFormat.OpenSSH
     )
 
-    return private_key, public_key
+    return SSHKeyPair(private_key, public_key)