geturls.py

#!/usr/bin/env python3
"""
nmap2urls.py

Takes an nmap XML file and prints a list of URLs discovered.
Optionally fetches each URL and extracts Title, Server banner, and detects login/auth.

Usage examples:
  python3 nmap2urls.py --info -t 5 scan.xml
  python3 nmap2urls.py -v --workers 20 scan.xml
"""
from __future__ import annotations

import argparse
import csv
import html
import re
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Optional, Tuple, List

import requests
import xmltodict
from requests.adapters import HTTPAdapter
from requests.exceptions import ConnectionError, Timeout, RequestException
from urllib3.util import Retry

# Quiet insecure warnings by default (existing behaviour)
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# ---- CLI ----
parser = argparse.ArgumentParser(
    description=(
        "Take an nmap XML file and emit URLs. "
        "With --info, perform GET requests and output CSV with title, server banner and "
        "a basic 'login detected' heuristic."
    )
)
parser.add_argument("nmapxmlfile", type=argparse.FileType("r"), help="Nmap XML file")
parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output to stderr")
parser.add_argument("-t", "--timeout", type=float, default=3.0, help="Timeout for HTTP GET (seconds)")
parser.add_argument("--info", action="store_true", help="Fetch each URL and output CSV with extra info")
parser.add_argument("--verify", action="store_true", help="Verify TLS certificates (default: OFF)")
parser.add_argument("--workers", type=int, default=10, help="Number of concurrent workers when fetching (default: 10)")
parser.add_argument("--max-retries", type=int, default=2, help="Number of retries for each request (default: 2)")
args = parser.parse_args()

# ---- Regexes ----
TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)
LOGIN_RE = re.compile(
    r"(<input[^>]+type\s*=\s*['\"]?password['\"]?[^>]*>|"
    r"\b(login|log in|sign in|sign-in|signin|authenti?cate)\b)",
    re.IGNORECASE,
)


def eprint(*vals, **kwargs):
    print(*vals, file=sys.stderr, **kwargs)


# ---- Helpers ----
def build_session(max_retries: int) -> requests.Session:
    """
    Build a requests.Session with a Retry-backed adapter.
    This function auto-detects whether urllib3's Retry expects 'allowed_methods'
    (newer versions) or 'method_whitelist' (older versions).
    """
    sess = requests.Session()
    retry_args = {
        "total": max_retries,
        "backoff_factor": 0.5,
        "status_forcelist": (429, 500, 502, 503, 504),
        "raise_on_status": False,
    }

    # Try newer parameter name first, fall back to older if TypeError is raised.
    try:
        retry_args["allowed_methods"] = ("GET", "HEAD")
        retries = Retry(**retry_args)
    except TypeError:
        retry_args.pop("allowed_methods", None)
        retry_args["method_whitelist"] = ("GET", "HEAD")
        retries = Retry(**retry_args)

    adapter = HTTPAdapter(max_retries=retries)
    sess.mount("http://", adapter)
    sess.mount("https://", adapter)
    sess.headers.update(
        {
            "User-Agent": "nmap2urls/1.0 (+https://example.invalid/)",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        }
    )
    return sess


def ip_repr(addr: str) -> str:
    """Return an IP suitable for embedding in a URL (bracket IPv6)."""
    if ":" in addr and not addr.startswith("["):
        return f"[{addr}]"
    return addr


def maybe_strip_default_port(proto: str, port: int) -> str:
    """Return '' if port is default for proto so the URL will omit :port."""
    if (proto == "http" and port == 80) or (proto == "https" and port == 443):
        return ""
    return f":{port}"


def find_ipv4_or_ipaddress(host: dict) -> Optional[str]:
    """Return first IPv4 or first address found in nmap host dict."""
    addrs = host.get("address")
    if addrs is None:
        return None
    if isinstance(addrs, list):
        for a in addrs:
            if a.get("@addrtype") in ("ipv4", "ipv6", "mac", "ip"):
                return a.get("@addr")
        return addrs[0].get("@addr")
    else:
        return addrs.get("@addr")


def hostname_from_nmap(host: dict) -> Optional[str]:
    """Extract hostname (if any) from nmap hostnames structure."""
    hns = host.get("hostnames")
    if not hns:
        return None
    hn = hns.get("hostname")
    if not hn:
        return None
    if isinstance(hn, list):
        first = hn[0]
    else:
        first = hn
    return first.get("@name")


