from __future__ import annotations import argparse import json import sys from datetime import datetime, timezone from modules.banners import banners from modules import VERSION, CVE_IDS from modules.colors import C, w from modules.scanner import ( AUTO_PORTS, DEFAULT_TENANT, DEFAULT_NAMESPACE, DEFAULT_TIMEOUT, RCE_POLL_INTERVAL, RCE_POLL_RETRIES, print_single, print_bulk, run_bulk, scan_target, parse_and_normalize, expand_targets, load_targets_from_file, ) BANNER = f"""{C.C}{C.B} ╔══════════════════════════════════════════════════════════════╗ ║ Kestra Auth-Bypass Scanner — {', '.join(CVE_IDS)} ║ ║ CVSS 10.0 — Unauthenticated RCE via /configs bypass ║ ╚══════════════════════════════════════════════════════════════╝{C.N} """ def _build_report(all_results, args, targets): results = [] for r in all_results: entry = { "target": r.target, "timestamp": r.timestamp, "vulnerable": r.vulnerable, "verdict": r.verdict, "probes": { k: { "url": p.url, "status_code": p.status_code, "error": p.error, "bypass_successful": p.bypass_successful, } for k, p in r.probes.items() }, } # Enrichment fields if r.version is not None: entry["version"] = r.version if r.version_in_range is not None: entry["version_in_affected_range"] = r.version_in_range if r.rce_confirmed is not None: entry["rce_confirmed"] = r.rce_confirmed if r.rce_method: entry["rce_method"] = r.rce_method if r.rce_evidence: entry["rce_evidence"] = r.rce_evidence if r.ssrf_detected is not None: entry["ssrf_detected"] = r.ssrf_detected if r.ssrf_evidence: entry["ssrf_evidence"] = r.ssrf_evidence if r.delete_bypass is not None: entry["delete_bypass"] = r.delete_bypass results.append(entry) return { "scan_info": { "scanner": f"kestra_cve v{VERSION}", "cve": CVE_IDS, "source": args.file or args.target, "total_targets": len(targets), "timestamp": datetime.now(timezone.utc).isoformat(), "aggressive": args.aggressive, "rce": args.rce, "ssrf": args.ssrf, "destructive": args.destructive, }, "results": results, "summary": { "total": len(all_results), "vulnerable": sum(1 for r in all_results if r.vulnerable), "rce_confirmed": sum(1 for r in all_results if r.rce_confirmed is True), "ssrf_detected": sum(1 for r in all_results if r.ssrf_detected is True), "delete_bypass": sum(1 for r in all_results if r.delete_bypass is True), "safe": sum(1 for r in all_results if not r.vulnerable and "INCONCLUSIVE" not in r.verdict and not r.verdict.startswith("ERROR")), "inconclusive": sum( 1 for r in all_results if "INCONCLUSIVE" in r.verdict or r.verdict.startswith("ERROR") ), }, } def main() -> None: parser = argparse.ArgumentParser( description=f"Kestra {', '.join(CVE_IDS)} Scanner", epilog=( "Examples:\n" " python main.py http://localhost:8080\n" " python main.py -f targets.txt\n" " python main.py -f targets.txt --aggressive\n" " python main.py -f targets.txt -j -o report.json\n" " python main.py -f targets.txt --tenant default\n" " python main.py http://target:8080 --rce --ssrf\n" " python main.py http://target:8080 --rce --destructive\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) tg = parser.add_mutually_exclusive_group(required=True) tg.add_argument("target", nargs="?", default=None, help="Single target URL (e.g. http://localhost:8080)") tg.add_argument("--file", "-f", help="File with targets (one per line)") parser.add_argument("--timeout", "-t", type=int, default=DEFAULT_TIMEOUT, help=f"Request timeout (default: {DEFAULT_TIMEOUT}s)") parser.add_argument("--verify-ssl", action="store_true", help="Verify TLS certificates") parser.add_argument("--json", "-j", action="store_true", help="Output JSON") parser.add_argument("--output", "-o", help="Write JSON to file") parser.add_argument("--tenant", type=str, default=DEFAULT_TENANT, help=f'Tenant name (default: "{DEFAULT_TENANT}")') parser.add_argument("--namespace", type=str, default=DEFAULT_NAMESPACE, help=f'Namespace for bypass paths (default: "{DEFAULT_NAMESPACE}")') parser.add_argument("--workers", "-w", type=int, default=10, help="Thread count (default: 10)") parser.add_argument("--aggressive", "-a", action="store_true", help="Also send PUT requests to confirm write bypass (may create resources)") parser.add_argument("--auto-ports", "-P", nargs="+", type=int, default=None, help=f"Ports for auto-expand (default: {AUTO_PORTS})") # New v2.0 flags probe_group = parser.add_argument_group("advanced probes") probe_group.add_argument("--rce", "-r", action="store_true", help="Enable RCE verification chain (creates harmless shell/python flow, " "triggers execution, retrieves logs to confirm code execution)") probe_group.add_argument("--ssrf", "-s", action="store_true", help="Enable SSRF detection (creates flow with Pebble http() to reach " "cloud metadata endpoints, checks logs for evidence)") probe_group.add_argument("--destructive", "-d", action="store_true", help="Enable DELETE probes to confirm destructive operation bypass " "(WILL delete resources on vulnerable instances)") probe_group.add_argument("--poll-interval", type=int, default=RCE_POLL_INTERVAL, help=f"Seconds between log polls for RCE/SSRF chains (default: {RCE_POLL_INTERVAL})") probe_group.add_argument("--poll-retries", type=int, default=RCE_POLL_RETRIES, help=f"Max log poll attempts for RCE/SSRF chains (default: {RCE_POLL_RETRIES})") args = parser.parse_args() print(BANNER) ports = args.auto_ports if args.auto_ports else AUTO_PORTS # Common scan kwargs scan_kwargs = dict( timeout=args.timeout, verify_ssl=args.verify_ssl, tenant=args.tenant, namespace=args.namespace, aggressive=args.aggressive, rce=args.rce, ssrf=args.ssrf, destructive=args.destructive, poll_interval=args.poll_interval, poll_retries=args.poll_retries, ) if args.file: targets = load_targets_from_file(args.file) if args.auto_ports: raw = [] with open(args.file, encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#"): raw.append(line) targets = expand_targets(raw, auto_ports=ports) if not targets: print(f"{w('[!]', C.R)} No valid targets found.") sys.exit(1) print(f"{w('[*]', C.C)} Total scan targets: {w(str(len(targets)), C.B)}") print(f"{w('[*]', C.C)} Tenant: {args.tenant}") print(f"{w('[*]', C.C)} Namespace: {args.namespace}") print(f"{w('[*]', C.C)} Aggressive: {args.aggressive}") print(f"{w('[*]', C.C)} RCE verify: {args.rce}") print(f"{w('[*]', C.C)} SSRF detect: {args.ssrf}") print(f"{w('[*]', C.C)} Destructive: {args.destructive}") print(f"{w('[*]', C.C)} SSL verify: {args.verify_ssl}") print(f"{w('[*]', C.C)} Workers: {w(str(args.workers), C.B)}\n") all_results = run_bulk( targets, workers=args.workers, **scan_kwargs, ) if args.json or args.output: report = _build_report(all_results, args, targets) blob = json.dumps(report, indent=2, ensure_ascii=False) if args.output: with open(args.output, "w", encoding="utf-8") as f: f.write(blob) print(f"\n{w('[+]', C.G)} JSON report saved to: {args.output}") else: print(blob) if not args.json or args.output: print_bulk(all_results) sys.exit(1 if any(r.vulnerable for r in all_results) else 0) else: raw = args.target scheme, host, port = parse_and_normalize(raw) if scheme is None: print(f"{w('[!]', C.R)} Invalid target.") sys.exit(1) if port is not None: targets = [f"{scheme}://{host}:{port}"] else: targets = [f"{scheme}://{host}:{p}" for p in ports] print(f"{w('[*]', C.Y)} No port specified — trying: {ports}\n") if len(targets) == 1: print(f"{w('[*]', C.C)} Scanning: {w(targets[0], C.B)}") else: print(f"{w('[*]', C.C)} Scanning {w(str(len(targets)), C.B)} " f"port(s) for {w(host, C.B)}\n") all_results = [ scan_target(t, **scan_kwargs) for t in targets ] if len(targets) == 1: if args.json: sr = all_results[0] entry = { "target": sr.target, "timestamp": sr.timestamp, "vulnerable": sr.vulnerable, "verdict": sr.verdict, "version": sr.version, "version_in_affected_range": sr.version_in_range, "rce_confirmed": sr.rce_confirmed, "rce_method": sr.rce_method, "rce_evidence": sr.rce_evidence, "ssrf_detected": sr.ssrf_detected, "ssrf_evidence": sr.ssrf_evidence, "delete_bypass": sr.delete_bypass, "probes": { k: {"url": p.url, "status_code": p.status_code, "error": p.error, "bypass_successful": p.bypass_successful} for k, p in sr.probes.items() }, } print(json.dumps(entry, indent=2)) else: print_single(all_results[0]) else: if not args.json or args.output: print_bulk(all_results) if args.output: report = _build_report(all_results, args, targets) with open(args.output, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"\n{w('[+]', C.G)} JSON report saved to: {args.output}") has_vuln = any(r.vulnerable for r in all_results) if has_vuln: sys.exit(1) elif all("INCONCLUSIVE" in r.verdict or r.verdict.startswith("ERROR") for r in all_results): sys.exit(2) else: sys.exit(0) if __name__ == "__main__": banners() main()