#!/usr/bin/env python3 """ Auditor de correo de dominios — Ciberkit ============================================== Comprueba, para uno o muchos dominios, si alguien puede enviar correos haciéndose pasar por ellos: SPF (con el recuento real de consultas DNS, también dentro de los include), DMARC (con la herencia del dominio organizativo y la autorización de informes externos), DKIM (selectores habituales y tamaño de la clave), MX, MTA-STS, TLS-RPT, BIMI y DNSSEC. Usa los mismos criterios que la página «¿Pueden suplantar tu dominio?» de la suite. Las consultas van cifradas por DNS-over-HTTPS a Cloudflare (y, si falla, a Google). Solo se consulta información pública del DNS: no se envía ningún correo ni se conecta con los servidores del dominio. Sin dependencias: Python 3.8+ en Windows, Linux o macOS. Ejemplos -------- python auditar_dominios.py empresa.com python auditar_dominios.py empresa.com filial.com.co --selector s2024 python auditar_dominios.py --lista dominios.txt --csv informe.csv --json informe.json python auditar_dominios.py --lista dominios.txt --sin-dkim # más rápido Código de salida: 0 todos protegidos, 1 alguno parcialmente protegido, 2 alguno suplantable, 3 algún dominio no se pudo auditar (útil en CI). """ import argparse import base64 import csv import json import re import sys import threading import time import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor FORMATO = "suite-redes.correo-dominios/1" TIPOS = {"A": 1, "CNAME": 5, "SOA": 6, "MX": 15, "TXT": 16, "AAAA": 28} PROVEEDORES_DOH = [ ("cloudflare", "https://cloudflare-dns.com/dns-query?name={n}&type={t}", {"accept": "application/dns-json"}), ("google", "https://dns.google/resolve?name={n}&type={t}", {}), ] DKIM_SELECTORES = [ "google", "selector1", "selector2", "k1", "k2", "k3", "s1", "s2", "default", "dkim", "mail", "smtp", "mx", "zoho", "zmail", "fm1", "fm2", "fm3", "protonmail", "protonmail2", "protonmail3", "sig1", "mandrill", "mailjet", "everlytickey1", "everlytickey2", "hs1", "hs2", "cm", "key1", "dkim1", "mxvault", "pm", "scph0122", "turbo-smtp", ] DOS_NIVELES = { "co.uk", "org.uk", "ac.uk", "com.co", "net.co", "org.co", "edu.co", "gov.co", "com.mx", "org.mx", "gob.mx", "edu.mx", "com.ar", "gob.ar", "com.br", "com.pe", "gob.pe", "com.ve", "com.ec", "gob.ec", "com.es", "org.es", "gob.es", "com.uy", "com.py", "com.bo", "com.gt", "com.pa", "co.cr", "com.do", "com.sv", "com.hn", "com.ni", "cl", "co.za", "com.au", "co.jp", "com.tr", "co.in", } PROVEEDORES_CORREO = [ (r"(^|\.)(google|googlemail)\.com$", "Google Workspace", ["google"]), (r"\.mail\.protection\.outlook\.com$", "Microsoft 365", ["selector1", "selector2"]), (r"(^|\.)zoho\.(com|eu|in)$", "Zoho Mail", ["zmail", "zoho"]), (r"(^|\.)protonmail\.ch$", "Proton Mail", ["protonmail", "protonmail2", "protonmail3"]), (r"(^|\.)messagingengine\.com$", "Fastmail", ["fm1", "fm2", "fm3"]), (r"(^|\.)icloud\.com$", "iCloud", ["sig1"]), (r"(^|\.)mimecast\.com$", "Mimecast", []), (r"(^|\.)pphosted\.com$", "Proofpoint", []), (r"(^|\.)secureserver\.net$", "GoDaddy", []), (r"(^|\.)hostinger\.(com|es)$", "Hostinger", []), (r"(^|\.)yandex\.(net|ru)$", "Yandex", []), ] ORDEN = {"critico": 0, "aviso": 1, "info": 2, "ok": 3} ESTADO = {"protegido": "Protegido", "parcial": "Parcialmente protegido", "suplantable": "Suplantable"} RE_DOMINIO = re.compile(r"^(?=.{1,253}$)([a-z0-9_](?:[a-z0-9_-]{0,61}[a-z0-9])?\.)+[a-z0-9-]{2,63}$") # ───────────────────────── DNS ───────────────────────── def txt_data(data): """«"v=spf1 ip4:1.2.3.4 " "include:x ~all"» → «v=spf1 ip4:1.2.3.4 include:x ~all».""" s = str(data or "").strip() if not s.startswith('"'): return s return "".join(re.sub(r"\\(.)", r"\1", m) for m in re.findall(r'"((?:[^"\\]|\\.)*)"', s)) def normalizar_dominio(texto): """Acepta un dominio, una URL o un correo y devuelve el dominio en minúsculas (o lanza ValueError).""" s = str(texto or "").strip().lower() s = re.sub(r"^[a-z][a-z0-9+.-]*://", "", s) s = re.sub(r"^[^@\s]*@", "", s) s = re.split(r"[/?#:\s]", s)[0].rstrip(".") if not s: raise ValueError("dominio vacío") if not RE_DOMINIO.match(s) and not s.startswith("xn--"): raise ValueError(f"«{s}» no es un dominio válido") return s class ResolutorDoH: """resolve(nombre, tipo) → {status: ok|nodata|nxdomain|error, records, ad, cname}, con caché.""" def __init__(self, timeout=8.0, abrir=None): self.timeout = timeout self.abrir = abrir or urllib.request.urlopen self.cache = {} self.consultas = 0 self._lock = threading.Lock() def _una(self, plantilla, cabeceras, nombre, tipo): url = plantilla.format(n=urllib.parse.quote(nombre, safe=""), t=tipo) req = urllib.request.Request(url, headers={**cabeceras, "user-agent": "suite-redes-correo/1"}) with self.abrir(req, timeout=self.timeout) as r: return json.loads(r.read().decode("utf-8")) def __call__(self, nombre, tipo): clave = (nombre.lower(), tipo) with self._lock: if clave in self.cache: return self.cache[clave] self.consultas += 1 datos, error = None, None for _, plantilla, cabeceras in PROVEEDORES_DOH: for intento in range(2): try: datos = self._una(plantilla, cabeceras, nombre, tipo) break except Exception as e: # red, HTTP o JSON error = e time.sleep(0.3 * (intento + 1)) if datos is not None: break if datos is None: res = {"status": "error", "records": [], "ad": False, "cname": None, "error": str(error)} else: num = TIPOS[tipo] resp = datos.get("Answer") or [] regs = [txt_data(a.get("data")) if tipo == "TXT" else str(a.get("data", "")).rstrip(".") for a in resp if a.get("type") == num] cname = next((str(a.get("data", "")).rstrip(".") for a in resp if a.get("type") == TIPOS["CNAME"]), None) st = datos.get("Status") estado = "nxdomain" if st == 3 else "error" if st != 0 else "ok" if regs else "nodata" res = {"status": estado, "records": regs, "ad": bool(datos.get("AD")), "cname": cname} with self._lock: self.cache[clave] = res return res # ───────────────────────── SPF ───────────────────────── MECANISMOS_CON_CONSULTA = {"include", "a", "mx", "ptr", "exists"} RE_TERMINO = re.compile(r"^([+\-~?]?)(all|include|a|mx|ptr|ip4|ip6|exists)(?::([^/]+))?(/\d+(?://\d+)?)?$", re.I) def parse_spf(registro): out = {"terms": [], "redirect": None, "errors": []} partes = str(registro).strip().split() if not partes or partes[0].lower() != "v=spf1": out["errors"].append("No empieza por «v=spf1».") return out for raw in partes[1:]: mod = re.match(r"^([a-z][a-z0-9_.-]*)=(.*)$", raw, re.I) if mod: if mod.group(1).lower() == "redirect": out["redirect"] = mod.group(2) continue m = RE_TERMINO.match(raw) if not m: out["errors"].append(f"Término no válido: «{raw}».") continue mech = m.group(2).lower() valor = m.group(3) if mech in ("include", "exists") and not valor: out["errors"].append(f"«{raw}» necesita un dominio.") if mech == "ip4" and not re.match(r"^\d{1,3}(\.\d{1,3}){3}$", valor or ""): out["errors"].append(f"IP no válida en «{raw}».") if mech == "ip6" and ":" not in (valor or ""): out["errors"].append(f"IPv6 no válida en «{raw}».") out["terms"].append({"q": m.group(1) or "+", "mech": mech, "value": valor}) return out def evaluar_spf(dominio, resolve, estado=None): """Sigue include y redirect contando consultas DNS. Devuelve un nodo del árbol SPF.""" if estado is None: estado = {"lookups": 0, "voids": 0, "visitados": set(), "profundidad": 0} nodo = {"domain": dominio, "record": None, "status": "ok", "parsed": None, "children": [], "errors": []} if dominio in estado["visitados"]: nodo["status"] = "loop" nodo["errors"].append("Bucle: este dominio ya se incluyó antes.") return nodo if estado["profundidad"] > 10: nodo["status"] = "error" nodo["errors"].append("Demasiados niveles de include.") return nodo estado["visitados"].add(dominio) txt = resolve(dominio, "TXT") if txt["status"] == "error": nodo["status"] = "error" nodo["errors"].append("No se pudo consultar el DNS.") return nodo spf = [r for r in txt["records"] if re.match(r"^v=spf1(\s|$)", r, re.I)] if txt["status"] == "nxdomain": nodo["status"] = "nxdomain" return nodo if not spf: nodo["status"] = "none" return nodo if len(spf) > 1: nodo["status"] = "multiple" nodo["errors"].append(f"Tiene {len(spf)} registros SPF: es un error permanente (permerror) y los receptores ignoran todos.") nodo["record"] = spf[0] nodo["parsed"] = p = parse_spf(spf[0]) nodo["errors"].extend(p["errors"]) def vacia(nombre, tipos): rs = [resolve(nombre, t) for t in tipos] if all(r["status"] in ("nxdomain", "nodata") for r in rs): estado["voids"] += 1 return rs for t in p["terms"]: if t["mech"] in MECANISMOS_CON_CONSULTA: estado["lookups"] += 1 destino = t["value"] or dominio if "%{" in destino: continue # macros: no se pueden evaluar sin un correo concreto if t["mech"] == "include": estado["profundidad"] += 1 hijo = evaluar_spf(destino, resolve, estado) estado["profundidad"] -= 1 if hijo["status"] in ("none", "nxdomain"): hijo["errors"].append(f"«include:{destino}» apunta a un dominio sin SPF: es un error permanente.") nodo["children"].append(hijo) elif t["mech"] == "a": vacia(destino, ["A", "AAAA"]) elif t["mech"] == "mx": mx = vacia(destino, ["MX"]) if len(mx[0]["records"]) > 10: nodo["errors"].append(f"«mx» con {len(mx[0]['records'])} registros MX: el límite es 10.") elif t["mech"] == "exists": vacia(destino, ["A"]) elif t["mech"] == "ptr": nodo["errors"].append("Usa «ptr», desaconsejado (RFC 7208): es lento y poco fiable.") if p["redirect"] and not any(t["mech"] == "all" for t in p["terms"]): estado["lookups"] += 1 estado["profundidad"] += 1 hijo = evaluar_spf(p["redirect"], resolve, estado) estado["profundidad"] -= 1 hijo["via_redirect"] = True nodo["children"].append(hijo) return nodo def spf_all(nodo): """Calificador del «all» final (siguiendo redirect), o None.""" if not nodo or not nodo.get("parsed"): return None for t in nodo["parsed"]["terms"]: if t["mech"] == "all": return t["q"] red = next((c for c in nodo["children"] if c.get("via_redirect")), None) return spf_all(red) if red else None def errores_spf(nodo, out=None): out = [] if out is None else out out.extend({"domain": nodo["domain"], "text": e} for e in nodo["errors"]) for c in nodo["children"]: errores_spf(c, out) return out # ───────────────────────── DMARC y DKIM ───────────────────────── def parse_dmarc(registro): out = {"tags": {}, "errors": []} partes = [x.strip() for x in str(registro).split(";") if x.strip()] for i, p in enumerate(partes): m = re.match(r"^([a-z]+)\s*=\s*(.*)$", p, re.I) if not m: out["errors"].append(f"Etiqueta no válida: «{p}».") continue k, v = m.group(1).lower(), m.group(2).strip() out["tags"][k] = v if i == 0 and not (k == "v" and v == "DMARC1"): out["errors"].append("Debe empezar exactamente por «v=DMARC1».") t = out["tags"] if "p" not in t: out["errors"].append("Falta la política «p=» (none, quarantine o reject).") elif t["p"].lower() not in ("none", "quarantine", "reject"): out["errors"].append(f"p={t['p']} no es válido.") if t.get("sp") and t["sp"].lower() not in ("none", "quarantine", "reject"): out["errors"].append(f"sp={t['sp']} no es válido.") if t.get("pct"): try: ok = 0 <= float(t["pct"]) <= 100 except ValueError: ok = False if not ok: out["errors"].append(f"pct={t['pct']} debe estar entre 0 y 100.") return out def dominio_organizativo(dominio): l = dominio.split(".") return ".".join(l[-3:]) if ".".join(l[-2:]) in DOS_NIVELES else ".".join(l[-2:]) def bits_rsa(b64): """Longitud en bits del módulo RSA de una clave DER en base64 (SubjectPublicKeyInfo o RSAPublicKey).""" try: b = base64.b64decode(re.sub(r"\s+", "", str(b64)), validate=False) except Exception: return None mejor = 0 def recorrer(ini, fin): nonlocal mejor i = ini while i < fin: tag, largo, off = b[i], b[i + 1], 2 if largo & 0x80: n = largo & 0x7F largo = int.from_bytes(b[i + 2:i + 2 + n], "big") off = 2 + n s = i + off if tag == 0x02: j = s while j < s + largo and b[j] == 0: j += 1 if s + largo - j > 0: mejor = max(mejor, (s + largo - j - 1) * 8 + b[j].bit_length()) elif tag == 0x30: recorrer(s, s + largo) elif tag == 0x03: recorrer(s + 1, s + largo) i = s + largo try: recorrer(0, len(b)) except Exception: return None return mejor or None def parse_dkim(registro): tags = {} for p in str(registro).split(";"): m = re.match(r"^\s*([a-z]+)\s*=\s*(.*?)\s*$", p, re.I | re.S) if m: tags[m.group(1).lower()] = re.sub(r"\s+", "", m.group(2)) k = (tags.get("k") or "rsa").lower() revocada = "p" in tags and tags["p"] == "" bits = 0 if revocada else 256 if k == "ed25519" else bits_rsa(tags.get("p", "")) return {"k": k, "revoked": revocada, "bits": bits, "testing": bool(re.search(r"(^|:)y(:|$)", tags.get("t", "")))} def proveedor_de(host): for patron, nombre, selectores in PROVEEDORES_CORREO: if re.search(patron, host, re.I): return {"name": nombre, "dkim": selectores} return None # ───────────────────────── Auditoría ───────────────────────── def auditar(dominio, resolve, selectores_extra=(), dkim=True, hilos_dkim=8): hallazgos = [] puntos = [100] def add(sev, area, texto, ref=None): hallazgos.append({"severity": sev, "area": area, "text": texto, "ref": ref}) def resta(n): puntos[0] -= n mx_r = resolve(dominio, "MX") dmarc_r = resolve(f"_dmarc.{dominio}", "TXT") mta_r = resolve(f"_mta-sts.{dominio}", "TXT") tls_r = resolve(f"_smtp._tls.{dominio}", "TXT") bimi_r = resolve(f"default._bimi.{dominio}", "TXT") soa_r = resolve(dominio, "SOA") if soa_r["status"] == "nxdomain": raise LookupError(f"El dominio {dominio} no existe en el DNS.") if soa_r["status"] == "error" and mx_r["status"] == "error": raise ConnectionError("No se pudo consultar el DNS (sin conexión o bloqueado).") # MX mx = [] for r in mx_r["records"]: partes = r.split() try: mx.append({"prio": int(partes[0]), "host": (partes[1] if len(partes) > 1 else "").rstrip(".")}) except ValueError: continue mx.sort(key=lambda m: m["prio"]) mx_nulo = len(mx) == 1 and mx[0]["host"] in ("", ".") proveedor = next((p for p in (proveedor_de(m["host"]) for m in mx) if p), None) recibe = bool(mx) and not mx_nulo if mx_nulo: add("info", "MX", "MX nulo («0 .»): el dominio declara que no recibe correo.") elif not mx: add("info", "MX", "Sin registros MX: el dominio no recibe correo (o lo recibe por el registro A).") else: add("ok", "MX", f"{len(mx)} servidor{'es' if len(mx) > 1 else ''} de correo" + (f" de {proveedor['name']}" if proveedor else "") + ".") # SPF estado = {"lookups": 0, "voids": 0, "visitados": set(), "profundidad": 0} arbol = evaluar_spf(dominio, resolve, estado) errores = errores_spf(arbol) todo = spf_all(arbol) if arbol["status"] == "none": add("critico", "SPF", "No tiene registro SPF: cualquier servidor puede enviar correo en nombre del dominio.", "RFC 7208") resta(25) elif arbol["status"] == "multiple": add("critico", "SPF", arbol["errors"][0], "RFC 7208 §3.2") resta(25) elif arbol["status"] == "error": add("aviso", "SPF", "No se pudo consultar el SPF.") else: n = estado["lookups"] if n > 10: add("critico", "SPF", f"Necesita {n} consultas DNS y el límite es 10: los receptores lo dan por erróneo (permerror).", "RFC 7208 §4.6.4") resta(25) elif n >= 8: add("aviso", "SPF", f"Usa {n} de las 10 consultas DNS permitidas: un proveedor más y dejará de funcionar.", "RFC 7208 §4.6.4") else: add("ok", "SPF", f"{n} de 10 consultas DNS.") if estado["voids"] > 2: add("aviso", "SPF", f"{estado['voids']} consultas vacías (límite 2): algunos receptores lo dan por erróneo.", "RFC 7208 §4.6.4") resta(10) if todo == "+": add("critico", "SPF", "«+all» autoriza a cualquier servidor del mundo: equivale a no tener SPF.", "RFC 7208") resta(30) elif todo in ("?", None): add("aviso", "SPF", ("«?all» (neutral)" if todo == "?" else "Sin «all» al final") + ": no indica qué hacer con los servidores no autorizados.") resta(10) elif todo == "~": add("info", "SPF", "«~all» (softfail): correcto si DMARC está en quarantine o reject.") else: add("ok", "SPF", "«-all»: los servidores no autorizados se rechazan.") for e in errores: add("aviso", "SPF", ("" if e["domain"] == dominio else f"En {e['domain']}: ") + e["text"]) resta(15 if "permanente" in e["text"] else 3) # DMARC (con herencia del dominio organizativo) dmarc_regs = [r for r in dmarc_r["records"] if re.match(r"^v=DMARC1", r, re.I)] registro = dmarc_regs[0] if dmarc_regs else None origen = dominio org = dominio_organizativo(dominio) if not registro and org != dominio: org_r = resolve(f"_dmarc.{org}", "TXT") registro = next((r for r in org_r["records"] if re.match(r"^v=DMARC1", r, re.I)), None) if registro: origen = org dmarc = parse_dmarc(registro) if registro else None tags = dmarc["tags"] if dmarc else {} heredado = origen != dominio pol = ((tags.get("sp") or tags.get("p")) if heredado else tags.get("p")) or None pol = pol.lower() if pol else None try: pct = float(tags["pct"]) if tags.get("pct") else 100 except ValueError: pct = 100 if len(dmarc_regs) > 1: add("critico", "DMARC", "Hay varios registros DMARC: los receptores los ignoran todos.") resta(30) if not dmarc: add("critico", "DMARC", "Sin DMARC: los receptores no saben qué hacer con los correos que falsifican tu dominio en el «De:».", "RFC 7489") resta(35) else: for e in dmarc["errors"]: add("critico", "DMARC", e) resta(15) if heredado: add("info", "DMARC", f"Hereda el DMARC de {org}" + (f" (sp={tags['sp']})" if tags.get("sp") else "") + ".") if pol == "none": add("aviso", "DMARC", "p=none: solo observa. Los correos falsificados siguen llegando.", "RFC 7489") resta(25) elif pol == "quarantine": add("aviso" if pct < 100 else "info", "DMARC", f"p=quarantine" + (f" al {pct:g} %" if pct < 100 else "") + ": los falsos van a spam. El objetivo final es p=reject.") resta(12 if pct < 100 else 8) elif pol == "reject": if pct < 100: add("aviso", "DMARC", f"p=reject solo al {pct:g} %: el resto de falsos se trata con una política más suave.") resta(8) else: add("ok", "DMARC", "p=reject: los correos que falsifican el dominio se rechazan.") if not heredado and tags.get("sp", "").lower() == "none" and pol != "none": add("aviso", "DMARC", "sp=none: los subdominios se pueden falsificar aunque el dominio esté protegido.") resta(8) if not tags.get("rua"): add("aviso", "DMARC", "Sin «rua»: no recibes informes de quién envía en tu nombre.") resta(5) else: externos = [] for u in tags["rua"].split(","): m = re.search(r"mailto:[^@]+@([^!,\s]+)", u, re.I) if m: d = m.group(1).lower() if dominio_organizativo(d) != dominio_organizativo(dominio) and d not in externos: externos.append(d) for d in externos: aut = resolve(f"{dominio}._report._dmarc.{d}", "TXT") ok = any(re.match(r"^v=DMARC1", r, re.I) for r in aut["records"]) if not ok: ok = any(re.match(r"^v=DMARC1", r, re.I) for r in resolve(f"*._report._dmarc.{d}", "TXT")["records"]) if ok: add("ok", "DMARC", f"Informes autorizados en {d}.") else: add("aviso", "DMARC", f"Los informes van a {d}, pero ese dominio no ha autorizado recibirlos (falta {dominio}._report._dmarc.{d}).", "RFC 7489 §7.1") resta(5) # DKIM claves = [] probados = 0 if dkim: extra = [s.strip().lower() for s in selectores_extra if re.match(r"^[a-z0-9._-]+$", s.strip().lower())] selectores = list(dict.fromkeys(extra + (proveedor["dkim"] if proveedor else []) + DKIM_SELECTORES)) probados = len(selectores) def probar(sel): r = resolve(f"{sel}._domainkey.{dominio}", "TXT") reg = next((x for x in r["records"] if re.search(r"(^|;)\s*(v=DKIM1|k=|p=)", x, re.I)), None) return {"selector": sel, **parse_dkim(reg)} if reg else None with ThreadPoolExecutor(max_workers=hilos_dkim) as ex: claves = [c for c in ex.map(probar, selectores) if c] if not claves: if recibe or arbol["status"] == "ok": add("info", "DKIM", f"No se encontró DKIM con {probados} selectores habituales, lo que no prueba que no exista (indica el tuyo con --selector).") for k in claves: if k["revoked"]: add("info", "DKIM", f"Selector «{k['selector']}» revocado (p= vacío).") elif k["bits"] and k["bits"] < 1024: add("critico", "DKIM", f"Selector «{k['selector']}»: clave RSA de {k['bits']} bits, rompible.") resta(15) elif k["bits"] and k["bits"] < 2048 and k["k"] == "rsa": add("aviso", "DKIM", f"Selector «{k['selector']}»: RSA de {k['bits']} bits. Se recomienda 2048.", "RFC 8301") resta(5) else: add("ok", "DKIM", f"Selector «{k['selector']}»: " + ("Ed25519" if k["k"] == "ed25519" else f"RSA de {k['bits'] or '?'} bits") + ".") if k["testing"]: add("info", "DKIM", f"Selector «{k['selector']}» en modo de pruebas (t=y).") # Extras mtasts = any(re.match(r"^v=STSv1", r, re.I) for r in mta_r["records"]) tlsrpt = any(re.match(r"^v=TLSRPTv1", r, re.I) for r in tls_r["records"]) bimi = any(re.match(r"^v=BIMI1", r, re.I) for r in bimi_r["records"]) if recibe: if mtasts: add("ok", "MTA-STS", "MTA-STS publicado.") else: add("info", "MTA-STS", "Sin MTA-STS: un atacante en la red podría forzar que el correo entrante viaje sin cifrar.", "RFC 8461") resta(2) if tlsrpt: add("ok", "TLS-RPT", "TLS-RPT publicado.") else: add("info", "TLS-RPT", "Sin TLS-RPT: no te enteras de fallos de cifrado en la entrega.", "RFC 8460") resta(1) if bimi: add("ok", "BIMI", "BIMI publicado.") if soa_r["ad"]: add("ok", "DNSSEC", "DNSSEC validado.") else: add("info", "DNSSEC", "Sin DNSSEC: las respuestas DNS del dominio no están firmadas.") resta(2) # Veredicto aplica = bool(dmarc) and not dmarc["errors"] and pol in ("reject", "quarantine") veredicto = "suplantable" if not aplica else "protegido" if pol == "reject" and pct == 100 else "parcial" nota = max(0, min(100, round(puntos[0]))) if veredicto == "suplantable": nota = min(nota, 59) letra = "A" if nota >= 90 else "B" if nota >= 75 else "C" if nota >= 60 else "D" if nota >= 40 else "F" hallazgos.sort(key=lambda h: ORDEN[h["severity"]]) return { "dominio": dominio, "veredicto": veredicto, "veredicto_texto": ESTADO[veredicto], "nota": nota, "letra": letra, "mx": mx, "mx_nulo": mx_nulo, "proveedor": proveedor["name"] if proveedor else None, "spf": {"registro": arbol["record"], "estado": arbol["status"], "consultas": estado["lookups"], "vacias": estado["voids"], "all": todo}, "dmarc": {"registro": registro, "politica": pol, "pct": pct, "heredado_de": origen if heredado else None, "rua": tags.get("rua")} if dmarc else None, "dkim": {"selectores_probados": probados, "encontrados": [{"selector": k["selector"], "tipo": k["k"], "bits": k["bits"], "revocada": k["revoked"]} for k in claves]} if dkim else None, "mta_sts": mtasts, "tls_rpt": tlsrpt, "bimi": bimi, "dnssec": bool(soa_r["ad"]), "hallazgos": hallazgos, } # ───────────────────────── Salida ───────────────────────── COLUMNAS_CSV = ["dominio", "veredicto", "nota", "letra", "proveedor", "spf_all", "spf_consultas", "dmarc_politica", "dmarc_pct", "dkim_selectores", "mta_sts", "tls_rpt", "dnssec", "criticos", "avisos", "error"] def _celda(v): s = "" if v is None else str(v) return "'" + s if s[:1] in ("=", "+", "-", "@") else s # evita fórmulas en Excel def fila_csv(r): if "error" in r: return {"dominio": _celda(r["dominio"]), "error": _celda(r["error"])} sev = [h["severity"] for h in r["hallazgos"]] return { "dominio": r["dominio"], "veredicto": r["veredicto"], "nota": r["nota"], "letra": r["letra"], "proveedor": r["proveedor"] or "", "spf_all": (r["spf"]["all"] + "all") if r["spf"]["all"] else "", "spf_consultas": r["spf"]["consultas"], "dmarc_politica": (r["dmarc"] or {}).get("politica") or "", "dmarc_pct": (r["dmarc"] or {}).get("pct", ""), "dkim_selectores": " ".join(k["selector"] for k in (r["dkim"] or {}).get("encontrados", [])), "mta_sts": "sí" if r["mta_sts"] else "no", "tls_rpt": "sí" if r["tls_rpt"] else "no", "dnssec": "sí" if r["dnssec"] else "no", "criticos": sev.count("critico"), "avisos": sev.count("aviso"), "error": "", } def codigo_salida(resultados): if any("error" in r for r in resultados): return 3 if any(r["veredicto"] == "suplantable" for r in resultados): return 2 if any(r["veredicto"] == "parcial" for r in resultados): return 1 return 0 def imprimir(r, detalle, color): def c(cod, s): return f"\033[{cod}m{s}\033[0m" if color else s if "error" in r: print(f" {c('90', '·')} {r['dominio']:<34} {c('90', 'ERROR')} {r['error']}") return tono = {"protegido": "32", "parcial": "33", "suplantable": "31"}[r["veredicto"]] spf = (r["spf"]["all"] + "all") if r["spf"]["all"] else ("sin SPF" if r["spf"]["estado"] == "none" else "—") dm = f"p={r['dmarc']['politica']}" if r["dmarc"] else "sin DMARC" print(f" {c(tono, '●')} {r['dominio']:<34} {c(tono, r['veredicto_texto']):<{31 if color else 22}} {r['letra']} {r['nota']:>3}/100 SPF {spf} ({r['spf']['consultas']}/10) · DMARC {dm}") if detalle: for h in r["hallazgos"]: if h["severity"] in ("critico", "aviso"): marca = c("31", "✖") if h["severity"] == "critico" else c("33", "▲") print(f" {marca} {h['area']}: {h['text']}") def leer_lista(ruta): with open(ruta, encoding="utf-8-sig") as f: return [l.split("#")[0].strip() for l in f if l.split("#")[0].strip()] def preparar_salida(): """Evita que el script falle en consolas o redirecciones sin UTF-8 (p. ej. cp1252 en Windows).""" for flujo in (sys.stdout, sys.stderr): try: if not flujo.isatty(): flujo.reconfigure(encoding="utf-8", errors="replace") else: flujo.reconfigure(errors="replace") except (AttributeError, ValueError): pass def main(argv=None): preparar_salida() ap = argparse.ArgumentParser(description="Audita SPF, DKIM y DMARC de uno o muchos dominios (solo DNS público).") ap.add_argument("dominios", nargs="*", help="dominios (también URL o correo: se queda con el dominio)") ap.add_argument("--lista", metavar="FICHERO", help="fichero con un dominio por línea (# para comentarios)") ap.add_argument("--selector", action="append", default=[], metavar="S", help="selector DKIM propio (se puede repetir)") ap.add_argument("--sin-dkim", action="store_true", help="no probar selectores DKIM (unas 35 consultas menos por dominio)") ap.add_argument("--csv", metavar="FICHERO", help="guarda un resumen por dominio en CSV") ap.add_argument("--json", metavar="FICHERO", help="guarda el informe completo en JSON") ap.add_argument("--hilos", type=int, default=4, help="dominios en paralelo (por defecto 4)") ap.add_argument("--resumen", action="store_true", help="una línea por dominio, sin detalle") a = ap.parse_args(argv) entradas = list(a.dominios) + (leer_lista(a.lista) if a.lista else []) if not entradas: ap.error("indica al menos un dominio o --lista") dominios, resultados = [], [] for e in entradas: try: d = normalizar_dominio(e) if d not in dominios: dominios.append(d) except ValueError as err: resultados.append({"dominio": str(e), "error": str(err)}) resolve = ResolutorDoH() def uno(d): try: return auditar(d, resolve, a.selector, dkim=not a.sin_dkim) except Exception as err: return {"dominio": d, "error": str(err)} color = sys.stdout.isatty() print(f"Auditando {len(dominios)} dominio{'s' if len(dominios) != 1 else ''} por DNS-over-HTTPS…\n") with ThreadPoolExecutor(max_workers=max(1, min(a.hilos, 16))) as ex: for r in ex.map(uno, dominios): resultados.append(r) imprimir(r, detalle=not a.resumen, color=color) ok = [r for r in resultados if "error" not in r] cuenta = {v: sum(1 for r in ok if r["veredicto"] == v) for v in ESTADO} print(f"\n{cuenta['protegido']} protegidos · {cuenta['parcial']} parciales · {cuenta['suplantable']} suplantables" f" · {len(resultados) - len(ok)} con error · {resolve.consultas} consultas DNS") if a.csv: with open(a.csv, "w", newline="", encoding="utf-8-sig") as f: w = csv.DictWriter(f, fieldnames=COLUMNAS_CSV, restval="") w.writeheader() for r in resultados: w.writerow(fila_csv(r)) print(f"CSV guardado en {a.csv}") if a.json: with open(a.json, "w", encoding="utf-8") as f: json.dump({"formato": FORMATO, "fecha": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "resultados": resultados}, f, ensure_ascii=False, indent=2) print(f"JSON guardado en {a.json}") return codigo_salida(resultados) if __name__ == "__main__": sys.exit(main())