# ---- Parse XML ----
xml_text = args.nmapxmlfile.read()
try:
    nmap = xmltodict.parse(
        xml_text,
        dict_constructor=dict,
        force_list=("host", "address", "port", "hostname", "portservice"),
    )
except Exception as exc:
    eprint("Error parsing XML document:", exc)
    sys.exit(2)

hosts = nmap.get("nmaprun", {}).get("host", [])
if not isinstance(hosts, list):
    hosts = [hosts]

# Build a list of (url, meta) to process
targets: List[Tuple[str, dict]] = []

for host in hosts:
    ip = find_ipv4_or_ipaddress(host)
    if not ip:
        if args.verbose:
            eprint("Skipping host with no address element in XML")
        continue

    ports_container = host.get("ports", {})
    ports = ports_container.get("port")
    if not ports:
        if args.verbose:
            eprint(f"No ports for host {ip}, skipping")
        continue
    if not isinstance(ports, list):
        ports = [ports]

    for p in ports:
        state = p.get("state", {}).get("@state")
        if state != "open":
            continue

        try:
            portnum = int(p.get("@portid"))
        except (TypeError, ValueError):
            continue

        servicename = None
        svc = p.get("service")
        if svc:
            servicename = svc.get("@name")

        # Determine proto
        proto = None
        if servicename in ("http", "http-alt", "www", "http-proxy"):
            proto = "http"
            if svc and svc.get("@tunnel") == "ssl":
                proto = "https"
        elif servicename in ("https", "ssl/http"):
            proto = "https"
        else:
            # Best-effort fallback based on port number
            if portnum == 80:
                proto = "http"
            elif portnum == 443:
                proto = "https"

        if not proto:
            continue

        host_display = hostname_from_nmap(host) or ip
        port_str = maybe_strip_default_port(proto, portnum)
        url = f"{proto}://{ip_repr(ip)}{port_str}/"
        targets.append((url, {"ip": ip, "port": portnum, "proto": proto, "host": host_display}))

# If --info, prepare CSV header
if args.info:
    csv_writer = csv.writer(sys.stdout)
    csv_writer.writerow(["URL", "Title", "Server Banner", "Login Detected", "Notes"])

# If not info, just print URLs (one per line)
if not args.info:
    for url, _meta in targets:
        print(url)
    sys.exit(0)

# ---- Fetching logic ----
session = build_session(max_retries=args.max_retries)


def fetch_info(url: str, timeout: float, verify: bool) -> Tuple[str, Optional[str], Optional[str], bool, str]:
    """
    Returns tuple: (url, title_or_None, server_or_None, login_detected_bool, notes)
    Notes contains error messages or empty string.
    """
    title = None
    server = None
    login = False
    notes = ""

    try:
        resp = session.get(url, timeout=timeout, verify=verify, allow_redirects=True)
        text = resp.text or ""
        m = TITLE_RE.search(text)
        if m:
            title = html.unescape(m.group(1).strip())

        server = resp.headers.get("server")

        # Basic auth header (case-insensitive) and login form detection
        if any(h.lower() == "www-authenticate" for h in resp.headers):
            login = True
        if LOGIN_RE.search(text):
            login = True

    except Timeout:
        notes = "ERROR - request timed out"
    except ConnectionError as ce:
        notes = f"ERROR - connection failed ({ce.__class__.__name__})"
    except RequestException as rexc:
        notes = f"ERROR - request exception: {rexc}"
    except Exception as exc:
        notes = f"ERROR - unexpected: {exc}"

    return url, title, server, login, notes


# Concurrent fetch
workers = max(1, args.workers)
futures = []
with ThreadPoolExecutor(max_workers=workers) as ex:
    for url, meta in targets:
        futures.append(ex.submit(fetch_info, url, args.timeout, args.verify))

    for fut in as_completed(futures):
        url, title, server, login, notes = fut.result()
        csv_writer.writerow(
            [
                url,
                title or "",
                server or "",
                "yes" if login else "no",
                notes or "",
            ]
        )
        try:
            sys.stdout.flush()
        except Exception:
            pass

if args.verbose:
    eprint(f"Processed {len(targets)} targets with {workers} worker(s).")