mirror of
https://github.com/YunoHost/yunohost.git
synced 2024-09-03 20:06:10 +02:00
Update acme_tiny.py to 5350420
This commit is contained in:
Alexandre Aubin
parent
158aa08022
commit
984be7987b
1 changed files with 131 additions and 132 deletions
263
src/yunohost/vendor/acme_tiny/acme_tiny.py
vendored
263
src/yunohost/vendor/acme_tiny/acme_tiny.py
vendored
|
|
@ -1,73 +1,94 @@
|
|||
#!/usr/bin/env python
|
||||
# Copyright Daniel Roesler, under MIT license, see LICENSE at github.com/diafygi/acme-tiny
|
||||
import argparse, subprocess, json, os, sys, base64, binascii, time, hashlib, re, copy, textwrap, logging
|
||||
try:
|
||||
from urllib.request import urlopen # Python 3
|
||||
from urllib.request import urlopen, Request # Python 3
|
||||
except ImportError:
|
||||
from urllib2 import urlopen # Python 2
|
||||
from urllib2 import urlopen, Request # Python 2
|
||||
|
||||
#DEFAULT_CA = "https://acme-staging.api.letsencrypt.org"
|
||||
DEFAULT_CA = "https://acme-v01.api.letsencrypt.org"
|
||||
DEFAULT_CA = "https://acme-v02.api.letsencrypt.org" # DEPRECATED! USE DEFAULT_DIRECTORY_URL INSTEAD
|
||||
DEFAULT_DIRECTORY_URL = "https://acme-v02.api.letsencrypt.org/directory"
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
LOGGER.addHandler(logging.StreamHandler())
|
||||
LOGGER.setLevel(logging.INFO)
|
||||
|
||||
def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA, no_checks=False):
|
||||
# helper function base64 encode for jose spec
|
||||
def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA, disable_check=False, directory_url=DEFAULT_DIRECTORY_URL, contact=None):
|
||||
directory, acct_headers, alg, jwk = None, None, None, None # global variables
|
||||
|
||||
# helper functions - base64 encode for jose spec
|
||||
def _b64(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"):
|
||||
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"}))
|
||||
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 = _b64(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'))
|
||||
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))
|
||||
|
||||
# helper function - poll until complete
|
||||
def _poll_until_not(url, pending_statuses, err_msg):
|
||||
while True:
|
||||
result, _, _ = _do_request(url, err_msg=err_msg)
|
||||
if result['status'] in pending_statuses:
|
||||
time.sleep(2)
|
||||
continue
|
||||
return result
|
||||
|
||||
# parse account key to get public key
|
||||
log.info("Parsing account key...")
|
||||
proc = subprocess.Popen(["openssl", "rsa", "-in", account_key, "-noout", "-text"],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
out, err = proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
raise IOError("OpenSSL Error: {0}".format(err))
|
||||
pub_hex, pub_exp = re.search(
|
||||
r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
|
||||
out.decode('utf8'), re.MULTILINE|re.DOTALL).groups()
|
||||
out = _cmd(["openssl", "rsa", "-in", account_key, "-noout", "-text"], err_msg="OpenSSL Error")
|
||||
pub_pattern = r"modulus:\n\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
|
||||
header = {
|
||||
"alg": "RS256",
|
||||
"jwk": {
|
||||
"e": _b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
|
||||
"kty": "RSA",
|
||||
"n": _b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
|
||||
},
|
||||
alg = "RS256"
|
||||
jwk = {
|
||||
"e": _b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
|
||||
"kty": "RSA",
|
||||
"n": _b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
|
||||
}
|
||||
accountkey_json = json.dumps(header['jwk'], sort_keys=True, separators=(',', ':'))
|
||||
accountkey_json = json.dumps(jwk, sort_keys=True, separators=(',', ':'))
|
||||
thumbprint = _b64(hashlib.sha256(accountkey_json.encode('utf8')).digest())
|
||||
|
||||
# helper function make signed requests
|
||||
def _send_signed_request(url, payload):
|
||||
payload64 = _b64(json.dumps(payload).encode('utf8'))
|
||||
protected = copy.deepcopy(header)
|
||||
protected["nonce"] = urlopen(CA + "/directory").headers['Replay-Nonce']
|
||||
protected64 = _b64(json.dumps(protected).encode('utf8'))
|
||||
proc = subprocess.Popen(["openssl", "dgst", "-sha256", "-sign", account_key],
|
||||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
out, err = proc.communicate("{0}.{1}".format(protected64, payload64).encode('utf8'))
|
||||
if proc.returncode != 0:
|
||||
raise IOError("OpenSSL Error: {0}".format(err))
|
||||
data = json.dumps({
|
||||
"header": header, "protected": protected64,
|
||||
"payload": payload64, "signature": _b64(out),
|
||||
})
|
||||
try:
|
||||
resp = urlopen(url, data.encode('utf8'))
|
||||
return resp.getcode(), resp.read()
|
||||
except IOError as e:
|
||||
return getattr(e, "code", None), getattr(e, "read", e.__str__)()
|
||||
|
||||
# find domains
|
||||
log.info("Parsing CSR...")
|
||||
proc = subprocess.Popen(["openssl", "req", "-in", csr, "-noout", "-text"],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
out, err = proc.communicate()
|
||||
if proc.returncode != 0:
|
||||
raise IOError("Error loading {0}: {1}".format(csr, err))
|
||||
out = _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:
|
||||
|
|
@ -77,122 +98,100 @@ def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA, no_checks=Fal
|
|||
for san in subject_alt_names.group(1).split(", "):
|
||||
if san.startswith("DNS:"):
|
||||
domains.add(san[4:])
|
||||
log.info("Found domains: {0}".format(", ".join(domains)))
|
||||
|
||||
# get the certificate domains and expiration
|
||||
# get the ACME directory of urls
|
||||
log.info("Getting directory...")
|
||||
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...")
|
||||
code, result = _send_signed_request(CA + "/acme/new-reg", {
|
||||
"resource": "new-reg",
|
||||
"agreement": json.loads(urlopen(CA + "/directory").read().decode('utf8'))['meta']['terms-of-service'],
|
||||
})
|
||||
if code == 201:
|
||||
log.info("Registered!")
|
||||
elif code == 409:
|
||||
log.info("Already registered!")
|
||||
else:
|
||||
raise ValueError("Error registering: {0} {1}".format(code, result))
|
||||
reg_payload = {"termsOfServiceAgreed": True}
|
||||
account, code, acct_headers = _send_signed_request(directory['newAccount'], reg_payload, "Error registering")
|
||||
log.info("Registered!" if code == 201 else "Already registered!")
|
||||
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'])))
|
||||
|
||||
# verify each domain
|
||||
for domain in domains:
|
||||
# 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")
|
||||
log.info("Order created!")
|
||||
|
||||
# get the authorizations that need to be completed
|
||||
for auth_url in order['authorizations']:
|
||||
authorization, _, _ = _do_request(auth_url, err_msg="Error getting challenges")
|
||||
domain = authorization['identifier']['value']
|
||||
log.info("Verifying {0}...".format(domain))
|
||||
|
||||
# get new challenge
|
||||
code, result = _send_signed_request(CA + "/acme/new-authz", {
|
||||
"resource": "new-authz",
|
||||
"identifier": {"type": "dns", "value": domain},
|
||||
})
|
||||
if code != 201:
|
||||
raise ValueError("Error requesting challenges: {0} {1}".format(code, result))
|
||||
|
||||
# make the challenge file
|
||||
challenge = [c for c in json.loads(result.decode('utf8'))['challenges'] if c['type'] == "http-01"][0]
|
||||
# find the http-01 challenge and write the challenge file
|
||||
challenge = [c for c in authorization['challenges'] if c['type'] == "http-01"][0]
|
||||
token = re.sub(r"[^A-Za-z0-9_\-]", "_", challenge['token'])
|
||||
keyauthorization = "{0}.{1}".format(token, thumbprint)
|
||||
wellknown_path = os.path.join(acme_dir, token)
|
||||
with open(wellknown_path, "w") as wellknown_file:
|
||||
wellknown_file.write(keyauthorization)
|
||||
|
||||
if not no_checks: # sometime the local g
|
||||
# check that the file is in place
|
||||
# check that the file is in place
|
||||
try:
|
||||
wellknown_url = "http://{0}/.well-known/acme-challenge/{1}".format(domain, token)
|
||||
try:
|
||||
resp = urlopen(wellknown_url)
|
||||
resp_data = resp.read().decode('utf8').strip()
|
||||
assert resp_data == keyauthorization
|
||||
except (IOError, AssertionError):
|
||||
os.remove(wellknown_path)
|
||||
raise ValueError("Wrote file to {0}, but couldn't download {1}".format(
|
||||
wellknown_path, wellknown_url))
|
||||
assert(disable_check or _do_request(wellknown_url)[0] == keyauthorization)
|
||||
except (AssertionError, ValueError) as e:
|
||||
os.remove(wellknown_path)
|
||||
raise ValueError("Wrote file to {0}, but couldn't download {1}: {2}".format(wellknown_path, wellknown_url, e))
|
||||
|
||||
# notify challenge are met
|
||||
code, result = _send_signed_request(challenge['uri'], {
|
||||
"resource": "challenge",
|
||||
"keyAuthorization": keyauthorization,
|
||||
})
|
||||
if code != 202:
|
||||
raise ValueError("Error triggering challenge: {0} {1}".format(code, result))
|
||||
# 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))
|
||||
if authorization['status'] != "valid":
|
||||
raise ValueError("Challenge did not pass for {0}: {1}".format(domain, authorization))
|
||||
log.info("{0} verified!".format(domain))
|
||||
|
||||
# wait for challenge to be verified
|
||||
while True:
|
||||
try:
|
||||
resp = urlopen(challenge['uri'])
|
||||
challenge_status = json.loads(resp.read().decode('utf8'))
|
||||
except IOError as e:
|
||||
raise ValueError("Error checking challenge: {0} {1}".format(
|
||||
e.code, json.loads(e.read().decode('utf8'))))
|
||||
if challenge_status['status'] == "pending":
|
||||
time.sleep(2)
|
||||
elif challenge_status['status'] == "valid":
|
||||
log.info("{0} verified!".format(domain))
|
||||
os.remove(wellknown_path)
|
||||
break
|
||||
else:
|
||||
raise ValueError("{0} challenge did not pass: {1}".format(
|
||||
domain, challenge_status))
|
||||
|
||||
# get the new certificate
|
||||
# finalize the order with the csr
|
||||
log.info("Signing certificate...")
|
||||
proc = subprocess.Popen(["openssl", "req", "-in", csr, "-outform", "DER"],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
csr_der, err = proc.communicate()
|
||||
code, result = _send_signed_request(CA + "/acme/new-cert", {
|
||||
"resource": "new-cert",
|
||||
"csr": _b64(csr_der),
|
||||
})
|
||||
if code != 201:
|
||||
raise ValueError("Error signing certificate: {0} {1}".format(code, result))
|
||||
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")
|
||||
|
||||
# return signed certificate!
|
||||
# poll the order to monitor when it's done
|
||||
order = _poll_until_not(order_headers['Location'], ["pending", "processing"], "Error checking order status")
|
||||
if order['status'] != "valid":
|
||||
raise ValueError("Order failed: {0}".format(order))
|
||||
|
||||
# download the certificate
|
||||
certificate_pem, _, _ = _do_request(order['certificate'], err_msg="Certificate download failed")
|
||||
log.info("Certificate signed!")
|
||||
return """-----BEGIN CERTIFICATE-----\n{0}\n-----END CERTIFICATE-----\n""".format(
|
||||
"\n".join(textwrap.wrap(base64.b64encode(result).decode('utf8'), 64)))
|
||||
return certificate_pem
|
||||
|
||||
def main(argv):
|
||||
def main(argv=None):
|
||||
parser = argparse.ArgumentParser(
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
description=textwrap.dedent("""\
|
||||
This script automates the process of getting a signed TLS certificate from
|
||||
Let's Encrypt using the ACME protocol. It will need to be run on your server
|
||||
and have access to your private account key, so PLEASE READ THROUGH IT! It's
|
||||
only ~200 lines, so it won't take long.
|
||||
This script automates the process of getting a signed TLS certificate from Let's Encrypt using
|
||||
the ACME protocol. It will need to be run on your server and have access to your private
|
||||
account key, so PLEASE READ THROUGH IT! It's only ~200 lines, so it won't take long.
|
||||
|
||||
===Example Usage===
|
||||
python acme_tiny.py --account-key ./account.key --csr ./domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > signed.crt
|
||||
===================
|
||||
Example Usage:
|
||||
python acme_tiny.py --account-key ./account.key --csr ./domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > signed_chain.crt
|
||||
|
||||
===Example Crontab Renewal (once per month)===
|
||||
0 0 1 * * python /path/to/acme_tiny.py --account-key /path/to/account.key --csr /path/to/domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > /path/to/signed.crt 2>> /var/log/acme_tiny.log
|
||||
==============================================
|
||||
Example Crontab Renewal (once per month):
|
||||
0 0 1 * * python /path/to/acme_tiny.py --account-key /path/to/account.key --csr /path/to/domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > /path/to/signed_chain.crt 2>> /var/log/acme_tiny.log
|
||||
""")
|
||||
)
|
||||
parser.add_argument("--account-key", required=True, help="path to your Let's Encrypt account private key")
|
||||
parser.add_argument("--csr", required=True, help="path to your certificate signing request")
|
||||
parser.add_argument("--acme-dir", required=True, help="path to the .well-known/acme-challenge/ directory")
|
||||
parser.add_argument("--quiet", action="store_const", const=logging.ERROR, help="suppress output except for errors")
|
||||
parser.add_argument("--ca", default=DEFAULT_CA, help="certificate authority, default is Let's Encrypt")
|
||||
parser.add_argument("--disable-check", default=False, action="store_true", help="disable checking if the challenge file is hosted correctly before telling the CA")
|
||||
parser.add_argument("--directory-url", default=DEFAULT_DIRECTORY_URL, help="certificate authority directory url, default is Let's Encrypt")
|
||||
parser.add_argument("--ca", default=DEFAULT_CA, help="DEPRECATED! USE --directory-url INSTEAD!")
|
||||
parser.add_argument("--contact", metavar="CONTACT", default=None, nargs="*", help="Contact details (e.g. mailto:aaa@bbb.com) for your account-key")
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
LOGGER.setLevel(args.quiet or LOGGER.level)
|
||||
signed_crt = get_crt(args.account_key, args.csr, args.acme_dir, log=LOGGER, CA=args.ca)
|
||||
signed_crt = get_crt(args.account_key, args.csr, args.acme_dir, log=LOGGER, CA=args.ca, disable_check=args.disable_check, directory_url=args.directory_url, contact=args.contact)
|
||||
sys.stdout.write(signed_crt)
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue