Inititial commit

src/main.py

@@ -0,0 +1,189 @@
+import socket
+import time
+from typing import Optional
+import requests
+import json
+import argparse
+import logging
+import random
+import ipaddress
+import threading
+from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
+
+log = logging.getLogger(__name__)
+
+class Mode:
+    IPv4 = "ipv4"
+    IPv6 = "ipv6"
+
+def get_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-c", "--config", default="config.json")
+    parser.add_argument("-s", "--state", default="state.json")
+    parser.add_argument("-w", "--web", default="web")
+    parser.add_argument("-v", "--verbose", action="store_true")
+    return parser.parse_args()
+
+def get_config(path: str) -> dict:
+    with open(path, "r") as f:
+        return json.load(f)
+
+def resolve(domain: str, mode: str) -> Optional[str]:
+    try:
+        tuples = socket.getaddrinfo(domain, None, socket.AF_INET6 if mode == Mode.IPv6 else socket.AF_INET)
+        if tuples:
+            return tuples[0][4][0] # type: ignore
+    except Exception as e:
+        pass
+    return None
+
+def replace_placeholders(text: str, replace: dict) -> str:
+    for key, value in replace.items():
+        text = text.replace(key, value)
+    return text
+
+
+class Updater:
+    def __init__(self, config: dict, state_path: str):
+        self.config = config
+        self.state_path = state_path
+        self.addrs = {}
+        self.history = {}
+        for domain, update_info in config["domains"].items():
+            self.history[domain] = []
+            for mode in update_info.get("modes", [Mode.IPv4]):
+                if not mode in self.addrs:
+                    self.addrs[mode] = ""
+
+    def determine_ip(self, mode: str) -> Optional[str]:
+        urls = [*self.config["ip_lookup"][mode]]
+        random.shuffle(urls)
+        while urls:
+            url = urls.pop()
+            log.debug(f"Looking up local {mode} from {url}")
+            try:
+                response = requests.get(url, timeout=5)
+                response.raise_for_status()
+                addr = response.text.strip()
+                ipaddress.ip_address(addr)
+                log.debug(f"Resolved {mode} to {addr}")
+                return addr
+            except Exception as e:
+                log.error(e)
+        log.debug(f"Unable to determine {mode}")
+        return None
+
+    def detect_updates(self):
+        for mode in self.addrs:
+            addr = self.determine_ip(mode)
+            if not addr:
+                continue
+            if self.addrs[mode] != addr:
+                log.info(f"Detected new {mode}: {addr} (old: {self.addrs[mode]})")
+                self.addrs[mode] = addr
+
+    def add_to_history(self, domain: str, addrs: dict, success: bool, response_text: str):
+        if not domain in self.history:
+            self.history[domain] = []
+        self.history[domain].append({
+            "timestamp": time.time(),
+            "addrs": addrs,
+            "success": success,
+            "response_text": response_text
+        })
+        if len(self.history[domain]) > 10:
+            self.history[domain].pop(0)
+
+    def update_domain(self, domain: str, update_info: dict):
+        modes = update_info.get("modes", [Mode.IPv4])
+        update_required = False
+        for mode in modes:
+            log.debug(f"Resolving {domain} ({mode})")
+            resolved = resolve(domain, mode)
+            log.debug(f"Resolved to {resolved}")
+            if resolved != self.addrs[mode]:
+                update_required = True
+                break
+        if not update_required:
+            return
+        log.info(f"Updating {domain}")
+        replacements = {
+            "${DOMAIN}": domain,
+            "${IPV4}": self.addrs.get(Mode.IPv4, ""),
+            "${IPV6}": self.addrs.get(Mode.IPv6, "")
+        }
+        method = update_info.get("method", "GET")
+        url = replace_placeholders(update_info["url"], replacements)
+        params = update_info.get("params", {})
+        params = {k: replace_placeholders(v, replacements) for k, v in params.items()}
+        headers = update_info.get("headers", {})
+        headers = {k: replace_placeholders(v, replacements) for k, v in headers.items()}
+        response = requests.request(method, url, params=params, headers=headers, timeout=10)
+        success = response.status_code == 200
+        response_text = response.text
+        print(response_text)
+        log.info(f"Updated {domain}")
+        self.add_to_history(domain, self.addrs, success, response_text)
+
+    def update_domains(self):
+        for domain, update_info in self.config["domains"].items():
+            self.update_domain(domain, update_info)
+
+    def write_state(self):
+        with open(self.state_path, "w") as f:
+            json.dump({
+                "timestamp": time.time(),
+                "addrs": self.addrs,
+                "domains": self.history
+            }, f, indent=2)
+
+    def run(self):
+        while True:
+            try:
+                self.detect_updates()
+                self.update_domains()
+                self.write_state()
+            except Exception as e:
+                log.exception(e)
+            time.sleep(10*60)
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        assert isinstance(self.server, WebServer)
+        if self.path == '/':
+            path = self.server.args.web + '/index.html'
+        elif self.path == '/state':
+            path = self.server.args.state
+        else:
+            path = self.server.args.web + self.path
+        try:
+            data = open(path).read()
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(bytes(data, 'utf-8'))
+        except:
+            self.send_response(404)
+            self.end_headers()
+            self.wfile.write(bytes("File not found", 'utf-8'))
+
+class WebServer(ThreadingHTTPServer):
+    def __init__(self, server_address, RequestHandlerClass, args):
+        self.args = args
+        super().__init__(server_address, RequestHandlerClass)
+
+def main():
+    args = get_args()
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
+    log.info(f"Loading config from {args.config}")
+    config = get_config(args.config)
+    if args.web:
+        log.info(f"Serving web interface from {args.web}")
+        httpd = WebServer(('', 8080), Handler, args)
+        threading.Thread(target=httpd.serve_forever).start()
+    updater = Updater(config, args.state)
+    updater.run()
+
+
+if __name__ == "__main__":
+    main()