acmens.py: Add revoke_crt.

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

The revoke_crt.py will be removed.

1 files changed, 123 insertions, 5 deletions

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)