#!/usr/bin/env python3
"""
Recolector de la configuración de seguridad de una web — Ciberkit
=======================================================================

Reúne en un JSON lo que el navegador no puede ver por sí mismo y lo evalúa la
página «¿Qué tan blindada está tu web?» de la suite:

  - Cabeceras de la respuesta final y cadena de redirecciones.
  - Si http:// redirige a https://.
  - Versiones de TLS aceptadas (1.0, 1.1, 1.2, 1.3) y validez del certificado.
  - CORS: si la web refleja un origen ajeno (prueba con https://evil.example).
  - Métodos anunciados (OPTIONS) y /.well-known/security.txt.
  - Ficheros que no deberían ser públicos (.git, .env, copias de seguridad,
    server-status…). Se confirma por el CONTENIDO, no solo por el código 200,
    para no dar falsos positivos en webs que responden 200 a todo.

Hace unas 20 peticiones normales (GET, HEAD y OPTIONS), sin cargas de ataque.

    Úsalo SOLO en webs propias o con autorización.

Sin dependencias: Python 3.7+.

Ejemplos
--------
  python recoger_web.py https://tuweb.com
  python recoger_web.py https://tuweb.com/login --salida login.json
  python recoger_web.py https://tuweb.com --sin-ficheros      # sin comprobar ficheros expuestos
"""

import argparse
import datetime
import json
import re
import socket
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
import warnings

FORMATO = "suite-redes.web/1"
UA = "suite-redes-web/1 (auditoria autorizada)"
ORIGEN_PRUEBA = "https://evil.example"
MAX_CUERPO = 65536

# Ruta, qué es, y cómo se confirma que el fichero existe de verdad (por su contenido)
EXPOSICIONES = [
    ("/.git/HEAD", "repositorio Git", lambda b: re.match(rb"^(ref: refs/|[0-9a-f]{40}\s*$)", b) is not None),
    ("/.git/config", "configuración de Git", lambda b: b"[core]" in b),
    ("/.env", "variables de entorno", lambda b: not b.lstrip().startswith(b"<") and re.search(rb"(?m)^[A-Z][A-Z0-9_]{2,}=", b) is not None),
    ("/.htpasswd", "contraseñas de Apache", lambda b: not b.lstrip().startswith(b"<") and re.search(rb"(?m)^[\w.-]+:(\$apr1\$|\$2[aby]\$|\{SHA\}|[./0-9A-Za-z]{13}$)", b) is not None),
    ("/.DS_Store", "índice de carpetas de macOS", lambda b: b[:8] == b"\x00\x00\x00\x01Bud1"),
    ("/wp-config.php.bak", "copia de la configuración de WordPress", lambda b: b"<?php" in b and b"DB_" in b),
    ("/config.php.bak", "copia de configuración PHP", lambda b: b"<?php" in b),
    ("/backup.sql", "volcado de base de datos", lambda b: re.search(rb"CREATE TABLE|INSERT INTO", b, re.I) is not None and not b.lstrip().startswith(b"<")),
    ("/database.sql", "volcado de base de datos", lambda b: re.search(rb"CREATE TABLE|INSERT INTO", b, re.I) is not None and not b.lstrip().startswith(b"<")),
    ("/server-status", "estado de Apache", lambda b: b"Apache Server Status" in b),
    ("/phpinfo.php", "phpinfo()", lambda b: b"PHP Version" in b and b"phpinfo" in b.lower()),
    ("/actuator/env", "variables de Spring Boot", lambda b: b'"propertySources"' in b),
]


def contexto_auditoria():
    """Las peticiones de auditoría no validan el certificado: así se puede auditar una web con el
    certificado caducado o de otro dominio (justo cuando más hace falta). No se envían credenciales
    ni datos. La validez del certificado se comprueba aparte, en certificado(), y se informa."""
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


class SinRedirecciones(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def peticion(url, metodo="GET", cabeceras=None, timeout=10, leer=True, contexto=None):
    """Devuelve (status, [(cabecera, valor)], cuerpo) sin seguir redirecciones; los 4xx/5xx también."""
    req = urllib.request.Request(url, method=metodo, headers={"User-Agent": UA, **(cabeceras or {})})
    opener = urllib.request.build_opener(SinRedirecciones, urllib.request.HTTPSHandler(context=contexto or contexto_auditoria()))
    try:
        r = opener.open(req, timeout=timeout)
    except urllib.error.HTTPError as e:
        r = e
    try:
        cuerpo = r.read(MAX_CUERPO) if leer and metodo != "HEAD" else b""
        status = getattr(r, "status", None) or r.getcode()
        return status, list(r.headers.items()), cuerpo
    finally:
        r.close()


def cabecera(lista, nombre):
    return next((v for k, v in lista if k.lower() == nombre.lower()), None)


def seguir(url, timeout, contexto=None, max_saltos=10):
    """Sigue las redirecciones a mano para guardar la cadena."""
    cadena = []
    actual = url
    for _ in range(max_saltos):
        status, headers, _ = peticion(actual, timeout=timeout, leer=False, contexto=contexto)
        loc = cabecera(headers, "Location")
        if status in (301, 302, 303, 307, 308) and loc:
            cadena.append({"url": actual, "status": status, "location": loc})
            actual = urllib.parse.urljoin(actual, loc)
            continue
        return actual, status, headers, cadena
    raise RuntimeError("demasiadas redirecciones")


# ───────────────────────── TLS ─────────────────────────

VERSIONES = [("TLSv1", "TLSv1"), ("TLSv1.1", "TLSv1_1"), ("TLSv1.2", "TLSv1_2"), ("TLSv1.3", "TLSv1_3")]


def probar_version(host, puerto, nombre_attr, timeout=8):
    """True si el servidor acepta esa versión, False si la rechaza, None si este equipo no puede probarla."""
    version = getattr(ssl.TLSVersion, nombre_attr, None)
    if version is None:
        return None
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE  # solo se mide qué versión acepta; el certificado se valida en certificado()
    try:
        with warnings.catch_warnings():  # probar TLS 1.0/1.1 es justo lo que queremos: sin avisos de obsolescencia
            warnings.simplefilter("ignore", DeprecationWarning)
            ctx.minimum_version = version
            ctx.maximum_version = version
        if nombre_attr in ("TLSv1", "TLSv1_1"):
            ctx.set_ciphers("ALL:@SECLEVEL=0")
    except (ValueError, ssl.SSLError):
        return None
    try:
        with socket.create_connection((host, puerto), timeout=timeout) as s:
            with ctx.wrap_socket(s, server_hostname=host):
                return True
    except ssl.SSLError as e:
        texto = str(e).upper()
        if "NO_PROTOCOLS_AVAILABLE" in texto or "NO PROTOCOLS AVAILABLE" in texto:
            return None  # la librería local no permite esa versión: no sabemos qué haría el servidor
        return False  # el servidor la rechaza (alerta o responde con otra versión)
    except (OSError, socket.timeout):
        return False


def resumen_certificado(cert, ahora=None):
    """Datos útiles de getpeercert()."""
    ahora = ahora or datetime.datetime.now(datetime.timezone.utc)
    fin = datetime.datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z").replace(tzinfo=datetime.timezone.utc)
    emisor = dict(x[0] for x in cert.get("issuer", ()))
    return {
        "not_after": fin.strftime("%Y-%m-%d"), "days_left": (fin - ahora).days,
        "issuer": emisor.get("organizationName") or emisor.get("commonName"),
        "san": [v for k, v in cert.get("subjectAltName", ()) if k == "DNS"][:20],
    }


def certificado(host, puerto, timeout=8):
    ctx = ssl.create_default_context()
    try:
        with socket.create_connection((host, puerto), timeout=timeout) as s:
            with ctx.wrap_socket(s, server_hostname=host) as t:
                return resumen_certificado(t.getpeercert())
    except ssl.SSLCertVerificationError as e:
        return {"error": getattr(e, "verify_message", None) or str(e)}
    except (OSError, ssl.SSLError) as e:
        return {"error": str(e)}


# ───────────────────────── Recogida ─────────────────────────

def recoger(url, timeout=10, ficheros=True, tls=True, contexto=None, avisar=None):
    avisar = avisar or (lambda m: None)
    p = urllib.parse.urlsplit(url)
    if p.scheme not in ("http", "https") or not p.hostname:
        raise ValueError("La URL debe empezar por http:// o https://")
    base = "{}://{}".format(p.scheme, p.netloc)
    avisar("Cabeceras y redirecciones…")
    final, status, headers, cadena = seguir(url, timeout, contexto)
    fu = urllib.parse.urlsplit(final)
    origen_final = "{}://{}".format(fu.scheme, fu.netloc)
    datos = {
        "formato": FORMATO, "generado": datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
        "url": url, "final_url": final, "status": status, "headers": headers, "redirects": cadena,
    }

    # http:// → https://
    if p.scheme == "http":
        datos["http_redirect"] = {"status": cadena[0]["status"], "location": cadena[0]["location"]} if cadena else {"status": status, "location": None}
    else:
        try:
            st, hs, _ = peticion("http://{}/".format(p.hostname), timeout=timeout, leer=False, contexto=contexto)
            datos["http_redirect"] = {"status": st, "location": cabecera(hs, "Location")}
        except (OSError, urllib.error.URLError):
            datos["http_redirect"] = None

    # TLS y certificado
    if tls and fu.scheme == "https":
        avisar("TLS y certificado…")
        puerto = fu.port or 443
        versiones, sin_probar = {}, []
        for etiqueta, attr in VERSIONES:
            r = probar_version(fu.hostname, puerto, attr, timeout)
            if r is None:
                sin_probar.append(etiqueta)
            else:
                versiones[etiqueta] = r
        datos["tls"] = {"versions": versiones, "untested": sin_probar, "cert": certificado(fu.hostname, puerto, timeout)}

    # CORS
    avisar("CORS…")
    try:
        _, hs, _ = peticion(final, cabeceras={"Origin": ORIGEN_PRUEBA}, timeout=timeout, leer=False, contexto=contexto)
        acao = cabecera(hs, "Access-Control-Allow-Origin")
        datos["cors"] = {"origin": ORIGEN_PRUEBA, "acao": acao, "reflected": acao == ORIGEN_PRUEBA,
                         "acac": (cabecera(hs, "Access-Control-Allow-Credentials") or "").lower() == "true"}
    except (OSError, urllib.error.URLError):
        datos["cors"] = None

    # OPTIONS y security.txt
    try:
        _, hs, _ = peticion(final, metodo="OPTIONS", timeout=timeout, leer=False, contexto=contexto)
        datos["options"] = {"allow": cabecera(hs, "Allow")}
    except (OSError, urllib.error.URLError):
        datos["options"] = None
    try:
        st, _, cuerpo = peticion(origen_final + "/.well-known/security.txt", timeout=timeout, contexto=contexto)
        datos["security_txt"] = st == 200 and re.search(rb"(?im)^contact:", cuerpo) is not None
    except (OSError, urllib.error.URLError):
        datos["security_txt"] = None

    # Ficheros expuestos (confirmados por contenido)
    if ficheros:
        avisar("Ficheros que no deberían ser públicos…")
        exp = []
        for ruta, que, valida in EXPOSICIONES:
            try:
                st, _, cuerpo = peticion(origen_final + ruta, timeout=timeout, contexto=contexto)
            except (OSError, urllib.error.URLError):
                continue
            exp.append({"path": ruta, "what": que, "status": st, "match": bool(st == 200 and valida(cuerpo))})
        datos["exposures"] = exp
    return datos


def preparar_salida():
    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, contexto=None):
    preparar_salida()
    ap = argparse.ArgumentParser(description="Recoge cabeceras, TLS, CORS y ficheros expuestos de TU web para analizarlos en la suite.")
    ap.add_argument("url", help="https://tuweb.com (o la URL de una página concreta, como el login)")
    ap.add_argument("--salida", help="fichero JSON (por defecto web-HOST.json)")
    ap.add_argument("--sin-ficheros", action="store_true", help="no comprobar ficheros expuestos")
    ap.add_argument("--sin-tls", action="store_true", help="no probar versiones de TLS")
    ap.add_argument("--timeout", type=float, default=10)
    a = ap.parse_args(argv)
    url = a.url if re.match(r"^https?://", a.url, re.I) else "https://" + a.url
    try:
        datos = recoger(url, a.timeout, not a.sin_ficheros, not a.sin_tls, contexto, avisar=lambda m: print(m, file=sys.stderr))
    except (ValueError, RuntimeError, OSError, urllib.error.URLError) as e:
        print("Error: no se pudo conectar con {}: {}".format(url, getattr(e, "reason", e)), file=sys.stderr)
        return 1
    salida = a.salida or "web-{}.json".format(re.sub(r"[^\w.-]", "_", urllib.parse.urlsplit(url).hostname))
    with open(salida, "w", encoding="utf-8") as f:
        json.dump(datos, f, ensure_ascii=False, indent=2)
    t = datos.get("tls") or {}
    print("\n{} → {} ({})".format(url, datos["final_url"], datos["status"]))
    if t:
        print("  TLS: " + ", ".join("{} {}".format(k, "sí" if v else "no") for k, v in t["versions"].items()) + (" · sin probar: " + ", ".join(t["untested"]) if t["untested"] else ""))
        c = t["cert"]
        print("  Certificado: " + (c["error"] if c.get("error") else "caduca el {} ({} días)".format(c["not_after"], c["days_left"])))
    expuestos = [e["path"] for e in datos.get("exposures", []) if e["match"]]
    print("  Ficheros expuestos: " + (", ".join(expuestos) if expuestos else "ninguno"))
    print("Guardado en {}. Cárgalo en Ciberkit -> Pentest web -> Cabeceras de seguridad.".format(salida))
    return 0


if __name__ == "__main__":
    sys.exit(main())