revoke_crt.py: Refactor script to use acmev2.

1 files changed, 76 insertions, 81 deletions

diff --git a/revoke_crt.py b/revoke_crt.py
index 22d2660..49de151 100644
--- a/revoke_crt.py
+++ b/revoke_crt.py
@@ -5,17 +5,16 @@ import argparse, subprocess, json, os, urllib.request, sys, base64, binascii, co
 from urllib.request import urlopen
 from urllib.error import HTTPError
 
-def revoke_crt(pubkey, crt):
+def revoke_crt(account_key, crt):
     """Use the ACME protocol to revoke an ssl certificate signed by a
     certificate authority.
 
-    :param string pubkey: Path to the user account public key.
+    :param string account_key: Path to your Let's Encrypt account private key.
     :param string crt: Path to the signed certificate.
     """
-    #CA = "https://acme-staging.api.letsencrypt.org"
-    CA = "https://acme-v01.api.letsencrypt.org"
-    nonce_req = urllib.request.Request("{0}/directory".format(CA))
-    nonce_req.get_method = lambda : 'HEAD'
+    #CA = "https://acme-staging-v02.api.letsencrypt.org"
+    CA = "https://acme-v02.api.letsencrypt.org"
+    DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode('utf8'))
 
     def _b64(b):
         "Shortcut function to go from bytes to jwt base64 string"
@@ -28,14 +27,54 @@ def revoke_crt(pubkey, crt):
         "Shortcut function to go from jwt base64 string to bytes"
         return base64.urlsafe_b64decode(str(a + ("=" * (len(a) % 4))))
 
