sign_csr.py -> acmens.py

1 files changed, 0 insertions, 255 deletions

diff --git a/sign_csr.py b/sign_csr.py
deleted file mode 100644
index 2ab9d6b..0000000
--- a/sign_csr.py
+++ /dev/null
@@ -1,255 +0,0 @@
-#!/usr/bin/env python3
-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
-
-def sign_csr(account_key, csr, email=None):
-    """Use the ACME protocol to get an ssl certificate signed by a
-    certificate authority.
-
-    :param string account_key: Path to the user account key.
-    :param string csr: Path to the certificate signing request.
-    :param string email: An optional user account contact email
-                         (defaults to webmaster@<shortest_domain>)
-
-    :returns: Signed Certificate (PEM format)
-    :rtype: string
-
-    """
-    #CA = "https://acme-staging-v02.api.letsencrypt.org"
-    CA = "https://acme-v02.api.letsencrypt.org"
-    DIRECTORY = json.loads(urlopen(CA + "/directory").read().decode('utf8'))
-
-    def _b64(b):
-        "Shortcut function to go from bytes to jwt base64 string"
-        if type(b) is str:
-            b = b.encode()
-
-        return base64.urlsafe_b64encode(b).decode().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 = 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
-        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 = "" 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)})
-        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):
-        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
-
-
-    # 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")
-    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()
-    pub_mod = binascii.unhexlify(re.sub("(\s|:)", "", pub_hex))
-    pub_mod64 = _b64(pub_mod)
-    pub_exp = int(pub_exp)
-    pub_exp = "{0:x}".format(pub_exp)
-    pub_exp = "0{0}".format(pub_exp) if len(pub_exp) % 2 else pub_exp
-    pub_exp = binascii.unhexlify(pub_exp)
-    pub_exp64 = _b64(pub_exp)
-    jwk = {
-        "e": pub_exp64,
-        "kty": "RSA",
-        "n": pub_mod64,
-    }
-    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))
-    domains = set([])
-    common_name = re.search("Subject:.*? CN *= *([^\s,;/]+)", out.decode('utf8'))
-    if common_name is not None:
-        domains.add(common_name.group(1))
-    subject_alt_names = re.search("X509v3 Subject Alternative Name: \n +([^\n]+)\n", out.decode('utf8'), re.MULTILINE|re.DOTALL)
-    if subject_alt_names is not None:
-        for san in subject_alt_names.group(1).split(", "):
-            if san.startswith("DNS:"):
-                domains.add(san[4:])
-    sys.stderr.write("Found domains {0}\n".format(", ".join(domains)))
-
-    # Step 3: Ask user for contact email
-    if not email:
-        default_email = "webmaster@{0}".format(min(domains, key=len))
-        stdout = sys.stdout
-        sys.stdout = sys.stderr
-        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
-    }
-    acct_headers = None
-    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 each domain
-    for domain in domains:
-        sys.stderr.write("Making new order for {0}...\n".format(domain))
-        id = {
-            "identifiers": [{
-                "type": "dns",
-                "value": domain,
-            }],
-        }
-        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")
-
-        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_raw = "{}"
-
-        # challenge response for server
-        response = {
-            "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("""\
-Please update your server to serve the following file at this URL:
-
---------------
-URL: http://{0}/{1}
-File contents: \"{2}\"
---------------
-
-Notes:
-- Do not include the quotes in the file.
-- The file should be one line without any spaces.
-
-""".format(domain, response['uri'], response['data']))
-
-        stdout = sys.stdout
-        sys.stdout = sys.stderr
-        input("Press Enter when you've got the file hosted on your server...")
-        sys.stdout = stdout
-
-        # 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))
-        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")
-
-        # 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")
-
-        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")
-
-    sys.stderr.write("Received certificate!\n")
-    sys.stderr.write("You can remove the acme-challenge file from your webserver now.\n")
-
-    return signed_pem
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-        description="""\
-Get a SSL certificate signed by a Let's Encrypt (ACME) certificate
-authority and output that signed certificate. You do NOT need to run
-this script on your server, it is meant to be run on your
-computer. The script will request you to manually deploy the acme
-challenge on your server.
-
-NOTE: YOUR ACCOUNT KEY NEEDS TO BE DIFFERENT FROM YOUR DOMAIN KEY.
-
-Prerequisites:
-* openssl
-* python version 3
-
-Example: Generate an account keypair, a domain key and csr, and have the domain csr signed.
---------------
-$ openssl genrsa -aes256 4096 > user.key
-$ openssl rsa -in user.key -pubout > user.pub
-$ openssl genrsa -aes256 4096 > domain.key
-$ openssl req -new -sha256 -key domain.key -subj "/CN=example.com" > domain.csr
-$ python3 sign_csr.py --account-key user.key --email user@example.com domain.csr > signed.crt
---------------
-
-""")
-    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_path", help="path to your certificate signing request")
-
-    args = parser.parse_args()
-    signed_crt = sign_csr(args.account_key, args.csr_path, email=args.email)
-    sys.stdout.write(signed_crt)
-