intelliscan/modules/crawler.py

"""
Module 1: Crawler

Authenticated BFS crawler that discovers all reachable forms and URL parameters
on a target web application.  Handles CSRF tokens transparently.
"""

from __future__ import annotations

import json
import logging
from collections import deque
from collections.abc import Iterable
from dataclasses import asdict, dataclass, field
from pathlib import Path
from urllib.parse import parse_qsl, urljoin, urlparse

from bs4 import BeautifulSoup

from intelliscan.config import (
    BLACKLISTED_PATHS,
    MAX_CRAWL_DEPTH,
    MAX_PAGES,
    RESULTS_DIR,
)
from intelliscan.utils.http_client import HttpClient

log = logging.getLogger(__name__)


@dataclass
class Form:
    """An HTML form discovered during crawling."""

    url: str
    action: str
    method: str
    inputs: list[dict[str, str]] = field(default_factory=list)


@dataclass
class UrlParam:
    """A URL with query parameters that may be injectable."""

    url: str
    params: list[str] = field(default_factory=list)


@dataclass
class CrawlResult:
    """Aggregated output of a crawl run."""

    target: str
    pages_visited: int
    forms: list[Form]
    url_params: list[UrlParam]
    blocked_responses: int = 0  # pages that returned HTTP >= 400 (WAF/auth/missing)
    start_reachable: bool = True  # False when the start URL itself returned >= 400

    def to_dict(self) -> dict:
        return {
            "target": self.target,
            "pages_visited": self.pages_visited,
            "forms": [asdict(f) for f in self.forms],
            "url_params": [asdict(p) for p in self.url_params],
            "blocked_responses": self.blocked_responses,
            "start_reachable": self.start_reachable,
        }

    def save(self, path: str | Path) -> Path:
        path = Path(path)
        path.write_text(json.dumps(self.to_dict(), indent=2), encoding="utf-8")
        return path


class Crawler:
    """
    BFS crawler with optional authentication.

    Examples
    --------
    >>> c = Crawler("http://localhost:8080", auth=("admin", "password"))
    >>> result = c.run()
    >>> print(f"Found {len(result.forms)} forms")
    """

    def __init__(
        self,
        target: str,
        auth: tuple[str, str] | None = None,
        max_pages: int = MAX_PAGES,
        max_depth: int = MAX_CRAWL_DEPTH,
        verify_tls: bool = True,
        timeout: int | None = None,
    ) -> None:
        from intelliscan.config import DEFAULT_TIMEOUT

        self.target = target.rstrip("/")
        self.auth = auth
        self.max_pages = max_pages
        self.max_depth = max_depth
        self.http = HttpClient(
            target,
            verify_tls=verify_tls,
            timeout=timeout if timeout is not None else DEFAULT_TIMEOUT,
        )

        self._domain = urlparse(self.target).netloc
        self._visited: set[str] = set()
        self._forms: list[Form] = []
        self._params: list[UrlParam] = []
        self._blocked: int = 0
        self._start_reachable: bool = True

    # ── public API ────────────────────────────────────────────────────────
    def run(self) -> CrawlResult:
        """Authenticate, set DVWA security to low, then BFS-crawl."""
        if self.auth:
            self._login()
            self._set_dvwa_security_low()

        self._bfs(self.target)

        log.info(
            "Crawl complete: %d pages, %d forms, %d url-params",
            len(self._visited),
            len(self._forms),
            len(self._params),
        )
        return CrawlResult(
            target=self.target,
            pages_visited=len(self._visited),
            forms=self._forms,
            url_params=self._params,
            blocked_responses=self._blocked,
            start_reachable=self._start_reachable,
        )

    # ── authentication ────────────────────────────────────────────────────
    def _login(self) -> None:
        if self.auth is None:
            return
        username, password = self.auth
        log.info("Authenticating as %s", username)
        token = self.http.get_csrf_token(f"{self.target}/login.php")
        data = {"username": username, "password": password, "Login": "Login"}
        if token:
            data["user_token"] = token
        r = self.http.post(f"{self.target}/login.php", data=data)
        if "logout" not in r.text.lower() and "login failed" in r.text.lower():
            raise RuntimeError("Authentication failed: check credentials")

    def _set_dvwa_security_low(self) -> None:
        """If target is DVWA, set security level to low."""
        url = f"{self.target}/security.php"
        try:
            token = self.http.get_csrf_token(url)
            if token:
                self.http.post(
                    url,
                    data={"security": "low", "seclev_submit": "Submit", "user_token": token},
                )
                log.debug("DVWA security set to low")
        except Exception as e:  # noqa: BLE001
            log.debug("DVWA security not configured (target may not be DVWA): %s", e)

    # ── BFS crawl ─────────────────────────────────────────────────────────
    def _bfs(self, start: str) -> None:
        queue: deque[tuple[str, int]] = deque([(start, 0)])

        while queue and len(self._visited) < self.max_pages:
            url, depth = queue.popleft()
            url = url.split("#")[0]  # strip fragment

            if url in self._visited or depth > self.max_depth:
                continue
            if not self._same_domain(url) or self._is_blacklisted(url):
                continue

            try:
                r = self.http.get(url)
            except Exception as e:  # noqa: BLE001
                log.debug("Skip %s: %s", url, e)
                if url == start:
                    self._start_reachable = False
                continue

            # Error-status pages (WAF block, auth wall, 404, rate-limit) carry no
            # real injectable content. Counting them as "visited" and parsing their
            # bodies is what let WAF/error pages leak into the results as phantom
            # findings, so skip them here and record that the target pushed back.
            if r.status_code >= 400:
                self._blocked += 1
                if url == start:
                    self._start_reachable = False
                log.debug("Skip %s: HTTP %d", url, r.status_code)
                continue

            self._visited.add(url)
            soup = BeautifulSoup(r.text, "html.parser")
            self._extract_forms(url, soup)
            self._extract_url_params(url)

            for next_url in self._extract_links(url, soup):
                if next_url not in self._visited:
                    queue.append((next_url, depth + 1))

    # ── extractors ────────────────────────────────────────────────────────
    def _extract_links(self, page_url: str, soup: BeautifulSoup) -> Iterable[str]:
        for a in soup.find_all("a", href=True):
            # str(): bs4 returns a list for multi-valued attributes; coerce defensively.
            href = urljoin(page_url, str(a["href"])).split("#")[0]
            if self._same_domain(href) and not self._is_blacklisted(href):
                yield href

    def _extract_forms(self, page_url: str, soup: BeautifulSoup) -> None:
        for f in soup.find_all("form"):
            action = urljoin(page_url, str(f.get("action") or page_url)).split("#")[0]
            method = str(f.get("method") or "GET").upper()
            inputs: list[dict[str, str]] = []

            for inp in f.find_all(("input", "textarea", "select")):
                name = inp.get("name")
                if name:
                    inputs.append(
                        {
                            "name": str(name),
                            "type": str(inp.get("type", "text")),
                            "value": str(inp.get("value", "")),
                        }
                    )

            if inputs:
                self._forms.append(Form(url=page_url, action=action, method=method, inputs=inputs))

    def _extract_url_params(self, url: str) -> None:
        parsed = urlparse(url)
        params = [name for name, _ in parse_qsl(parsed.query)]
        if params:
            self._params.append(UrlParam(url=url, params=params))

    # ── helpers ───────────────────────────────────────────────────────────
    def _same_domain(self, url: str) -> bool:
        return urlparse(url).netloc == self._domain

    @staticmethod
    def _is_blacklisted(url: str) -> bool:
        path = urlparse(url).path.lower()
        return any(b in path for b in BLACKLISTED_PATHS)


# ── CLI helper ────────────────────────────────────────────────────────────
def main(target: str, auth: str | None = None, output: str = "results.json") -> None:
    """CLI entry point: ``python -m intelliscan.modules.crawler --target ...``"""
    auth_t: tuple[str, str] | None = None
    if auth and ":" in auth:
        user, password = auth.split(":", 1)
        auth_t = (user, password)
    c = Crawler(target, auth=auth_t)
    result = c.run()
    out = result.save(RESULTS_DIR / output)
    log.info("Saved -> %s", out)


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser(description="IntelliScan crawler")
    p.add_argument("--target", required=True, help="Target URL (e.g. http://localhost:8080)")
    p.add_argument("--auth", help="user:password (optional)")
    p.add_argument("--output", default="results.json")
    args = p.parse_args()

    logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
    main(args.target, args.auth, args.output)