#!/usr/bin/env python3 import gzip import hashlib import os import re import subprocess import sys import tempfile import time import urllib.request RPZ_DIR = "/etc/powerdns/rpz" MIRROR = "https://hagezi-mirror.dnsbunker.org" USER_AGENT = "DNSBunker-rpz-update/1.0" TIMEOUT = 120 RELOAD_TIMEOUT = 600 PENDING = os.path.join("/etc/powerdns/rpz", ".reload-pending") RPZ_FEEDS = { "30-pro.rpz": (f"{MIRROR}/rpz/pro.txt", 50_000), "40-tif.rpz": (f"{MIRROR}/rpz/tif.txt", 50_000), } ADBLOCK_FEEDS = { "10-allow.rpz": (f"{MIRROR}/share/hagezi-dns-allowlist.txt", "rpz-passthru.", 1), "20-deny.rpz": (f"{MIRROR}/share/hagezi-dns-denylist.txt", ".", 1), } DOMAIN_RE = re.compile( r"^(?=.{1,253}$)(?:[a-z0-9_](?:[a-z0-9_-]{0,61}[a-z0-9])?\.)+" r"(?:[a-z]{2,63}|xn--[a-z0-9-]{1,59})$" ) HOSTS_PREFIXES = ("0.0.0.0", "127.0.0.1", "::", "::1") def log(msg: str) -> None: print(msg, flush=True) def fetch(url: str) -> str: req = urllib.request.Request( url, headers={"User-Agent": USER_AGENT, "Accept-Encoding": "gzip"} ) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: data = resp.read() if resp.headers.get("Content-Encoding", "").lower() == "gzip": data = gzip.decompress(data) return data.decode("utf-8", errors="replace") def sha256(text: str) -> str: return hashlib.sha256(text.encode()).hexdigest() def marker_path(path: str) -> str: head, name = os.path.split(path) return os.path.join(head, f".{name}.sha256") def file_hash_marker(path: str) -> str | None: try: with open(marker_path(path), encoding="utf-8") as f: return f.read().strip() or None except FileNotFoundError: return None def strip_zone_comments(text: str) -> str: return "".join(line + "\n" for line in text.splitlines() if not line.lstrip().startswith(";")) def file_sha256(path: str) -> str | None: try: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() except FileNotFoundError: return None def write_atomic(path: str, content: str) -> None: fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path), prefix=".tmp-") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(content) os.chmod(tmp, 0o644) os.replace(tmp, path) except BaseException: if os.path.exists(tmp): os.unlink(tmp) raise def parse_adblock(text: str) -> set[str]: domains: set[str] = set() for raw in text.splitlines(): line = raw.strip() if not line or line[0] in "!#[": continue if line.startswith("@@"): line = line[2:] if line.startswith("||"): line = line[2:] line = line.split("$", 1)[0].rstrip("^|").strip() parts = line.split() if len(parts) == 2 and parts[0] in HOSTS_PREFIXES: line = parts[1] elif len(parts) != 1: continue domain = line.lower().rstrip(".") if domain.startswith("*."): domain = domain[2:] if DOMAIN_RE.match(domain): domains.add(domain) return domains def build_zone(domains: set[str], target: str) -> str: serial = int(time.time()) out = [ "$TTL 300", f"@ SOA localhost. hostmaster.dnsbunker.org. {serial} 3600 600 86400 300", " NS localhost.", ] for d in sorted(domains): out.append(f"{d} CNAME {target}") out.append(f"*.{d} CNAME {target}") return "\n".join(out) + "\n" def update_rpz_feed(name: str, url: str, min_lines: int) -> bool: path = os.path.join(RPZ_DIR, name) content = fetch(url) lines = content.count("\n") head = "\n".join(content.splitlines()[:30]) if " SOA " not in head and "\tSOA\t" not in head: raise ValueError(f"{name}: kein SOA im Kopf, keine gültige RPZ") if lines < min_lines: raise ValueError(f"{name}: nur {lines} Zeilen (Minimum {min_lines})") content = strip_zone_comments(content) if sha256(content) == file_sha256(path): log(f"{name}: unverändert ({lines} Zeilen)") return False write_atomic(path, content) log(f"{name}: aktualisiert ({lines} Zeilen)") return True def update_adblock_feed(name: str, url: str, target: str, min_domains: int) -> bool: path = os.path.join(RPZ_DIR, name) domains = parse_adblock(fetch(url)) if len(domains) < min_domains: raise ValueError(f"{name}: nur {len(domains)} Domains (Minimum {min_domains})") digest = sha256("\n".join(sorted(domains)) + target) if digest == file_hash_marker(path): log(f"{name}: unverändert ({len(domains)} Domains)") return False write_atomic(path, build_zone(domains, target)) write_atomic(marker_path(path), digest + "\n") log(f"{name}: aktualisiert ({len(domains)} Domains)") return True def main() -> int: os.makedirs(RPZ_DIR, exist_ok=True) changed = False failed = False for name, (url, min_lines) in RPZ_FEEDS.items(): try: changed |= update_rpz_feed(name, url, min_lines) except Exception as exc: failed = True log(f"FEHLER {name}: {exc}") for name, (url, target, min_domains) in ADBLOCK_FEEDS.items(): try: changed |= update_adblock_feed(name, url, target, min_domains) except Exception as exc: failed = True log(f"FEHLER {name}: {exc}") if changed: open(PENDING, "w").close() if os.path.exists(PENDING) and "--no-reload" not in sys.argv: start = time.monotonic() result = subprocess.run( ["rec_control", f"--timeout={RELOAD_TIMEOUT}", "reload-yaml"], capture_output=True, text=True, ) duration = time.monotonic() - start if result.returncode == 0: os.unlink(PENDING) log(f"Recursor neu geladen ({duration:.1f}s)") else: failed = True err = (result.stderr or result.stdout).strip() log(f"FEHLER reload-yaml nach {duration:.1f}s: {err}") return 1 if failed else 0 if __name__ == "__main__": sys.exit(main())