feat: add porkbun handler

nyacme/handlers/__init__.py

@@ -1,9 +1,11 @@
 from .cloudflare import CloudflareHandler
 from .hetzner import HetznerHandler
 from .http import HTTPHandler
+from .porkbun import PorkbunHandler
 
 __all__ = [
     'CloudflareHandler',
     'HetznerHandler',
-    'HTTPHandler'
+    'HTTPHandler',
+    'PorkbunHandler',
 ]

nyacme/handlers/porkbun.py

@@ -0,0 +1,61 @@
+import json
+import urllib.request
+from typing import Any, Optional
+
+from ..config import Config
+from .base import Handler
+
+
+class PorkbunHandler(Handler):
+    # discovered
+    nameservers: list[str]
+
+    def __init__(self, zone_name: str, config: Config, token: str) -> None:
+        super().__init__(zone_name, config, token)
+        self.apikey = config.get_secret('porkbun.apikey')
+        self.secretapikey = config.get_secret('porkbun.secretapikey')
+
+        self.nameservers = self.fetch(f'/domain/getNs/{self.zone}')['ns']
+
+    def fetch(self, url: str, data: Optional[dict[str, Any]] = None) -> Any:
+        req = urllib.request.Request('https://api.porkbun.com/api/json/v3' + url)
+        req.add_header('Auth-API-Token', self.secret)
+        if not data:
+            data = {}
+
+        data['apikey'] = self.apikey
+        data['secretapikey'] = self.secretapikey
+        req.data = json.dumps(data).encode('utf-8')
+        req.add_header('Content-Type', 'application/json;charset=utf-8')
+        req.method = 'POST'
+        try:
+            with urllib.request.urlopen(req) as f:
+                data = json.load(f)
+                if data['status'] != 'SUCCESS':
+                    raise Exception(data['message'])
+                return data
+        except urllib.error.HTTPError as ex:
+            self.log.error('cannot %s %s: %s', req.method, url, ex)
+            res = ex.fp.read().decode('utf-8')
+            try:
+                raise Exception(json.loads(res)['message'])
+            except json.JSONDecodeError:
+                raise Exception(res)
+
+    def create(self, record_name: str, record_value: str) -> None:
+        self.remove(record_name)
+        self.log.info('creating %s with value %s', record_name, record_value)
+        self.fetch(f'/dns/create/{self.zone}', {
+            'name': record_name,
+            'type': 'TXT',
+            'content': record_value,
+            'ttl': 300,
+        })
+
+    def remove(self, record_name: str) -> None:
+        full_record_name = record_name + '.' + self.zone
+        records = self.fetch(f'/dns/retrieve/{self.zone}')['records']
+        for record in records:
+            if record['name'] == full_record_name:
+                self.log.info('removing %s', full_record_name)
+                self.fetch(f'/dns/delete/{self.zone}/{record['id']}')

nyacme/hook.py

@@ -6,7 +6,7 @@ from itertools import chain
 import dns.resolver
 
 from .config import read_config
-from .handlers import CloudflareHandler, HetznerHandler, HTTPHandler
+from .handlers import CloudflareHandler, HetznerHandler, HTTPHandler, PorkbunHandler
 
 logging.basicConfig(level=logging.INFO, format='> [%(levelname)s] %(name)s: %(message)s')
 log = logging.getLogger('nyacme_hook')
@@ -16,6 +16,7 @@ handlers = {
     'cloudflare': CloudflareHandler,
     'hetzner': HetznerHandler,
     'http': HTTPHandler,
+    'porkbun': PorkbunHandler,
 }
 
 class Args: