modules/porkbun-ddns: init

Flake lock file updates:

• Updated input 'nixpkgs':
    'github:NixOS/nixpkgs/90e85bc7c1a6fc0760a94ace129d3a1c61c3d035' (2023-10-29)
  → 'github:NixOS/nixpkgs/a0b3b06b7a82c965ae0bb1d59f6e386fe755001d' (2023-11-05)

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

flake.lock

@@ -20,11 +20,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1697915759,
-        "narHash": "sha256-WyMj5jGcecD+KC8gEs+wFth1J1wjisZf8kVZH13f1Zo=",
+        "lastModified": 1699186365,
+        "narHash": "sha256-Pxrw5U8mBsL3NlrJ6q1KK1crzvSUcdfwb9083sKDrcU=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "51d906d2341c9e866e48c2efcaac0f2d70bfd43e",
+        "rev": "a0b3b06b7a82c965ae0bb1d59f6e386fe755001d",
         "type": "github"
       },
       "original": {

modules/services/networking/porkbun-ddns.nix

@@ -0,0 +1,74 @@
+{ config, lib, pkgs, ... }:
+let
+  inherit (lib) mkOption types;
+
+  cfg = config.services.porkbun-ddns;
+in
+{
+  options = {
+    services.porkbun-ddns = {
+      enable = lib.mkEnableOption "Porkbun dynamic DNS client";
+
+      package = mkOption {
+        # TODO: How do I use mkPackageOption when the package isn't in the
+        #  package set?
+        type = types.package;
+        default = (import ../../..).porkbun-ddns;
+        defaultText = "pkgs.porkbun-ddns";
+        description = lib.mdDoc "The porkbun-ddns package to use.";
+      };
+
+      interval = mkOption {
+        type = types.str;
+        default = "10m";
+        default = lib.mdDoc ''
+          Interval to update dynamic DNS records. The default is to update every
+          10 minutes. The format is described in {manpage}`systemd.time(7)`.
+        '';
+      };
+
+      domains = mkOption {
+        type = types.listOf types.str;
+        default = [ ];
+        description = lib.mdDoc "Domains to update.";
+      };
+
+      apiKeyFile = mkOption {
+        type = types.nullOr types.path;
+        description = lib.mdDoc ''
+          File containing the API key to use when running the client.
+        '';
+      };
+
+      secretApiKeyFile = mkOption {
+        type = types.nullOr types.path;
+        description = lib.mdDoc ''
+          File containing the secret API key to use when running the
+          client.
+        '';
+      };
+    };
+  };
+
+  config = lib.mkIf cfg.enable {
+    systemd.services.porkbun-ddns = {
+      description = "Porkbun dynamic DNS client";
+      script = ''
+        ${cfg.package}/bin/porkbun-ddns \
+          -K ${cfg.apiKeyFile} \
+          -S ${cfg.secretApiKeyFile} \
+          ${lib.concatStringsSep " " cfg.domains}
+      '';
+    };
+
+    systemd.timers.porkbun-ddns = {
+      description = "Porkbun dynamic DNS client";
+      wants = [ "network-online.target" ];
+      wantedBy = [ "timers.target" ];
+      timerConfig = {
+        OnBootSec = cfg.interval;
+        OnUnitActiveSec = cfg.interval;
+      };
+    };
+  };
+}

pkgs/by-name/po/porkbun-ddns/package.nix

@@ -0,0 +1,25 @@
+{ lib
+, stdenv
+, python3
+}:
+let
+  python = python3.withPackages (py: [ py.requests ]);
+in
+stdenv.mkDerivation {
+  name = "porkbun-ddns";
+
+  src = ./.;
+  inherit python;
+
+  installPhase = ''
+    mkdir -p $out/bin
+    install -Dm0755 $src/porkbun-ddns.py $out/bin/porkbun-ddns
+    substituteAllInPlace $out/bin/porkbun-ddns
+  '';
+
+  meta = {
+    description = "Porkbun dynamic DNS script";
+    license = lib.licenses.gpl3;
+    platforms = python.meta.platforms;
+  };
+}

pkgs/by-name/po/porkbun-ddns/porkbun-ddns.py

@@ -0,0 +1,176 @@
+#!@python@/bin/python
+
+import argparse
+import json
+import os
+import re
+import requests
+from dataclasses import dataclass, fields as datafields
+from enum import Enum, unique
+from typing import List, Optional
+
+APIBASE = "https://porkbun.com/api/json/v3/dns"
+
+
+def dataclass_from_dict(klass: object, d: dict):
+    try:
+        fieldtypes = {f.name: f.type for f in datafields(klass)}
+        return klass(**{f: dataclass_from_dict(fieldtypes[f], d[f]) for f in d})
+    except:
+        return d  # Not a dataclass field
+
+
+def remove_domain(domain: str, name: str):
+    return re.sub(f"\\.?{domain}$", "", name)
+
+
+@unique
+class RecordType(Enum):
+    a = "A"
+    aaaa = "AAAA"
+    cname = "CNAME"
+    mx = "MX"
+    srv = "SRV"
+    txt = "TXT"
+
+
+@dataclass
+class Record:
+    id: str
+    name: str
+    type: str
+    content: str
+    ttl: str
+    prio: str = ""
+    notes: str = ""
+
+
+@dataclass
+class Retrieval:
+    status: str
+    records: List[Record]
+
+
+class ApiError(Exception):
+    pass
+
+
+class ArgumentError(Exception):
+    pass
+
+
+class PorkbunClient:
+    def __init__(self, apikey: str, secretapikey: str):
+        self.apikey = apikey
+        self.secretapikey = secretapikey
+
+    def _make_payload(self, **kwargs):
+        return json.dumps(
+            {"apikey": self.apikey, "secretapikey": self.secretapikey, **kwargs}
+        )
+
+    def edit_record(
+        self,
+        domain: str,
+        record: Record,
+        name: Optional[str] = None,
+        type: Optional[RecordType] = None,
+        content: Optional[str] = None,
+        ttl: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> bool:
+        return self.edit(
+            domain,
+            record.id,
+            name=name or record.name,
+            type=type or RecordType(record.type),
+            content=content or record.content,
+            ttl=ttl or record.ttl,
+            priority=priority or record.prio,
+        )
+
+    def edit(
+        self,
+        domain: str,
+        id: str,
+        name: str,
+        type: RecordType,
+        content: str,
+        ttl: int = 300,
+        priority: Optional[str] = None,
+    ) -> bool:
+        # API returns FQN name rather than the actual prefix, so scrub it
+        name = remove_domain(domain, name)
+        payload = self._make_payload(
+            name=name, type=type.value, content=content, ttl=str(ttl), prio=priority
+        )
+        res = requests.post(f"{APIBASE}/edit/{domain}/{id}", data=payload)
+        body = res.json()
+        if body["status"] != "SUCCESS":
+            raise ApiError(body["message"])
+        return True
+
+    def delete(self, domain: str, id: str) -> bool:
+        payload = self._make_payload()
+        res = requests.post(f"{APIBASE}/delete/{domain}/{id}", data=payload)
+        body = res.json()
+        if body["status"] != "SUCCESS":
+            raise ApiError(body["message"])
+        return True
+
+    def retrieve(self, domain: str) -> List[Retrieval]:
+        payload = self._make_payload()
+        res = requests.post(f"{APIBASE}/retrieve/{domain}", data=payload)
+        body = res.json()
+        if body["status"] != "SUCCESS":
+            raise ApiError(body["message"])
+        return [dataclass_from_dict(Record, d) for d in body["records"]]
+
+
+def current_ip() -> str:
+    return requests.get("https://ifconfig.me").text
+
+
+def _load_key(key: Optional[str], keyfile: Optional[str]) -> str:
+    if keyfile is not None:
+        with open(keyfile) as f:
+            return f.read().strip()
+    if key is not None:
+        return key
+    raise ArgumentError("key or key file is required")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Wrapper around Porkbun DNS API")
+    keyarg = parser.add_mutually_exclusive_group(required=True)
+    keyarg.add_argument("-k", "--key", metavar="KEY", type=str, help="API key")
+    keyarg.add_argument(
+        "-K", "--key-file", metavar="FILE", type=str, help="API key file"
+    )
+    secretarg = parser.add_mutually_exclusive_group(required=True)
+    secretarg.add_argument(
+        "-s", "--secret", metavar="SECRET", type=str, help="secret API key"
+    )
+    secretarg.add_argument(
+        "-S", "--secret-file", metavar="FILE", type=str, help="secret API key file"
+    )
+    parser.add_argument("domains", type=str, nargs="+", help="domain(s) to update")
+
+    args = parser.parse_args()
+    try:
+        apikey = _load_key(args.key, args.key_file)
+        secretapikey = _load_key(args.secret, args.secret_file)
+    except Exception as e:
+        print("error: " + str(e))
+        parser.print_help()
+        exit(1)
+
+    current_ip = current_ip()
+    client = PorkbunClient(apikey, secretapikey)
+    for domain in args.domains:
+        recs = client.retrieve(domain)
+        arecs = [r for r in recs if r.type == RecordType.a.value]
+        for arec in arecs:
+            if arec.content != current_ip:
+                client.edit_record(domain, arec, content=current_ip)
+                print(f"Pointed '{arec.name}' to {current_ip}")