diff --git a/README.md b/README.md index c8a111a..deb3bec 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,539 @@ # GATEkeeper -capture everything – network requests, responses, console logs, and any hidden redirect. \ No newline at end of file +
+ +**GATEkeeper by ek0mssavior.dev** + +**Advanced Browser Automation for Gatekeeper Testing & Network Recon** + +[Features](#features) • [Installation](#installation) • [Usage](#usage) • [Bulk Scanning](#bulk-scanning) • [Options](#options) • [Output](#output) • [Disclaimer](#disclaimer) + +
+ +--- + +## Overview + +GATEkeeper is a Playwright-powered reconnaissance tool for authorized web application testing. + +It launches a Chromium browser, simulates human-like interaction, captures browser network activity, tracks redirects and URL changes, logs console messages, saves the final rendered page, and optionally dumps response bodies for offline analysis. + +GATEkeeper also performs lightweight preflight reconnaissance before browser interaction, including DNS resolution, target IP collection, common web port checks, service hints, TLS certificate metadata, and triage scoring. + +It is useful for testing applications where content, redirects, API calls, authentication workflows, or client-side behavior only appear after JavaScript execution, user interaction, authentication, or delayed page activity. + +--- + +## Features + +* Interactive prompt mode or command-line mode +* Single-target scanning +* Bulk scanning from a file of domains or URLs +* Headless or visible Chromium browser automation +* Lightweight preflight recon: + + * DNS resolution + * Target IP collection + * Canonical name and alias collection + * Common web port checks + * Service hints for discovered ports + * TLS certificate metadata + * Subject Alternative Name preview +* Simulated human-like interaction: + + * Mouse movement + * Page clicks + * Scrolling + * Keyboard events + * Viewport resizing +* Captures network requests and responses +* Captures failed requests +* Captures JavaScript console messages +* Tracks URL changes and redirect responses +* Handles binary/compressed POST bodies safely +* Loads cookies for authenticated testing +* Supports custom HTTP headers +* Supports custom User-Agent strings +* Waits for a specific CSS selector before interaction +* Optional response body dumping +* Basic technology fingerprinting +* Basic security header analysis +* Structured `report.json` generation +* Bulk `aggregate_report.json` generation +* Triage scoring to help prioritize interesting targets + +--- + +## Installation + +### Requirements + +* Python 3.10+ +* Playwright + +### Clone the Repository + +```bash +git clone https://github.com/ekomsSavior/GATEkeeper.git +cd GATEkeeper +``` + +### Install Playwright + +```bash +pip install playwright --break-system-packages +``` + +### Install Chromium + +```bash +python3 -m playwright install chromium +``` + +If Playwright reports missing browser dependencies, run: + +```bash +python3 -m playwright install-deps chromium +python3 -m playwright install chromium +``` + +--- + +## Usage + +GATEkeeper supports interactive mode, command-line mode, and bulk mode. + +### Interactive Mode + +```bash +python3 gatekeeper.py +``` + +You will be prompted for: + +* Target URL +* Output directory +* Headless mode +* Interaction duration + +--- + +## Command-Line Mode + +### Basic Scan + +```bash +python3 gatekeeper.py https://example.com +``` + +### Headless Scan + +```bash +python3 gatekeeper.py https://example.com --headless +``` + +### Longer Interaction Window + +```bash +python3 gatekeeper.py https://example.com --duration 90 +``` + +### Save Response Bodies and Generate Report + +```bash +python3 gatekeeper.py https://example.com --headless --duration 60 --save-bodies --report +``` + +### Authenticated Testing with Cookies + +```bash +python3 gatekeeper.py https://example.com/dashboard --cookies cookies.json --report +``` + +### Add Custom Headers + +```bash +python3 gatekeeper.py https://example.com/dashboard \ + --header "Authorization: Bearer TOKEN_HERE" \ + --header "X-Test-Mode: authorized" +``` + +### Custom User-Agent + +```bash +python3 gatekeeper.py https://example.com --user-agent "Mozilla/5.0 CustomTestAgent" +``` + +### Wait for a DOM Element Before Interacting + +```bash +python3 gatekeeper.py https://example.com/dashboard --wait-selector "#dashboard" +``` + +### Non-Interactive Mode + +```bash +python3 gatekeeper.py https://example.com --non-interactive --headless --report +``` + +--- + +## Bulk Scanning + +GATEkeeper can scan a list of targets from a text file. + +Create a file like: + +```text +app.example.com +api.example.com +portal.example.com +https://admin.example.com +``` + +Run: + +```bash +python3 gatekeeper.py -i subs.txt --headless --duration 15 --timeout 30000 --report --concurrency 3 --delay 1 +``` + +Bare domains are automatically normalized with `https://` unless another scheme is specified. + +Example with a full path: + +```bash +python3 gatekeeper.py \ + -i /home/ek0ms/kctcs/subs.txt \ + --headless \ + --duration 10 \ + --timeout 25000 \ + --report \ + --concurrency 3 \ + --delay 1 +``` + +Bulk mode creates one folder per target and writes a ranked aggregate report. + +Default bulk output: + +```text +gatekeeper_bulk_results/ +``` + +Example: + +```text +gatekeeper_bulk_results/ +├── aggregate_report.json +├── app.example.com/ +│ ├── network_capture.json +│ ├── report.json +│ ├── final_page.html +│ ├── final_screenshot.png +│ └── console_log.txt +└── api.example.com/ + ├── network_capture.json + ├── report.json + ├── final_page.html + ├── final_screenshot.png + └── console_log.txt +``` + +For large scopes, start without `--save-bodies` to keep scans fast and output smaller. Use `--save-bodies` later on the most interesting hosts. + +--- + +## Options + +| Argument | Description | +| ------------------------- | ------------------------------------------------------------------------------------------------------------- | +| `url` | Target URL. Optional if using interactive mode or bulk mode. | +| `-i`, `--input-file FILE` | File containing targets, one domain or URL per line. | +| `-o`, `--output DIR` | Output directory. Defaults to `_bang/` for single scans or `gatekeeper_bulk_results/` for bulk scans. | +| `--headless` | Run Chromium without a visible browser window. | +| `--duration N` | Interaction duration in seconds. Default: `45`. | +| `--timeout N` | Navigation timeout in milliseconds. Default: `60000`. | +| `--cookies FILE` | Load cookies from a Playwright-compatible JSON cookie file. | +| `--header "Name: Value"` | Add a custom HTTP header. Can be used multiple times. | +| `--user-agent STRING` | Override the default User-Agent. | +| `--wait-selector CSS` | Wait for a CSS selector before starting interaction. | +| `--save-bodies` | Save response bodies to disk. | +| `--report` | Generate `report.json`. | +| `--non-interactive` | Do not prompt for missing values. | +| `--concurrency N` | Number of targets to scan at once in bulk mode. Default: `2`. | +| `--delay N` | Delay in seconds after each target finishes. Default: `0`. | +| `--scheme http/https` | Default scheme for bare domains in input files. Default: `https`. | + +--- + +## Output + +By default, single-target results are saved in a directory named after the target domain. + +Example: + +```text +example.com_bang/ +``` + +Generated files: + +| File / Directory | Description | +| ----------------------- | ---------------------------------------------------------------------------------------- | +| `network_capture.json` | Full request, response, failure, URL change, console log, recon, and saved body metadata | +| `final_page.html` | Final rendered DOM after interaction | +| `final_screenshot.png` | Full-page screenshot | +| `console_log.txt` | JavaScript console messages | +| `report.json` | Structured report generated with `--report` | +| `response_bodies/` | Saved response bodies generated with `--save-bodies` | +| `aggregate_report.json` | Bulk scan summary generated in bulk mode | + +--- + +## Terminal Summary + +Each target prints a summary like: + +```text +============================================================ +GATEKEEPER SUMMARY +============================================================ +Initial URL: https://example.com +Final URL: https://example.com/ +Title: Example Domain +Requests: 12 +Responses: 12 +Failures: 0 +Console: 1 +URL changes: 0 +Bodies saved:0 + +------------------------------------------------------------ +LIGHTWEIGHT RECON +------------------------------------------------------------ +Host: example.com +IPs: ['93.184.216.34'] +Canonical: example.com +Aliases: [] +Open ports: [80, 443] + +[+] Service hints: + - 80: http + - 443: https + +[+] TLS info: + Not before: May 28 15:39:24 2026 GMT + Not after: Dec 12 15:39:24 2026 GMT + SAN count: 3 + - example.com + - www.example.com + +[+] No HTTP redirect responses captured. + +[-] No configured interesting endpoint terms captured. + +[+] Detected technologies: + - Nginx + +[!] Missing common security headers: + - content-security-policy + - permissions-policy + +[+] Final HTML captured and does not obviously look stuck on a spinner. + +[+] Triage score: 12 +``` + +--- + +## Report Contents + +When `--report` is used, GATEkeeper generates `report.json` with: + +* Target URL +* Final URL +* Final page title +* Triage score +* DNS and IP information +* Canonical name and aliases +* Open common web ports +* Service hints +* TLS certificate metadata +* Total requests +* Total responses +* Failed request count +* Console message count +* URL changes +* Redirects +* Status code summary +* Resource type summary +* Interesting endpoint matches +* Basic detected technologies +* Common security header analysis +* Saved response body metadata + +Example: + +```json +{ + "target": "https://example.com", + "final_url": "https://example.com/dashboard", + "final_title": "Dashboard", + "triage_score": 18, + "recon": { + "host": "example.com", + "dns": { + "canonical_name": "example.com", + "aliases": [], + "ips": ["93.184.216.34"] + }, + "open_ports": [80, 443], + "services": [ + {"port": 80, "hint": "http"}, + {"port": 443, "hint": "https"} + ], + "tls": { + "not_before": "May 28 15:39:24 2026 GMT", + "not_after": "Dec 12 15:39:24 2026 GMT", + "san": ["example.com", "www.example.com"] + } + }, + "summary": { + "total_requests": 134, + "total_responses": 129, + "failed_requests": 5, + "redirects_detected": 2, + "response_bodies_saved": 48 + }, + "technologies": [ + "Cloudflare", + "React", + "Google Analytics" + ], + "security_headers": { + "strict-transport-security": { + "present": true, + "value": "max-age=31536000" + }, + "content-security-policy": { + "present": false, + "value": null + } + } +} +``` + +--- + +## Aggregate Report + +Bulk mode writes: + +```text +aggregate_report.json +``` + +The aggregate report includes: + +* Total targets +* Completed scans +* Errors +* Results ranked by triage score +* Final URL +* Final title +* IPs +* Canonical name +* Aliases +* Open ports +* Service hints +* TLS expiration +* Request/response/failure counts +* URL change count +* Detected technologies +* Interesting endpoint count + +This makes it easier to triage large target lists without opening every folder manually. + +--- + +## Triage Score + +GATEkeeper assigns a lightweight triage score to help prioritize results. + +The score increases for signals such as: + +* Non-standard web ports +* URL changes +* Interesting endpoint terms +* Missing common security headers +* Final URL differing from the initial target +* Certain detected technologies + +The score is not a vulnerability rating. It is a sorting helper for recon workflows. + +--- + +## Cookie File Format + +The cookie file must be a JSON array of cookie objects accepted by Playwright. + +Example: + +```json +[ + { + "name": "sessionid", + "value": "abc123", + "domain": "example.com", + "path": "/", + "httpOnly": true, + "secure": true, + "sameSite": "Lax" + } +] +``` + +--- + +## How It Works + +1. Normalizes the target URL. +2. Runs lightweight preflight recon: + + * Resolves DNS + * Collects target IPs + * Checks common web ports + * Adds service hints + * Pulls TLS certificate metadata when HTTPS is open +3. Launches a Chromium browser with a realistic viewport and User-Agent. +4. Optionally loads cookies and custom headers. +5. Navigates to the target URL. +6. Optionally waits for a CSS selector. +7. Simulates browser interaction: + + * Mouse movement + * Clicks + * Scrolling + * Keyboard input + * Viewport resizing +8. Captures requests, responses, failed requests, redirects, URL changes, and console logs. +9. Saves final HTML and screenshot. +10. Optionally saves response bodies. +11. Optionally generates a structured report. +12. In bulk mode, writes an aggregate report ranked by triage score. + +--- + +## Disclaimer + +This tool is intended for authorized security testing, research, educational assessments only. + + + +--- + +
+ +**Happy Hacking — Authorized Testing Only** + +
diff --git a/gatekeeper.py b/gatekeeper.py new file mode 100644 index 0000000..c93d52e --- /dev/null +++ b/gatekeeper.py @@ -0,0 +1,1148 @@ +#!/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())