client.py

import time
import copy
import json
import base64
import hashlib
import logging
import binascii
import platform
import sys
import http_requests as requests
import OpenSSL
import cryptography
from . import __version__ as sewer_version
from .config import ACME_DIRECTORY_URL_PRODUCTION



class Client(object):
    """
    todo: improve documentation.

    usage:
        import sewer
        dns_class = sewer.CloudFlareDns(CLOUDFLARE_EMAIL='example@example.com',
                                        CLOUDFLARE_API_KEY='nsa-grade-api-key')

        1. to create a new certificate.
        client = sewer.Client(domain_name='example.com',
                              dns_class=dns_class)
        certificate = client.cert()
        certificate_key = client.certificate_key
        account_key = client.account_key

        with open('certificate.crt', 'w') as certificate_file:
            certificate_file.write(certificate)

        with open('certificate.key', 'w') as certificate_key_file:
            certificate_key_file.write(certificate_key)


        2. to renew a certificate:
        with open('account_key.key', 'r') as account_key_file:
            account_key = account_key_file.read()

        client = sewer.Client(domain_name='example.com',
                              dns_class=dns_class,
                              account_key=account_key)
        certificate = client.renew()
        certificate_key = client.certificate_key

    todo:
        - handle more exceptions
    """

    def __init__(
        self,
        domain_name,
        dns_class,
        domain_alt_names=None,
        contact_email=None,
        account_key=None,
        certificate_key=None,
        bits=2048,
        digest="sha256",
        ACME_REQUEST_TIMEOUT=7,
        ACME_AUTH_STATUS_WAIT_PERIOD=8,
        ACME_AUTH_STATUS_MAX_CHECKS=3,
        ACME_DIRECTORY_URL=ACME_DIRECTORY_URL_PRODUCTION,
        LOG_LEVEL="INFO",
    ):

        self.domain_name = domain_name
        self.dns_class = dns_class
        if not domain_alt_names:
            domain_alt_names = []
        self.domain_alt_names = domain_alt_names
        self.domain_alt_names = list(set(self.domain_alt_names))
        self.contact_email = contact_email
        self.bits = bits
        self.digest = digest
        self.ACME_REQUEST_TIMEOUT = ACME_REQUEST_TIMEOUT
        self.ACME_AUTH_STATUS_WAIT_PERIOD = ACME_AUTH_STATUS_WAIT_PERIOD
        self.ACME_AUTH_STATUS_MAX_CHECKS = ACME_AUTH_STATUS_MAX_CHECKS
        self.ACME_DIRECTORY_URL = ACME_DIRECTORY_URL
        self.LOG_LEVEL = LOG_LEVEL.upper()

        try:
            self.all_domain_names = copy.copy(self.domain_alt_names)
            self.all_domain_names.insert(0, self.domain_name)
            self.domain_alt_names = list(set(self.domain_alt_names))

            self.User_Agent = self.get_user_agent()
            acme_endpoints = self.get_acme_endpoints().json()
            self.ACME_GET_NONCE_URL = acme_endpoints["newNonce"]
            self.ACME_TOS_URL = acme_endpoints["meta"]["termsOfService"]
            self.ACME_KEY_CHANGE_URL = acme_endpoints["keyChange"]
            self.ACME_NEW_ACCOUNT_URL = acme_endpoints["newAccount"]
            self.ACME_NEW_ORDER_URL = acme_endpoints["newOrder"]
            self.ACME_REVOKE_CERT_URL = acme_endpoints["revokeCert"]

            # unique account identifier
            # https://tools.ietf.org/html/draft-ietf-acme-acme#section-6.2
            self.kid = None

            self.certificate_key = certificate_key or self.create_certificate_key()
            self.csr = self.create_csr()

            if not account_key:
                self.account_key = self.create_account_key()
                self.PRIOR_REGISTERED = False
            else:
                self.account_key = account_key
                self.PRIOR_REGISTERED = True

        except Exception as e:
            raise e

    @staticmethod
    def log_response(response):
        """
        renders response as json or as a string
        """
        # TODO: use this to handle all response logs.
        try:
            log_body = response.json()
        except ValueError:
            log_body = response.content[:30]
        return log_body

    @staticmethod
    def get_user_agent():
        return "python-requests/{requests_version} ({system}: {machine}) sewer {sewer_version} ({sewer_url})".format(
            requests_version=requests.__version__,
            system=platform.system(),
            machine=platform.machine(),
            sewer_version=sewer_version.__version__,
            sewer_url=sewer_version.__url__,
        )

    def get_acme_endpoints(self):
        headers = {"User-Agent": self.User_Agent}
        get_acme_endpoints = requests.get(
            self.ACME_DIRECTORY_URL, timeout=self.ACME_REQUEST_TIMEOUT, headers=headers,verify=False
        )
        if get_acme_endpoints.status_code not in [200, 201]:
            raise ValueError(
                "Error while getting Acme endpoints: status_code={status_code} response={response}".format(
                    status_code=get_acme_endpoints.status_code,
                    response=self.log_response(get_acme_endpoints),
                )
            )
        return get_acme_endpoints

    def create_certificate_key(self):
        return self.create_key().decode()

    def create_account_key(self):
        return self.create_key().decode()

    def create_key(self, key_type=OpenSSL.crypto.TYPE_RSA):
        key = OpenSSL.crypto.PKey()
        key.generate_key(key_type, self.bits)
        private_key = OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, key)
        return private_key

    def create_csr(self):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.4
        The CSR is sent in the base64url-encoded version of the DER format. (NB: this
        field uses base64url, and does not include headers, it is different from PEM.)
        """
        X509Req = OpenSSL.crypto.X509Req()
        X509Req.get_subject().CN = self.domain_name

        if self.domain_alt_names:
            SAN = "DNS:{0}, ".format(self.domain_name).encode("utf8") + ", ".join(
                "DNS:" + i for i in self.domain_alt_names
            ).encode("utf8")
        else:
            SAN = "DNS:{0}".format(self.domain_name).encode("utf8")

        X509Req.add_extensions(
            [
                OpenSSL.crypto.X509Extension(
                    "subjectAltName".encode("utf8"), critical=False, value=SAN
                )
            ]
        )
        pk = OpenSSL.crypto.load_privatekey(
            OpenSSL.crypto.FILETYPE_PEM, self.certificate_key.encode()
        )
        X509Req.set_pubkey(pk)
        X509Req.set_version(2)
        X509Req.sign(pk, self.digest)
        return OpenSSL.crypto.dump_certificate_request(OpenSSL.crypto.FILETYPE_ASN1, X509Req)

    def acme_register(self):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.3
        The server creates an account and stores the public key used to
        verify the JWS (i.e., the "jwk" element of the JWS header) to
        authenticate future requests from the account.
        The server returns this account object in a 201 (Created) response, with the account URL
        in a Location header field.
        This account URL will be used in subsequest requests to ACME, as the "kid" value in the acme header.
        If the server already has an account registered with the provided
        account key, then it MUST return a response with a 200 (OK) status
        code and provide the URL of that account in the Location header field.
        If there is an existing account with the new key
        provided, then the server SHOULD use status code 409 (Conflict) and
        provide the URL of that account in the Location header field
        """
        if self.PRIOR_REGISTERED:
            payload = {"onlyReturnExisting": True}
        elif self.contact_email:
            payload = {
                "termsOfServiceAgreed": True,
                "contact": ["mailto:{0}".format(self.contact_email)],
            }
        else:
            payload = {"termsOfServiceAgreed": True}

        url = self.ACME_NEW_ACCOUNT_URL
        acme_register_response = self.make_signed_acme_request(url=url, payload=payload)

        if acme_register_response.status_code not in [201, 200, 409]:
            raise ValueError(
                "Error while registering: status_code={status_code} response={response}".format(
                    status_code=acme_register_response.status_code,
                    response=self.log_response(acme_register_response),
                )
            )

        kid = acme_register_response.headers["Location"]
        setattr(self, "kid", kid)

        return acme_register_response

    def apply_for_cert_issuance(self):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.4
        The order object returned by the server represents a promise that if
        the client fulfills the server's requirements before the "expires"
        time, then the server will be willing to finalize the order upon
        request and issue the requested certificate.  In the order object,
        any authorization referenced in the "authorizations" array whose
        status is "pending" represents an authorization transaction that the
        client must complete before the server will issue the certificate.

        Once the client believes it has fulfilled the server's requirements,
        it should send a POST request to the order resource's finalize URL.
        The POST body MUST include a CSR:

        The date values seem to be ignored by LetsEncrypt although they are
        in the ACME draft spec; https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.4
        """
        identifiers = []
        for domain_name in self.all_domain_names:
            identifiers.append({"type": "dns", "value": domain_name})

        payload = {"identifiers": identifiers}
        url = self.ACME_NEW_ORDER_URL
        apply_for_cert_issuance_response = self.make_signed_acme_request(url=url, payload=payload)

        if apply_for_cert_issuance_response.status_code != 201:
            raise ValueError(
                "Error applying for certificate issuance: status_code={status_code} response={response}".format(
                    status_code=apply_for_cert_issuance_response.status_code,
                    response=self.log_response(apply_for_cert_issuance_response),
                )
            )

        apply_for_cert_issuance_response_json = apply_for_cert_issuance_response.json()
        finalize_url = apply_for_cert_issuance_response_json["finalize"]
        authorizations = apply_for_cert_issuance_response_json["authorizations"]

        return authorizations, finalize_url

    def get_identifier_authorization(self, url):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.5
        When a client receives an order from the server it downloads the
        authorization resources by sending GET requests to the indicated
        URLs.  If the client initiates authorization using a request to the
        new authorization resource, it will have already received the pending
        authorization object in the response to that request.

        This is also where we get the challenges/tokens.
        """
        headers = {"User-Agent": self.User_Agent}
        get_identifier_authorization_response = requests.get(
            url, timeout=self.ACME_REQUEST_TIMEOUT, headers=headers,verify=False
        )
        if get_identifier_authorization_response.status_code not in [200, 201]:
            raise ValueError(
                "Error getting identifier authorization: status_code={status_code} response={response}".format(
                    status_code=get_identifier_authorization_response.status_code,
                    response=self.log_response(get_identifier_authorization_response),
                )
            )
        res = get_identifier_authorization_response.json()
        domain = res["identifier"]["value"]
        wildcard = res.get("wildcard")
        if wildcard:
            domain = "*." + domain

        for i in res["challenges"]:
            if i["type"] == "dns-01":
                dns_challenge = i
        dns_token = dns_challenge["token"]
        dns_challenge_url = dns_challenge["url"]
        identifier_auth = {
            "domain": domain,
            "url": url,
            "wildcard": wildcard,
            "dns_token": dns_token,
            "dns_challenge_url": dns_challenge_url,
        }


        return identifier_auth

    def get_keyauthorization(self, dns_token):
        acme_header_jwk_json = json.dumps(
            self.get_acme_header("GET_THUMBPRINT")["jwk"], sort_keys=True, separators=(",", ":")
        )
        acme_thumbprint = self.calculate_safe_base64(
            hashlib.sha256(acme_header_jwk_json.encode("utf8")).digest()
        )
        acme_keyauthorization = "{0}.{1}".format(dns_token, acme_thumbprint)
        base64_of_acme_keyauthorization = self.calculate_safe_base64(
            hashlib.sha256(acme_keyauthorization.encode("utf8")).digest()
        )

        return acme_keyauthorization, base64_of_acme_keyauthorization

    def check_authorization_status(self, authorization_url, desired_status=None):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.5.1
        To check on the status of an authorization, the client sends a GET(polling)
        request to the authorization URL, and the server responds with the
        current authorization object.

        https://tools.ietf.org/html/draft-ietf-acme-acme#section-8.2
        Clients SHOULD NOT respond to challenges until they believe that the
        server's queries will succeed. If a server's initial validation
        query fails, the server SHOULD retry[intended to address things like propagation delays in
        HTTP/DNS provisioning] the query after some time.
        The server MUST provide information about its retry state to the
        client via the "errors" field in the challenge and the Retry-After
        """
        desired_status = desired_status or ["pending", "valid","invalid"]
        number_of_checks = 0
        while True:
            headers = {"User-Agent": self.User_Agent}
            check_authorization_status_response = requests.get(
                authorization_url, timeout=self.ACME_REQUEST_TIMEOUT, headers=headers,verify=False
            )
            a_auth = check_authorization_status_response.json()
            authorization_status = a_auth["status"]
            number_of_checks = number_of_checks + 1
            if number_of_checks == self.ACME_AUTH_STATUS_MAX_CHECKS:
                raise StopIteration(
                    "Checks done={0}. Max checks allowed={1}. Interval between checks={2}seconds. >>>> {3}".format(
                        number_of_checks,
                        self.ACME_AUTH_STATUS_MAX_CHECKS,
                        self.ACME_AUTH_STATUS_WAIT_PERIOD,
                        json.dumps(a_auth)
                    )
                )
            if authorization_status in desired_status:
                if authorization_status == "invalid":
                    try:
                        import panelLets
                        if 'error' in a_auth['challenges'][0]:
                            ret_title = a_auth['challenges'][0]['error']['detail']
                        elif 'error' in a_auth['challenges'][1]:
                            ret_title = a_auth['challenges'][1]['error']['detail']
                        elif 'error' in a_auth['challenges'][2]:
                            ret_title = a_auth['challenges'][2]['error']['detail']
                        else:
                            ret_title = str(a_auth)
                        ret_title = panelLets.panelLets().get_error(ret_title)
                    except:
                        ret_title = str(a_auth)
                    raise StopIteration(
                        "{0} >>>> {1}".format(
                            ret_title,
                            json.dumps(a_auth)
                        )
                    )
                break
            else:
                # for any other status, sleep then retry
                time.sleep(self.ACME_AUTH_STATUS_WAIT_PERIOD)

        return check_authorization_status_response

    def respond_to_challenge(self, acme_keyauthorization, dns_challenge_url):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.5.1
        To prove control of the identifier and receive authorization, the
        client needs to respond with information to complete the challenges.
        The server is said to "finalize" the authorization when it has
        completed one of the validations, by assigning the authorization a
        status of "valid" or "invalid".

        Usually, the validation process will take some time, so the client
        will need to poll the authorization resource to see when it is finalized.
        To check on the status of an authorization, the client sends a GET(polling)
        request to the authorization URL, and the server responds with the
        current authorization object.
        """
        payload = {"keyAuthorization": "{0}".format(acme_keyauthorization)}
        respond_to_challenge_response = self.make_signed_acme_request(dns_challenge_url, payload)

        return respond_to_challenge_response

    def send_csr(self, finalize_url):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-7.4
        Once the client believes it has fulfilled the server's requirements,
        it should send a POST request(include a CSR) to the order resource's finalize URL.
        A request to finalize an order will result in error if the order indicated does not have status "pending",
        if the CSR and order identifiers differ, or if the account is not authorized for the identifiers indicated in the CSR.
        The CSR is sent in the base64url-encoded version of the DER format(OpenSSL.crypto.FILETYPE_ASN1)

        A valid request to finalize an order will return the order to be finalized.
        The client should begin polling the order by sending a
        GET request to the order resource to obtain its current state.
        """
        payload = {"csr": self.calculate_safe_base64(self.csr)}
        send_csr_response = self.make_signed_acme_request(url=finalize_url, payload=payload)

        if send_csr_response.status_code not in [200, 201]:
            raise ValueError(
                "Error sending csr: status_code={status_code} response={response}".format(
                    status_code=send_csr_response.status_code,
                    response=self.log_response(send_csr_response),
                )
            )
        send_csr_response_json = send_csr_response.json()
        certificate_url = send_csr_response_json["certificate"]
        return certificate_url

    def download_certificate(self, certificate_url):
        download_certificate_response = self.make_signed_acme_request(
            certificate_url, payload="DOWNLOAD_Z_CERTIFICATE"
        )

        if download_certificate_response.status_code not in [200, 201]:
            raise ValueError(
                "Error fetching signed certificate: status_code={status_code} response={response}".format(
                    status_code=download_certificate_response.status_code,
                    response=self.log_response(download_certificate_response),
                )
            )


        pem_certificate = download_certificate_response.content
        if type(pem_certificate) == bytes: pem_certificate = pem_certificate.decode('utf-8')

        return pem_certificate

    def sign_message(self, message):
        pk = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, self.account_key.encode())
        return OpenSSL.crypto.sign(pk, message.encode("utf8"), self.digest)

    def get_nonce(self):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-6.4
        Each request to an ACME server must include a fresh unused nonce
        in order to protect against replay attacks.
        """
        headers = {"User-Agent": self.User_Agent}
        response = requests.get(
            self.ACME_GET_NONCE_URL, timeout=self.ACME_REQUEST_TIMEOUT, headers=headers,verify=False
        )
        nonce = response.headers["Replay-Nonce"]
        return nonce

    @staticmethod
    def stringfy_items(payload):
        """
        method that takes a dictionary and then converts any keys or values
        in that are of type bytes into unicode strings.
        This is necessary esp if you want to then turn that dict into a json string.
        """
        if isinstance(payload, str):
            return payload

        for k, v in payload.items():
            if isinstance(k, bytes):
                k = k.decode("utf-8")
            if isinstance(v, bytes):
                v = v.decode("utf-8")
            payload[k] = v
        return payload

    @staticmethod
    def calculate_safe_base64(un_encoded_data):
        """
        takes in a string or bytes
        returns a string
        """
        if sys.version_info[0] == 3:
            if isinstance(un_encoded_data, str):
                un_encoded_data = un_encoded_data.encode("utf8")
        r = base64.urlsafe_b64encode(un_encoded_data).rstrip(b"=")
        return r.decode("utf8")

    def get_acme_header(self, url):
        """
        https://tools.ietf.org/html/draft-ietf-acme-acme#section-6.2
        The JWS Protected Header MUST include the following fields:
        - "alg" (Algorithm)
        - "jwk" (JSON Web Key, only for requests to new-account and revoke-cert resources)
        - "kid" (Key ID, for all other requests). gotten from self.ACME_NEW_ACCOUNT_URL
        - "nonce". gotten from self.ACME_GET_NONCE_URL
        - "url"
        """
        header = {"alg": "RS256", "nonce": self.get_nonce(), "url": url}
        
        if url in [self.ACME_NEW_ACCOUNT_URL, self.ACME_REVOKE_CERT_URL, "GET_THUMBPRINT"]:
            private_key = cryptography.hazmat.primitives.serialization.load_pem_private_key(
                self.account_key.encode(),
                password=None,
                backend=cryptography.hazmat.backends.default_backend(),
            )
            public_key_public_numbers = private_key.public_key().public_numbers()
            # private key public exponent in hex format
            exponent = "{0:x}".format(public_key_public_numbers.e)
            exponent = "0{0}".format(exponent) if len(exponent) % 2 else exponent
            # private key modulus in hex format
            modulus = "{0:x}".format(public_key_public_numbers.n)
            jwk = {
                "kty": "RSA",
                "e": self.calculate_safe_base64(binascii.unhexlify(exponent)),
                "n": self.calculate_safe_base64(binascii.unhexlify(modulus)),
            }
            header["jwk"] = jwk
        else:
            header["kid"] = self.kid
        return header

    def make_signed_acme_request(self, url, payload):
        headers = {"User-Agent": self.User_Agent}
        payload = self.stringfy_items(payload)

        if payload in ["GET_Z_CHALLENGE", "DOWNLOAD_Z_CERTIFICATE"]:
            response = requests.get(url, timeout=self.ACME_REQUEST_TIMEOUT, headers=headers,verify=False)
        else:
            payload64 = self.calculate_safe_base64(json.dumps(payload))
            protected = self.get_acme_header(url)
            protected64 = self.calculate_safe_base64(json.dumps(protected))
            signature = self.sign_message(message="{0}.{1}".format(protected64, payload64))  # bytes
            signature64 = self.calculate_safe_base64(signature)  # str
            data = json.dumps(
                {"protected": protected64, "payload": payload64, "signature": signature64}
            )
            headers.update({"Content-Type": "application/jose+json"})
            response = requests.post(
                url, data=data.encode("utf8"), timeout=self.ACME_REQUEST_TIMEOUT, headers=headers ,verify=False
            )
        return response

    def get_certificate(self):
        domain_dns_value = "placeholder"
        dns_names_to_delete = []
        try:
            self.acme_register()
            authorizations, finalize_url = self.apply_for_cert_issuance()
            responders = []
            for url in authorizations:
                identifier_auth = self.get_identifier_authorization(url)
                authorization_url = identifier_auth["url"]
                dns_name = identifier_auth["domain"]
                dns_token = identifier_auth["dns_token"]
                dns_challenge_url = identifier_auth["dns_challenge_url"]

                acme_keyauthorization, domain_dns_value = self.get_keyauthorization(dns_token)
                self.dns_class.create_dns_record(dns_name, domain_dns_value)
                dns_names_to_delete.append(
                    {"dns_name": dns_name, "domain_dns_value": domain_dns_value}
                )
                responders.append(
                    {
                        "authorization_url": authorization_url,
                        "acme_keyauthorization": acme_keyauthorization,
                        "dns_challenge_url": dns_challenge_url,
                    }
                )

            # for a case where you want certificates for *.example.com and example.com
            # you have to create both dns records AND then respond to the challenge.
            # see issues/83
            for i in responders:
                # Make sure the authorization is in a status where we can submit a challenge
                # response. The authorization can be in the "valid" state before submitting
                # a challenge response if there was a previous authorization for these hosts
                # that was successfully validated, still cached by the server.
                auth_status_response = self.check_authorization_status(i["authorization_url"])
                if auth_status_response.json()["status"] == "pending":
                    self.respond_to_challenge(i["acme_keyauthorization"], i["dns_challenge_url"])

            for i in responders:
                # Before sending a CSR, we need to make sure the server has completed the
                # validation for all the authorizations
                self.check_authorization_status(i["authorization_url"], ["valid"])

            certificate_url = self.send_csr(finalize_url)
            certificate = self.download_certificate(certificate_url)
        except Exception as e:
            raise e
        finally:
            for i in dns_names_to_delete:
                self.dns_class.delete_dns_record(i["dns_name"], i["domain_dns_value"])

        return certificate

    def cert(self):
        """
        convenience method to get a certificate without much hassle
        """
        return self.get_certificate()

    def renew(self):
        """
        renews a certificate.
        A renewal is actually just getting a new certificate.
        An issuance request counts as a renewal if it contains the exact same set of hostnames as a previously issued certificate.
            https://letsencrypt.org/docs/rate-limits/
        """
        return self.cert()