"""verify_check.py - Standalone deployment verification module. Used by task_runner to check DNS records and port connectivity after deployment. """ import asyncio import dns.resolver async def run_verify(runner, options): """Verify deployed servers: DNS records + port connectivity. Args: runner: TaskRunner instance (for push() output) options: dict of panel options """ from store import get_servers servers = [s for s in get_servers() if s.get("status") == "\u6210\u529f"] if not servers: runner.push("\u6ca1\u6709\u5df2\u6210\u529f\u90e8\u7f72\u7684\u670d\u52a1\u5668\uff0c\u65e0\u9700\u9a8c\u8bc1\n", running=False, exit_code=0) return runner.push( "\n$ \u90e8\u7f72\u9a8c\u8bc1\n\n\u5171 {} \u53f0\uff0c\u5f00\u59cb\u9a8c\u8bc1...\n\n".format(len(servers)), running=True, task_name="\u90e8\u7f72\u9a8c\u8bc1" ) sem = asyncio.Semaphore(max(1, min(100, options.get("workers", 40)))) ok = [0] fail = [0] async def verify_one(server): async with sem: if runner._stop_flag: return ip = server.get("ip", "") hostname = server.get("hostname", "") domain = server.get("domain", hostname) dkim_sel = server.get("dkim_selector", "default") dkim_pub = server.get("dkim_public", "") loop = asyncio.get_event_loop() passed = True runner.push("\n==== {} ({}) ====\n".format(hostname, ip), running=True) # 1. A record try: answers = await loop.run_in_executor(None, lambda: dns.resolver.resolve(hostname, "A")) a_ips = [str(r) for r in answers] if ip in a_ips: runner.push(" [OK] A: {} -> {}\n".format(hostname, ", ".join(a_ips)), running=True) else: runner.push(" [FAIL] A mismatch: got {} expected {}\n".format(", ".join(a_ips), ip), running=True) fail[0] += 1 return except Exception as e: runner.push(" [FAIL] A lookup failed: {} -> {}\n".format(hostname, e), running=True) fail[0] += 1 return # 2. MX try: answers = await loop.run_in_executor(None, lambda: dns.resolver.resolve(hostname, "MX")) mx_list = ["{} (pri={})".format(str(r.exchange), r.preference) for r in answers] runner.push(" [OK] MX: {}\n".format(", ".join(mx_list)), running=True) except Exception as e: runner.push(" [WARN] MX not found: {}\n".format(e), running=True) # 3. SPF try: answers = await loop.run_in_executor(None, lambda: dns.resolver.resolve(hostname, "TXT")) spf_found = [] for r in answers: txt = "".join(s.decode() if isinstance(s, bytes) else s for s in r.strings) if "v=spf1" in txt.lower(): spf_found.append(txt[:120]) if spf_found: runner.push(" [OK] SPF: {}\n".format(spf_found[0]), running=True) else: runner.push(" [WARN] SPF not found\n", running=True) except Exception as e: runner.push(" [WARN] TXT lookup failed: {}\n".format(e), running=True) # 4. DKIM dkim_name = "{}._domainkey.{}".format(dkim_sel, hostname) try: answers = await loop.run_in_executor(None, lambda: dns.resolver.resolve(dkim_name, "TXT")) dkim_txt = "" for r in answers: dkim_txt = "".join(s.decode() if isinstance(s, bytes) else s for s in r.strings) break if dkim_pub and dkim_pub in dkim_txt: runner.push(" [OK] DKIM: {} key matches\n".format(dkim_name), running=True) elif dkim_txt: runner.push(" [WARN] DKIM key mismatch: {}\n".format(dkim_name), running=True) else: runner.push(" [WARN] DKIM not found: {}\n".format(dkim_name), running=True) except Exception as e: runner.push(" [WARN] DKIM lookup failed: {}\n".format(dkim_name), running=True) # 5. DMARC dmarc_name = "_dmarc.{}".format(hostname) try: answers = await loop.run_in_executor(None, lambda: dns.resolver.resolve(dmarc_name, "TXT")) dmarc_txt = "" for r in answers: dmarc_txt = "".join(s.decode() if isinstance(s, bytes) else s for s in r.strings) break runner.push(" [OK] DMARC: {} -> {}\n".format(dmarc_name, dmarc_txt[:100]), running=True) except Exception as e: runner.push(" [WARN] DMARC not found: {}\n".format(dmarc_name), running=True) # 6. Port checks for port in [25, 465, 587]: try: _, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=8) writer.close() try: await writer.wait_closed() except Exception: pass runner.push(" [OK] Port {} open\n".format(port), running=True) except asyncio.TimeoutError: runner.push(" [WARN] Port {} timeout\n".format(port), running=True) except (ConnectionRefusedError, OSError) as e: runner.push(" [FAIL] Port {} closed\n".format(port), running=True) passed = False except Exception as e: runner.push(" [WARN] Port {}: {}\n".format(port, e), running=True) # 7. CNAME for sub in ["imap", "smtp", "pop3"]: sub_name = "{}.{}".format(sub, domain) try: answers = await loop.run_in_executor(None, lambda sn=sub_name: dns.resolver.resolve(sn, "CNAME")) cname_val = str(answers[0].target).rstrip(".") if cname_val == hostname or cname_val.rstrip(".") == hostname: runner.push(" [OK] CNAME: {} -> {}\n".format(sub_name, cname_val), running=True) else: runner.push(" [WARN] CNAME mismatch: {} -> {} expected {}\n".format( sub_name, cname_val, hostname), running=True) except Exception: runner.push(" [WARN] CNAME not found: {}\n".format(sub_name), running=True) if passed: ok[0] += 1 runner.push(" >> PASS: {}\n".format(hostname), running=True) else: runner.push(" >> PARTIAL: {} (check warnings above)\n".format(hostname), running=True) tasks = [verify_one(s) for s in servers] await asyncio.gather(*tasks, return_exceptions=True) total = ok[0] + fail[0] if fail[0] == 0: summary = "Verify: {}/{} OK".format(ok[0], total) else: summary = "Verify: {}/{} OK ({} FAIL)".format(ok[0], total, fail[0]) runner.push( "\n===== Verify Complete =====\nTotal {}: OK {}, FAIL {}\n\n".format(total, ok[0], fail[0]), running=False, exit_code=0 if fail[0] == 0 else 1, summary=summary )