acme_tiny.py 8.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200
  1. #!/usr/bin/env python
  2. import argparse, subprocess, json, os, urllib2, sys, base64, binascii, time, \
  3. hashlib, re, copy, textwrap
  4. #CA = "https://acme-staging.api.letsencrypt.org"
  5. CA = "https://acme-v01.api.letsencrypt.org"
  6. def get_crt(account_key, csr, acme_dir):
  7. # helper function base64 encode for jose spec
  8. def _b64(b):
  9. return base64.urlsafe_b64encode(b).replace("=", "")
  10. # parse account key to get public key
  11. sys.stderr.write("Parsing account key...")
  12. proc = subprocess.Popen(["openssl", "rsa", "-in", account_key, "-noout", "-text"],
  13. stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  14. out, err = proc.communicate()
  15. if proc.returncode != 0:
  16. raise IOError("OpenSSL Error: {0}".format(err))
  17. pub_hex, pub_exp = re.search(
  18. "modulus:\n\s+00:([a-f0-9\:\s]+?)\npublicExponent: ([0-9]+)",
  19. out, re.MULTILINE|re.DOTALL).groups()
  20. pub_mod = binascii.unhexlify(re.sub("(\s|:)", "", pub_hex))
  21. pub_mod64 = _b64(pub_mod)
  22. pub_exp = "{0:x}".format(int(pub_exp))
  23. pub_exp = "0{0}".format(pub_exp) if len(pub_exp) % 2 else pub_exp
  24. pub_exp64 = _b64(binascii.unhexlify(pub_exp))
  25. header = {
  26. "alg": "RS256",
  27. "jwk": {
  28. "e": pub_exp64,
  29. "kty": "RSA",
  30. "n": pub_mod64,
  31. },
  32. }
  33. accountkey_json = json.dumps(header['jwk'], sort_keys=True, separators=(',', ':'))
  34. thumbprint = _b64(hashlib.sha256(accountkey_json).digest())
  35. sys.stderr.write("parsed!\n")
  36. # helper function make signed requests
  37. def _send_signed_request(url, payload):
  38. nonce = urllib2.urlopen(CA + "/directory").headers['Replay-Nonce']
  39. payload64 = _b64(json.dumps(payload))
  40. protected = copy.deepcopy(header)
  41. protected.update({"nonce": nonce})
  42. protected64 = _b64(json.dumps(protected))
  43. proc = subprocess.Popen(["openssl", "dgst", "-sha256", "-sign", account_key],
  44. stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  45. out, err = proc.communicate("{0}.{1}".format(protected64, payload64))
  46. if proc.returncode != 0:
  47. raise IOError("OpenSSL Error: {0}".format(err))
  48. data = json.dumps({
  49. "header": header,
  50. "protected": protected64,
  51. "payload": payload64,
  52. "signature": _b64(out),
  53. })
  54. try:
  55. resp = urllib2.urlopen(url, data)
  56. return resp.getcode(), resp.read()
  57. except urllib2.HTTPError as e:
  58. return e.code, e.read()
  59. # find domains
  60. sys.stderr.write("Parsing CSR...")
  61. proc = subprocess.Popen(["openssl", "req", "-in", csr, "-noout", "-text"],
  62. stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  63. out, err = proc.communicate()
  64. if proc.returncode != 0:
  65. raise IOError("Error loading {0}: {1}".format(csr, err))
  66. domains = set([])
  67. common_name = re.search("Subject:.*? CN=([^\s,;/]+)", out)
  68. if common_name is not None:
  69. domains.add(common_name.group(1))
  70. subject_alt_names = re.search("X509v3 Subject Alternative Name: \n +([^\n]+)\n", out, re.MULTILINE|re.DOTALL)
  71. if subject_alt_names is not None:
  72. for san in subject_alt_names.group(1).split(", "):
  73. if san.startswith("DNS:"):
  74. domains.add(san[4:])
  75. sys.stderr.write("parsed!\n")
  76. # get the certificate domains and expiration
  77. sys.stderr.write("Registering account...")
  78. code, result = _send_signed_request(CA + "/acme/new-reg", {
  79. "resource": "new-reg",
  80. "agreement": "https://letsencrypt.org/documents/LE-SA-v1.0.1-July-27-2015.pdf",
  81. })
  82. if code == 201:
  83. sys.stderr.write("registered!\n")
  84. elif code == 409:
  85. sys.stderr.write("already registered!\n")
  86. else:
  87. raise ValueError("Error registering: {0} {1}".format(code, result))
  88. # verify each domain
  89. for domain in domains:
  90. sys.stderr.write("Verifying {0}...".format(domain))
  91. # get new challenge
  92. code, result = _send_signed_request(CA + "/acme/new-authz", {
  93. "resource": "new-authz",
  94. "identifier": {
  95. "type": "dns",
  96. "value": domain,
  97. },
  98. })
  99. if code != 201:
  100. raise ValueError("Error registering: {0} {1}".format(code, result))
  101. # make the challenge file
  102. challenge = [c for c in json.loads(result)['challenges'] if c['type'] == "http-01"][0]
  103. keyauthorization = "{0}.{1}".format(challenge['token'], thumbprint)
  104. acme_dir = acme_dir[:-1] if acme_dir.endswith("/") else acme_dir
  105. wellknown_path = "{0}/{1}".format(acme_dir, challenge['token'])
  106. wellknown_file = open(wellknown_path, "w")
  107. wellknown_file.write(keyauthorization)
  108. wellknown_file.close()
  109. # check that the file is in place
  110. wellknown_url = "http://{0}/.well-known/acme-challenge/{1}".format(
  111. domain, challenge['token'])
  112. try:
  113. resp = urllib2.urlopen(wellknown_url)
  114. assert resp.read().strip() == keyauthorization
  115. except (urllib2.HTTPError, urllib2.URLError, AssertionError):
  116. os.remove(wellknown_path)
  117. raise ValueError("Wrote file to {0}, but couldn't download {1}".format(
  118. wellknown_path, wellknown_url))
  119. # notify challenge are met
  120. code, result = _send_signed_request(challenge['uri'], {
  121. "resource": "challenge",
  122. "keyAuthorization": keyauthorization,
  123. })
  124. if code != 202:
  125. raise ValueError("Error triggering challenge: {0} {1}".format(code, result))
  126. # wait for challenge to be verified
  127. while True:
  128. try:
  129. resp = urllib2.urlopen(challenge['uri'])
  130. challenge_status = json.loads(resp.read())
  131. except urllib2.HTTPError as e:
  132. raise ValueError("Error checking challenge: {0} {1}".format(
  133. e.code, json.loads(e.read())))
  134. if challenge_status['status'] == "pending":
  135. time.sleep(2)
  136. elif challenge_status['status'] == "valid":
  137. sys.stderr.write("verified!\n")
  138. os.remove(wellknown_path)
  139. break
  140. else:
  141. raise ValueError("{0} challenge did not pass: {1}".format(
  142. domain, challenge_status))
  143. # get the new certificate
  144. sys.stderr.write("Signing certificate...")
  145. proc = subprocess.Popen(["openssl", "req", "-in", csr, "-outform", "DER"],
  146. stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  147. csr_der, err = proc.communicate()
  148. code, result = _send_signed_request(CA + "/acme/new-cert", {
  149. "resource": "new-cert",
  150. "csr": _b64(csr_der),
  151. })
  152. if code != 201:
  153. raise ValueError("Error signing certificate: {0} {1}".format(code, result))
  154. # return signed certificate!
  155. sys.stderr.write("signed!\n")
  156. return """-----BEGIN CERTIFICATE-----\n{0}\n-----END CERTIFICATE-----\n""".format(
  157. "\n".join(textwrap.wrap(base64.b64encode(result), 64)))
  158. if __name__ == "__main__":
  159. parser = argparse.ArgumentParser(
  160. formatter_class=argparse.RawDescriptionHelpFormatter,
  161. description=textwrap.dedent(
  162. """\
  163. This script automates the process of getting a signed TLS certificate from
  164. Let's Encrypt using the ACME protocol. It will need to be run on your server
  165. and have access to your private account key, so PLEASE READ THROUGH IT! It's
  166. only ~200 lines, so it won't take long.
  167. ===Example Usage===
  168. python acme_tiny.py --account-key ./account.key --csr ./domain.csr --acme-dir /usr/share/nginx/html/.well-known/acme-challenge/ > signed.crt
  169. ===================
  170. ===Example Crontab Renewal (once per month)===
  171. 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
  172. ==============================================
  173. """
  174. )
  175. )
  176. parser.add_argument("--account-key", required=True, help="path to your Let's Encrypt account private key")
  177. parser.add_argument("--csr", required=True, help="path to your certificate signing request")
  178. parser.add_argument("--acme-dir", required=True, help="path to the .well-known/acme-challenge/ directory")
  179. args = parser.parse_args()
  180. signed_crt = get_crt(args.account_key, args.csr, args.acme_dir)
  181. sys.stdout.write(signed_crt)