use x509 to check if server certificate has expired
authorThomasV <thomasv@gitorious>
Fri, 27 Jun 2014 09:36:51 +0000 (11:36 +0200)
committerThomasV <thomasv@gitorious>
Fri, 27 Jun 2014 09:36:51 +0000 (11:36 +0200)
lib/interface.py
lib/paymentrequest.py
lib/x509.py

index 0b1cba4..4424a0c 100644 (file)
@@ -27,48 +27,12 @@ from version import ELECTRUM_VERSION, PROTOCOL_VERSION
 from util import print_error, print_msg
 from simple_config import SimpleConfig
 
+import x509
 
 DEFAULT_TIMEOUT = 5
 proxy_modes = ['socks4', 'socks5', 'http']
 
 
-def check_cert(host, cert):
-    from OpenSSL import crypto as c
-    _cert = c.load_certificate(c.FILETYPE_PEM, cert)
-
-    m = "host: %s\n"%host
-    m += "has_expired: %s\n"% _cert.has_expired()
-    m += "pubkey: %s bits\n" % _cert.get_pubkey().bits()
-    m += "serial number: %s\n"% _cert.get_serial_number() 
-    #m += "issuer: %s\n"% _cert.get_issuer()
-    #m += "algo: %s\n"% _cert.get_signature_algorithm() 
-    m += "version: %s\n"% _cert.get_version()
-    print_msg(m)
-
-
-def cert_has_expired(cert_path):
-    try:
-        import OpenSSL
-    except Exception:
-        print_error("Warning: cannot import OpenSSL")
-        return False
-    from OpenSSL import crypto as c
-    with open(cert_path) as f:
-        cert = f.read()
-    _cert = c.load_certificate(c.FILETYPE_PEM, cert)
-    return _cert.has_expired()
-
-
-def check_certificates():
-    config = SimpleConfig()
-    mydir = os.path.join(config.path, "certs")
-    certs = os.listdir(mydir)
-    for c in certs:
-        print c
-        p = os.path.join(mydir,c)
-        with open(p) as f:
-            cert = f.read()
-        check_cert(c, cert)
     
 
 def cert_verify_hostname(s):
@@ -392,11 +356,23 @@ class Interface(threading.Thread):
                         os.unlink(rej)
                     os.rename(temporary_path, rej)
                 else:
-                    if cert_has_expired(cert_path):
+                    with open(cert_path) as f:
+                        cert = f.read()
+                    try:
+                        x = x509.X509()
+                        x.parse(cert)
+                        x.slow_parse()
+                    except:
+                        traceback.print_exc(file=sys.stdout)
+                        print_error("wrong certificate", self.host)
+                        return
+                    try:
+                        x.check_date()
+                    except:
                         print_error("certificate has expired:", cert_path)
                         os.unlink(cert_path)
-                    else:
-                        print_error("wrong certificate", self.host)
+                        return
+                    print_error("wrong certificate", self.host)
                 return
             except Exception:
                 print_error("wrap_socket failed", self.host)
@@ -614,6 +590,38 @@ class Interface(threading.Thread):
         return out
 
 
-if __name__ == "__main__":
 
-    check_certificates()
+
+def check_cert(host, cert):
+    try:
+        x = x509.X509()
+        x.parse(cert)
+        x.slow_parse()
+    except:
+        traceback.print_exc(file=sys.stdout)
+        return
+
+    try:
+        x.check_date()
+        expired = False
+    except:
+        expired = True
+
+    m = "host: %s\n"%host
+    m += "has_expired: %s\n"% expired
+    print_msg(m)
+
+
+def test_certificates():
+    config = SimpleConfig()
+    mydir = os.path.join(config.path, "certs")
+    certs = os.listdir(mydir)
+    for c in certs:
+        print c
+        p = os.path.join(mydir,c)
+        with open(p) as f:
+            cert = f.read()
+        check_cert(c, cert)
+
+if __name__ == "__main__":
+    test_certificates()
index 8f96966..9d4138a 100644 (file)
@@ -169,8 +169,11 @@ class PaymentRequest:
             x.slow_parse()
             x509_chain.append(x)
             if i == 0:
-                if not x.check_name(self.domain):
-                    self.error = "Certificate Domain Mismatch"
+                try:
+                    x.check_date()
+                    x.check_name(self.domain)
+                except Exception as e:
+                    self.error = str(e)
                     return
             else:
                 if not x.check_ca():
index 0656820..31d5c80 100644 (file)
@@ -170,9 +170,8 @@ class X509(tlslite.X509):
             return None
         return not_after - datetime.utcnow()
 
-    def check_name(self, expected):
+    def check_date(self):
         not_before, not_after = self.extract_dates()
-        cert_names = self.extract_names()
         now = datetime.utcnow()
         if not_before > now:
             raise CertificateError(
@@ -180,6 +179,9 @@ class X509(tlslite.X509):
         if not_after <= now:
             raise CertificateError(
                 'Certificate has expired.')
+
+    def check_name(self, expected):
+        cert_names = self.extract_names()
         if '.' in expected:
             expected_wild = expected[expected.index('.'):]
         else: