From d40793074febfca1ef9d71fb155113897115cdd9 Mon Sep 17 00:00:00 2001 From: rafal Date: Thu, 29 Jan 2026 20:37:35 +0000 Subject: [PATCH] Add update_ddns.py --- update_ddns.py | 153 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 update_ddns.py diff --git a/update_ddns.py b/update_ddns.py new file mode 100644 index 0000000..df32ddb --- /dev/null +++ b/update_ddns.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +import requests +import json +import logging +import sys +import os +import time + +# --- KONFIGURACJA --- +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +CONFIG_FILE = os.path.join(BASE_DIR, "config.json") +DOMAINS_FILE = os.path.join(BASE_DIR, "domains.txt") +LOG_FILE = "/var/log/ionos-ddns.log" +INTERVAL = 600 # Czas w sekundach (10 minut) + +# --- LOGOWANIE --- +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', + handlers=[ + logging.FileHandler(LOG_FILE), + logging.StreamHandler(sys.stdout) + ] +) + +class IonosClient: + def __init__(self, prefix, secret): + self.base_url = "https://api.hosting.ionos.com/dns/v1" + self.headers = { + "X-API-Key": f"{prefix}.{secret}", + "Content-Type": "application/json", + "Accept": "application/json" + } + + def get_zones(self): + try: + resp = requests.get(f"{self.base_url}/zones", headers=self.headers) + if resp.status_code == 200: + return resp.json() + logging.error(f"API Error (get_zones): {resp.status_code} {resp.text}") + except Exception as e: + logging.error(f"Connection Error (get_zones): {e}") + return [] + + def get_zone_records(self, zone_id): + try: + resp = requests.get(f"{self.base_url}/zones/{zone_id}", headers=self.headers) + if resp.status_code == 200: + return resp.json().get('records', []) + except Exception as e: + logging.error(f"Connection Error (get_zone_records): {e}") + return [] + + def create_record(self, zone_id, name, ip): + data = [{"name": name, "type": "A", "content": ip, "ttl": 3600}] + logging.info(f" [ACTION] CREATING: {name} -> {ip}") + try: + resp = requests.post(f"{self.base_url}/zones/{zone_id}/records", headers=self.headers, json=data) + if resp.status_code == 201: + logging.info(" [SUCCESS] Record created.") + else: + logging.error(f" [ERROR] API: {resp.text}") + except Exception as e: + logging.error(f" [ERROR] Connection: {e}") + + def update_record(self, zone_id, record_id, name, ip): + data = {"name": name, "type": "A", "content": ip, "ttl": 3600} + logging.info(f" [ACTION] UPDATING: {name} -> {ip}") + try: + resp = requests.put(f"{self.base_url}/zones/{zone_id}/records/{record_id}", headers=self.headers, json=data) + if resp.status_code == 200: + logging.info(" [SUCCESS] Updated.") + else: + logging.error(f" [ERROR] API: {resp.text}") + except Exception as e: + logging.error(f" [ERROR] Connection: {e}") + +def load_config(): + if not os.path.exists(CONFIG_FILE): + logging.error(f"CRITICAL: Config file missing: {CONFIG_FILE}") + return None + try: + with open(CONFIG_FILE, 'r') as f: + return json.load(f) + except Exception as e: + logging.error(f"CRITICAL: Config read error: {e}") + return None + +def get_public_ip(url): + try: + return requests.get(url, timeout=10).text.strip() + except Exception as e: + logging.error(f"Public IP fetch error: {e}") + return None + +def find_zone_for_domain(full_domain, zones): + best_match = None + best_len = 0 + for zone in zones: + zone_name = zone['name'] + if full_domain == zone_name or full_domain.endswith("." + zone_name): + if len(zone_name) > best_len: + best_len = len(zone_name) + best_match = zone + return best_match + +def main(): + logging.info("=== STARTING IONOS DDNS SERVICE ===") + + config = load_config() + if not config: + sys.exit(1) + + ionos = IonosClient(config['api_prefix'], config['api_secret']) + + while True: + try: + if not os.path.exists(DOMAINS_FILE): + logging.error(f"Domains file missing: {DOMAINS_FILE}") + else: + with open(DOMAINS_FILE, 'r') as f: + target_domains = [line.strip() for line in f if line.strip()] + + public_ip = get_public_ip(config['check_ip_url']) + + if public_ip and target_domains: + zones = ionos.get_zones() + if zones: + for full_domain in target_domains: + zone = find_zone_for_domain(full_domain, zones) + if not zone: + logging.warning(f" No zone found for: {full_domain}") + continue + + records = ionos.get_zone_records(zone['id']) + target_record = next((r for r in records if r['name'] == full_domain and r['type'] == 'A'), None) + + if target_record: + if target_record['content'] != public_ip: + ionos.update_record(zone['id'], target_record['id'], full_domain, public_ip) + else: + # Opcjonalnie można zakomentować, żeby nie spamowało logów co 10 min + logging.info(f" [OK] {full_domain} is up to date.") + else: + ionos.create_record(zone['id'], full_domain, public_ip) + except Exception as e: + logging.critical(f"UNEXPECTED LOOP ERROR: {e}") + + time.sleep(INTERVAL) + +if __name__ == "__main__": + main() \ No newline at end of file