161 lines
6.3 KiB
Python
161 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any
|
|
from urllib import parse, request, error
|
|
|
|
|
|
def api_request(base_url: str, method: str, path: str, *, token: str | None = None, query: dict | None = None) -> tuple[int, Any]:
|
|
url = base_url.rstrip('/') + path
|
|
if query:
|
|
url += '?' + parse.urlencode(query, doseq=True)
|
|
headers = {'Accept': 'application/json'}
|
|
if token:
|
|
headers['Authorization'] = f'Bearer {token}'
|
|
req = request.Request(url, headers=headers, method=method)
|
|
try:
|
|
with request.urlopen(req, timeout=30) as resp:
|
|
return resp.status, json.load(resp)
|
|
except error.HTTPError as exc:
|
|
body = exc.read().decode('utf-8', 'replace')
|
|
try:
|
|
payload = json.loads(body)
|
|
except Exception:
|
|
payload = body
|
|
return exc.code, payload
|
|
|
|
|
|
def expect_ok(status, payload, context):
|
|
if status != 200 or payload.get('status') != 'ok':
|
|
print(f'{context} failed: http={status}', file=sys.stderr)
|
|
print(json.dumps(payload, indent=2, sort_keys=True), file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def bool_s(v: bool) -> str:
|
|
return 'true' if v else 'false'
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description='Idempotent Technitium baseline config for the homelab pilot')
|
|
ap.add_argument('--base-url', default='http://127.0.0.1:5380')
|
|
ap.add_argument('--admin-user', default='admin')
|
|
ap.add_argument('--admin-pass', default=os.environ.get('DNS_SERVER_ADMIN_PASSWORD'))
|
|
ap.add_argument('--server-domain', default='dns.home.paccoco.com')
|
|
ap.add_argument('--zone', default='home.paccoco.com')
|
|
ap.add_argument('--record-host', default='dns.home.paccoco.com')
|
|
ap.add_argument('--record-ip', default='10.5.30.8')
|
|
ap.add_argument('--web-port', type=int, default=80)
|
|
ap.add_argument('--blocklist', action='append', default=None)
|
|
args = ap.parse_args()
|
|
|
|
if not args.admin_pass:
|
|
print('Missing admin password: pass --admin-pass or set DNS_SERVER_ADMIN_PASSWORD', file=sys.stderr)
|
|
return 1
|
|
|
|
desired_blocklists = args.blocklist or ['https://big.oisd.nl/']
|
|
|
|
st, login = api_request(args.base_url, 'GET', '/api/user/login', query={
|
|
'user': args.admin_user,
|
|
'pass': args.admin_pass,
|
|
'includeInfo': 'true',
|
|
})
|
|
expect_ok(st, login, 'login')
|
|
token = login['token']
|
|
|
|
st, settings = api_request(args.base_url, 'GET', '/api/settings/get', token=token)
|
|
expect_ok(st, settings, 'settings/get')
|
|
resp = settings['response']
|
|
|
|
query = {
|
|
'dnsServerDomain': args.server_domain,
|
|
'webServiceLocalAddresses': ','.join(resp.get('webServiceLocalAddresses') or ['[::]']),
|
|
'webServiceHttpPort': str(args.web_port),
|
|
'webServiceEnableTls': bool_s(False),
|
|
'webServiceTlsPort': str(resp.get('webServiceTlsPort', 53443)),
|
|
'webServiceHttpToTlsRedirect': bool_s(False),
|
|
'webServiceUseSelfSignedTlsCertificate': bool_s(False),
|
|
'preferIPv6': bool_s(bool(resp.get('preferIPv6', False))),
|
|
'dnssecValidation': bool_s(True),
|
|
'enableBlocking': bool_s(True),
|
|
'allowTxtBlockingReport': bool_s(True),
|
|
'blockListUrls': ','.join(desired_blocklists),
|
|
'blockListUpdateIntervalHours': str(resp.get('blockListUpdateIntervalHours', 24)),
|
|
'logQueries': bool_s(False),
|
|
'allowRecursion': bool_s(True),
|
|
'allowRecursionOnlyForPrivateNetworks': bool_s(True),
|
|
'qnameMinimization': bool_s(True),
|
|
'serveStale': bool_s(True),
|
|
}
|
|
|
|
st, updated = api_request(args.base_url, 'GET', '/api/settings/set', token=token, query=query)
|
|
expect_ok(st, updated, 'settings/set')
|
|
|
|
verify_base_url = args.base_url
|
|
parsed_base = parse.urlparse(args.base_url)
|
|
if parsed_base.port and parsed_base.port != args.web_port:
|
|
host = parsed_base.hostname or '127.0.0.1'
|
|
verify_base_url = f'{parsed_base.scheme}://{host}:{args.web_port}'
|
|
|
|
st, zones = api_request(verify_base_url, 'GET', '/api/zones/list', token=token)
|
|
expect_ok(st, zones, 'zones/list')
|
|
zone_names = {z['name'] for z in zones['response']['zones']}
|
|
if args.zone not in zone_names:
|
|
st, created = api_request(verify_base_url, 'GET', '/api/zones/create', token=token, query={
|
|
'zone': args.zone,
|
|
'type': 'Primary',
|
|
})
|
|
expect_ok(st, created, 'zones/create')
|
|
|
|
st, current_records = api_request(verify_base_url, 'GET', '/api/zones/records/get', token=token, query={
|
|
'domain': args.record_host,
|
|
'zone': args.zone,
|
|
'listZone': 'false',
|
|
})
|
|
expect_ok(st, current_records, 'zones/records/get')
|
|
for record in current_records['response'].get('records', []):
|
|
if record.get('type') == 'A':
|
|
ip = (record.get('rData') or {}).get('ipAddress')
|
|
if ip and ip != args.record_ip:
|
|
st, deleted = api_request(verify_base_url, 'GET', '/api/zones/records/delete', token=token, query={
|
|
'domain': args.record_host,
|
|
'zone': args.zone,
|
|
'type': 'A',
|
|
'ipAddress': ip,
|
|
})
|
|
expect_ok(st, deleted, 'zones/records/delete')
|
|
|
|
st, added = api_request(verify_base_url, 'GET', '/api/zones/records/add', token=token, query={
|
|
'domain': args.record_host,
|
|
'zone': args.zone,
|
|
'type': 'A',
|
|
'ipAddress': args.record_ip,
|
|
'overwrite': 'true',
|
|
'ttl': '300',
|
|
})
|
|
expect_ok(st, added, 'zones/records/add')
|
|
|
|
st, verify = api_request(verify_base_url, 'GET', '/api/settings/get', token=token)
|
|
expect_ok(st, verify, 'settings/get verify')
|
|
v = verify['response']
|
|
summary = {
|
|
'dnsServerDomain': v.get('dnsServerDomain'),
|
|
'webServiceHttpPort': v.get('webServiceHttpPort'),
|
|
'webServiceEnableTls': v.get('webServiceEnableTls'),
|
|
'recursion': v.get('recursion'),
|
|
'dnssecValidation': v.get('dnssecValidation'),
|
|
'enableBlocking': v.get('enableBlocking'),
|
|
'allowTxtBlockingReport': v.get('allowTxtBlockingReport'),
|
|
'blockListUrls': v.get('blockListUrls'),
|
|
'zone': args.zone,
|
|
'record': {args.record_host: args.record_ip},
|
|
}
|
|
print(json.dumps(summary, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|