From 273d1906e1d2f8a0dcab024b53adc53f4af2dbb3 Mon Sep 17 00:00:00 2001 From: xeals Date: Thu, 9 Nov 2023 09:36:40 +1100 Subject: [PATCH] porkbun-ddns: init --- pkgs/by-name/po/porkbun-ddns/package.nix | 25 +++ pkgs/by-name/po/porkbun-ddns/porkbun-ddns.py | 176 +++++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 pkgs/by-name/po/porkbun-ddns/package.nix create mode 100644 pkgs/by-name/po/porkbun-ddns/porkbun-ddns.py diff --git a/pkgs/by-name/po/porkbun-ddns/package.nix b/pkgs/by-name/po/porkbun-ddns/package.nix new file mode 100644 index 0000000..ef48900 --- /dev/null +++ b/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; + }; +} diff --git a/pkgs/by-name/po/porkbun-ddns/porkbun-ddns.py b/pkgs/by-name/po/porkbun-ddns/porkbun-ddns.py new file mode 100644 index 0000000..a343562 --- /dev/null +++ b/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}")