acmens.py: Add revoke_crt.

Move certicate revocation function -- revoke_crt -- into acmens.py.

The revoke_crt.py will be removed.

2 files changed, 131 insertions, 19 deletions

diff --git a/README.md b/README.md
index 8f59d6d..49850f3 100644
--- a/README.md
+++ b/README.md
@@ -4,24 +4,18 @@ A fork of [acme-nosudo][]. It uses ACMEv2 protocol and requires Python 3.
 
 [acme-nosudo]: https://github.com/diafygi/acme-nosudo
 
-acmens has two scripts:
+The `acmens` may be used for getting a new SSL certificate or renewing
+a SSL certificate for a domain and revoking a certificate for a
+domain.
 
- - acmens.py
- - revoke_crt.py
-
-The `acmens.py` is for getting a new SSL certificate or renewing a
-SSL certificate for a domain.
-
-The `revoke_crt.py` is for revoking a certificate for a domain.
-
-Both scripts are meant to be run locally from your computer.
+It's meant to be run locally from your computer.
 
 ## Prerequisites
 
 * openssl
 * python3
 
-## How to use the signing script
+## getting/renewing a certificate
 
 First, you need to generate an user account key for Let's Encrypt.
 This is the key that you use to register with Let's Encrypt. If you
@@ -69,18 +63,18 @@ When you run the script, it will:
  - Will write the certificate to `signed.crt` if ACME HTTP challenge is
    successful.
 
-## How to use the revocation script
+## revoking a certificate
 
 First, you will need to the user account key for Let's Encrypt that was used
 when the certifacate was signed.
 
 Second, you will need the PEM encoded signed certificate that was produced by
-`acmens.py`.
+`acmens`.
 
 Third, you run the script using python and passing in the path to your user
 account key and the signed domain certificate. The paths can be relative or
 absolute.
 
 ```sh
-python3 revoke_crt.py -k user.key domain.crt
+python3 acmens.py --revoke -k user.key --crt domain.crt
 ```
diff --git a/acmens.py b/acmens.py
index e371432..7078023 100644
--- a/acmens.py
+++ b/acmens.py
@@ -219,6 +219,109 @@ Notes:
 
     return signed_pem
 
+
+def revoke_crt(account_key, crt):
+    """Use the ACME protocol to revoke an ssl certificate signed by a
+    certificate authority.
+
+    :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-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("=", "")
+
+    def _a64(a):
+        "Shortcut function to go from jwt base64 string to bytes"
+        return base64.urlsafe_b64decode(str(a + ("=" * (len(a) % 4))))
+
+    # 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))
+
+    # 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,
+    }
+    sys.stderr.write("Found public key!\n")
+
+    # Step 2: Get account info.
+    sys.stderr.write("Getting account info...\n")
+    reg = {
+        "onlyReturnExistiing": True
+    }
+    acct_headers = None
+    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_der64 = _b64(crt_der)
+    rvk_payload = {
+        "certificate": crt_der64,
+    }
+    _send_signed_request(DIRECTORY['revokeCert'], rvk_payload, "Error revoking certificate")
+    sys.stderr.write("Certificate revoked!\n")
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -229,6 +332,9 @@ 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.
 
+You may also revoke a signed Let's Encrypt (ACME) certificate.
+
+
 NOTE: YOUR ACCOUNT KEY NEEDS TO BE DIFFERENT FROM YOUR DOMAIN KEY.
 
 Prerequisites:
@@ -241,19 +347,31 @@ $ 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 acmens.py --account-key user.key --email user@example.com domain.csr > signed.crt
+$ python3 acmens.py --account-key user.key --email user@example.com --csr domain.csr > signed.crt
 --------------
 
-""")
+Example: Revoking a signed certificate:
+--------------
+$ python3 acmens.py --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("--csr", help="path to your certificate signing request")
+    parser.add_argument("--crt", help="path to your signed certificate")
 
     args = parser.parse_args()
-    if args.csr is None:
+    if (not args.revoke) and (args.csr is None):
         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.exit(1)
 
-    signed_crt = sign_csr(args.account_key, args.csr, email=args.email)
-    sys.stdout.write(signed_crt)
+    if args.revoke:
+        revoke_crt(args.account_key, args.crt)
+    else:
+        signed_crt = sign_csr(args.account_key, args.csr, email=args.email)
+        sys.stdout.write(signed_crt)