| """Keyless Retire.js sampling and a bounded Wapiti profile.""" |
| import asyncio |
| import json |
| import re |
| import tempfile |
| import time |
| from html.parser import HTMLParser |
| from pathlib import Path |
| from urllib.parse import urljoin, urlsplit |
|
|
| import httpx |
|
|
| from . import config |
| from .coverage import profile as coverage_settings |
| from .pipeline import execute |
| from .scope import public_host |
| from .web_modules import same_origin, read_report, alert |
|
|
|
|
| from .sampling import FetchProblem, fetch_page |
|
|
|
|
| async def fetch_sample(url, agent, maximum): |
| return await fetch_page(url, agent, maximum, public_host) |
|
|
|
|
| class ScriptInventory(HTMLParser): |
| def __init__(self, target, maximum=None): |
| super().__init__() |
| self.maximum = maximum or config.RETIRE_MAX_FILE_BYTES |
| self.inline_oversized = 0 |
| self.target, self.urls, self.inline = target, [], [] |
| self.current = None |
| self.inline_bytes = 0 |
|
|
| def handle_starttag(self, tag, attrs): |
| if tag != "script": |
| return |
| attrs = dict(attrs) |
| src = attrs.get("src") |
| if src: |
| try: |
| candidate = urljoin(self.target, src) |
| if same_origin(candidate, self.target): |
| self.urls.append(candidate) |
| except ValueError: |
| pass |
| self.current = None |
| elif str(attrs.get("type", "")).lower() in ("", "text/javascript", "application/javascript", "module"): |
| self.current = [] |
|
|
| def handle_data(self, data): |
| if self.current is not None and self.inline_bytes < self.maximum * 16: |
| self.current.append(data) |
| self.inline_bytes += len(data.encode("utf-8")) |
|
|
| def handle_endtag(self, tag): |
| if tag == "script" and self.current is not None: |
| code = "".join(self.current).encode("utf-8") |
| if code.strip() and len(code) <= self.maximum: |
| self.inline.append(code) |
| elif code.strip(): |
| self.inline_oversized += 1 |
| self.current = None |
|
|
|
|
| def parse_retire(report, mapping): |
| results = [] |
| entries = report.get("data", []) if isinstance(report, dict) else [] |
| if not isinstance(entries, list): |
| return [] |
| for entry in entries: |
| if not isinstance(entry, dict) or not isinstance(entry.get("file"), str): |
| continue |
| target = mapping.get(str(Path(entry["file"]).resolve())) |
| if target is None: |
| continue |
| for component in entry.get("results", []): |
| if not isinstance(component, dict): |
| continue |
| for vuln in component.get("vulnerabilities", []): |
| if not isinstance(vuln, dict): |
| continue |
| ids = vuln.get("identifiers") or {} |
| if not isinstance(ids, dict): |
| ids = {} |
| cves = ids.get("CVE") or [] |
| references = ", ".join(str(v) for v in cves[:8]) if isinstance(cves, list) else str(cves) |
| risk = str(vuln.get("severity") or "info").title() |
| if risk not in ("Low", "Medium", "High", "Critical"): |
| risk = "Informational" |
| title = str(component.get("component") or "JavaScript library") + " " + str(component.get("version") or "") |
| description = ("Unverified library/version advisory match. " + str(ids.get("summary") or vuln.get("details") or "") + |
| ("; " + references if references else "")) |
| results.append(alert("retirejs", title, target, description, risk, references)) |
| return results[:config.WEB_FINDINGS_LIMIT] |
|
|
|
|
| async def retirejs(url, host, agent, delay, discovered, profile="current"): |
| settings = coverage_settings(profile) |
| limit = config.RETIRE_MAX_FILES if profile == "current" else settings["retire_files"] |
| maximum = config.RETIRE_MAX_FILE_BYTES if profile == "current" else settings["retire_bytes"] |
| attempts_limit = limit if profile == "current" else settings["retire_attempts"] |
| reasons, samples = [], [] |
| policy_limits = set() |
| byte_limit = settings["retire_total_bytes"] |
| sampled_bytes = 0 |
| parser = ScriptInventory(url, maximum) |
| started = time.monotonic() |
| coverage = {"profile": profile, "max_files": limit, "max_file_bytes": maximum, |
| "attempted_scripts": 0, "oversized": 0, "fetch_failures": 0, |
| "inline_available": 0, "external_available": 0, |
| "max_total_bytes": byte_limit, "missing_inventory_scripts": 0} |
|
|
| def add_sample(source, body): |
| nonlocal sampled_bytes |
| if sampled_bytes + len(body) > byte_limit: |
| policy_limits.add("total_bytes") |
| return False |
| samples.append((source, body)) |
| sampled_bytes += len(body) |
| return True |
|
|
| async def sample(candidate, cap): |
| remaining = settings["retire_sampling_seconds"] - (time.monotonic() - started) |
| if remaining <= 0: |
| raise asyncio.TimeoutError() |
| return await asyncio.wait_for(fetch_sample(candidate, agent, cap), remaining) |
|
|
| try: |
| page_cap = config.WEB_PREFLIGHT_MAX_BYTES if profile == "current" else maximum |
| body, content_type, page_url = await sample(url, page_cap) |
| parser = ScriptInventory(page_url, maximum) |
| if "html" in content_type or body.lstrip().lower().startswith((b"<!doctype html", b"<html")): |
| parser.feed(body.decode("utf-8", "replace")) |
| elif "javascript" in content_type or urlsplit(url).path.lower().endswith(".js"): |
| add_sample(page_url, body) |
| except (FetchProblem, httpx.HTTPError, asyncio.TimeoutError, ValueError) as exc: |
| reasons.append("Page sampling: " + (str(exc) if isinstance(exc, FetchProblem) else exc.__class__.__name__)) |
| coverage["fetch_failures"] += 1 |
| candidates = list(dict.fromkeys(parser.urls + [ |
| item for item in discovered if isinstance(item, str) and same_origin(item, url) and |
| urlsplit(item).path.lower().endswith((".js", ".mjs"))])) |
| already = {source for source, _ in samples} |
| candidates = [candidate for candidate in candidates if candidate not in already] |
| coverage["eligible_files"] = len(candidates) + len(parser.inline) + len(samples) |
| coverage.update(external_available=len(candidates), inline_available=len(parser.inline), |
| oversized=parser.inline_oversized) |
| |
| |
| initial_inline = 0 if profile == "current" else min(len(parser.inline), limit // 2) |
| for body in parser.inline[:initial_inline]: |
| add_sample(parser.target, body) |
| attempted = 0 |
| for candidate in candidates: |
| if len(samples) >= limit or attempted >= attempts_limit: |
| policy_limits.add("file_or_attempt_count") |
| break |
| if time.monotonic() - started + max(config.REQUEST_DELAY_MIN, float(delay)) >= settings["retire_sampling_seconds"]: |
| reasons.append("JavaScript sampling time budget reached") |
| policy_limits.add("sampling_time") |
| break |
| await asyncio.sleep(max(config.REQUEST_DELAY_MIN, float(delay))) |
| attempted += 1 |
| coverage["attempted_scripts"] = attempted |
| try: |
| body, content_type, sampled_url = await sample(candidate, maximum) |
| if "html" in content_type or body.lstrip().lower().startswith((b"<!doctype html", b"<html")): |
| raise FetchProblem("HTML returned instead of JavaScript") |
| if "javascript" not in content_type and not urlsplit(candidate).path.lower().endswith((".js", ".mjs")): |
| raise FetchProblem("Response not identified as JavaScript") |
| if body: |
| add_sample(sampled_url, body) |
| except (FetchProblem, httpx.HTTPError, asyncio.TimeoutError, ValueError) as exc: |
| detail = str(exc) if isinstance(exc, FetchProblem) else exc.__class__.__name__ |
| if detail in ("HTTP 404", "HTTP 410") and candidate not in parser.urls: |
| coverage["missing_inventory_scripts"] += 1 |
| coverage["eligible_files"] -= 1 |
| continue |
| if "byte limit" in detail: |
| coverage["oversized"] += 1 |
| policy_limits.add("file_bytes") |
| else: |
| coverage["fetch_failures"] += 1 |
| reasons.append("Script sampling: " + detail) |
| remaining_inline = parser.inline[initial_inline:] |
| inline_limit = max(0, limit - len(samples)) |
| for body in remaining_inline[:inline_limit]: |
| add_sample(parser.target, body) |
| omitted_external = max(0, len(candidates) - attempted) |
| omitted_inline = max(0, len(remaining_inline) - inline_limit) |
| if omitted_external: |
| reasons.append("JavaScript URL sample limited to " + str(attempted)) |
| policy_limits.add("external_sample_count") |
| if omitted_inline: |
| reasons.append("Inline JavaScript sample limited") |
| policy_limits.add("inline_sample_count") |
| if parser.inline_oversized: |
| reasons.append("Oversized inline JavaScript excluded") |
| policy_limits.add("inline_bytes") |
| if "total_bytes" in policy_limits: |
| reasons.append("Aggregate JavaScript byte budget reached") |
| coverage.update(sampled_bytes=sampled_bytes, policy_limits=sorted(policy_limits), sampled_files=len(samples), omitted_external=omitted_external, |
| omitted_inline=omitted_inline) |
| if not samples: |
| reason = "; ".join(reasons) or "No same-origin or inline JavaScript available" |
| return {"status": "skipped", "source": "retirejs", "alerts": [], "target": url, |
| "reason": reason, "coverage": coverage, "output": "Retire.js skipped: " + reason} |
| with tempfile.TemporaryDirectory(prefix="asfb-retirejs-") as tmp: |
| root = Path(tmp) |
| assets = root / "assets" |
| assets.mkdir() |
| mapping = {} |
| for index, (source, body) in enumerate(samples): |
| folder = assets / str(index) |
| folder.mkdir() |
| name = re.sub(r"[^A-Za-z0-9_.-]", "_", Path(urlsplit(source).path).name)[:100] |
| if not name.lower().endswith(".js"): |
| name = "sample.js" |
| filename = folder / name |
| filename.write_bytes(body) |
| mapping[str(filename.resolve())] = source |
| report_path = str(root / "report.json") |
| ignore_path = root / "ignore.json" |
| ignore_path.write_text("[]", encoding="utf-8") |
| rows, status = await execute([ |
| "/usr/local/bin/node", "/opt/retire/lib/cli.js", "--path", str(assets), |
| "--jsrepo", "/opt/retire-db/jsrepository-v5.json", "--cachedir", str(root / "cache"), |
| "--ignorefile", str(ignore_path), "--outputformat", "json", |
| "--outputpath", report_path, "--exitwith", "13"], |
| config.RETIRE_TIMEOUT_SECONDS) |
| report = read_report(report_path) |
| findings = parse_retire(report, mapping) |
| diagnostic = [row[:180] for row in rows if row.startswith(("[stderr]", "[diagnostic]"))] |
| code13 = any(row.strip() == "[diagnostic] Process exit code: 13" for row in rows) |
| if code13 and isinstance(report, dict) and findings and not report.get("errors"): |
| status = "findings" |
| diagnostic = [s for s in diagnostic if s != "[diagnostic] Process exit code: 13"] |
| if not isinstance(report, dict) or not isinstance(report.get("data"), list): |
| if status == "completed": |
| status = "failed" |
| reasons.append("No readable Retire.js JSON report") |
| elif report.get("errors"): |
| reasons.append("Retire.js reported scanner errors") |
| if status in ("completed", "findings"): |
| status = "partial" if findings else "failed" |
| execution_state = status |
| early_stop = (bool(policy_limits) and coverage["fetch_failures"] == 0 |
| and status in ("completed", "findings") and isinstance(report, dict) |
| and not report.get("errors")) |
| if reasons and status in ("completed", "findings") and not early_stop: |
| status = "partial" |
| if early_stop: |
| status = "findings" if findings else "completed" |
| reasons.insert(0, "Early stop: configured JavaScript sampling limits; coverage is incomplete") |
| coverage["engine_status"] = execution_state |
| coverage["coverage_limited"] = bool(reasons) |
| reasons.extend(diagnostic) |
| if coverage["missing_inventory_scripts"]: |
| reasons.append(str(coverage["missing_inventory_scripts"]) + " unavailable inventory-only script URLs excluded (HTTP 404/410)") |
| reasons.append("Bounded JavaScript sample using the image-bundled advisory snapshot") |
| return {"status": status, "source": "retirejs", "target": url, "alerts": findings, |
| "sampled_files": len(samples), "coverage": coverage, |
| **({"early_stop": True, "stop_reason": "sampling_policy", "coverage_complete": False} if early_stop else {}), |
| "reason": "; ".join(dict.fromkeys(reasons))[:500], |
| "output": f"Retire.js: engine {execution_state}; {len(samples)}/{coverage['eligible_files']} eligible files sampled; {len(findings)} unverified advisory matches; coverage: {'limited' if coverage['coverage_limited'] else 'complete'}"} |
|
|
|
|
| def parse_wapiti(report, target): |
| findings = [] |
| groups = report.get("vulnerabilities", {}) if isinstance(report, dict) else {} |
| if not isinstance(groups, dict): |
| return [] |
| risks = {0: "Informational", 1: "Low", 2: "Medium", 3: "High", 4: "Critical"} |
| for category, entries in groups.items(): |
| if not isinstance(entries, list): |
| continue |
| for item in entries: |
| if not isinstance(item, dict) or str(item.get("method", "GET")).upper() not in ("GET", "HEAD"): |
| continue |
| candidate = urljoin(target, str(item.get("path") or target)) |
| if same_origin(candidate, target): |
| findings.append(alert("wapiti", "Wapiti: " + str(category), candidate, |
| item.get("info") or "Unverified Wapiti observation", |
| risks.get(item.get("level"), "Informational"), |
| item.get("module", ""))) |
| return findings[:config.WEB_FINDINGS_LIMIT] |
|
|
|
|
| async def wapiti(url, host, agent, delay): |
| with tempfile.TemporaryDirectory(prefix="asfb-wapiti-") as tmp: |
| root = Path(tmp) |
| report_path, stats_path = str(root / "report.json"), str(root / "stats.json") |
| args = ["/opt/wapiti/bin/python", "/app/app/wapiti_runner.py", |
| "--asfb-stats", stats_path, "--asfb-delay", str(delay), |
| "--asfb-budget", str(config.WAPITI_REQUEST_LIMIT), |
| "-u", url, "--scope", "url", "-m", "csp,http_headers,cookieflags", |
| "--headless", "no", "--tasks", "1", "-d", "0", |
| "--max-links-per-page", "1", "--max-files-per-dir", "1", |
| "--max-scan-time", str(config.WAPITI_SCAN_SECONDS), "--max-attack-time", "5", |
| "--max-parameters", "5", "-t", "5", "-A", agent, "--verify-ssl", "1", |
| "--store-session", str(root / "session"), "--store-config", str(root / "config"), |
| "--no-bugreport", "-f", "json", "-o", report_path] |
| rows, status = await execute(args, config.WAPITI_TIMEOUT_SECONDS) |
| report, stats = read_report(report_path), read_report(stats_path) |
| findings = parse_wapiti(report, url) |
| reasons = [row[:180] for row in rows if row.startswith(("[stderr]", "[diagnostic]"))] |
| if not isinstance(report, dict) or not isinstance(report.get("vulnerabilities"), dict): |
| if status == "completed": |
| status = "failed" |
| reasons.append("No valid Wapiti JSON report produced") |
| if isinstance(stats, dict): |
| if stats.get("limit_reached") or stats.get("response_limited") or stats.get("blocked"): |
| if status == "completed": |
| status = "partial" |
| reasons.append("Origin/method/request/response guard limited coverage") |
| else: |
| if status == "completed": |
| status = "partial" |
| reasons.append("Wapiti guard telemetry unavailable") |
| if any("maximum scan time" in row.lower() or "max scan time" in row.lower() for row in rows): |
| if status == "completed": |
| status = "partial" |
| reasons.append("Wapiti scan deadline reached") |
| reasons.append("Limited GET/HEAD CSP, header and cookie profile; no browser or injection modules") |
| return {"status": status, "source": "wapiti", "target": url, "alerts": findings, |
| "request_count": stats.get("requests", 0) if isinstance(stats, dict) else 0, |
| "reason": "; ".join(reasons)[:500], |
| "output": f"Wapiti: {len(findings)} unverified findings; status: {status}"} |
|
|
