From d95db24bd7ee1da6524c9260700486888919e1a4 Mon Sep 17 00:00:00 2001 From: Ruben Vorderman Date: Sat, 6 Nov 2021 11:22:29 +0100 Subject: [PATCH] Remove redundant comments --- acme_tiny.py | 53 +++++++++++++++++++++------------------------------- 1 file changed, 21 insertions(+), 32 deletions(-) diff --git a/acme_tiny.py b/acme_tiny.py index d992d02d..ea92640f 100755 --- a/acme_tiny.py +++ b/acme_tiny.py @@ -16,19 +16,16 @@ def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA, disable_check=False, directory_url=DEFAULT_DIRECTORY_URL, contact=None, check_port=None): directory, acct_headers, alg, jwk = None, None, None, None # global variables - # helper functions - base64 encode for jose spec - def _b64(b): + def _b64_encode_jose(b): return base64.urlsafe_b64encode(b).decode('utf8').replace("=", "") - # helper function - run external commands - def _cmd(cmd_list, stdin=None, cmd_input=None, err_msg="Command Line Error"): + def _run_external_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 = urlopen(Request(url, data=data, headers={"Content-Type": "application/jose+json", "User-Agent": "acme-tiny"})) @@ -46,23 +43,21 @@ def _do_request(url, data=None, err_msg="Error", depth=0): 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')) + payload64 = "" if payload is None else _b64_encode_jose(json.dumps(payload).encode('utf8')) new_nonce = _do_request(directory['newNonce'])[2]['Replay-Nonce'] protected = {"url": url, "alg": alg, "nonce": new_nonce} protected.update({"jwk": jwk} if acct_headers is None else {"kid": acct_headers['Location']}) - protected64 = _b64(json.dumps(protected).encode('utf8')) + protected64 = _b64_encode_jose(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)}) + out = _run_external_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_encode_jose(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)) - # helper function - poll until complete - def _poll_until_not(url, pending_statuses, err_msg): + def _poll_until_complete(url, pending_statuses, err_msg): result, t0 = None, time.time() while result is None or result['status'] in pending_statuses: assert (time.time() - t0 < 3600), "Polling timeout" # 1 hour timeout @@ -70,24 +65,22 @@ def _poll_until_not(url, pending_statuses, err_msg): result, _, _ = _send_signed_request(url, None, err_msg) return result - # parse account key to get public key - log.info("Parsing account key...") - out = _cmd(["openssl", "rsa", "-in", account_key, "-noout", "-text"], err_msg="OpenSSL Error") + log.info("Parsing account key to get public key...") + out = _run_external_cmd(["openssl", "rsa", "-in", account_key, "-noout", "-text"], err_msg="OpenSSL Error") pub_pattern = r"modulus:[\s]+?00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)" pub_hex, pub_exp = re.search(pub_pattern, out.decode('utf8'), re.MULTILINE|re.DOTALL).groups() pub_exp = "{0:x}".format(int(pub_exp)) pub_exp = "0{0}".format(pub_exp) if len(pub_exp) % 2 else pub_exp alg, jwk = "RS256", { - "e": _b64(binascii.unhexlify(pub_exp.encode("utf-8"))), + "e": _b64_encode_jose(binascii.unhexlify(pub_exp.encode("utf-8"))), "kty": "RSA", - "n": _b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))), + "n": _b64_encode_jose(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))), } accountkey_json = json.dumps(jwk, sort_keys=True, separators=(',', ':')) - thumbprint = _b64(hashlib.sha256(accountkey_json.encode('utf8')).digest()) + thumbprint = _b64_encode_jose(hashlib.sha256(accountkey_json.encode('utf8')).digest()) - # find domains - log.info("Parsing CSR...") - out = _cmd(["openssl", "req", "-in", csr, "-noout", "-text"], err_msg="Error loading {0}".format(csr)) + log.info("Parsing CSR to find domains...") + out = _run_external_cmd(["openssl", "req", "-in", csr, "-noout", "-text"], err_msg="Error loading {0}".format(csr)) domains = set([]) common_name = re.search(r"Subject:.*? CN\s?=\s?([^\s,;/]+)", out.decode('utf8')) if common_name is not None: @@ -99,14 +92,12 @@ def _poll_until_not(url, pending_statuses, err_msg): domains.add(san[4:]) log.info(u"Found domains: {0}".format(", ".join(domains))) - # get the ACME directory of urls - log.info("Getting directory...") + log.info("Getting ACME directory of urls...") directory_url = CA + "/directory" if CA != DEFAULT_CA else directory_url # backwards compatibility with deprecated CA kwarg directory, _, _ = _do_request(directory_url, err_msg="Error getting directory") log.info("Directory found!") - # create account, update contact details (if any), and set the global key identifier - log.info("Registering account...") + log.info("Registering account, updating contact details and setting the global key identifier...") reg_payload = {"termsOfServiceAgreed": True} if contact is None else {"termsOfServiceAgreed": True, "contact": contact} account, code, acct_headers = _send_signed_request(directory['newAccount'], reg_payload, "Error registering") log.info("{0} Account ID: {1}".format("Registered!" if code == 201 else "Already registered!", acct_headers['Location'])) @@ -114,7 +105,6 @@ def _poll_until_not(url, pending_statuses, err_msg): account, _, _ = _send_signed_request(acct_headers['Location'], {"contact": contact}, "Error updating contact details") log.info("Updated contact details:\n{0}".format("\n".join(account['contact']))) - # create a new order log.info("Creating new order...") order_payload = {"identifiers": [{"type": "dns", "value": d} for d in domains]} order, _, order_headers = _send_signed_request(directory['newOrder'], order_payload, "Error creating new order") @@ -125,8 +115,7 @@ def _poll_until_not(url, pending_statuses, err_msg): authorization, _, _ = _send_signed_request(auth_url, None, "Error getting challenges") domain = authorization['identifier']['value'] - # skip if already valid - if authorization['status'] == "valid": + if authorization['status'] == "valid": # skip if already valid log.info("Already verified: {0}, skipping...".format(domain)) continue log.info("Verifying {0}...".format(domain)) @@ -148,7 +137,7 @@ def _poll_until_not(url, pending_statuses, err_msg): # say the challenge is done _send_signed_request(challenge['url'], {}, "Error submitting challenges: {0}".format(domain)) - authorization = _poll_until_not(auth_url, ["pending"], "Error checking challenge status for {0}".format(domain)) + authorization = _poll_until_complete(auth_url, ["pending"], "Error checking challenge status for {0}".format(domain)) if authorization['status'] != "valid": raise ValueError("Challenge did not pass for {0}: {1}".format(domain, authorization)) os.remove(wellknown_path) @@ -156,11 +145,11 @@ def _poll_until_not(url, pending_statuses, err_msg): # finalize the order with the csr log.info("Signing certificate...") - csr_der = _cmd(["openssl", "req", "-in", csr, "-outform", "DER"], err_msg="DER Export Error") - _send_signed_request(order['finalize'], {"csr": _b64(csr_der)}, "Error finalizing order") + csr_der = _run_external_cmd(["openssl", "req", "-in", csr, "-outform", "DER"], err_msg="DER Export Error") + _send_signed_request(order['finalize'], {"csr": _b64_encode_jose(csr_der)}, "Error finalizing order") # poll the order to monitor when it's done - order = _poll_until_not(order_headers['Location'], ["pending", "processing"], "Error checking order status") + order = _poll_until_complete(order_headers['Location'], ["pending", "processing"], "Error checking order status") if order['status'] != "valid": raise ValueError("Order failed: {0}".format(order))