Merge pull request #104 from alkor/cleanup-crypto

Cleanup crypto

opcua/client/client.py

@@ -15,7 +15,7 @@ from opcua.crypto import security_policies
 use_crypto = True
 try:
     from opcua.crypto import uacrypto
-except:
+except ImportError:
     print("cryptography is not installed, use of crypto disabled")
     use_crypto = False
 
@@ -53,9 +53,9 @@ class KeepAlive(Thread):
 
     def stop(self):
         self.logger.debug("stoping keepalive thread")
+        self._dostop = True
         with self._cond:
             self._cond.notify_all()
-        self._dostop = True
 
 
 class Client(object):

opcua/crypto/uacrypto.py

@@ -27,12 +27,6 @@ def x509_from_der(data):
     return x509.load_der_x509_certificate(data, default_backend())
 
 
-def x509_to_der(cert):
-    if not cert:
-        return b''
-    return cert.public_bytes(serialization.Encoding.DER)
-
-
 def load_private_key(path):
     _, ext = os.path.splitext(path)
     with open(path, "rb") as f:
@@ -162,6 +156,23 @@ def p_sha1(secret, seed, sizes=()):
     return tuple(parts)
 
 
+def x509_name_to_string(name):
+    parts = ["{}={}".format(attr.oid._name, attr.value) for attr in name]
+    return ', '.join(parts)
+
+
+def x509_to_string(cert):
+    """
+    Convert x509 certificate to human-readable string
+    """
+    if cert.subject == cert.issuer:
+        issuer = ' (self-signed)'
+    else:
+        issuer = ', issuer: {}'.format(x509_name_to_string(cert.issuer))
+    # TODO: show more information
+    return "{}{}, {} - {}".format(x509_name_to_string(cert.subject), issuer, cert.not_valid_before, cert.not_valid_after)
+
+
 if __name__ == "__main__":
     # Convert from PEM to DER
     cert = load_certificate("../examples/server_cert.pem")

opcua/server/server.py

@@ -22,7 +22,7 @@ from opcua.crypto import security_policies
 use_crypto = True
 try:
     from opcua.crypto import uacrypto
-except:
+except ImportError:
     print("cryptography is not installed, use of crypto disabled")
     use_crypto = False
 

opcua/tools.py

@@ -403,11 +403,22 @@ def application_to_strings(app):
     return result  # ['{}: {}'.format(n, v) for (n, v) in result]
 
 
+def cert_to_string(der):
+    if not der:
+        return '[no certificate]'
+    try:
+        from opcua.crypto import uacrypto
+    except ImportError:
+        return "{} bytes".format(len(der))
+    cert = uacrypto.x509_from_der(der)
+    return uacrypto.x509_to_string(cert)
+
+
 def endpoint_to_strings(ep):
     result = [('Endpoint URL', ep.EndpointUrl)]
     result += application_to_strings(ep.Server)
     result += [
-        ('Server Certificate', len(ep.ServerCertificate)),
+        ('Server Certificate', cert_to_string(ep.ServerCertificate)),
         ('Security Mode', str(ep.SecurityMode)),
         ('Security Policy URI', ep.SecurityPolicyUri)]
     for tok in ep.UserIdentityTokens: