Compare commits

..

No commits in common. "5b9c6c5a289f8d4c7c33e78befc2a76c2c24a3ff" and "870d0e90bd13a77b81798318b1d4d3e6c64899f1" have entirely different histories.

4 changed files with 3 additions and 278 deletions

6
flake.lock generated
View File

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

View File

@ -1,74 +0,0 @@
{ 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;
};
};
};
}

View File

@ -1,25 +0,0 @@
{ 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;
};
}

View File

@ -1,176 +0,0 @@
#!@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}")