#!/usr/bin/env python3 """ GATEkeeper - Browser interaction + network recon for authorized testing. Features: - Single URL scanning & Bulk target scanning from file - Interactive prompts or CLI arguments - Cookie loading - Custom headers - Wait selector support - Request/response/failure/console capture - Optional response body dumping - Technology fingerprinting - Security header analysis - Lightweight DNS/IP/port/TLS recon - JSON report generation - Aggregate bulk report generation """ import argparse import asyncio import hashlib import json import random import re import socket import ssl import sys from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from urllib.parse import urlparse from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError MAX_POST_CHARS = 5000 MAX_BODY_BYTES = 10 * 1024 * 1024 COMMON_PORTS = [80, 443, 8080, 8443, 8000, 8888, 3000, 5000] INTERESTING_TERMS = [ "auth", "login", "logout", "oauth", "saml", "sso", "callback", "token", "jwt", "session", "csrf", "xsrf", "api", "graphql", "admin", "dashboard", "internal", "gateway", "redirect", "collect", "track", "analytics", "chronos", "dune", "ct" ] TECH_SIGNATURES = { "Cloudflare": ["cf-ray", "cf-cache-status", "__cf_bm", "cloudflare"], "Akamai": ["akamai", "akamaighost", "ak_bmsc"], "Fastly": ["fastly", "x-served-by", "x-cache-hits"], "Nginx": ["nginx"], "Apache": ["apache"], "IIS": ["microsoft-iis", "asp.net"], "React": ["react", "__react", "react-dom"], "Next.js": ["next.js", "_next/static", "x-nextjs"], "Vue": ["vue", "__vue"], "Angular": ["angular", "ng-version"], "jQuery": ["jquery"], "WordPress": ["wp-content", "wp-includes", "wordpress"], "Drupal": ["drupal"], "Google Tag Manager": ["googletagmanager.com", "gtm.js"], "Google Analytics": ["google-analytics.com", "gtag/js", "analytics.google.com"], "Microsoft Clarity": ["clarity.ms"], } SECURITY_HEADERS = [ "strict-transport-security", "content-security-policy", "x-frame-options", "x-content-type-options", "referrer-policy", "permissions-policy", "cross-origin-opener-policy", "cross-origin-resource-policy", "cross-origin-embedder-policy", ] PORT_SERVICE_HINTS = { 80: "http", 443: "https", 8080: "http-alt/proxy", 8443: "https-alt", 8000: "dev-http", 8888: "dev-http/proxy", 3000: "node/react/dev", 5000: "flask/dev-api", } async def tcp_check(host: str, port: int, timeout: float = 2.0): try: reader, writer = await asyncio.wait_for( asyncio.open_connection(host, port), timeout=timeout, ) writer.close() await writer.wait_closed() return True except Exception: return False def resolve_host(host: str): result = { "host": host, "canonical_name": None, "aliases": [], "ips": [], "error": None, } try: cname, aliases, ips = socket.gethostbyname_ex(host) result["canonical_name"] = cname result["aliases"] = aliases result["ips"] = sorted(set(ips)) except Exception as e: result["error"] = f"{type(e).__name__}: {e}" return result def get_tls_info(host: str, port: int = 443, timeout: float = 3.0): result = { "host": host, "port": port, "subject": None, "issuer": None, "not_before": None, "not_after": None, "san": [], "error": None, } try: context = ssl.create_default_context() with socket.create_connection((host, port), timeout=timeout) as sock: with context.wrap_socket(sock, server_hostname=host) as ssock: cert = ssock.getpeercert() result["subject"] = cert.get("subject") result["issuer"] = cert.get("issuer") result["not_before"] = cert.get("notBefore") result["not_after"] = cert.get("notAfter") san = cert.get("subjectAltName", []) result["san"] = [value for key, value in san if key.lower() == "dns"] except Exception as e: result["error"] = f"{type(e).__name__}: {e}" return result async def light_recon(target_url: str, ports=None): ports = ports or COMMON_PORTS parsed = urlparse(target_url) host = parsed.hostname recon = { "target_url": target_url, "host": host, "dns": {}, "open_ports": [], "services": [], "tls": None, "error": None, } if not host: recon["error"] = "Could not parse host from URL" return recon recon["dns"] = resolve_host(host) checks = await asyncio.gather( *(tcp_check(host, port) for port in ports), return_exceptions=True, ) for port, is_open in zip(ports, checks): if is_open is True: recon["open_ports"].append(port) recon["services"].append( { "port": port, "hint": PORT_SERVICE_HINTS.get(port, "unknown"), } ) if 443 in recon["open_ports"]: recon["tls"] = get_tls_info(host, 443) return recon class GatekeeperBanger: def __init__( self, target_url: str, output_dir: Path, headless: bool = False, timeout: int = 60000, interaction_duration: int = 45, cookies_file: Path | None = None, custom_headers: dict | None = None, user_agent: str | None = None, wait_selector: str | None = None, save_bodies: bool = False, report: bool = False, ): self.target_url = target_url self.output_dir = output_dir self.headless = headless self.timeout = timeout self.interaction_duration = interaction_duration self.cookies_file = cookies_file self.custom_headers = custom_headers or {} self.user_agent = user_agent self.wait_selector = wait_selector self.save_bodies = save_bodies self.report = report self.captured_requests = [] self.captured_responses = [] self.failed_requests = [] self.console_logs = [] self.url_changes = [] self.saved_bodies = [] self.final_html = None self.final_url = None self.final_title = None self.main_response_headers = {} self.recon = {} self.output_dir.mkdir(parents=True, exist_ok=True) self.bodies_dir = self.output_dir / "response_bodies" if self.save_bodies: self.bodies_dir.mkdir(parents=True, exist_ok=True) self.capture_file = self.output_dir / "network_capture.json" self.report_file = self.output_dir / "report.json" self.html_file = self.output_dir / "final_page.html" self.screenshot_file = self.output_dir / "final_screenshot.png" self.console_log_file = self.output_dir / "console_log.txt" async def run(self): print("[*] Running lightweight target recon...") self.recon = await light_recon(self.target_url) async with async_playwright() as p: browser = await p.chromium.launch( headless=self.headless, args=[ "--disable-blink-features=AutomationControlled", "--no-sandbox", "--disable-dev-shm-usage", "--ignore-certificate-errors", ], ) default_headers = { "Accept-Language": "en-US,en;q=0.9", "DNT": "1", } default_headers.update(self.custom_headers) context = await browser.new_context( viewport={"width": 1920, "height": 1080}, locale="en-US", timezone_id="America/New_York", user_agent=self.user_agent or ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/134.0.0.0 Safari/537.36" ), extra_http_headers=default_headers, java_script_enabled=True, ignore_https_errors=True, ) await self.load_cookies(context) await context.add_init_script( """ Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); Object.defineProperty(navigator, 'plugins', { get: () => [1,2,3,4,5] }); Object.defineProperty(navigator, 'languages', { get: () => ['en-US','en'] }); window.chrome = { runtime: {} }; """ ) page = await context.new_page() page.on("request", self._on_request) page.on("response", self._on_response) page.on("requestfailed", self._on_request_failed) page.on("console", self._on_console) print(f"[*] Navigating to {self.target_url}") try: main_response = await page.goto( self.target_url, wait_until="domcontentloaded", timeout=self.timeout, ) if main_response: self.main_response_headers = dict(main_response.headers) except PlaywrightTimeoutError: print("[!] Initial DOM load timeout, continuing anyway...") except Exception as e: print(f"[!] Initial navigation error, continuing anyway: {type(e).__name__}: {e}") if self.wait_selector: print(f"[*] Waiting for selector: {self.wait_selector}") try: await page.wait_for_selector(self.wait_selector, timeout=self.timeout) print("[+] Selector appeared.") except PlaywrightTimeoutError: print("[!] Selector wait timed out, continuing anyway.") except Exception as e: print(f"[!] Selector wait failed: {type(e).__name__}: {e}") await self.safe_wait(page, 2000) print(f"[*] Starting interaction simulation for {self.interaction_duration} seconds...") await self.simulate_interactions(page) self.final_url = page.url try: self.final_title = await page.title() except Exception: self.final_title = None try: self.final_html = await page.content() except Exception as e: print(f"[!] Could not capture final HTML: {type(e).__name__}: {e}") try: await page.screenshot(path=str(self.screenshot_file), full_page=True) except Exception as e: print(f"[!] Screenshot failed: {type(e).__name__}: {e}") await browser.close() self._save_results() self._print_summary() async def load_cookies(self, context): if not self.cookies_file: return try: with open(self.cookies_file, "r", encoding="utf-8") as f: cookies = json.load(f) if not isinstance(cookies, list): print("[!] Cookie file must be a JSON array of cookie objects.") return await context.add_cookies(cookies) print(f"[+] Loaded {len(cookies)} cookies from {self.cookies_file}") except Exception as e: print(f"[!] Failed to load cookies: {type(e).__name__}: {e}") async def simulate_interactions(self, page): try: for _ in range(30): x = random.randint(100, 1600) y = random.randint(100, 900) await page.mouse.move(x, y, steps=random.randint(3, 8)) await self.safe_wait(page, random.randint(150, 450)) await self.safe_click(page, "body", 300, 300) await self.safe_wait(page, 1000) for scroll in range(0, 1600, 100): await self.safe_eval(page, f"window.scrollTo(0, {scroll})") await self.safe_wait(page, 120) for scroll in range(1600, 0, -100): await self.safe_eval(page, f"window.scrollTo(0, {scroll})") await self.safe_wait(page, 120) for key in ["ArrowDown", "ArrowDown", "ArrowUp", "Tab", "Enter", "Escape"]: try: await page.keyboard.press(key) except Exception: pass await self.safe_wait(page, 200) try: await page.set_viewport_size({"width": 1280, "height": 800}) await self.safe_wait(page, 500) await page.set_viewport_size({"width": 1920, "height": 1080}) except Exception: pass start_time = datetime.now() last_url = page.url no_change_count = 0 while (datetime.now() - start_time).seconds < self.interaction_duration: await self.safe_wait(page, 2000) current_url = page.url if current_url != last_url: print(f"[*] URL changed: {last_url} -> {current_url}") self.url_changes.append( { "from": last_url, "to": current_url, "timestamp": datetime.now().isoformat(), } ) last_url = current_url no_change_count = 0 else: no_change_count += 1 if no_change_count > 4: await self.safe_click(page, "body", 500, 500) await self.safe_eval(page, "window.dispatchEvent(new Event('mousemove'))") no_change_count = 0 except Exception as e: print(f"[!] Interaction loop error, continuing to save results: {type(e).__name__}: {e}") async def safe_wait(self, page, ms: int): try: await page.wait_for_timeout(ms) except Exception: pass async def safe_click(self, page, selector: str, x: int, y: int): try: await page.click(selector, position={"x": x, "y": y}, force=True, timeout=5000) except Exception as e: print(f"[!] Click failed: {type(e).__name__}: {e}") async def safe_eval(self, page, script: str): try: await page.evaluate(script) except Exception: pass def safe_post_data(self, request): if request.method != "POST": return None try: data = request.post_data if data is None: return None if len(data) > MAX_POST_CHARS: return data[:MAX_POST_CHARS] + f"... [truncated {len(data) - MAX_POST_CHARS} chars]" return data except UnicodeDecodeError as e: return f"[binary/compressed POST body unreadable as UTF-8: {e}]" except Exception as e: return f"[POST body unavailable: {type(e).__name__}: {e}]" def _on_request(self, request): try: req_data = { "url": request.url, "method": request.method, "resource_type": request.resource_type, "headers": dict(request.headers), "post_data": self.safe_post_data(request), "timestamp": datetime.now().isoformat(), } self.captured_requests.append(req_data) print(f" [req] {request.method} {request.url[:140]}") except Exception as e: print(f"[!] Request capture error ignored: {type(e).__name__}: {e}") def _on_response(self, response): try: resp_data = { "url": response.url, "status": response.status, "headers": dict(response.headers), "timestamp": datetime.now().isoformat(), } self.captured_responses.append(resp_data) print(f" [res] {response.status} {response.url[:140]}") if self.save_bodies: asyncio.create_task(self.save_response_body(response)) except Exception as e: print(f"[!] Response capture error ignored: {type(e).__name__}: {e}") async def save_response_body(self, response): try: content_type = response.headers.get("content-type", "").lower() body = await response.body() if not body: return if len(body) > MAX_BODY_BYTES: return parsed = urlparse(response.url) path = parsed.path or "/" suffix = self.guess_extension(content_type, path) raw_name = f"{response.status}_{response.request.method}_{response.url}" digest = hashlib.sha256(raw_name.encode("utf-8", errors="ignore")).hexdigest()[:16] filename = f"resp_{digest}{suffix}" file_path = self.bodies_dir / filename with open(file_path, "wb") as f: f.write(body) self.saved_bodies.append( { "url": response.url, "status": response.status, "content_type": content_type, "size": len(body), "file": str(file_path), "timestamp": datetime.now().isoformat(), } ) except Exception: pass def guess_extension(self, content_type: str, path: str) -> str: path_lower = path.lower() for ext in [".js", ".css", ".json", ".html", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp", ".xml", ".txt"]: if path_lower.endswith(ext): return ext if "javascript" in content_type: return ".js" if "css" in content_type: return ".css" if "json" in content_type: return ".json" if "html" in content_type: return ".html" if "png" in content_type: return ".png" if "jpeg" in content_type or "jpg" in content_type: return ".jpg" if "svg" in content_type: return ".svg" if "xml" in content_type: return ".xml" if "text" in content_type: return ".txt" return ".bin" def _on_request_failed(self, request): try: failure = request.failure or {} fail_data = { "url": request.url, "method": request.method, "resource_type": request.resource_type, "failure": failure, "timestamp": datetime.now().isoformat(), } self.failed_requests.append(fail_data) print(f" [fail] {request.method} {request.url[:120]} {failure}") except Exception as e: print(f"[!] Failed-request capture error ignored: {type(e).__name__}: {e}") def _on_console(self, msg): try: log_entry = f"[{msg.type}] {msg.text}" self.console_logs.append(log_entry) print(f" [console] {log_entry[:180]}") except Exception: pass def detect_technologies(self): haystack_parts = [] if self.final_html: haystack_parts.append(self.final_html) for req in self.captured_requests: haystack_parts.append(req.get("url", "")) haystack_parts.append(json.dumps(req.get("headers", {}), default=str)) for resp in self.captured_responses: haystack_parts.append(resp.get("url", "")) haystack_parts.append(json.dumps(resp.get("headers", {}), default=str)) haystack = "\n".join(haystack_parts).lower() detected = [] for tech, signatures in TECH_SIGNATURES.items(): if any(sig.lower() in haystack for sig in signatures): detected.append(tech) return sorted(set(detected)) def analyze_security_headers(self): headers = {k.lower(): v for k, v in self.main_response_headers.items()} result = {} for header in SECURITY_HEADERS: result[header] = { "present": header in headers, "value": headers.get(header), } return result def find_interesting_endpoints(self): matches = [] for req in self.captured_requests: url = req.get("url", "") url_lower = url.lower() hit_terms = [term for term in INTERESTING_TERMS if term in url_lower] if hit_terms: matches.append( { "method": req.get("method"), "url": url, "resource_type": req.get("resource_type"), "terms": sorted(set(hit_terms)), } ) return matches def triage_score(self): score = 0 nonstandard_ports = [p for p in self.recon.get("open_ports", []) if p not in (80, 443)] score += len(nonstandard_ports) * 10 score += len(self.url_changes) * 8 score += len(self.find_interesting_endpoints()) * 4 missing_headers = [ h for h, v in self.analyze_security_headers().items() if not v["present"] ] score += len(missing_headers) if self.final_url and self.final_url != self.target_url: score += 5 tech = self.detect_technologies() if any(t in tech for t in ["WordPress", "Drupal", "IIS"]): score += 5 return score def build_report(self): status_counter = Counter(r["status"] for r in self.captured_responses) resource_counter = Counter(r["resource_type"] for r in self.captured_requests) endpoints = defaultdict(int) for req in self.captured_requests: endpoints[(req["method"], req["url"])] += 1 top_endpoints = [ {"method": method, "url": url, "count": count} for (method, url), count in sorted(endpoints.items(), key=lambda x: x[1], reverse=True)[:50] ] redirects = [ { "status": r["status"], "url": r["url"], "location": r["headers"].get("location", "N/A"), } for r in self.captured_responses if r["status"] in (301, 302, 303, 307, 308) ] return { "generated_at": datetime.now().isoformat(), "target": self.target_url, "final_url": self.final_url, "final_title": self.final_title, "triage_score": self.triage_score(), "recon": self.recon, "summary": { "total_requests": len(self.captured_requests), "total_responses": len(self.captured_responses), "failed_requests": len(self.failed_requests), "console_messages": len(self.console_logs), "url_changes": len(self.url_changes), "redirects_detected": len(redirects), "response_bodies_saved": len(self.saved_bodies), }, "status_codes": dict(status_counter), "resource_types": dict(resource_counter), "technologies": self.detect_technologies(), "security_headers": self.analyze_security_headers(), "redirects": redirects, "interesting_endpoints": self.find_interesting_endpoints(), "top_endpoints": top_endpoints, "url_changes": self.url_changes, "saved_bodies": self.saved_bodies, } def _save_results(self): capture = { "timestamp": datetime.now().isoformat(), "target": self.target_url, "final_url": self.final_url, "final_title": self.final_title, "triage_score": self.triage_score(), "recon": self.recon, "url_changes": self.url_changes, "requests": self.captured_requests, "responses": self.captured_responses, "failed_requests": self.failed_requests, "console_logs": self.console_logs, "saved_bodies": self.saved_bodies, } with open(self.capture_file, "w", encoding="utf-8") as f: json.dump(capture, f, indent=2, ensure_ascii=False) print(f"[*] Network capture saved to {self.capture_file}") if self.final_html: with open(self.html_file, "w", encoding="utf-8") as f: f.write(self.final_html) print(f"[*] Final HTML saved to {self.html_file}") if self.console_logs: with open(self.console_log_file, "w", encoding="utf-8") as f: f.write("\n".join(self.console_logs)) print(f"[*] Console log saved to {self.console_log_file}") if self.report: report_data = self.build_report() with open(self.report_file, "w", encoding="utf-8") as f: json.dump(report_data, f, indent=2, ensure_ascii=False) print(f"[*] Report saved to {self.report_file}") print(f"[*] Screenshot path: {self.screenshot_file}") def _print_summary(self): print("\n" + "=" * 60) print("GATEKEEPER SUMMARY") print("=" * 60) print(f"Initial URL: {self.target_url}") print(f"Final URL: {self.final_url}") print(f"Title: {self.final_title}") print(f"Requests: {len(self.captured_requests)}") print(f"Responses: {len(self.captured_responses)}") print(f"Failures: {len(self.failed_requests)}") print(f"Console: {len(self.console_logs)}") print(f"URL changes: {len(self.url_changes)}") print(f"Bodies saved:{len(self.saved_bodies)}") print("\n" + "-" * 60) print("LIGHTWEIGHT RECON") print("-" * 60) print(f"Host: {self.recon.get('host')}") print(f"IPs: {self.recon.get('dns', {}).get('ips', [])}") print(f"Canonical: {self.recon.get('dns', {}).get('canonical_name')}") print(f"Aliases: {self.recon.get('dns', {}).get('aliases', [])}") print(f"Open ports: {self.recon.get('open_ports', [])}") services = self.recon.get("services", []) if services: print("\n[+] Service hints:") for service in services: print(f" - {service.get('port')}: {service.get('hint')}") tls = self.recon.get("tls") or {} if tls: print("\n[+] TLS info:") print(f" Not before: {tls.get('not_before')}") print(f" Not after: {tls.get('not_after')}") if tls.get("san"): print(f" SAN count: {len(tls.get('san', []))}") for san in tls.get("san", [])[:10]: print(f" - {san}") if tls.get("error"): print(f" TLS error: {tls.get('error')}") redirects = [r for r in self.captured_responses if r["status"] in (301, 302, 303, 307, 308)] if redirects: print(f"\n[!] Redirects detected: {len(redirects)}") for r in redirects[:20]: location = r["headers"].get("location", "N/A") print(f" {r['status']} {r['url'][:100]} -> {location}") else: print("\n[+] No HTTP redirect responses captured.") interesting = self.find_interesting_endpoints() if interesting: print(f"\n[+] Interesting endpoints detected: {len(interesting)}") for item in interesting[:20]: print(f" {item['method']} {item['url'][:120]} [{', '.join(item['terms'])}]") else: print("\n[-] No configured interesting endpoint terms captured.") technologies = self.detect_technologies() if technologies: print("\n[+] Detected technologies:") for tech in technologies: print(f" - {tech}") else: print("\n[-] No configured technology signatures detected.") security = self.analyze_security_headers() missing = [h for h, v in security.items() if not v["present"]] present = [h for h, v in security.items() if v["present"]] if present: print("\n[+] Present common security headers:") for header in present: print(f" - {header}: {security[header].get('value')}") if missing: print("\n[!] Missing common security headers:") for header in missing: print(f" - {header}") else: print("\n[+] All configured common security headers are present.") if self.final_html: html_lower = self.final_html.lower() loading_words = ["spinner", "loading", "getting things ready"] if any(word in html_lower for word in loading_words): print("\n[!] Final page may still contain loading indicators.") else: print("\n[+] Final HTML captured and does not obviously look stuck on a spinner.") else: print("\n[-] No final HTML captured.") print(f"\n[+] Triage score: {self.triage_score()}") def normalize_url(value: str, default_scheme: str = "https") -> str: value = value.strip().strip(",") if not value: return "" if not value.startswith(("http://", "https://")): value = f"{default_scheme}://{value}" return value def load_targets(input_file: Path, default_scheme: str = "https") -> list[str]: targets = [] with open(input_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue url = normalize_url(line, default_scheme=default_scheme) if url: targets.append(url) seen = set() deduped = [] for target in targets: if target not in seen: seen.add(target) deduped.append(target) return deduped def sanitize_domain_name(url: str) -> str: parsed = urlparse(url if "://" in url else f"http://{url}") domain = parsed.netloc or parsed.path.split("/")[0] domain = re.sub(r"[^a-zA-Z0-9.-]", "_", domain) return domain or "gatekeeper_output" def parse_header(header_values): headers = {} for item in header_values or []: if ":" not in item: print(f"[!] Ignoring malformed header: {item}") continue name, value = item.split(":", 1) name = name.strip() value = value.strip() if name: headers[name] = value return headers async def run_single_target(target_url, args, custom_headers, output_base=None): domain_name = sanitize_domain_name(target_url) if output_base: output_dir = output_base / domain_name elif args.output: output_dir = Path(args.output) else: output_dir = Path(f"./{domain_name}_bang") banger = GatekeeperBanger( target_url=target_url, output_dir=output_dir, headless=args.headless, timeout=args.timeout, interaction_duration=args.duration or 45, cookies_file=args.cookies, custom_headers=custom_headers, user_agent=args.user_agent, wait_selector=args.wait_selector, save_bodies=args.save_bodies, report=args.report, ) try: await banger.run() tls = banger.recon.get("tls") or {} return { "target": target_url, "output_dir": str(output_dir), "final_url": banger.final_url, "final_title": banger.final_title, "triage_score": banger.triage_score(), "ips": banger.recon.get("dns", {}).get("ips", []), "canonical_name": banger.recon.get("dns", {}).get("canonical_name"), "aliases": banger.recon.get("dns", {}).get("aliases", []), "open_ports": banger.recon.get("open_ports", []), "services": banger.recon.get("services", []), "tls_not_before": tls.get("not_before"), "tls_not_after": tls.get("not_after"), "requests": len(banger.captured_requests), "responses": len(banger.captured_responses), "failures": len(banger.failed_requests), "console": len(banger.console_logs), "url_changes": len(banger.url_changes), "bodies_saved": len(banger.saved_bodies), "technologies": banger.detect_technologies(), "interesting_endpoints": len(banger.find_interesting_endpoints()), "status": "completed", } except Exception as e: return { "target": target_url, "output_dir": str(output_dir), "status": "error", "error": f"{type(e).__name__}: {e}", } async def run_bulk_targets(targets, args, custom_headers): output_base = Path(args.output) if args.output else Path("./gatekeeper_bulk_results") output_base.mkdir(parents=True, exist_ok=True) semaphore = asyncio.Semaphore(args.concurrency) results = [] async def worker(target): async with semaphore: print("\n" + "=" * 80) print(f"[*] Bulk scan target: {target}") print("=" * 80) result = await run_single_target(target, args, custom_headers, output_base) results.append(result) if args.delay > 0: await asyncio.sleep(args.delay) await asyncio.gather(*(worker(target) for target in targets)) results = sorted( results, key=lambda x: x.get("triage_score", 0), reverse=True, ) aggregate_file = output_base / "aggregate_report.json" with open(aggregate_file, "w", encoding="utf-8") as f: json.dump( { "generated_at": datetime.now().isoformat(), "total_targets": len(targets), "completed": len([r for r in results if r.get("status") == "completed"]), "errors": len([r for r in results if r.get("status") == "error"]), "results_ranked_by_triage_score": results, }, f, indent=2, ensure_ascii=False, ) print(f"\n[+] Bulk scan complete. Aggregate report saved to: {aggregate_file}") def build_parser(): parser = argparse.ArgumentParser( description="GATEkeeper - Browser interaction and network recon for authorized testing." ) parser.add_argument("url", nargs="?", help="Target URL") parser.add_argument("-i", "--input-file", type=Path, help="File containing targets, one domain or URL per line") parser.add_argument("-o", "--output", help="Output directory") parser.add_argument("--headless", action="store_true", help="Run browser in headless mode") parser.add_argument("--duration", type=int, default=None, help="Interaction duration in seconds") parser.add_argument("--timeout", type=int, default=60000, help="Navigation timeout in milliseconds") parser.add_argument("--cookies", type=Path, help="Load Playwright cookies from JSON file") parser.add_argument("--header", action="append", help='Custom header, example: --header "Authorization: Bearer TOKEN"') parser.add_argument("--user-agent", help="Override User-Agent") parser.add_argument("--wait-selector", help="Wait for CSS selector before interactions") parser.add_argument("--save-bodies", action="store_true", help="Save response bodies to disk") parser.add_argument("--report", action="store_true", help="Generate report.json") parser.add_argument("--non-interactive", action="store_true", help="Do not prompt for missing values") parser.add_argument("--concurrency", type=int, default=2, help="Number of targets to scan at once in bulk mode") parser.add_argument("--delay", type=float, default=0, help="Delay in seconds after each target finishes") parser.add_argument("--scheme", default="https", choices=["http", "https"], help="Default scheme for bare domains") return parser async def main(): parser = build_parser() args = parser.parse_args() print("=" * 60) print("GATEkeeper - Interactive Browser Automation") print("Authorized testing only. Captures browser network activity.") print("=" * 60) custom_headers = parse_header(args.header) if args.input_file: targets = load_targets(args.input_file, default_scheme=args.scheme) if not targets: print("[!] No targets found in input file.") sys.exit(1) print(f"[*] Loaded {len(targets)} targets from {args.input_file}") print(f"[*] Concurrency: {args.concurrency}") await run_bulk_targets(targets, args, custom_headers) return target_url = args.url if not target_url and not args.non_interactive: target_url = input("\nEnter target URL (e.g., https://example.com): ").strip() if not target_url: print("[!] No URL provided. Exiting.") sys.exit(1) target_url = normalize_url(target_url, default_scheme=args.scheme) domain_name = sanitize_domain_name(target_url) default_output = f"./{domain_name}_bang" output_dir = Path(args.output) if args.output else None if output_dir is None and not args.non_interactive: output_dir_input = input(f"Output directory (default: {default_output}): ").strip() output_dir = Path(output_dir_input) if output_dir_input else Path(default_output) if output_dir is None: output_dir = Path(default_output) headless = args.headless if not args.headless and not args.non_interactive: headless_input = input("Run in headless mode? (y/N): ").strip().lower() headless = headless_input == "y" interaction_duration = args.duration if interaction_duration is None and not args.non_interactive: duration_input = input("Interaction duration in seconds (default: 45): ").strip() try: interaction_duration = int(duration_input) if duration_input else 45 except ValueError: interaction_duration = 45 print("[!] Invalid number, using 45 seconds.") if interaction_duration is None: interaction_duration = 45 print("\n[*] Starting browser...") banger = GatekeeperBanger( target_url=target_url, output_dir=output_dir, headless=headless, timeout=args.timeout, interaction_duration=interaction_duration, cookies_file=args.cookies, custom_headers=custom_headers, user_agent=args.user_agent, wait_selector=args.wait_selector, save_bodies=args.save_bodies, report=args.report, ) await banger.run() print(f"\n[+] Script finished. Results saved in: {output_dir}") if __name__ == "__main__": asyncio.run(main())