Procházet zdrojové kódy

Fix CNAME handling: validate dig output as IP addresses

dig +short follows CNAME chains and returns the final A/AAAA
records, but it also echoes the intermediate CNAMEs into the output
(e.g. get.pnpm.io -> cname.vercel-dns.com). These lines are no IP
addresses and were previously taken as-is, which would have pushed
invalid CIDRs into the Proxmox IPSet.

All dig output lines are now validated with the ipaddress module;
anything that is not a valid IP address is dropped.
Paul Klumpp před 6 dny
rodič
revize
bcf1fa59cd
1 změnil soubory, kde provedl 20 přidání a 4 odebrání
  1. 20 4
      manage_ipset.py

+ 20 - 4
manage_ipset.py

@@ -42,6 +42,7 @@ Requires: python3 (stdlib only), pvesh (pve-manager), dig (dnsutils).
 """
 """
 
 
 import argparse
 import argparse
+import ipaddress
 import json
 import json
 import shutil
 import shutil
 import subprocess
 import subprocess
@@ -67,6 +68,24 @@ def dig(*args, timeout=15):
     return result.stdout
     return result.stdout
 
 
 
 
+def extract_ips(lines):
+    """Return the valid IP addresses from dig +short output.
+
+    dig follows CNAME chains and returns the final A/AAAA records, but
+    it also echoes the intermediate CNAMEs into the output. Anything
+    that is not a valid IP address (CNAMEs, TXT, noise) is dropped."""
+    ips = set()
+    for line in lines.splitlines():
+        line = line.strip()
+        if not line or line.startswith(";"):
+            continue
+        try:
+            ips.add(str(ipaddress.ip_address(line)))
+        except ValueError:
+            continue
+    return ips
+
+
 def pvesh(*args):
 def pvesh(*args):
     """Run pvesh, return stdout as text. None on error."""
     """Run pvesh, return stdout as text. None on error."""
     try:
     try:
@@ -100,10 +119,7 @@ def resolve_host(host, max_rounds=50, stale_after=10, timeout=15,
                           timeout=timeout)
                           timeout=timeout)
             except (OSError, subprocess.TimeoutExpired):
             except (OSError, subprocess.TimeoutExpired):
                 out = ""
                 out = ""
-            for line in out.splitlines():
-                line = line.strip()
-                if line and not line.startswith(";"):
-                    ips.add(line)
+            ips |= extract_ips(out)
         new = ips - collected
         new = ips - collected
         if new:
         if new:
             collected |= new
             collected |= new