add acme script

.gitignore

@@ -1,2 +1,3 @@
 *.secret
 **/__pycache__
+certbot

README.md

@@ -0,0 +1,41 @@
+# hatecomputers.club dns updater & certbot plugin
+
+this is a simple wrapper over hatecomputers.club's dns api
+
+## dns creation steps
+
+1. obtain an api key at [hatecomputers.club](https://hatecomputers.club)
+2. put it in `apikey.secret`
+3. modify `records.json` to your liking
+4. `./main.py --create --records-file=records.json`
+
+## certbot plugin
+
+follow the above to generate an api key.
+
+if you use the split-zone dns provided by hatecomputers.club and run your own certificate
+authority, you can try something like:
+```bash
+REQUESTS_CA_BUNDLE=~/armin/roots.pem certbot certonly \
+  --manual --manual-auth-hook ./plugin.sh \
+  --preferred-challenges dns \
+  -d *.internal.simponic.xyz \
+  --config-dir ./certbot \
+  --work-dir ./certbot \
+  --logs-dir ./certbot \
+  --server https://ca.internal.simponic.xyz/acme/ACME/directory \
+  --email simponic@hatecomputers.club \
+  --agree-tos \
+   --no-eff-email
+```
+
+otherwise:
+```bash
+sudo certbot certonly \
+  --manual --manual-auth-hook ./plugin.sh \
+  --preferred-challenges dns \
+  -d *.simponic.hatecomputers.club \
+  --email simponic@hatecomputers.club \
+  --agree-tos \
+  --no-eff-email
+```

args.py

@@ -4,26 +4,56 @@ import argparse
 def get_args():
     parser = argparse.ArgumentParser()
 
-    parser.add_argument("--endpoint", default="https://hatecomputers.club")
+    parser.add_argument(
-    parser.add_argument("--api-key-file", default="apikey.secret")
+        "--endpoint", default="https://hatecomputers.club", help="API endpoint"
+    )
+    parser.add_argument(
+        "--api-key-file",
+        default="apikey.secret",
+        help="path to file containing the API key",
+    )
     parser.add_argument(
         "--log-level",
         default="INFO",
         choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
     )
 
-    parser.add_argument("--certbot", action="store_true", default=False)
+    parser.add_argument(
-    parser.add_argument("--acme-url", required=False)
+        "--public-suffixes",
-    parser.add_argument("--acme-storage", default="acme.json")
+        default="hatecomputers.club",
+        help="comma separated list of public suffixes",
+    )
+    parser.add_argument(
+        "--dns-propogate-time",
+        default=20,
+        type=int,
+        help="time to sleep to allow DNS to propogate",
+    )
+    parser.add_argument(
+        "--certbot", action="store_true", default=False, help="enable certbot mode"
+    )
+    parser.add_argument(
+        "--certbot-domain", required=False, help="splat/domain to validate with certbot"
+    )
+    parser.add_argument(
+        "--certbot-validation", required=False, help="validation token for certbot"
+    )
 
-    parser.add_argument("--sync", action="store_true", default=False)
+    parser.add_argument(
-    parser.add_argument("--records-file", default="records.json")
+        "--create",
+        action="store_true",
+        default=False,
+        help="upload records file to API to sync",
+    )
+    parser.add_argument("--records-file", default="records.json", help="records file")
 
     args = parser.parse_args()
 
-    if (args.certbot) and (not args.acme_url):
+    if (args.certbot) and (not args.certbot_domain):
-        parser.error("--acme-url is required when --certbot is used")
+        parser.error("--certbot-domain is required when --certbot is used")
-    if (args.certbot) and (args.sync):
+    if (args.certbot) and (not args.certbot_validation):
-        parser.error("--sync cannot be used in combination with --certbot")
+        parser.error("--certbot-validation is required when --certbot is used")
+    if args.certbot:
+        args.public_suffixes = args.public_suffixes.split(",")
 
     return args

main.py

@@ -1,14 +1,37 @@
+#!/usr/bin/env python3
+
+import os
 import json
 import logging
+import time
 
 from updater.adapter import HatecomputersDNSAdapter
+from updater.utils import record_transformer
 from args import get_args
 
 
-def sync_records(adapter, records_path):
+def certbot_mode(args, dns_api_adapter, record_transformer):
-    records_file = open(records_path, "r")
+    domain = args.certbot_domain
-    dns_records = json.load(records_file)
+    if domain.startswith("*."):
-    adapter.post_records(dns_records)
+        domain = domain[2:]
+    logging.info(f"processing domain {domain}")
+
+    record = {
+        "ttl": 60,
+        "name": "_acme-challenge." + domain,
+        "type": "TXT",
+        "content": args.certbot_validation,
+    }
+    record = record_transformer(record)
+    logging.info(f"creating record {record}")
+    dns_api_adapter.post_record(record)
+
+    logging.info(
+        f"eeping out for {args.dns_propogate_time}s, to allow DNS propogation. look at this cute little guy 🐢 until then!!"
+    )
+    time.sleep(args.dns_propogate_time)
+
+    logging.info(f"updating record for {domain} with {args.certbot_validation}")
 
 
 if __name__ == "__main__":
@@ -17,10 +40,18 @@ if __name__ == "__main__":
     logging.root.setLevel(args.log_level)
 
     api_key = open(args.api_key_file, "r").read().strip()
-    adapter = HatecomputersDNSAdapter(args.endpoint, api_key)
+    dns_api_adapter = HatecomputersDNSAdapter(args.endpoint, api_key)
 
-    if args.sync:
+    if args.create:
-        sync_records(adapter, args.records_file)
+        records_file = open(args.records_file, "r")
+        dns_records = json.load(records_file)
+        dns_api_adapter.post_records(dns_records)
 
     if args.certbot:
-        logging.info("certbot mode")
+        certbot_mode(
+            args,
+            dns_api_adapter,
+            record_transformer(args.public_suffixes),
+        )
+
+    logging.info("done")

plugin.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+unset REQUESTS_CA_BUNDLE
+
+API_KEY_FILE=/Users/lizzy/git/simponic/dns-updater/apikey.secret
+ENDPOINT=https://hatecomputers.club
+PUBLIC_SUFFIXES=.hatecomputers.club
+
+./main.py --certbot \
+          --public-suffixes=$PUBLIC_SUFFIXES \
+          --certbot-domain=$CERTBOT_DOMAIN \
+          --certbot-validation=$CERTBOT_VALIDATION \
+          --endpoint=$ENDPOINT \
+          --api-key-file=$API_KEY_FILE

updater/acmedns.py

@@ -1,164 +0,0 @@
-#!/usr/bin/env python
-import json
-import os
-import requests
-import sys
-
-### EDIT THESE: Configuration values ###
-
-# URL to acme-dns instance
-ACMEDNS_URL = "https://auth.acme-dns.io"
-# Path for acme-dns credential storage
-STORAGE_PATH = "/etc/letsencrypt/acmedns.json"
-# Whitelist for address ranges to allow the updates from
-# Example: ALLOW_FROM = ["192.168.10.0/24", "::1/128"]
-ALLOW_FROM = []
-# Force re-registration. Overwrites the already existing acme-dns accounts.
-FORCE_REGISTER = False
-
-###   DO NOT EDIT BELOW THIS POINT   ###
-###         HERE BE DRAGONS          ###
-
-DOMAIN = os.environ["CERTBOT_DOMAIN"]
-if DOMAIN.startswith("*."):
-    DOMAIN = DOMAIN[2:]
-VALIDATION_DOMAIN = "_acme-challenge." + DOMAIN
-VALIDATION_TOKEN = os.environ["CERTBOT_VALIDATION"]
-
-
-class AcmeDnsClient(object):
-    """
-    Handles the communication with ACME-DNS API
-    """
-
-    def __init__(self, acmedns_url):
-        self.acmedns_url = acmedns_url
-
-    def register_account(self, allowfrom):
-        """Registers a new ACME-DNS account"""
-
-        if allowfrom:
-            # Include whitelisted networks to the registration call
-            reg_data = {"allowfrom": allowfrom}
-            res = requests.post(
-                self.acmedns_url + "/register", data=json.dumps(reg_data)
-            )
-        else:
-            res = requests.post(self.acmedns_url + "/register")
-        if res.status_code == 201:
-            # The request was successful
-            return res.json()
-        else:
-            # Encountered an error
-            msg = (
-                "Encountered an error while trying to register a new acme-dns "
-                "account. HTTP status {}, Response body: {}"
-            )
-            print(msg.format(res.status_code, res.text))
-            sys.exit(1)
-
-    def update_txt_record(self, account, txt):
-        """Updates the TXT challenge record to ACME-DNS subdomain."""
-        update = {"subdomain": account["subdomain"], "txt": txt}
-        headers = {
-            "X-Api-User": account["username"],
-            "X-Api-Key": account["password"],
-            "Content-Type": "application/json",
-        }
-        res = requests.post(
-            self.acmedns_url + "/update", headers=headers, data=json.dumps(update)
-        )
-        if res.status_code == 200:
-            # Successful update
-            return
-        else:
-            msg = (
-                "Encountered an error while trying to update TXT record in "
-                "acme-dns. \n"
-                "------- Request headers:\n{}\n"
-                "------- Request body:\n{}\n"
-                "------- Response HTTP status: {}\n"
-                "------- Response body: {}"
-            )
-            s_headers = json.dumps(headers, indent=2, sort_keys=True)
-            s_update = json.dumps(update, indent=2, sort_keys=True)
-            s_body = json.dumps(res.json(), indent=2, sort_keys=True)
-            print(msg.format(s_headers, s_update, res.status_code, s_body))
-            sys.exit(1)
-
-
-class Storage(object):
-    def __init__(self, storagepath):
-        self.storagepath = storagepath
-        self._data = self.load()
-
-    def load(self):
-        """Reads the storage content from the disk to a dict structure"""
-        data = dict()
-        filedata = ""
-        try:
-            with open(self.storagepath, "r") as fh:
-                filedata = fh.read()
-        except IOError as e:
-            if os.path.isfile(self.storagepath):
-                # Only error out if file exists, but cannot be read
-                print("ERROR: Storage file exists but cannot be read")
-                sys.exit(1)
-        try:
-            data = json.loads(filedata)
-        except ValueError:
-            if len(filedata) > 0:
-                # Storage file is corrupted
-                print("ERROR: Storage JSON is corrupted")
-                sys.exit(1)
-        return data
-
-    def save(self):
-        """Saves the storage content to disk"""
-        serialized = json.dumps(self._data)
-        try:
-            with os.fdopen(
-                os.open(self.storagepath, os.O_WRONLY | os.O_CREAT, 0o600), "w"
-            ) as fh:
-                fh.truncate()
-                fh.write(serialized)
-        except IOError as e:
-            print("ERROR: Could not write storage file.")
-            sys.exit(1)
-
-    def put(self, key, value):
-        """Puts the configuration value to storage and sanitize it"""
-        # If wildcard domain, remove the wildcard part as this will use the
-        # same validation record name as the base domain
-        if key.startswith("*."):
-            key = key[2:]
-        self._data[key] = value
-
-    def fetch(self, key):
-        """Gets configuration value from storage"""
-        try:
-            return self._data[key]
-        except KeyError:
-            return None
-
-
-if __name__ == "__main__":
-    # Init
-    client = AcmeDnsClient(ACMEDNS_URL)
-    storage = Storage(STORAGE_PATH)
-
-    # Check if an account already exists in storage
-    account = storage.fetch(DOMAIN)
-    if FORCE_REGISTER or not account:
-        # Create and save the new account
-        account = client.register_account(ALLOW_FROM)
-        storage.put(DOMAIN, account)
-        storage.save()
-
-        # Display the notification for the user to update the main zone
-        msg = "Please add the following CNAME record to your main DNS zone:\n{}"
-        cname = "{} CNAME {}.".format(VALIDATION_DOMAIN, account["fulldomain"])
-        print(msg.format(cname))
-
-    # Update the TXT record in acme-dns instance
-    client.update_txt_record(account, VALIDATION_TOKEN)

updater/utils.py

@@ -0,0 +1,21 @@
+import logging
+
+
+def record_transformer(public_suffixes):
+    def transform(record):
+        name = record["name"]
+        suffixes = [suffix for suffix in public_suffixes if name.endswith(suffix)]
+        suffix = suffixes[0] if suffixes else None
+
+        if suffix:
+            logging.debug(f"stripping {suffix} from {name} as it is a public suffix")
+
+            record["name"] = name[: -len(suffix)]
+            record["internal"] = "off"
+            return record
+
+        logging.debug(f"keeping {name} as it is not a public suffix")
+        record["internal"] = "on"
+        return record
+
+    return transform