path: root/simpleddns.py

#!/usr/bin/env python3
'''Simple Dynamic DNS updater'''

import os
import sys
import subprocess
import argparse
import ast
import dataclasses
from time import sleep
from abc import ABC, abstractmethod
from typing import NoReturn, Optional, Any
from pathlib import Path
import requests

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''')
	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')
	return parser.parse_args()

def fatal_error(message: str) -> NoReturn:
	'''Output error message & exit'''
	sys.stderr.write(f'Fatal error: {message}\n')
	sys.exit(1)

def warn(message: str) -> None:
	'''Print warning message'''
	sys.stderr.write(f'WARNING: {message}\n')

def setup_config(config_path: Path) -> None:
	'''Interactively set up configuration file'''
	if config_path.exists():
		should_delete = input('Configuration already exists. Delete it [y/n]? ')
		if not should_delete.strip().lower().startswith('y'):
			print('Aborting.')
			return
		config_path.unlink()
	default_get_ip = 'curl --no-progress-meter ifconfig.co'
	domain_name = input('Domain name? ').strip()
	if not domain_name:
		fatal_error('Domain name must be set')
	get_ip = input(f'Command for getting IP address [default: {default_get_ip}]? ').strip()
	if not get_ip:
		get_ip = default_get_ip
	if any(c.isspace() for c in domain_name):
		fatal_error('Domain name must not contain any whitespace')
	print('1. Linode Domains')
	# TODO: AWS
	domain_type = input('Select a domain type from the list above [1-1]: ').strip()
	if domain_type == '1':
		access_token = input('Enter personal access token: ')
		options = f'''# Personal access token (secret!!)
    access_token = {repr(access_token)}'''
		type_name = 'linode'
	else:
		print('Invalid choice')
		return
	fd = os.open(config_path, os.O_CREAT | os.O_WRONLY | os.O_EXCL, 0o600)
	with os.fdopen(fd, 'w') as config:
		config.write(f'''Settings
    # Interval in seconds between checking IP address for changes
    #  (API calls are only made when it changes)
    interval = 15
    # Timeout to wait for API responses
    timeout = 20
Domain {type_name} {domain_name}
    # Command(s) for getting IP address(es)
    getip = [{repr(get_ip)}]
    {options}
