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?
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)
with req(order["certificate"]) as resp:
return resp.read().decode("us-ascii")
+class maybeopen(object):
+ def __init__(self, name, mode):
+ if name == "-":
+ self.opened = False
+ if mode == "r":
+ self.fp = sys.stdin
+ elif mode == "w":
+ self.fp = sys.stdout
+ else:
+ raise ValueError(mode)
+ else:
+ self.opened = True
+ self.fp = open(name, mode)
+
+ def __enter__(self):
+ return self.fp
+
+ def __exit__(self, *excinfo):
+ if self.opened:
+ self.fp.close()
+ return False
+
+class usageerr(Exception):
+ pass
+
+commands = {}
+
+def cmd_reg(args):
+ "usage: acmecert reg [OUTPUT-FILE]"
+ acct = register()
+ os.umask(0o077)
+ with maybeopen(args[1] if len(args) > 1 else "-", "w") as fp:
+ acct.write(fp)
+commands["reg"] = cmd_reg
+
+def cmd_validate_acct(args):
+ "usage: acmecert validate-acct ACCOUNT-FILE"
+ if len(args) < 2: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ account.read(fp).validate()
+commands["validate-acct"] = cmd_validate_acct
+
+def cmd_acct_info(args):
+ "usage: acmecert acct-info ACCOUNT-FILE"
+ if len(args) < 2: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ pprint.pprint(account.read(fp).getinfo())
+commands["acct-info"] = cmd_validate_acct
+
+def cmd_order(args):
+ "usage: acmecert order ACCOUNT-FILE CSR [OUTPUT-FILE]"
+ if len(args) < 4: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ acct = account.read(fp)
+ with maybeopen(args[2], "r") as fp:
+ csr = signreq.read(fp)
+ order = mkorder(acct, csr)
+ with maybeopen(args[3] if len(args) > 3 else "-", "w") as fp:
+ fp.write("%s\n" % (order["acmecert.location"]))
+commands["order"] = cmd_order
+
+def cmd_http_auth(args):
+ "usage: acmecert http-auth ACCOUNT-FILE HTTP-CONFIG {ORDER-ID|ORDER-FILE}"
+ if len(args) < 4: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ acct = account.read(fp)
+ with maybeopen(args[2], "r") as fp:
+ htconf = htconfig.read(fp)
+ if "://" in args[3]:
+ orderid = args[3]
+ else:
+ with maybeopen(args[3], "r") as fp:
+ orderid = fp.readline().strip()
+ authorder(acct, htconf, orderid)
+commands["http-auth"] = cmd_http_auth
+
+def cmd_get(args):
+ "usage: acmecert get ACCOUNT-FILE CSR {ORDER-ID|ORDER-FILE}"
+ if len(args) < 4: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ acct = account.read(fp)
+ with maybeopen(args[2], "r") as fp:
+ csr = signreq.read(fp)
+ if "://" in args[3]:
+ orderid = args[3]
+ else:
+ with maybeopen(args[3], "r") as fp:
+ orderid = fp.readline().strip()
+ sys.stdout.write(finalize(acct, csr, orderid))
+commands["get"] = cmd_get
+
+def cmd_http_order(args):
+ "usage: acmecert http-order ACCOUNT-FILE CSR HTTP-CONFIG [OUTPUT-FILE]"
+ if len(args) < 4: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ acct = account.read(fp)
+ with maybeopen(args[2], "r") as fp:
+ csr = signreq.read(fp)
+ with maybeopen(args[3], "r") as fp:
+ htconf = htconfig.read(fp)
+ orderid = mkorder(acct, csr)["acmecert.location"]
+ authorder(acct, htconf, orderid)
+ with maybeopen(args[4] if len(args) > 4 else "-", "w") as fp:
+ fp.write(finalize(acct, csr, orderid))
+commands["http-order"] = cmd_http_order
+
+def cmd_check_cert(args):
+ "usage: acmecert check-cert CERT-FILE TIME-SPEC"
+ if len(args) < 3: raise usageerr()
+ with maybeopen(args[1], "r") as fp:
+ crt = certificate.read(fp)
+ sys.exit(1 if crt.expiring(args[2]) else 0)
+commands["check-cert"] = cmd_check_cert
+
+def cmd_directory(args):
+ "usage: acmecert directory"
+ pprint.pprint(directory())
+commands["directory"] = cmd_directory
+
def usage(out):
- out.write("usage: acmecert [-h] [-D SERVICE]\n")
+ out.write("usage: acmecert [-D SERVICE] COMMAND [ARGS...]\n")
+ out.write(" acmecert -h [COMMAND]\n")
+ buf = " COMMAND is any of: "
+ f = True
+ for cmd in commands:
+ if len(buf) + len(cmd) > 70:
+ out.write("%s\n" % (buf,))
+ buf = " "
+ f = True
+ if not f:
+ buf += ", "
+ buf += cmd
+ f = False
+ if not f:
+ out.write("%s\n" % (buf,))
def main(argv):
global service
opts, args = getopt.getopt(argv[1:], "hD:")
for o, a in opts:
if o == "-h":
- usage(sys.stdout)
+ if len(args) > 0:
+ cmd = commands.get(args[0])
+ if cmd is None:
+ sys.stderr.write("acmecert: unknown command: %s\n" % (args[0],))
+ sys.exit(1)
+ sys.stdout.write("%s\n" % (cmd.__doc__,))
+ else:
+ usage(sys.stdout)
sys.exit(0)
elif o == "-D":
service = a
if len(args) < 1:
usage(sys.stderr)
sys.exit(1)
- if args[0] == "reg":
- register().write(sys.stdout)
- elif args[0] == "validate-acct":
- with open(args[1], "r") as fp:
- account.read(fp).validate()
- elif args[0] == "acctinfo":
- with open(args[1], "r") as fp:
- pprint.pprint(account.read(fp).getinfo())
- elif args[0] == "order":
- with open(args[1], "r") as fp:
- acct = account.read(fp)
- with open(args[2], "r") as fp:
- csr = signreq.read(fp)
- order = mkorder(acct, csr)
- with open(args[3], "w") as fp:
- fp.write("%s\n" % (order["acmecert.location"]))
- elif args[0] == "http-auth":
- with open(args[1], "r") as fp:
- acct = account.read(fp)
- with open(args[2], "r") as fp:
- htconf = htconfig.read(fp)
- with open(args[3], "r") as fp:
- orderid = fp.readline().strip()
- authorder(acct, htconf, orderid)
- elif args[0] == "get":
- with open(args[1], "r") as fp:
- acct = account.read(fp)
- with open(args[2], "r") as fp:
- csr = signreq.read(fp)
- with open(args[3], "r") as fp:
- orderid = fp.readline().strip()
- sys.stdout.write(finalize(acct, csr, orderid))
- elif args[0] == "http-order":
- with open(args[1], "r") as fp:
- acct = account.read(fp)
- with open(args[2], "r") as fp:
- csr = signreq.read(fp)
- with open(args[3], "r") as fp:
- htconf = htconfig.read(fp)
- orderid = mkorder(acct, csr)["acmecert.location"]
- authorder(acct, htconf, orderid)
- sys.stdout.write(finalize(acct, csr, orderid))
- elif args[0] == "directory":
- pprint.pprint(directory())
- else:
+ cmd = commands.get(args[0])
+ if cmd is None:
sys.stderr.write("acmecert: unknown command: %s\n" % (args[0],))
usage(sys.stderr)
sys.exit(1)
+ try:
+ cmd(args)
+ except usageerr:
+ sys.stderr.write("%s\n" % (cmd.__doc__,))
+ sys.exit(1)
if __name__ == "__main__":
try:
Dolda2000 GitWeb / utils.git / blobdiff