acme_tiny.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. #!/usr/bin/env python
  2. # Copyright Daniel Roesler, under MIT license, see LICENSE at github.com/diafygi/acme-tiny
  3. import argparse, subprocess, json, os, sys, base64, binascii, time, hashlib, re, copy, textwrap, logging
  4. try:
  5. from urllib.request import urlopen, Request # Python 3
  6. except ImportError:
  7. from urllib2 import urlopen, Request # Python 2
  8. DEFAULT_CA = "https://acme-v02.api.letsencrypt.org" # DEPRECATED! USE DEFAULT_DIRECTORY_URL INSTEAD
  9. DEFAULT_DIRECTORY_URL = "https://acme-v02.api.letsencrypt.org/directory"
  10. LOGGER = logging.getLogger(__name__)
  11. LOGGER.addHandler(logging.StreamHandler())
  12. LOGGER.setLevel(logging.INFO)
  13. def get_crt(account_key, csr, acme_dir, log=LOGGER, CA=DEFAULT_CA, disable_check=False, directory_url=DEFAULT_DIRECTORY_URL, contact=None):
  14. directory, acct_headers, alg, jwk = None, None, None, None # global variables
  15. # helper functions - base64 encode for jose spec
  16. def _b64(b):
  17. return base64.urlsafe_b64encode(b).decode('utf8').replace("=", "")
  18. # helper function - run external commands
  19. def _cmd(cmd_list, stdin=None, cmd_input=None, err_msg="Command Line Error"):
  20. proc = subprocess.Popen(cmd_list, stdin=stdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  21. out, err = proc.communicate(cmd_input)
  22. if proc.returncode != 0:
  23. raise IOError("{0}\n{1}".format(err_msg, err))
  24. return out
  25. # helper function - make request and automatically parse json response
  26. def _do_request(url, data=None, err_msg="Error", depth=0):
  27. try:
  28. resp = urlopen(Request(url, data=data, headers={"Content-Type": "application/jose+json", "User-Agent": "acme-tiny"}))
  29. resp_data, code, headers = resp.read().decode("utf8"), resp.getcode(), resp.headers
  30. resp_data = json.loads(resp_data) # try to parse json results
  31. except ValueError:
  32. pass # ignore json parsing errors
  33. except IOError as e:
  34. resp_data = e.read().decode("utf8") if hasattr(e, "read") else str(e)
  35. code, headers = getattr(e, "code", None), {}
  36. if depth < 100 and code == 400 and json.loads(resp_data)['type'] == "urn:ietf:params:acme:error:badNonce":
  37. raise IndexError(resp_data) # allow 100 retrys for bad nonces
  38. if code not in [200, 201, 204]:
  39. raise ValueError("{0}:\nUrl: {1}\nData: {2}\nResponse Code: {3}\nResponse: {4}".format(err_msg, url, data, code, resp_data))
  40. return resp_data, code, headers
  41. # helper function - make signed requests
  42. def _send_signed_request(url, payload, err_msg, depth=0):
  43. payload64 = _b64(json.dumps(payload).encode('utf8'))
  44. new_nonce = _do_request(directory['newNonce'])[2]['Replay-Nonce']
  45. protected = {"url": url, "alg": alg, "nonce": new_nonce}
  46. protected.update({"jwk": jwk} if acct_headers is None else {"kid": acct_headers['Location']})
  47. protected64 = _b64(json.dumps(protected).encode('utf8'))
  48. protected_input = "{0}.{1}".format(protected64, payload64).encode('utf8')
  49. out = _cmd(["openssl", "dgst", "-sha256", "-sign", account_key], stdin=subprocess.PIPE, cmd_input=protected_input, err_msg="OpenSSL Error")
  50. data = json.dumps({"protected": protected64, "payload": payload64, "signature": _b64(out)})
  51. try:
  52. return _do_request(url, data=data.encode('utf8'), err_msg=err_msg, depth=depth)
  53. except IndexError: # retry bad nonces (they raise IndexError)
  54. return _send_signed_request(url, payload, err_msg, depth=(depth + 1))
  55. # helper function - poll until complete
  56. def _poll_until_not(url, pending_statuses, err_msg):
  57. while True:
  58. result, _, _ = _do_request(url, err_msg=err_msg)
  59. if result['status'] in pending_statuses:
  60. time.sleep(2)
  61. continue
  62. return result
  63. # parse account key to get public key
  64. log.info("Parsing account key...")
  65. out = _cmd(["openssl", "rsa", "-in", account_key, "-noout", "-text"], err_msg="OpenSSL Error")
  66. pub_pattern = r"modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)"
  67. pub_hex, pub_exp = re.search(pub_pattern, out.decode('utf8'), re.MULTILINE|re.DOTALL).groups()
  68. pub_exp = "{0:x}".format(int(pub_exp))
  69. pub_exp = "0{0}".format(pub_exp) if len(pub_exp) % 2 else pub_exp
  70. alg = "RS256"
  71. jwk = {
  72. "e": _b64(binascii.unhexlify(pub_exp.encode("utf-8"))),
  73. "kty": "RSA",
  74. "n": _b64(binascii.unhexlify(re.sub(r"(\s|:)", "", pub_hex).encode("utf-8"))),
  75. }
  76. accountkey_json = json.dumps(jwk, sort_keys=True, separators=(',', ':'))
  77. thumbprint = _b64(hashlib.sha256(accountkey_json.encode('utf8')).digest())
  78. # find domains
  79. log.info("Parsing CSR...")
  80. out = _cmd(["openssl", "req", "-in", csr, "-noout", "-text"], err_msg="Error loading {0}".format(csr))
  81. domains = set([])
  82. common_name = re.search(r"Subject:.*? CN\s?=\s?([^\s,;/]+)", out.decode('utf8'))
  83. if common_name is not None:
  84. domains.add(common_name.group(1))
  85. subject_alt_names = re.search(r"X509v3 Subject Alternative Name: \n +([^\n]+)\n", out.decode('utf8'), re.MULTILINE|re.DOTALL)
  86. if subject_alt_names is not None:
  87. for san in subject_alt_names.group(1).split(", "):
  88. if san.startswith("DNS:"):
  89. domains.add(san[4:])
  90. log.info("Found domains: {0}".format(", ".join(domains)))
  91. # get the ACME directory of urls
  92. log.info("Getting directory...")
  93. directory_url = CA + "/directory" if CA != DEFAULT_CA else directory_url # backwards compatibility with deprecated CA kwarg
  94. directory, _, _ = _do_request(directory_url, err_msg="Error getting directory")
  95. log.info("Directory found!")
  96. # create account, update contact details (if any), and set the global key identifier
  97. log.info("Registering account...")
  98. reg_payload = {"termsOfServiceAgreed": True}
  99. account, code, acct_headers = _send_signed_request(directory['newAccount'], reg_payload, "Error registering")
  100. log.info("Registered!" if code == 201 else "Already registered!")
  101. if contact is not None:
  102. account, _, _ = _send_signed_request(acct_headers['Location'], {"contact": contact}, "Error updating contact details")
  103. log.info("Updated contact details:\n{0}".format("\n".join(account['contact'])))
  104. # create a new order
  105. log.info("Creating new order...")
  106. order_payload = {"identifiers": [{"type": "dns", "value": d} for d in domains]}
  107. order, _, order_headers = _send_signed_request(directory['newOrder'], order_payload, "Error creating new order")
  108. log.info("Order created!")
  109. # get the authorizations that need to be completed
  110. for auth_url in order['authorizations']:
  111. authorization, _, _ = _do_request(auth_url, err_msg="Error getting challenges")
  112. domain = authorization['identifier']['value']
  113. log.info("Verifying {0}...".format(domain))
  114. # find the http-01 challenge and write the challenge file
  115. challenge = [c for c in authorization['challenges'] if c['type'] == "http-01"][0]
  116. token = re.sub(r"[^A-Za-z0-9_\-]", "_", challenge['token'])
  117. keyauthorization = "{0}.{1}".format(token, thumbprint)
  118. wellknown_path = os.path.join(acme_dir, token)
  119. with open(wellknown_path, "w") as wellknown_file:
  120. wellknown_file.write(keyauthorization)
  121. # check that the file is in place
  122. try:
  123. wellknown_url = "http://{0}/.well-known/acme-challenge/{1}".format(domain, token)
  124. assert(disable_check or _do_request(wellknown_url)[0] == keyauthorization)
  125. except (AssertionError, ValueError) as e:
  126. os.remove(wellknown_path)
  127. raise ValueError("Wrote file to {0}, but couldn't download {1}: {2}".format(wellknown_path, wellknown_url, e))
  128. # say the challenge is done
  129. _send_signed_request(challenge['url'], {}, "Error submitting challenges: {0}".format(domain))
  130. authorization = _poll_until_not(auth_url, ["pending"], "Error checking challenge status for {0}".format(domain))
  131. if authorization['status'] != "valid":
  132. raise ValueError("Challenge did not pass for {0}: {1}".format(domain, authorization))
  133. log.info("{0} verified!".format(domain))
  134. # finalize the order with the csr
  135. log.info("Signing certificate...")
  136. csr_der = _cmd(["openssl", "req", "-in", csr, "-outform", "DER"], err_msg="DER Export Error")
  137. _send_signed_request(order['finalize'], {"csr": _b64(csr_der)}, "Error finalizing order")
  138. # poll the order to monitor when it's done
  139. order = _poll_until_not(order_headers['Location'], ["pending", "processing"], "Error checking order status")
  140. if order['status'] != "valid":
  141. raise ValueError("Order failed: {0}".format(order))
  142. # download the certificate
  143. certificate_pem, _, _ = _do_request(order['certificate'], err_msg="Certificate download failed")
  144. log.info("Certificate signed!")
  145. return certificate_pem
  146. def main(argv=None):
  147. parser = argparse.ArgumentParser(
  148. formatter_class=argparse.RawDescriptionHelpFormatter,
  149. description=textwrap.dedent("""\
  150. This script automates the process of getting a signed TLS certificate from Let's Encrypt using
  151. the ACME protocol. It will need to be run on your server and have access to your private
  152. account key, so PLEASE READ THROUGH IT! It's only ~200 lines, so it won't take long.
  153. Example Usage:
  154. python acme_tiny.py --account-key ./account.key --csr ./domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > signed_chain.crt
  155. Example Crontab Renewal (once per month):
  156. 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
  157. """)
  158. )
  159. parser.add_argument("--account-key", required=True, help="path to your Let's Encrypt account private key")
  160. parser.add_argument("--csr", required=True, help="path to your certificate signing request")
  161. parser.add_argument("--acme-dir", required=True, help="path to the .well-known/acme-challenge/ directory")
  162. parser.add_argument("--quiet", action="store_const", const=logging.ERROR, help="suppress output except for errors")
  163. parser.add_argument("--disable-check", default=False, action="store_true", help="disable checking if the challenge file is hosted correctly before telling the CA")
  164. parser.add_argument("--directory-url", default=DEFAULT_DIRECTORY_URL, help="certificate authority directory url, default is Let's Encrypt")
  165. parser.add_argument("--ca", default=DEFAULT_CA, help="DEPRECATED! USE --directory-url INSTEAD!")
  166. parser.add_argument("--contact", metavar="CONTACT", default=None, nargs="*", help="Contact details (e.g. mailto:aaa@bbb.com) for your account-key")
  167. args = parser.parse_args(argv)
  168. LOGGER.setLevel(args.quiet or LOGGER.level)
  169. 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)
  170. sys.stdout.write(signed_crt)
  171. if __name__ == "__main__": # pragma: no cover
  172. main(sys.argv[1:])