''')
	print(f'Configuration created successfully at {config_path}')

@dataclasses.dataclass
class Settings:
	'''Global (i.e. not per-domain) settings'''
	dry_run: bool = False
	interval: int = 15
	timeout: int = 20

class Domain(ABC):
	'''A domain whose DNS will be updated'''
	full_domain: str
	root_domain: str
	subdomain: str
	getip: list[str]
	settings: Settings
	last_ips: list[str]
	def __init__(self, settings: Settings, domain: str):
		self.settings = settings
		self.full_domain = domain
		last_dot = domain.rfind('.')
		if last_dot == -1:
			raise ValueError(f'Domain {domain} has no . in it')
		second_last_dot = domain.rfind('.', 0, last_dot)
		if second_last_dot == -1:
			self.subdomain = ''
			self.root_domain = domain
		else:
			self.subdomain = domain[:second_last_dot]
			self.root_domain = domain[second_last_dot+1:]
		self.last_ips = []
		self._init()
	def _info(self, message: str) -> None:
		'''Print informational message prefixed with domain name'''
		print(f'{self.full_domain}: {message}')
	@abstractmethod
	def _init(self) -> None:
		'''Extra provider-specific initialization'''
	@abstractmethod
	def update(self, ips: list[str]) -> bool:
		'''Make necessary API request to update domain to use new IP addresses.
		   Returns False on failure.'''
	@abstractmethod
	def validate_specifics(self) -> str:
		'''Validate provider-specific settings for domain'''
	def validate(self) -> str:
		'''Validate this object (ensure all required settings are set)'''
		if not getattr(self, 'getip', ''):
			return 'getip not set'
		if not isinstance(self.getip, list):
			return 'getip should be a list.'
		return self.validate_specifics()
	def get_ips(self) -> list[str]:
		'''Get IP addresses using the registered commands.'''
		ips = []
		for cmd in self.getip:
			result = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, check=False)
			if result.returncode:
				warn(f'{self.getip} failed (exit code {result.returncode})')
				return []
			ip = result.stdout.decode(errors='replace').strip()
			# this test isn't perfect, but it should catch most cases of weird stuff
			if not ip or '..' in ip or any(c not in '0123456789abcdefABCDEF:.' for c in ip):
				warn(f'IP address {repr(ip)} is invalid')
				return []
			ips.append(ip)
		ips.sort()
		return ips
	def check_for_update(self) -> bool:
		'''Update DNS records if IP has changed. Returns False on error.'''
		ips = self.get_ips()
		if ips == self.last_ips:
			return True
		self._info('Dealing with new IP address(es)... ')
		sys.stdout.flush()
		if self.update(ips):
			self.last_ips = ips
			return True
		return False

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:
		self._id = None
	def __repr__(self) -> str:
		return f'<LinodeDomain domain={self.full_domain} ' \
			'getip={repr(self.getip)} ' \
			'access_token={repr(self.access_token)}>'
	def _headers(self, has_body: bool = False) -> dict[str, str]:
		'''Get HTTP headers for making requests'''
		headers = {'Accept': 'application/json', 'Authorization': f'Bearer {self.access_token}'}
		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:
		headers = self._headers(has_body = body is not None)
		options: dict[Any, Any] = {
			'headers': headers,
			'timeout': self.settings.timeout,
			'method': method,
			'url': url
		}
		if body is not None:
			options['json'] = body
		try:
			response = requests.request(**options)
			if not response.ok:
				self._error(f'''Got error response from endpoint {repr(url)} (code {response.status_code}):
{response.text}''')
			response_json = response.json()
			if response_json is None:
				self._error(f'Got null response from endpoint {repr(url)}')
			return response_json
		except requests.JSONDecodeError as e:
			self._error(f'Invalid JSON at endpoint {repr(url)}: {e}')
			return None
		except requests.RequestException as e:
			self._error(f'Error making request to {repr(url)}: {e}')
			return None
	def _paginated_get(self, url: str) -> list[Any]:
		'''Get results for a Linode paginated GET API endpoint.'''
		results = []
		page_size = 500
		for page in range(1, 100):
			page_url = f'{url}{"&" if "?" in url else "?"}page={page}&page_size={page_size}'
			sleep(0.33) # should prevent us from hitting Linode's rate limit
			response_json = self._make_request('GET', page_url)
			if response_json is None:
				break
			try:
				response_data = response_json['data']
				if not isinstance(response_data, list):
					raise ValueError('"data" member is not a list')
			except (KeyError, ValueError) as e:
				self._error(f'Bad JSON format at endpoint {repr(page_url)}: {e}')
				break
			results.extend(response_data)
			if len(response_data) < page_size:
				# Reached last page, presumably
				break
			if page == 99:
				self._error(f'Giving up after 99 pages of responses to API endpoint {repr(url)}')
		return results
	def update(self, ips: list[str]) -> bool:
		self._had_error = False
		if self._id is None:
			domains = self._paginated_get('https://api.linode.com/v4/domains')
			if self._had_error:
				return False
			for domain in domains:
				domain_name = domain['domain']
				if self.root_domain == domain_name:
					# this is it!
					self._id = domain['id']
			if self._id is None:
				warn(f'Domain {self.root_domain} not found in Linode. Are you sure it is set up there?')
				return False
		records = self._paginated_get(f'https://api.linode.com/v4/domains/{self._id}/records')
		remaining_ips = set(ips)
		unused_records: dict[str, list[int]] = {'A': [], 'AAAA': []}
		for record in records:
			if record['type'] not in ['A', 'AAAA']:
				continue
			if record['name'] != self.subdomain:
				continue
			if record['target'] in remaining_ips:
				# We're all good
				remaining_ips.remove(record['target'])
				continue
			unused_records[record['type']].append(record['id'])
		could_update = set()
		# Update existing records if possible (to save on API calls)
		for ip in remaining_ips:
			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}')
				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}')
		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}')
		return not self._had_error
	def validate_specifics(self) -> str:
		if not getattr(self, 'access_token', ''):
			return 'Access token not set'
		# OK
		return ''

def parse_config(config_path: Path) -> tuple[Settings, list[Domain]]:
	'''Parse configuration file'''
	curr_section: Settings | Domain | None = None
	settings = Settings()
	domains: list[Domain] = []
	with open(config_path, encoding='utf-8') as config:
		for line in config:
			line = line.strip()
			if line[0] == '#':
				# comment
				continue
			if line.startswith('Settings'):
				if isinstance(curr_section, Domain):
					domains.append(curr_section)
				curr_section = settings
			elif line.startswith('Domain '):
				if isinstance(curr_section, Domain):
					domains.append(curr_section)
				parts = line.split()
				if len(parts) != 3:
					fatal_error('Domain declaration should have exactly ' \
						'two space-separated arguments (type and name)')
				kind = parts[1]
				domain_name = parts[2]
				if kind == 'linode':
					curr_section = LinodeDomain(settings, domain_name)
				else:
					fatal_error(f'No such domain type: {kind}')
			else:
				if not curr_section:
					fatal_error('First non-empty line of configuration must be a Domain declaration or Settings')
				parts = [part.strip() for part in line.split('=', maxsplit=1)]
				if len(parts) != 2:
					fatal_error(f'Invalid syntax (want key = value): {line}')
				[key, value] = parts
				setattr(curr_section, key, ast.literal_eval(value))
	if isinstance(curr_section, Domain):
		domains.append(curr_section)
	for domain in domains:
		err = domain.validate()
		if err:
			fatal_error(f'In domain {domain.full_domain}: {err}')
	return (settings, domains)

def main() -> None:
	'''Run simpleddns'''
	args = parse_args()
	simpleddns_config_dir = os.getenv('SIMPLEDDNS_CONFIG_DIR')
	xdg_config_dir = os.getenv('XDG_CONFIG_DIR')
	if simpleddns_config_dir:
		config_dir = Path(simpleddns_config_dir)
	elif xdg_config_dir:
		config_dir = Path(xdg_config_dir, 'simpleddns')
	else:
		config_dir = Path('~/.config/simpleddns').expanduser()
	os.makedirs(config_dir, exist_ok = True)
	config_path = Path(config_dir, 'config')
	if args.setup:
		setup_config(config_path)
		sys.exit(0)
	if not config_path.exists():
		fatal_error("Configuration doesn't exist. Try running with --setup first.")
	if os.name == 'posix':
		config_mode = config_path.stat().st_mode
		if config_mode & 0o4:
			fatal_error(f'''DANGER: configuration file {config_path} allows
	reading by other users (mode {config_mode & 0o777:o}).
	If there are API tokens in there, revoke them immediately!!''')

	settings, domains = parse_config(config_path)
	settings.dry_run = args.dry_run
	if not domains:
		fatal_error('No domains defined. Try running with --setup?')
	while True:
		print('Checking if IP changed... ', end='')
		sys.stdout.flush()
		all_ok = True
		for domain in domains:
			all_ok = domain.check_for_update() and all_ok
		if all_ok:
			print('All OK')
		sleep(settings.interval)

if __name__ == '__main__':
	main()