summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorrsiddharth <s@ricketyspace.net>2020-06-17 21:06:15 -0400
committerrsiddharth <s@ricketyspace.net>2020-06-17 21:06:15 -0400
commit9eb4aa1223a36564f9129984671620fbe0072fb1 (patch)
tree43c2e49597a261bc90a9d99d0b8634e9fb6d9c7a
parente88223916180ebed436b640440954076a380a578 (diff)
acmens.py: Format it via black.
-rw-r--r--acmens.py307
1 files changed, 214 insertions, 93 deletions
diff --git a/acmens.py b/acmens.py
index f82e2f0..0cb037b 100644
--- a/acmens.py
+++ b/acmens.py
@@ -6,14 +6,13 @@
# Copyright © 2020 rsiddharth <s@ricketyspace.net>
#
-import argparse, subprocess, json, os, urllib.request, sys, base64, binascii, \
- time, hashlib, tempfile, re, copy, textwrap
+import argparse, subprocess, json, os, urllib.request, sys, base64, binascii, time, hashlib, tempfile, re, copy, textwrap
from urllib.request import urlopen
from urllib.error import HTTPError
-__version__ = '0.1.2'
+__version__ = "0.1.3.dev1"
def sign_csr(account_key, csr, email=None):
@@ -29,9 +28,9 @@ def sign_csr(account_key, csr, email=None):
:rtype: string
"""
- #CA = "https://acme-staging-v02.api.letsencrypt.org"
+ # CA = "https://acme-staging-v02.api.letsencrypt.org"
CA = "https://acme-v02.api.letsencrypt.org"
- DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode('utf8'))
+ DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode("utf8"))
def _b64(b):
"Shortcut function to go from bytes to jwt base64 string"
@@ -42,7 +41,9 @@ def sign_csr(account_key, csr, email=None):
# 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)
+ 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))
@@ -51,52 +52,88 @@ def sign_csr(account_key, csr, email=None):
# 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
+ 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
+ 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
+ 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))
+ 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']
+ 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)})
+ 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 _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):
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
+ 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
-
# 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")
+ 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()
+ 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)
@@ -109,15 +146,18 @@ def sign_csr(account_key, csr, email=None):
"kty": "RSA",
"n": pub_mod64,
}
- accountkey_json = json.dumps(jwk, sort_keys=True, separators=(',', ':'))
+ accountkey_json = json.dumps(jwk, sort_keys=True, separators=(",", ":"))
thumbprint = _b64(hashlib.sha256(accountkey_json.encode()).digest())
sys.stderr.write("Found public key!\n")
# Step 2: Get the domain names to be certified
sys.stderr.write("Reading csr file...\n")
- out = _cmd(["openssl", "req", "-in", csr, "-noout", "-text"], err_msg="Error loading {}".format(csr))
+ out = _cmd(
+ ["openssl", "req", "-in", csr, "-noout", "-text"],
+ err_msg="Error loading {}".format(csr),
+ )
domain = None
- common_name = re.search("Subject:.*? CN *= *([^\s,;/]+)", out.decode('utf8'))
+ common_name = re.search("Subject:.*? CN *= *([^\s,;/]+)", out.decode("utf8"))
if common_name is not None:
domain = common_name.group(1)
sys.stderr.write("Found domain {0}\n".format(domain))
@@ -127,54 +167,56 @@ def sign_csr(account_key, csr, email=None):
default_email = "webmaster@{0}".format(domain)
stdout = sys.stdout
sys.stdout = sys.stderr
- input_email = input("STEP 1: What is your contact email? ({0}) ".format(default_email))
+ input_email = input(
+ "STEP 1: What is your contact email? ({0}) ".format(default_email)
+ )
email = input_email if input_email else default_email
sys.stdout = stdout
-
# Step 4: Generate the payload for registering user and initiate registration.
sys.stderr.write("Registering {0}...\n".format(email))
- reg = {
- "termsOfServiceAgreed": True
- }
+ reg = {"termsOfServiceAgreed": True}
acct_headers = None
- result, code, acct_headers = _send_signed_request(DIRECTORY['newAccount'], reg, "Error registering")
+ result, code, acct_headers = _send_signed_request(
+ DIRECTORY["newAccount"], reg, "Error registering"
+ )
if code == 201:
sys.stderr.write("Registered!\n")
else:
sys.stderr.write("Already registered!\n")
-
# Step 5: Request challenges for domain
sys.stderr.write("Making new order for {0}...\n".format(domain))
id = {
- "identifiers": [{
- "type": "dns",
- "value": domain,
- }],
+ "identifiers": [{"type": "dns", "value": domain,}],
}
- order, order_code, order_headers = _send_signed_request(DIRECTORY['newOrder'], id, "Error creating new order")
+ order, order_code, order_headers = _send_signed_request(
+ DIRECTORY["newOrder"], id, "Error creating new order"
+ )
# Request challenges
sys.stderr.write("Requesting challenges...\n")
- chl_result, chl_code, chl_headers = _send_signed_request(order['authorizations'][0], None, "Error getting challenges")
+ chl_result, chl_code, chl_headers = _send_signed_request(
+ order["authorizations"][0], None, "Error getting challenges"
+ )
- challenge = [c for c in chl_result['challenges'] if c['type'] == "http-01"][0]
- token = re.sub(r"[^A-Za-z0-9_\-]", "_", challenge['token'])
- keyauthorization = "{0}.{1}".format(challenge['token'], thumbprint)
+ challenge = [c for c in chl_result["challenges"] if c["type"] == "http-01"][0]
+ token = re.sub(r"[^A-Za-z0-9_\-]", "_", challenge["token"])
+ keyauthorization = "{0}.{1}".format(challenge["token"], thumbprint)
# build request for the server to test this challenge.
- test_url = challenge['url']
+ test_url = challenge["url"]
test_raw = "{}"
# challenge response for server
response = {
- "uri": ".well-known/acme-challenge/{0}".format(challenge['token']),
+ "uri": ".well-known/acme-challenge/{0}".format(challenge["token"]),
"data": keyauthorization,
}
# Step 6: Ask the user to host the token on their server
- sys.stderr.write("""\
+ sys.stderr.write(
+ """\
Please update your server to serve the following file at this URL:
--------------
@@ -186,7 +228,10 @@ Notes:
- Do not include the quotes in the file.
- The file should be one line without any spaces.
-""".format(domain, response['uri'], response['data']))
+""".format(
+ domain, response["uri"], response["data"]
+ )
+ )
stdout = sys.stdout
sys.stdout = sys.stderr
@@ -195,31 +240,51 @@ Notes:
# Step 7: Let the CA know you're ready for the challenge
sys.stderr.write("Requesting verification for {0}...\n".format(domain))
- _send_signed_request(challenge['url'], {}, "Error requesting challenge verfication: {0}".format(domain))
- chl_verification = _poll_until_not(challenge['url'], ["pending"], "Error checking challenge verification")
- if chl_verification['status'] != "valid":
- raise ValueError("Challenge did not pass for {0}: {1}".format(domain, chl_verification))
+ _send_signed_request(
+ challenge["url"],
+ {},
+ "Error requesting challenge verfication: {0}".format(domain),
+ )
+ chl_verification = _poll_until_not(
+ challenge["url"], ["pending"], "Error checking challenge verification"
+ )
+ if chl_verification["status"] != "valid":
+ raise ValueError(
+ "Challenge did not pass for {0}: {1}".format(domain, chl_verification)
+ )
sys.stderr.write("{} verified!\n".format(domain))
# Step 8: Finalize
- csr_der = _cmd(["openssl", "req", "-in", csr, "-outform", "DER"], err_msg="DER Export Error")
- fnlz_resp, fnlz_code, fnlz_headers = _send_signed_request(order['finalize'], {"csr": _b64(csr_der)}, "Error finalizing order")
+ csr_der = _cmd(
+ ["openssl", "req", "-in", csr, "-outform", "DER"], err_msg="DER Export Error"
+ )
+ fnlz_resp, fnlz_code, fnlz_headers = _send_signed_request(
+ order["finalize"], {"csr": _b64(csr_der)}, "Error finalizing order"
+ )
# Step 9: Wait for CA to mark test as valid
sys.stderr.write("Waiting for {0} challenge to pass...\n".format(domain))
- order = _poll_until_not(order_headers['Location'], ["pending", "processing"], "Error checking order status")
+ order = _poll_until_not(
+ order_headers["Location"],
+ ["pending", "processing"],
+ "Error checking order status",
+ )
- if order['status'] == "valid":
+ if order["status"] == "valid":
sys.stderr.write("Passed {0} challenge!\n".format(domain))
else:
raise ValueError("'{0}' challenge did not pass: {1}".format(domain, order))
# Step 10: Get the certificate.
sys.stderr.write("Getting certificate...\n")
- signed_pem, _, _ = _send_signed_request(order['certificate'], None, "Error getting certificate")
+ signed_pem, _, _ = _send_signed_request(
+ order["certificate"], None, "Error getting certificate"
+ )
sys.stderr.write("Received certificate!\n")
- sys.stderr.write("You can remove the acme-challenge file from your webserver now.\n")
+ sys.stderr.write(
+ "You can remove the acme-challenge file from your webserver now.\n"
+ )
return signed_pem
@@ -231,9 +296,9 @@ def revoke_crt(account_key, crt):
: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-staging-v02.api.letsencrypt.org"
CA = "https://acme-v02.api.letsencrypt.org"
- DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode('utf8'))
+ DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode("utf8"))
def _b64(b):
"Shortcut function to go from bytes to jwt base64 string"
@@ -248,7 +313,9 @@ def revoke_crt(account_key, crt):
# 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)
+ 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))
@@ -257,43 +324,80 @@ def revoke_crt(account_key, crt):
# 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
+ 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
+ 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
+ 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))
+ 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']
+ 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)})
+ 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 _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")
+ 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()
+ 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)
@@ -310,19 +414,23 @@ def revoke_crt(account_key, crt):
# Step 2: Get account info.
sys.stderr.write("Getting account info...\n")
- reg = {
- "onlyReturnExistiing": True
- }
+ reg = {"onlyReturnExistiing": True}
acct_headers = None
- result, code, acct_headers = _send_signed_request(DIRECTORY['newAccount'], reg, "Error getting account info")
+ 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_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")
+ _send_signed_request(
+ DIRECTORY["revokeCert"], rvk_payload, "Error revoking certificate"
+ )
sys.stderr.write("Certificate revoked!\n")
@@ -358,19 +466,32 @@ Example: Revoking a signed certificate:
--------------
$ acmens --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@<shortest_domain>")
+""",
+ )
+ 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@<shortest_domain>",
+ )
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 (not args.revoke) and (args.csr is None):
- sys.stderr.write('Error: Path to CSR required\n')
+ 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.stderr.write("Error: Path to signed cert required\n")
sys.exit(1)
if args.revoke: