#!/usr/bin/env python3
"""altruia-suplantacion — mide si un dominio puede ser suplantado por correo.

Esta es LA MISMA herramienta con la que ALTRUIA mide el informe público de
suplantación en España (altruia.es/informe-suplantacion). Se publica entera para
que cualquiera pueda reproducir la cifra en lugar de tener que creérsela.

Ningún fabricante de este sector publica su instrumento de medida: sus porcentajes
son cajas negras. Un dato que no se puede verificar vale menos que uno pequeño que
sí, así que aquí está el nuestro.

QUÉ MIDE
    Si un tercero puede enviar correo poniendo tu dominio EXACTO en el remitente.
    Un dominio se cuenta como suplantable cuando no publica una política DMARC que
    ordene bloquear (cuarentena o rechazo) los mensajes que no se autentican:
      · sin registro DMARC            → suplantable
      · DMARC con p=none              → suplantable (observa, pero deja pasar)
      · DMARC con p=quarantine/reject → no suplantable
    Es exactamente la regla del informe, sin excepciones ni ajustes.

QUÉ **NO** MIDE
    · No mide incidentes: que un dominio sea suplantable no significa que lo hayan
      atacado, igual que una puerta sin cerrar no implica un robo.
    · No detecta dominios *parecidos* al tuyo registrados por otro.
    · No detecta suplantación solo del nombre visible con otra dirección detrás.
    · No dice si tu correo llega bien ni si estás en listas negras.

CÓMO FUNCIONA
    Consulta pública y pasiva: una única consulta DNS de tipo TXT a `_dmarc.<dominio>`.
    No envía correo, no escanea puertos, no accede a nada privado y no toca ningún
    sistema de terceros. Es legal y es lo que cualquiera puede ver desde fuera.

USO
    python3 altruia_suplantacion.py ejemplo.es
    python3 altruia_suplantacion.py ejemplo.es otro.es --json
    python3 altruia_suplantacion.py --lista dominios.txt --resumen
    python3 altruia_suplantacion.py --lista dominios.txt --csv salida.csv

    `--lista` acepta un fichero con un dominio por línea (o un CSV cuya primera
    columna sea el dominio). Las líneas que empiezan por # se ignoran.

REQUISITOS
    Python 3.9+ y nada más. Usa `dnspython` si está instalado y, si no, cae a
    DNS-over-HTTPS con la biblioteca estándar, para que funcione en cualquier
    máquina sin instalar dependencias.

LICENCIA
    Dominio público / CC0. Cópialo, cámbialo y publica tus propios datos.
    Si publicas resultados obtenidos con esta herramienta, cita la versión de la
    metodología que aparece abajo: sin ella los números no son comparables.
"""
from __future__ import annotations

import argparse
import concurrent.futures
import csv
import json
import re
import sys
import urllib.parse
import urllib.request

# Subir esta versión CUALQUIER vez que cambie la regla de decisión. Dos cifras
# calculadas con versiones distintas no son comparables, y decirlo es parte del
# dato.
METODOLOGIA = "1.0"

# Los mismos resolvers que usa el informe. Se fijan a propósito: el resolver del
# sistema puede tener caché sucia o respuestas filtradas, y eso movería la cifra.
RESOLVERS = ("1.1.1.1", "8.8.8.8")
DOH = ("https://cloudflare-dns.com/dns-query", "https://dns.google/resolve")
TIMEOUT = 6.0

SUPLANTABLE = "suplantable"
PROTEGIDO = "protegido"
PARCIAL = "parcial"
ERROR = "error"

_EXPLICA = {
    SUPLANTABLE: "Un tercero puede enviar correo con este dominio en el remitente.",
    PROTEGIDO: "El dominio pide que se bloquee el correo que no se autentica.",
    PARCIAL: "Consultado, pero la respuesta no permite decidir con seguridad.",
    ERROR: "No se ha podido consultar el dominio.",
}


# ── Resolución DNS ──────────────────────────────────────────────────────────

def _txt_dnspython(nombre: str) -> list[str] | None:
    """Devuelve los TXT, o None si dnspython no está disponible."""
    try:
        import dns.resolver  # type: ignore
    except ImportError:
        return None
    try:
        r = dns.resolver.Resolver()
        r.timeout = TIMEOUT
        r.lifetime = TIMEOUT * 2
        r.nameservers = list(RESOLVERS)
        return ["".join(s.decode() if isinstance(s, bytes) else s for s in rr.strings)
                for rr in r.resolve(nombre, "TXT")]
    except Exception:
        # NXDOMAIN, NoAnswer, timeout… todo significa lo mismo aquí: sin TXT.
        return []


def _txt_doh(nombre: str) -> list[str]:
    """DNS-over-HTTPS con la biblioteca estándar: cero dependencias."""
    for base in DOH:
        url = f"{base}?{urllib.parse.urlencode({'name': nombre, 'type': 'TXT'})}"
        req = urllib.request.Request(url, headers={"accept": "application/dns-json"})
        try:
            with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
                datos = json.loads(resp.read().decode())
        except Exception:
            continue
        salida = []
        for ans in datos.get("Answer") or []:
            if ans.get("type") != 16:      # 16 = TXT
                continue
            # El JSON entrega el TXT entrecomillado y, si venía troceado en varias
            # cadenas, concatenadas con comillas por medio.
            salida.append(re.sub(r'"\s*"', "", (ans.get("data") or "").strip('"')))
        return salida
    raise RuntimeError("ningún resolver DNS-over-HTTPS respondió")


def txt(nombre: str) -> list[str]:
    v = _txt_dnspython(nombre)
    return _txt_doh(nombre) if v is None else v


# ── Regla de decisión ───────────────────────────────────────────────────────

def politica_dmarc(registros: list[str]) -> tuple[str | None, str | None]:
    """Extrae (política, registro) del TXT de DMARC.

    Devuelve (None, None) si no hay registro DMARC. Si lo hay pero no declara p=,
    la política efectiva es `none`: así lo dice la especificación y así lo trata
    un receptor real.
    """
    reg = next((t for t in registros if t.strip().lower().startswith("v=dmarc1")), None)
    if reg is None:
        return None, None
    m = re.search(r"\bp\s*=\s*(\w+)", reg, re.I)
    return (m.group(1).lower() if m else "none"), reg


def analizar(dominio: str) -> dict:
    """Veredicto para un dominio. Nunca lanza: los fallos se devuelven como dato."""
    dominio = dominio.strip().lower().rstrip(".")
    dominio = re.sub(r"^https?://", "", dominio).split("/")[0].split("@")[-1]
    if not dominio or "." not in dominio:
        return {"dominio": dominio, "veredicto": ERROR, "politica": None,
                "registro": None, "detalle": "no parece un dominio"}
    try:
        registros = txt(f"_dmarc.{dominio}")
    except Exception as exc:
        return {"dominio": dominio, "veredicto": ERROR, "politica": None,
                "registro": None, "detalle": f"fallo de consulta: {exc}"}

    politica, registro = politica_dmarc(registros)
    if politica is None:
        veredicto, detalle = SUPLANTABLE, "sin registro DMARC"
    elif politica == "none":
        veredicto, detalle = SUPLANTABLE, "DMARC en modo observación (p=none): no bloquea"
    elif politica in ("quarantine", "reject"):
        veredicto, detalle = PROTEGIDO, f"DMARC con p={politica}"
    else:
        # Un valor de p= no reconocido: no se cuenta como protegido ni como
        # suplantable. Inventar un veredicto aquí sería ensuciar el agregado.
        veredicto, detalle = PARCIAL, f"política no reconocida: p={politica}"

    return {"dominio": dominio, "veredicto": veredicto, "politica": politica,
            "registro": registro, "detalle": detalle}


# ── Agregado ────────────────────────────────────────────────────────────────

def resumir(resultados: list[dict]) -> dict:
    """Agrega igual que el informe: los errores salen del denominador.

    Un dominio que no se ha podido consultar no es ni seguro ni inseguro. Contarlo
    en cualquiera de los dos lados falsearía el porcentaje.
    """
    medidos = [r for r in resultados if r["veredicto"] in (SUPLANTABLE, PROTEGIDO, PARCIAL)]
    n = len(medidos)
    sup = sum(1 for r in medidos if r["veredicto"] == SUPLANTABLE)
    sin_dmarc = sum(1 for r in medidos if r["politica"] is None
                    and r["veredicto"] == SUPLANTABLE)
    solo_observa = sum(1 for r in medidos if r["politica"] == "none")
    return {
        "metodologia": METODOLOGIA,
        "consultados": len(resultados),
        "medidos": n,
        "no_medibles": len(resultados) - n,
        "suplantables": sup,
        "pct_suplantables": round(100 * sup / n) if n else 0,
        "sin_dmarc": sin_dmarc,
        "pct_sin_dmarc": round(100 * sin_dmarc / n) if n else 0,
        "solo_observacion": solo_observa,
        "pct_solo_observacion": round(100 * solo_observa / n) if n else 0,
    }


def _cargar_lista(ruta: str) -> list[str]:
    dominios = []
    with open(ruta, encoding="utf-8", errors="replace") as fh:
        for linea in fh:
            linea = linea.strip()
            if not linea or linea.startswith("#"):
                continue
            dominios.append(linea.split(",")[0].split(";")[0].strip().strip('"'))
    # Sin duplicados y conservando el orden: un dominio repetido inflaría el total.
    return list(dict.fromkeys(d for d in dominios if d))


# ── CLI ─────────────────────────────────────────────────────────────────────

_SIMBOLO = {SUPLANTABLE: "SUPLANTABLE", PROTEGIDO: "protegido",
            PARCIAL: "sin decidir", ERROR: "error"}


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(
        description="Mide si un dominio puede ser suplantado por correo. "
                    "Consulta pública y pasiva; no envía correo ni escanea nada.",
        epilog=f"Metodología {METODOLOGIA} · altruia.es/metodologia")
    ap.add_argument("dominios", nargs="*", help="uno o más dominios")
    ap.add_argument("--lista", metavar="FICHERO",
                    help="fichero con un dominio por línea (o CSV con el dominio en la 1ª columna)")
    ap.add_argument("--csv", metavar="SALIDA", help="escribe el detalle por dominio en CSV")
    ap.add_argument("--json", action="store_true", help="salida en JSON")
    ap.add_argument("--resumen", action="store_true", help="solo el agregado")
    ap.add_argument("--hilos", type=int, default=12, help="consultas en paralelo (por defecto 12)")
    args = ap.parse_args(argv)

    dominios = list(args.dominios)
    if args.lista:
        dominios += _cargar_lista(args.lista)
    dominios = list(dict.fromkeys(dominios))
    if not dominios:
        ap.error("indica al menos un dominio, o --lista FICHERO")

    hilos = max(1, min(args.hilos, 32))
    with concurrent.futures.ThreadPoolExecutor(max_workers=hilos) as pool:
        resultados = list(pool.map(analizar, dominios))

    resumen = resumir(resultados)

    if args.csv:
        with open(args.csv, "w", newline="", encoding="utf-8") as fh:
            w = csv.writer(fh)
            w.writerow([f"# altruia-suplantacion metodologia={METODOLOGIA}"])
            w.writerow(["dominio", "veredicto", "politica_dmarc", "detalle"])
            for r in resultados:
                w.writerow([r["dominio"], r["veredicto"], r["politica"] or "", r["detalle"]])
        print(f"CSV escrito en {args.csv}", file=sys.stderr)

    if args.json:
        print(json.dumps({"resumen": resumen,
                          "resultados": [] if args.resumen else resultados},
                         ensure_ascii=False, indent=2))
        return 0

    if not args.resumen:
        ancho = max((len(r["dominio"]) for r in resultados), default=10)
        for r in resultados:
            print(f"{r['dominio']:<{ancho}}  {_SIMBOLO[r['veredicto']]:<12}  {r['detalle']}")
        print()

    if len(dominios) > 1 or args.resumen:
        print(f"Metodología {resumen['metodologia']} · {resumen['medidos']} dominios medidos"
              + (f" ({resumen['no_medibles']} no medibles, fuera del cálculo)"
                 if resumen["no_medibles"] else ""))
        print(f"  suplantables:        {resumen['suplantables']:>5}  ({resumen['pct_suplantables']}%)")
        print(f"    · sin DMARC:       {resumen['sin_dmarc']:>5}  ({resumen['pct_sin_dmarc']}%)")
        print(f"    · solo observación:{resumen['solo_observacion']:>5}  ({resumen['pct_solo_observacion']}%)")
        print("\nMide exposición, no incidentes: que un dominio sea suplantable no")
        print("significa que haya sido atacado.")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())