From e458921c8ab0713bbdcf824ee0b3279765b46fcb Mon Sep 17 00:00:00 2001 From: pommicket Date: Sun, 31 Aug 2025 12:15:18 -0400 Subject: Linode domain stuff is all working --- simpleddns.py | 117 ++++++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 86 insertions(+), 31 deletions(-) (limited to 'simpleddns.py') diff --git a/simpleddns.py b/simpleddns.py index 0fd5679..3cc3b32 100755 --- a/simpleddns.py +++ b/simpleddns.py @@ -13,16 +13,25 @@ from typing import NoReturn, Optional, Any from pathlib import Path import requests +VERSION = '0.1.0' + def parse_args() -> argparse.Namespace: '''Parse command-line arguments''' parser = argparse.ArgumentParser(prog='simpleddns', description='Simple dynamic DNS updater', - epilog='''ENVIRONMENT VARIABLES - SIMPLEDDNS_CONFIG_DIR - directory where configburation files are stored''') + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog='''configuration: + Configure SimpleDDNS by running with --setup or editing the file + ~/.config/simpleddns/config (or $SIMPLEDDNS_CONFIG_DIR/config). + There you should find comments documenting the various options. + +environment variables: + SIMPLEDDNS_CONFIG_DIR - directory where configburation files are stored''') parser.add_argument('--setup', help='Set up configuration', action='store_true') parser.add_argument('--dry-run', help='Print what API calls would be made, without actually making any non-GET calls.', action='store_true') + parser.add_argument('--version', action='version', version=f'%(prog)s {VERSION}') return parser.parse_args() def fatal_error(message: str) -> NoReturn: @@ -70,6 +79,13 @@ def setup_config(config_path: Path) -> None: interval = 15 # Timeout to wait for API responses timeout = 20 + # Time To Live (in seconds) with which to create DNS records + ttl = 300 + # Delay between API calls (seconds), to avoid being rate limited. + # 0.4 seconds should be more than enough for both Linode and AWS, + # but you may want to set this higher if you have other things + # making frequent API calls. + request_delay = 0.4 Domain {type_name} {domain_name} # Command(s) for getting IP address(es) getip = [{repr(get_ip)}] @@ -83,6 +99,8 @@ class Settings: dry_run: bool = False interval: int = 15 timeout: int = 20 + ttl: int = 300 + request_delay: float = 0.4 class Domain(ABC): '''A domain whose DNS will be updated''' @@ -92,6 +110,7 @@ class Domain(ABC): getip: list[str] settings: Settings last_ips: list[str] + _had_error: bool def __init__(self, settings: Settings, domain: str): self.settings = settings self.full_domain = domain @@ -110,6 +129,19 @@ class Domain(ABC): def _info(self, message: str) -> None: '''Print informational message prefixed with domain name''' print(f'{self.full_domain}: {message}') + def _error(self, message: str) -> None: + '''Print warning and set _had_error field to True.''' + self._had_error = True + warn(message) + def _ttl(self) -> int: + '''Get TTL value to use for records for this domain''' + return getattr(self, 'ttl', 0) or self.settings.ttl + def _timeout(self) -> int: + '''Get API request timeout''' + return getattr(self, 'timeout', 0) or self.settings.timeout + def _request_delay(self) -> float: + '''Get delay between requests (seconds)''' + return getattr(self, 'request_delay', 0) or self.settings.request_delay @abstractmethod def _init(self) -> None: '''Extra provider-specific initialization''' @@ -158,7 +190,6 @@ class Domain(ABC): class LinodeDomain(Domain): '''Domain registered with Linode Domains''' access_token: str - _had_error: bool # Domain ID for Linode API _id: Optional[int] def _init(self) -> None: @@ -173,15 +204,12 @@ class LinodeDomain(Domain): if has_body: headers['Content-Type'] = 'application/json' return headers - def _error(self, message: str) -> None: - self._had_error = True - warn(message) def _make_request(self, method: str, url: str, body: Any = None) -> Any: - sleep(0.33) # should prevent us from hitting Linode's rate limit + sleep(self._request_delay()) headers = self._headers(has_body = body is not None) options: dict[Any, Any] = { 'headers': headers, - 'timeout': self.settings.timeout, + 'timeout': self._timeout(), 'method': method, 'url': url } @@ -225,6 +253,50 @@ class LinodeDomain(Domain): if page == 99: self._error(f'Giving up after 99 pages of responses to API endpoint {repr(url)}') return results + + def _update_record(self, record_id: int, target: str) -> None: + '''Update A/AAAA record to point to new IP address.''' + if self.settings.dry_run: + self._info(f'Update DNS record {record_id} (ttl = {self._ttl()}, target = {target})') + else: + payload = { + 'name': self.subdomain, + 'target': target, + 'ttl_sec': self._ttl() + } + url = f'https://api.linode.com/v4/domains/{self._id}/records/{record_id}' + resp = self._make_request('PUT', url, payload) + if isinstance(resp, dict): + self._info(f'Successfully updated record {record_id} ' \ + f'with TTL = {self._ttl()}, target = {target}') + + def _delete_record(self, record_id: int) -> None: + '''Delete DNS record''' + if self.settings.dry_run: + self._info(f'Delete DNS record {record_id}') + else: + url = f'https://api.linode.com/v4/domains/{self._id}/records/{record_id}' + resp = self._make_request('DELETE', url) + if isinstance(resp, dict): + self._info(f'Successfully deleted record {record_id}') + + def _create_record(self, target: str) -> None: + '''Create A/AAAA record for target IP address''' + kind = 'AAAA' if ':' in target else 'A' + if self.settings.dry_run: + self._info(f'Add {kind} DNS record (ttl = {self._ttl()}, target = {target})') + else: + payload = { + 'type': kind, + 'name': self.subdomain, + 'target': target, + 'ttl_sec': self._ttl(), + } + url = f'https://api.linode.com/v4/domains/{self._id}/records' + resp = self._make_request('POST', url, payload) + if isinstance(resp, dict): + self._info(f'Successfully created {kind} record with TTL = {self._ttl()}, target = {target}') + def update(self, ips: list[str]) -> bool: self._had_error = False if self._id is None: @@ -247,8 +319,8 @@ class LinodeDomain(Domain): continue if record['name'] != self.subdomain: continue - if record['target'] in remaining_ips: - # We're all good + if record['target'] in remaining_ips and record['ttl_sec'] == self._ttl(): + # This record covers an IP address we want. remaining_ips.remove(record['target']) continue unused_records[record['type']].append(record['id']) @@ -261,32 +333,15 @@ class LinodeDomain(Domain): kind = 'AAAA' if ':' in ip else 'A' if unused_records[kind]: record_id = unused_records[kind].pop() - if self.settings.dry_run: - self._info(f'Update DNS record {record_id} to point to {ip}') - else: - print(f'TODO: update DNS record {record_id} to point to {ip}') + self._update_record(record_id, ip) could_update.add(ip) remaining_ips -= could_update for record_id in (x for rs in unused_records.values() for x in rs): - if self.settings.dry_run: - self._info(f'Delete DNS record {record_id}') - else: - print(f'TODO: delete DNS record {record_id}') + self._delete_record(record_id) for ip in remaining_ips: - kind = 'AAAA' if ':' in ip else 'A' - if self.settings.dry_run: - self._info(f'Add {kind} DNS record pointing to {ip}') - else: - payload = { - 'type': kind, - 'name': self.subdomain, - 'target': ip, - 'ttl': 300, # TODO: make this configurable - } - self._make_request('POST', f'https://api.linode.com/v4/domains/{self._id}/records', payload) - if not self._had_error: - self._info(f'Successfully created {kind} record pointing to {ip}') + self._create_record(ip) return not self._had_error + def validate_specifics(self) -> str: if not getattr(self, 'access_token', ''): return 'Access token not set' -- cgit v1.2.3