""" 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)