+    # helper function - run external commands
+    def _cmd(cmd_list, stdin=None, cmd_input=None, err_msg="Command Line Error"):
+        proc = subprocess.Popen(cmd_list, stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        out, err = proc.communicate(cmd_input)
+        if proc.returncode != 0:
+            raise IOError("{0}\n{1}".format(err_msg, err))
+        return out
+
+    # helper function - make request and automatically parse json response
+    def _do_request(url, data=None, err_msg="Error", depth=0):
+        try:
+            resp = urllib.request.urlopen(urllib.request.Request(url, data=data, headers={"Content-Type": "application/jose+json", "User-Agent": "acme-nosudo"}))
+            resp_data, code, headers = resp.read().decode("utf8"), resp.getcode(), resp.headers
+        except IOError as e:
+            resp_data = e.read().decode("utf8") if hasattr(e, "read") else str(e)
+            code, headers = getattr(e, "code", None), {}
+        try:
+            resp_data = json.loads(resp_data) # try to parse json results
+        except ValueError:
+            pass # ignore json parsing errors
+        if depth < 100 and code == 400 and resp_data['type'] == "urn:ietf:params:acme:error:badNonce":
+            raise IndexError(resp_data) # allow 100 retrys for bad nonces
+        if code not in [200, 201, 204]:
+            raise ValueError("{0}:\nUrl: {1}\nData: {2}\nResponse Code: {3}\nResponse: {4}".format(err_msg, url, data, code, resp_data))
+        return resp_data, code, headers
+
+    # helper function - make signed requests
+    def _send_signed_request(url, payload, err_msg, depth=0):
+        payload64 = "" if payload is None else _b64(json.dumps(payload).encode('utf8'))
+        new_nonce = _do_request(DIRECTORY['newNonce'])[2]['Replay-Nonce']
+        protected = {"url": url, "alg": "RS256", "nonce": new_nonce}
+        protected.update({"jwk": jwk} if acct_headers is None else {"kid": acct_headers['Location']})
+        protected64 = _b64(json.dumps(protected).encode('utf8'))
+        protected_input = "{0}.{1}".format(protected64, payload64).encode('utf8')
+        out = _cmd(["openssl", "dgst", "-sha256", "-sign", account_key], stdin=subprocess.PIPE, cmd_input=protected_input, err_msg="OpenSSL Error")
+        data = json.dumps({"protected": protected64, "payload": payload64, "signature": _b64(out)})
+        try:
+            return _do_request(url, data=data.encode('utf8'), err_msg=err_msg, depth=depth)
+        except IndexError: # retry bad nonces (they raise IndexError)
+            return _send_signed_request(url, payload, err_msg, depth=(depth + 1))
+
     # Step 1: Get account public key
     sys.stderr.write("Reading pubkey file...\n")
-    proc = subprocess.Popen(["openssl", "rsa", "-pubin", "-in", pubkey, "-noout", "-text"],
-        stdout=subprocess.PIPE, universal_newlines=True)
-    out, err = proc.communicate()
-    if proc.returncode != 0:
-        raise IOError("Error loading {0}".format(pubkey))
-    pub_hex, pub_exp = re.search("Modulus\:\s+00:([a-f0-9\:\s]+?)Exponent\: ([0-9]+)", out, re.MULTILINE|re.DOTALL).groups()
+    out = _cmd(["openssl", "rsa", "-in", account_key, "-noout", "-text"], err_msg="Error reading account public key")
+
+    pub_hex, pub_exp = re.search(
+        r"modulus:[\s]+?00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
+        out.decode('utf8'), re.MULTILINE|re.DOTALL).groups()
     pub_mod = binascii.unhexlify(re.sub("(\s|:)", "", pub_hex))
     pub_mod64 = _b64(pub_mod)
     pub_exp = int(pub_exp)
@@ -43,70 +82,28 @@ def revoke_crt(pubkey, crt):
     pub_exp = "0{0}".format(pub_exp) if len(pub_exp) % 2 else pub_exp
     pub_exp = binascii.unhexlify(pub_exp)
     pub_exp64 = _b64(pub_exp)
-    header = {
-        "alg": "RS256",
-        "jwk": {
-            "e": pub_exp64,
-            "kty": "RSA",
-            "n": pub_mod64,
-        },
+    jwk = {
+        "e": pub_exp64,
+        "kty": "RSA",
+        "n": pub_mod64,
     }
-    sys.stderr.write("Found public key!\n".format(header))
+    sys.stderr.write("Found public key!\n")
 
-    # Step 2: Generate the payload that needs to be signed
-    # revokation request
-    proc = subprocess.Popen(["openssl", "x509", "-in", crt, "-outform", "DER"],
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    crt_der, err = proc.communicate()
+    # Step 2: Get account info.
+    sys.stderr.write("Getting account info...\n")
+    reg = {
+        "onlyReturnExistiing": True
+    }
+    acct_headers = None
+    result, code, acct_headers = _send_signed_request(DIRECTORY['newAccount'], reg, "Error getting account info")
+
+    # Step 3: Generate the payload.
+    crt_der = _cmd(["openssl", "x509", "-in", crt, "-outform", "DER"], err_msg="DER export error")
     crt_der64 = _b64(crt_der)
-    crt_raw = json.dumps({
-        "resource": "revoke-cert",
+    rvk_payload = {
         "certificate": crt_der64,
-    }, sort_keys=True, indent=4)
-    crt_b64 = _b64(crt_raw)
-    crt_protected = copy.deepcopy(header)
-    crt_protected.update({"nonce": urlopen(nonce_req).headers['Replay-Nonce']})
-    crt_protected64 = _b64(json.dumps(crt_protected, sort_keys=True, indent=4))
-    crt_file = tempfile.NamedTemporaryFile(dir=".", prefix="revoke_", suffix=".json")
-    crt_file.write("{0}.{1}".format(crt_protected64, crt_b64).encode())
-    crt_file.flush()
-    crt_file_name = os.path.basename(crt_file.name)
-    crt_file_sig = tempfile.NamedTemporaryFile(dir=".", prefix="revoke_", suffix=".sig")
-    crt_file_sig_name = os.path.basename(crt_file_sig.name)
-
-    # Step 3: Ask the user to sign the revocation request
-    sys.stderr.write("""\
-STEP 1: You need to sign a file (replace 'user.key' with your user private key)
-
-openssl dgst -sha256 -sign user.key -out {0} {1}
-
-""".format(crt_file_sig_name, crt_file_name))
-
-    temp_stdout = sys.stdout
-    sys.stdout = sys.stderr
-    input("Press Enter when you've run the above command in a new terminal window...")
-    sys.stdout = temp_stdout
-
-    # Step 4: Load the signature and send the revokation request
-    sys.stderr.write("Requesting revocation...\n")
-    crt_file_sig.seek(0)
-    crt_sig64 = _b64(crt_file_sig.read())
-    crt_data = json.dumps({
-        "header": header,
-        "protected": crt_protected64,
-        "payload": crt_b64,
-        "signature": crt_sig64,
-    }, sort_keys=True, indent=4)
-    try:
-        resp = urlopen("{0}/acme/revoke-cert".format(CA), crt_data.encode())
-        signed_der = resp.read()
-    except HTTPError as e:
-        sys.stderr.write("Error: crt_data:\n")
-        sys.stderr.write(crt_data)
-        sys.stderr.write("\n")
-        sys.stderr.write(e.read().decode())
-        sys.stderr.write("\n")
-        raise
+    }
+    _send_signed_request(DIRECTORY['revokeCert'], rvk_payload, "Error revoking certificate")
     sys.stderr.write("Certificate revoked!\n")
 
 
@@ -115,27 +112,25 @@ if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         formatter_class=argparse.RawDescriptionHelpFormatter,
         description="""\
-Get a SSL certificate revoked by a Let's Encrypt (ACME) certificate authority.
-You do NOT need to run this script on your server and this script does not ask
-for your private keys. It will print out commands that you need to run with
-your private key, which gives you a chance to review the commands instead of
-trusting this script.
+Get a SSL certificate revoked by a Let's Encrypt (ACME) certificate
+authority.  You do NOT need to run this script on your server, it is
+meant to be run on your computer.
 
 NOTE: YOUR PUBLIC KEY NEEDS TO BE THE SAME KEY USED TO ISSUE THE CERTIFICATE.
 
 Prerequisites:
 * openssl
-* python
+* python 3
 
 Example:
 --------------
-$ python revoke_crt.py --public-key user.pub domain.crt
+$ python3 revoke_crt.py --public-key user.pub domain.crt
 --------------
 
 """)
-    parser.add_argument("-p", "--public-key", required=True, help="path to your account public key")
+    parser.add_argument("-k", "--account-key", required=True, help="path to your Let's Encrypt account private key")
     parser.add_argument("crt_path", help="path to your signed certificate")
 
     args = parser.parse_args()
-    revoke_crt(args.public_key, args.crt_path)
+    revoke_crt(args.account_key, args.crt_path)