Add Technitium backup sync automation

technitium-serenity/bin/configure_technitium.py

@@ -0,0 +1,160 @@
+#!/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())