62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
import urllib.request
|
|
import json
|
|
from typing import Optional, Any
|
|
|
|
from .base import Handler
|
|
from ..config import Config
|
|
|
|
class HetznerHandler(Handler):
|
|
# discovered
|
|
zone_id: str
|
|
nameservers: list[str]
|
|
|
|
def __init__(self, zone_name: str, config: Config, token: str) -> None:
|
|
super().__init__(zone_name, config, token)
|
|
self.secret = config.get_secret('hetzner')
|
|
|
|
zones = self.fetch('/zones')['zones']
|
|
for zone in zones:
|
|
if zone['name'] == zone_name:
|
|
self.zone_id = zone['id']
|
|
self.nameservers = zone['ns']
|
|
break
|
|
|
|
def fetch(self, url: str, data: Optional[Any] = None, **kwargs: str) -> Any:
|
|
req = urllib.request.Request('https://dns.hetzner.com/api/v1' + url)
|
|
req.add_header('Auth-API-Token', self.secret)
|
|
method = 'GET'
|
|
if data:
|
|
req.data = json.dumps(data).encode('utf-8')
|
|
req.add_header('Content-Type', 'application/json;charset=utf-8')
|
|
method = 'POST'
|
|
|
|
req.method = kwargs.get('method', method)
|
|
try:
|
|
with urllib.request.urlopen(req) as f:
|
|
return json.load(f)
|
|
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)['error'])
|
|
except Exception:
|
|
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('/records', {
|
|
'value': record_value,
|
|
'ttl': 300,
|
|
'type': 'TXT',
|
|
'name': record_name,
|
|
'zone_id': self.zone_id,
|
|
})
|
|
|
|
def remove(self, record_name: str) -> None:
|
|
records = self.fetch(f'/records?zone_id={self.zone_id}')['records']
|
|
for record in records:
|
|
if record['name'] == record_name:
|
|
self.log.info('removing %s', record_name)
|
|
self.fetch(f'/records/{record["id"]}', method='DELETE')
|