diff options
| -rw-r--r-- | revoke_crt.py | 157 |
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) |