Script to download the latest lets encrypt certificate and key from DNSimple and apply them to your heroku endpoints

update-heroku-certs.py

#!/usr/bin/env python3

"""
Copyright 2021 Trail of Bits

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import os
import sys
import requests
from urllib.parse import (
    urlencode, unquote, urlparse, parse_qsl, ParseResult
)
import tempfile

DNSIMPLE_API_TOKEN = os.environ['DNSIMPLE_API_TOKEN']
HEROKU_API_TOKEN = os.environ['HEROKU_API_TOKEN']
DOMAIN="your-domain.here"
apps = ("your-app-1", "your-app-2")

def main():
    certs, key = fetch_dnsimple_certificates()

    for app in apps:
        print(f"[-] {app}")
        update_heroku_certificates(app, certs, key)

def fetch_dnsimple_certificates():
    headers = {
        "Accept" : "application/json",
        "Authorization": f"Bearer {DNSIMPLE_API_TOKEN}"
    }

    # Get my account ID
    r = requests.get("https://api.dnsimple.com/v2/whoami", headers=headers)
    data = r.json()["data"]
    account = data["account"]
    account_email = account["email"]
    account_id = account["id"]

    print(f"{account_email=}\n{account_id=}")

    # Get a list of certificates (grab the one with the furthest into the future expiration time)
    certificates = get_all_pages(f"https://api.dnsimple.com/v2/{account_id}/domains/{DOMAIN}/certificates?sort=expiration:desc", headers=headers)
    certificates = sorted(certificates, key=lambda x: x['expires_at'], reverse=True)
    certificate_id = [x for x in certificates if x['state'] == 'issued'][0]['id']
    print(f"{certificate_id=}")


    # Download the public part and certificate chain
    r = requests.get(f"https://api.dnsimple.com/v2/{account_id}/domains/{DOMAIN}/certificates/{certificate_id}/download", headers=headers)
    certificate = r.json()['data']
    certificate_data = certificate['server'] + '\n' + '\n'.join(certificate['chain'])
    print("[+] Got certificate chain")

    # Download the private part of the certificate
    r = requests.get(f"https://api.dnsimple.com/v2/{account_id}/domains/{DOMAIN}/certificates/{certificate_id}/private_key", headers=headers)
    private_key = r.json()['data']['private_key']
    print("[+] Got private key")

    return (certificate_data, private_key)

def update_heroku_certificates(app_name, certs, keys):
    headers = {
        "Accept" : "application/vnd.heroku+json; version=3",
        "Authorization": f"Bearer {HEROKU_API_TOKEN}"
    }

    # Get app data
    r = requests.get("https://api.heroku.com/apps", headers=headers)
    json = r.json()
    app = [x for x in json if x['name'] == app_name][0]
    app_id = app['id']
    print(f"[{app_name}] {app_id=}")

    # List SNI SSL Endpoints for app
    r = requests.get(f"https://api.heroku.com/apps/{app_id}/sni-endpoints", headers=headers)
    json = r.json()
    sni_ids = [x['id'] for x in json]
    print(f"[{app_name}] {sni_ids=}")

    # Update certificate and key for all SNI endpoints
    data = {
            'certificate_chain' : certs,
            'private_key' : keys
        }
    for sni_id in sni_ids:
        r = requests.patch(f"https://api.heroku.com/apps/{app_id}/sni-endpoints/{sni_id}", headers=headers, data=data)
        assert(r.status_code == 202)
        print(f"[{app_name}] New certificate and key for {sni_id} accepted")

def add_url_params(url, params):
    """ Add GET params to provided URL being aware of existing.
    :param url: string of target URL
    :param params: dict containing requested params to be added
    :return: string with updated URL
    >> url = 'http://stackoverflow.com/test?answers=true'
    >> new_params = {'answers': False, 'data': ['some','values']}
    >> add_url_params(url, new_params)
    'http://stackoverflow.com/test?data=some&data=values&answers=false'
    """
    # Unquoting URL first so we don't loose existing args
    url = unquote(url)
    # Extracting url info
    parsed_url = urlparse(url)
    # Extracting URL arguments from parsed URL
    get_args = parsed_url.query
    # Converting URL arguments to dict
    parsed_get_args = dict(parse_qsl(get_args))
    # Merging URL arguments dict with new params
    parsed_get_args.update(params)

    # Bool and Dict values should be converted to json-friendly values
    # you may throw this part away if you don't like it :)
    parsed_get_args.update(
        {k: dumps(v) for k, v in parsed_get_args.items()
         if isinstance(v, (bool, dict))}
    )

    # Converting URL argument to proper query string
    encoded_get_args = urlencode(parsed_get_args, doseq=True)
    # Creating new parsed result object based on provided with new
    # URL arguments. Same thing happens inside of urlparse.
    new_url = ParseResult(
        parsed_url.scheme, parsed_url.netloc, parsed_url.path,
        parsed_url.params, encoded_get_args, parsed_url.fragment
    ).geturl()

    return new_url

def get_all_pages(url, headers):
    items = []

    page = 1
    per_page = 100
    while True:
        request_url = add_url_params(url, {'page': page, 'per_page': per_page })
        r = requests.get(url, headers=headers)
        json = r.json()

        items += json['data']

        pagination = json['pagination']
        if pagination['total_pages'] == page:
            break

        page += 1

    return items

if __name__ == '__main__':
    main()