From 58fcbdba99212e5c58aed687dfe56655f9713c78 Mon Sep 17 00:00:00 2001 From: rsiddharth Date: Sun, 12 Apr 2020 13:29:34 -0400 Subject: acmens.py: Add revoke_crt. Move certicate revocation function -- revoke_crt -- into acmens.py. The revoke_crt.py will be removed. --- acmens.py | 128 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 123 insertions(+), 5 deletions(-) (limited to 'acmens.py') diff --git a/acmens.py b/acmens.py index e371432..7078023 100644 --- a/acmens.py +++ b/acmens.py @@ -219,6 +219,109 @@ Notes: return signed_pem + +def revoke_crt(account_key, crt): + """Use the ACME protocol to revoke an ssl certificate signed by a + certificate authority. + + :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-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" + if type(b) is str: + b = b.encode() + + return base64.urlsafe_b64encode(b).decode().replace("=", "") + + def _a64(a): + "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") + 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) + pub_exp = "{0:x}".format(pub_exp) + 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) + jwk = { + "e": pub_exp64, + "kty": "RSA", + "n": pub_mod64, + } + sys.stderr.write("Found public key!\n") + + # 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) + rvk_payload = { + "certificate": crt_der64, + } + _send_signed_request(DIRECTORY['revokeCert'], rvk_payload, "Error revoking certificate") + sys.stderr.write("Certificate revoked!\n") + + if __name__ == "__main__": parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, @@ -229,6 +332,9 @@ this script on your server, it is meant to be run on your computer. The script will request you to manually deploy the acme challenge on your server. +You may also revoke a signed Let's Encrypt (ACME) certificate. + + NOTE: YOUR ACCOUNT KEY NEEDS TO BE DIFFERENT FROM YOUR DOMAIN KEY. Prerequisites: @@ -241,19 +347,31 @@ $ openssl genrsa -aes256 4096 > user.key $ openssl rsa -in user.key -pubout > user.pub $ openssl genrsa -aes256 4096 > domain.key $ openssl req -new -sha256 -key domain.key -subj "/CN=example.com" > domain.csr -$ python3 acmens.py --account-key user.key --email user@example.com domain.csr > signed.crt +$ python3 acmens.py --account-key user.key --email user@example.com --csr domain.csr > signed.crt -------------- -""") +Example: Revoking a signed certificate: +-------------- +$ python3 acmens.py --revoke --account-key user.key --crt domain.crt +-------------- +"""); + parser.add_argument("--revoke", action="store_true", help="Revoke a signed certificate") parser.add_argument("-k", "--account-key", required=True, help="path to your Let's Encrypt account private key") parser.add_argument("-e", "--email", default=None, help="contact email, default is webmaster@") parser.add_argument("--csr", help="path to your certificate signing request") + parser.add_argument("--crt", help="path to your signed certificate") args = parser.parse_args() - if args.csr is None: + if (not args.revoke) and (args.csr is None): sys.stderr.write('Error: Path to CSR required\n') sys.exit(1) + if args.revoke and args.crt is None: + sys.stderr.write('Error: Path to signed cert required\n') + sys.exit(1) - signed_crt = sign_csr(args.account_key, args.csr, email=args.email) - sys.stdout.write(signed_crt) + if args.revoke: + revoke_crt(args.account_key, args.crt) + else: + signed_crt = sign_csr(args.account_key, args.csr, email=args.email) + sys.stdout.write(signed_crt) -- cgit v1.2.3