Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Remove redundant comments #273

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Remove redundant comments #273

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 21 additions & 32 deletions acme_tiny.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}))
Expand All @@ -46,48 +43,44 @@ 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
time.sleep(0 if result is None else 2)
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:
Expand All @@ -99,22 +92,19 @@ 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']))
if contact is not None:
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")
Expand All @@ -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))
Expand All @@ -148,19 +137,19 @@ 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)
log.info("{0} verified!".format(domain))

# 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))

Expand Down