acmecert: Added certificate expiry check command.

diff --git a/acmecert b/acmecert
index 91fcece..9b30c44 100755 (executable)
--- a/acmecert
+++ b/acmecert
@@ -50,6 +50,42 @@ def jreq(url, data, auth):
     with req(url, data=enc) as resp:
         return json.loads(resp.read().decode("utf-8")), resp.headers
 
+class certificate(object):
+    @property
+    def enddate(self):
+        # No X509 parser for Python?
+        import subprocess, re, calendar
+        with subprocess.Popen(["openssl", "x509", "-noout", "-enddate"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) as openssl:
+            openssl.stdin.write(self.data.encode("us-ascii"))
+            openssl.stdin.close()
+            resp = openssl.stdout.read().decode("utf-8")
+            if openssl.wait() != 0:
+                raise Exception("openssl error")
+        m = re.search(r"notAfter=(.*)$", resp)
+        if m is None: raise Exception("unexpected openssl reply: %r" % (resp,))
+        return calendar.timegm(time.strptime(m.group(1), "%b %d %H:%M:%S %Y GMT"))
+
+    def expiring(self, timespec):
+        if timespec.endswith("y"):
+            timespec = int(timespec[:-1]) * 365 * 86400
+        elif timespec.endswith("m"):
+            timespec = int(timespec[:-1]) * 30 * 86400
+        elif timespec.endswith("w"):
+            timespec = int(timespec[:-1]) * 7 * 86400
+        elif timespec.endswith("d"):
+            timespec = int(timespec[:-1]) * 86400
+        elif timespec.endswith("h"):
+            timespec = int(timespec[:-1]) * 3600
+        else:
+            timespec = int(timespec)
+        return (self.enddate - time.time()) < timespec
+
+    @classmethod
+    def read(cls, fp):
+        self = cls()
+        self.data = fp.read()
+        return self
+
 class signreq(object):
     def domains(self):
         # No PCKS10 parser for Python?
@@ -57,7 +93,7 @@ class signreq(object):
         with subprocess.Popen(["openssl", "req", "-noout", "-text"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) as openssl:
             openssl.stdin.write(self.data.encode("us-ascii"))
             openssl.stdin.close()
-            resp = openssl.stdout.read().decode("utf8")
+            resp = openssl.stdout.read().decode("utf-8")
             if openssl.wait() != 0:
                 raise Exception("openssl error")
         m = re.search(r"X509v3 Subject Alternative Name:[^\n]*\n\s*((\w+:\S+,\s*)*\w+:\S+)\s*\n", resp)
@@ -300,6 +336,10 @@ def main(argv):
         orderid = mkorder(acct, csr)["acmecert.location"]
         authorder(acct, htconf, orderid)
         sys.stdout.write(finalize(acct, csr, orderid))
+    elif args[0] == "check-cert":
+        with open(args[1], "r") as fp:
+            crt = certificate.read(fp)
+        sys.exit(1 if crt.expiring(args[2]) else 0)
     elif args[0] == "directory":
         pprint.pprint(directory())
     else: