-
- 
-
- 
- 
- 
- 
- 
-
- One modular CLI for network scanning, service enumeration, OSINT, - vulnerability correlation, protocol intelligence, and passive dark-web reconnaissance. -
- -- Why SpectraScan ยท - Features ยท - Installation ยท - Quick Start ยท - Architecture ยท - Roadmap -
- -Target: {target}
-Resolved IP: {resolved_ip}
-Scan Type: {scan_type}
-Timestamp: {timestamp}
-Duration: {duration}s
-Open Ports: {len(results.get('open_ports', []))}
-Vulnerabilities: {len(results.get('vulnerabilities', []))}
-| Port | Service | State | Banner |
|---|---|---|---|
| {port_num} | {service} | " - f"{state} | {banner} |
Target: {target}
Resolved IP: {resolved_ip}
+Scan type: {scan_type}
Timestamp: {timestamp}
+Duration: {duration}s
+| Port | Protocol | State | Service | Banner |
|---|---|---|---|---|
| {html.escape(str(cell))} | " for cell in cells) + "
" + pretty + "" + return atomic_write_text(output_path, document) + if fmt != "csv": + raise ValueError("Report format must be json, html, or csv") + stream = io.StringIO(newline="") + writer = csv.writer(stream) + writer.writerow(["Target", "Port", "Protocol", "State", "Service", "Banner", "Classification"]) + rows = list(report.get("open_ports", [])) + for host in report.get("hosts", []): + for port in host.get("ports", []): + rows.append({"target": host.get("ip"), **port}) + for item in rows: + service = item.get("service", "") + if isinstance(service, tuple): + service = " ".join(str(part) for part in service if part) + writer.writerow([ + safe_csv_cell(item.get("target", report.get("target", ""))), + safe_csv_cell(item.get("port", "")), + safe_csv_cell(item.get("protocol", report.get("scan_type", "tcp"))), + safe_csv_cell(item.get("state", "")), safe_csv_cell(service), + safe_csv_cell((item.get("banner", "") or "")[:500]), + safe_csv_cell(item.get("classification", "")), + ]) + return atomic_write_text(output_path, stream.getvalue()) + + +def main(argv: Optional[List[str]] = None): + if argv is None: + argv = sys.argv[1:] + # Keep the interactive menu for an argument-free launch. + if argv: + parser = _cli_parser() + args = parser.parse_args(argv) + + def option_was_provided(*names): + return any(token in argv for token in names) + + port_options = ( + option_was_provided("-p", "--ports", "-T", "--timing", "--timeout", + "--async-scan", "--async-concurrency", "--async-timeout", + "--scan-type", "--vuln-scan", "--cve-correlate", "--max-cve-lookups", "--os-detect", "--web-enum", + "--max-targets", "--exclude") + ) + non_port_target = any((args.domain, args.ip, args.email, args.phone, args.image)) + module_target = any((args.smb, args.ldap, args.brute_force)) + if non_port_target and port_options: + parser.error("port-scan options cannot be combined with domain, IP, email, phone, or image modes") + if module_target and port_options: + parser.error("port-scan options cannot be combined with SMB/LDAP/credential-validation modes") + if args.web_enum and not args.target: + parser.error("--web-enum requires -t/--target") + if args.web_enum and args.brute_force: + parser.error("--web-enum and --brute-force must run separately") + if args.vuln_scan and args.async_scan: + parser.error("--vuln-scan is not supported by --async-scan yet") + if args.cve_correlate and args.async_scan: + parser.error("--cve-correlate requires legacy service/version detection; remove --async-scan") + if not 1 <= args.max_cve_lookups <= 20: + parser.error("--max-cve-lookups must be between 1 and 20") + if args.max_cve_lookups != 3 and not args.cve_correlate: + parser.error("--max-cve-lookups requires --cve-correlate") + if args.vuln_scan and args.scan_type == "udp": + parser.error("--vuln-scan is not supported by UDP probes") + if args.wordlist and not (args.web_enum or args.brute_force): + parser.error("--wordlist requires --web-enum or --brute-force") + if not any((args.target, args.domain, args.ip, args.email, args.phone, args.image)): + parser.error("choose one target using -t/--target, -d/--domain, -i/--ip, -e/--email, --phone, or --image") + if args.shodan and not args.ip: + parser.error("--shodan requires -i/--ip") + if args.links and not args.domain: + parser.error("--links requires -d/--domain") + if args.dns_zone and not args.domain: + parser.error("--dns-zone requires -d/--domain") + if (args.smb or args.ldap or args.brute_force) and not args.target: + parser.error("--smb, --ldap, and --brute-force require -t/--target") + if args.brute_force: + if not args.wordlist or not args.service or args.port is None: + parser.error("--brute-force requires --wordlist, --service ssh|ftp, and --port") + if not args.yes: + parser.error("--brute-force requires explicit confirmation with --yes") + if not 1 <= args.port <= 65535: + parser.error("--port must be between 1 and 65535") + if args.max_attempts is None: + args.max_attempts = 100 + if not 1 <= args.max_attempts <= 1000: + parser.error("--max-attempts must be between 1 and 1000") + if args.service != "ssh" and args.ssh_host_fingerprint: + parser.error("--ssh-host-fingerprint is only valid with --service ssh") + if args.allow_insecure_ftp and args.service != "ftp": + parser.error("--allow-insecure-ftp requires --service ftp") + elif args.service or args.port is not None or args.ssh_known_hosts or args.ssh_host_fingerprint or args.allow_insecure_ftp or args.max_attempts is not None: + parser.error("--service, --port, SSH host-key options, --allow-insecure-ftp, and --max-attempts require --brute-force") + if args.email and not EMAIL_REGEX.fullmatch(args.email): + parser.error("--email must be a valid email address") + if args.phone and detect_target_type(args.phone).get("type") != "phone": + parser.error("--phone must be a phone number in a recognized format") + if not 1 <= args.max_targets <= 65536: + parser.error("--max-targets must be between 1 and 65536") + if args.threads is not None and not 1 <= args.threads <= 128: + parser.error("--threads must be between 1 and 128") + if args.timeout is not None and (not math.isfinite(args.timeout) or not 0 < args.timeout <= 120): + parser.error("--timeout must be greater than 0 and at most 120 seconds") + if args.async_scan and args.scan_type != "tcp": + parser.error("--async-scan supports only --scan-type tcp") + if args.async_scan and args.timeout is not None: + parser.error("use --async-timeout with --async-scan instead of --timeout") + if args.scan_type == "syn" and not SYNScan().is_admin: + parser.error("--scan-type syn requires root/administrator privileges; no TCP-connect fallback is performed") + if args.web_enum and not args.wordlist: + parser.error("--web-enum requires --wordlist") + if args.format and not args.output: + parser.error("--format requires --output") + if args.output and not args.format and os.path.splitext(args.output)[1].lower() not in {".json", ".html", ".csv"}: + parser.error("cannot infer report format; specify --format json|html|csv") + if args.output and args.format and os.path.splitext(args.output)[1] and os.path.splitext(args.output)[1].lower() != "." + args.format: + parser.error("--format conflicts with output file extension") + + manager = ReportManager() + + def finish_report(report): + if args.output: + saved = _write_cli_report(report, args.output, args.format) + console.print(f"[green][+] Report saved: {saved}[/green]") + else: + console.print(json.dumps(report, indent=2, ensure_ascii=False, default=str), markup=False) + return 0 + + if args.email: + EmailScanner.scan(args.email, manager) + return finish_report({ + "module": "Email Intelligence", "target": args.email, + "timestamp": datetime.now().isoformat(), "output": manager._buffer, + }) + if args.phone: + PhoneScanner.scan(args.phone, manager) + return finish_report({ + "module": "Phone Intelligence", "target": args.phone, + "timestamp": datetime.now().isoformat(), "output": manager._buffer, + }) + if args.image: + ImageScanner.scan(args.image, manager) + return finish_report({ + "module": "Image Metadata", "target": args.image, + "timestamp": datetime.now().isoformat(), "output": manager._buffer, + }) + if args.domain: + domain_report = DomainScanner.scan(args.domain, manager) + if domain_report is None: + parser.error("domain scan could not validate the requested domain") + report = domain_report + if args.links: + LinkScanner.scan(args.domain, manager) + report = {"target": args.domain, "domain_intelligence": domain_report, "link_output": manager._buffer} + if args.dns_zone: + report = { + "target": args.domain, "domain_intelligence": domain_report, + "dns_zone": DNSZone.scan(args.domain), "output": manager._buffer, + } + return finish_report(report) + if args.ip: + try: + ipaddress.ip_address(args.ip) + except ValueError: + parser.error("--ip must be a valid IPv4 or IPv6 address") + IPScanner.scan(args.ip, manager, include_shodan=args.shodan) + return finish_report({ + "module": "IP Intelligence", "target": args.ip, + "shodan_requested": args.shodan, + "timestamp": datetime.now().isoformat(), "output": manager._buffer, + }) + + target = normalize_target(args.target) + if args.smb or args.ldap or args.brute_force: + if "/" in target: + parser.error("protocol enumeration and credential validation require a single host, not a CIDR") + try: + resolved_ip = resolve_host(target) + except (ValueError, socket.gaierror) as exc: + parser.error(str(exc)) + modules = {} + if args.smb: + modules["smb"] = SMBEnumerator.scan(resolved_ip) + if args.ldap: + modules["ldap"] = LDAPEnumerator.scan(resolved_ip) + if args.brute_force: + brute = BruteForcer( + resolved_ip, args.port, args.service, args.wordlist, + max_threads=min(args.threads or 10, 32), + known_hosts_path=args.ssh_known_hosts, + expected_host_fingerprint=args.ssh_host_fingerprint, + max_passwords=args.max_attempts, + allow_insecure_ftp=args.allow_insecure_ftp, + ) + modules["credential_validation"] = { + "target": resolved_ip, "port": args.port, "service": args.service, + "passwords_checked_per_username": len(brute.load_wordlist()), + "results": brute.run(), + "error_counts": dict(brute.error_counts), + } + return finish_report({ + "target": target, "resolved_ip": resolved_ip, "modules": modules, + "timestamp": datetime.now().isoformat(), + }) + if not target: + parser.error("target cannot be empty") + try: + network = ipaddress.ip_network(target, strict=False) + is_network = "/" in target + except ValueError: + network = None + is_network = False + if args.links: + parser.error("--links is supported with --domain; use -d/--domain") + if args.shodan or args.dns_zone or args.smb or args.ldap or args.brute_force: + parser.error("the selected module option is incompatible with this scan") + if args.phone or args.image or args.email: + parser.error("OSINT options cannot be combined with a port scan") + if args.scan_type == "udp" and args.ports.strip().lower() == "common": + ports = sorted(UDPScan().common_udp_ports) + else: + ports = parse_ports(args.ports) + if ports is None: + parser.error(f"invalid --ports value: {args.ports!r}; use comma-separated ports, ranges, common, or all") + if len(ports) > 5000 and not args.yes: + parser.error("large port ranges require explicit confirmation; add --yes") + if args.async_scan: + if not 1 <= args.async_concurrency <= 512: + parser.error("--async-concurrency must be between 1 and 512") + if not math.isfinite(args.async_timeout) or not 0 < args.async_timeout <= 120: + parser.error("--async-timeout must be greater than 0 and at most 120 seconds") + if len(ports) > 10000 and not args.yes: + parser.error("async scans over 10000 ports require --yes") + + scan_kwargs = { + "timeout": args.timeout if args.timeout is not None else TIMING_PROFILES[args.timing]["timeout"], + "threads": args.threads if args.threads is not None else TIMING_PROFILES[args.timing]["threads"], + "scan_type": args.scan_type, + "timing": "custom", + "ports": ports, "decoys": [], "check_vulns": args.vuln_scan, "rate_limit": 0, + } + if is_network and args.cve_correlate: + parser.error("--cve-correlate is restricted to a single host to bound NVD API requests") + if is_network and (args.web_enum or args.os_detect): + parser.error("--web-enum and --os-detect require a single host target, not a CIDR network") + if is_network: + if network.version != 4: + parser.error("CIDR port sweeps currently support IPv4 only; use a single IPv6 target") + if network.num_addresses > args.max_targets: + parser.error(f"network has {network.num_addresses} addresses, exceeding --max-targets={args.max_targets}") + if network.num_addresses > 64 and not args.yes: + parser.error("large network scan requires explicit confirmation; add --yes") + results = run_network_port_scan( + str(network), scan_kwargs, exclusions=args.exclude, + max_targets=args.max_targets, large_scan_confirmed=args.yes, + ) + report = { + "target": str(network), "scan_type": args.scan_type, + "timestamp": datetime.now().isoformat(), "hosts": results, "open_ports": [], + } + else: + resolved_ip = None + if args.async_scan: + scanner = AsyncPortScanner( + target, ports=ports, timeout=args.async_timeout, + concurrency=args.async_concurrency, + ) + try: scanner.scan() - scanner.print_summary() - except Exception as exc: - console.print( - f"[!] Scan failed: {exc}", - style="bold red", - ) - sys.exit(1) - - return - - print_banner() - - while True: - print_main_menu() - try: - choice = input(f"{CYAN}root@spectra:~#{RESET} Select Option: ").strip() - except (KeyboardInterrupt, EOFError): - console.print("\n[*] Exiting SpectraScan. Stay Anonymous.", style="yellow") - break - - if choice == "4": - console.print("[*] Exiting SpectraScan. Stay Anonymous.", style="yellow") - break - elif choice in MENU_ACTIONS: - MENU_ACTIONS[choice][1]() - else: - console.print("[!] Invalid Option. Please try again.", style="red") - - -if __name__ == "__main__": + resolved_ip = scanner.resolved_ip + report = scanner.get_results() + except (ValueError, OSError, asyncio.TimeoutError) as exc: + parser.error(f"async scan failed: {exc}") + else: + try: + resolved_ip = resolve_host(target) + if ipaddress.ip_address(resolved_ip).version == 6: + parser.error("single IPv6 targets require --async-scan; legacy TCP/SYN/UDP scanners are IPv4-only") + except ValueError as exc: + parser.error(str(exc)) + scanner = PortScanner(target, **scan_kwargs) + scanner.resolved_ip = resolved_ip + scanner.scan() + report = scanner.get_results() + if args.cve_correlate: + report["cve_correlation"] = correlate_open_port_cves( + report.get("open_ports", []), + max_lookups=args.max_cve_lookups, + api_key=os.environ.get("NVD_API_KEY"), + ) + if args.os_detect: + try: + report["os_fingerprint_ttl"] = OSFingerprint.get_ttl(resolved_ip) + report["os_fingerprint_note"] = "TTL heuristic only; not definitive OS identification." + except Exception: + report["os_fingerprint_ttl"] = None + if args.web_enum: + base_url = target if target.lower().startswith(("http://", "https://")) else "http://" + target + report["web_enumeration"] = WebEnumerator(base_url, args.wordlist).run() + if args.output: + saved = _write_cli_report(report, args.output, args.format) + console.print(f"[green][+] Report saved: {saved}[/green]") + elif args.scan_type == "udp": + console.print("[yellow]UDP silence is reported as open|filtered; it is not proof of an open port.[/yellow]") + return 0 + + print_banner() + while True: + print_main_menu() + try: + choice = input(f"{CYAN}root@spectra:~#{RESET} Select Option: ").strip() + except (KeyboardInterrupt, EOFError): + console.print("\n[*] Exiting SpectraScan. Stay Anonymous.", style="yellow") + break + if choice == "4": + console.print("[*] Exiting SpectraScan. Stay Anonymous.", style="yellow") + break + elif choice in MENU_ACTIONS: + MENU_ACTIONS[choice][1]() + else: + console.print("[!] Invalid Option. Please try again.", style="red") + return 0 + + +if __name__ == "__main__": main() \ No newline at end of file diff --git a/modules/async_scanner.py b/modules/async_scanner.py index 6135a31..4239b21 100644 --- a/modules/async_scanner.py +++ b/modules/async_scanner.py @@ -1,29 +1,23 @@ -"""Asynchronous TCP/HTTP scanning engine for SpectraScan. - -This module is additive: the legacy PortScanner remains unchanged and callers -can opt into AsyncPortScanner when they want asyncio/aiohttp concurrency. -""" - +"""Bounded asynchronous TCP scanner with verified HTTP/TLS enrichment.""" from __future__ import annotations import asyncio +import ipaddress +import math import socket import time -from typing import Any, Dict, Iterable, List, Optional, Tuple +import urllib.parse +from typing import Any, Dict, Iterable, List, Optional import aiohttp - HTTP_PORTS = {80, 443, 8000, 8080, 8081, 8443, 8888} +MAX_CONCURRENCY = 512 +MAX_TIMEOUT = 120.0 class AsyncPortScanner: - """Concurrent TCP scanner with optional aiohttp HTTP enrichment. - - The result schema intentionally mirrors the useful subset of PortScanner: - target, resolved_ip, scan_type, timestamp, duration, open_ports and - vulnerabilities. - """ + """Concurrent TCP scanner with optional read-only HTTP metadata collection.""" def __init__( self, @@ -34,81 +28,101 @@ def __init__( check_http: bool = True, user_agent: str = "SpectraScan/async", ) -> None: - if timeout <= 0: - raise ValueError("timeout must be greater than 0") - if concurrency <= 0: - raise ValueError("concurrency must be greater than 0") - - self.target = target - self.ports = sorted({int(port) for port in (ports or [])}) + if not target or not str(target).strip(): + raise ValueError("target must not be empty") + if not math.isfinite(float(timeout)) or not 0 < float(timeout) <= MAX_TIMEOUT: + raise ValueError(f"timeout must be between 0 and {MAX_TIMEOUT:g} seconds") + if isinstance(concurrency, bool) or not 1 <= int(concurrency) <= MAX_CONCURRENCY: + raise ValueError(f"concurrency must be between 1 and {MAX_CONCURRENCY}") + raw_ports = list(ports or []) + if any(isinstance(port, bool) for port in raw_ports): + raise ValueError("ports must be integers in the range 1..65535") + try: + normalized_ports = sorted({int(port) for port in raw_ports}) + except (TypeError, ValueError, OverflowError) as exc: + raise ValueError("ports must be integers in the range 1..65535") from exc + if not normalized_ports: + raise ValueError("at least one port must be supplied") + if any(isinstance(port, bool) or port < 1 or port > 65535 for port in normalized_ports): + raise ValueError("ports must be in the range 1..65535") + self.target = str(target).strip() + self.ports = normalized_ports self.timeout = float(timeout) self.concurrency = int(concurrency) - self.check_http = check_http + self.check_http = bool(check_http) self.user_agent = user_agent self.resolved_ip: Optional[str] = None self.results: List[Dict[str, Any]] = [] self.vulnerabilities: List[Dict[str, Any]] = [] self.start_time: Optional[float] = None + def _hostname(self) -> str: + value = self.target.strip() + if "://" in value: + hostname = urllib.parse.urlsplit(value).hostname + if not hostname: + raise ValueError(f"Invalid target URL: {self.target}") + return hostname + if value.startswith("[") and "]" in value: + return value[1:value.index("]")] + try: + ipaddress.IPv6Address(value) + return value + except ValueError: + pass + if value.count(":") == 1: + host, port = value.rsplit(":", 1) + if port.isdigit() and host: + return host + return value.rstrip("/") + async def resolve(self) -> str: - """Resolve the target without blocking the event loop.""" + """Resolve IPv4 or IPv6 asynchronously.""" + hostname = self._hostname() + try: + ipaddress.ip_address(hostname) + self.resolved_ip = hostname + return hostname + except ValueError: + pass loop = asyncio.get_running_loop() - infos = await loop.getaddrinfo( - self.target, - None, - family=socket.AF_INET, - type=socket.SOCK_STREAM, - ) + try: + infos = await loop.getaddrinfo( + hostname, None, family=socket.AF_UNSPEC, type=socket.SOCK_STREAM + ) + except socket.gaierror as exc: + raise ValueError(f"Cannot resolve target: {hostname}") from exc if not infos: - raise ValueError(f"Cannot resolve target: {self.target}") - + raise ValueError(f"Cannot resolve target: {hostname}") self.resolved_ip = infos[0][4][0] return self.resolved_ip - async def _probe_tcp( - self, - port: int, - semaphore: asyncio.Semaphore, - ) -> Dict[str, Any]: + async def _probe_tcp(self, port: int, semaphore: asyncio.Semaphore) -> Dict[str, Any]: result: Dict[str, Any] = { - "port": port, - "protocol": "tcp", - "state": "closed", - "service": self._service_name(port), - "banner": "", - "error": None, + "port": port, "protocol": "tcp", "state": "closed", + "service": self._service_name(port), "banner": "", "error": None, } - async with semaphore: writer: Optional[asyncio.StreamWriter] = None try: reader, writer = await asyncio.wait_for( - asyncio.open_connection(self.resolved_ip, port), - timeout=self.timeout, + asyncio.open_connection(self.resolved_ip, port), timeout=self.timeout ) result["state"] = "open" - - if port in HTTP_PORTS: + if port in HTTP_PORTS and port not in {443, 8443}: + host = self._hostname() request = ( - f"HEAD / HTTP/1.1\r\n" - f"Host: {self.target}\r\n" + f"HEAD / HTTP/1.1\r\nHost: {host}\r\n" f"User-Agent: {self.user_agent}\r\n" "Connection: close\r\n\r\n" ) - writer.write(request.encode("ascii", errors="ignore")) + writer.write(request.encode("ascii", errors="replace")) await writer.drain() - try: - data = await asyncio.wait_for( - reader.read(1024), - timeout=min(self.timeout, 1.5), - ) - result["banner"] = data.decode( - "utf-8", errors="replace" - ).strip()[:4096] + data = await asyncio.wait_for(reader.read(1024), timeout=min(self.timeout, 1.5)) + result["banner"] = data.decode("utf-8", errors="replace").strip()[:4096] except asyncio.TimeoutError: pass - except asyncio.TimeoutError: result["state"] = "filtered" except (ConnectionRefusedError, ConnectionResetError): @@ -118,33 +132,29 @@ async def _probe_tcp( result["state"] = "closed" else: result["state"] = "error" - result["error"] = str(exc) + result["error"] = f"{type(exc).__name__}: {exc}" finally: if writer is not None: writer.close() try: await writer.wait_closed() - except (AttributeError, OSError): + except (AttributeError, OSError, asyncio.CancelledError): pass - return result async def _http_enrich( - self, - result: Dict[str, Any], - session: aiohttp.ClientSession, + self, result: Dict[str, Any], session: aiohttp.ClientSession, semaphore: asyncio.Semaphore, ) -> Dict[str, Any]: port = int(result["port"]) scheme = "https" if port in {443, 8443} else "http" - url = f"{scheme}://{self.target}:{port}/" - + hostname = self._hostname() + formatted_host = f"[{hostname}]" if ":" in hostname and not hostname.startswith("[") else hostname + url = f"{scheme}://{formatted_host}:{port}/" async with semaphore: try: async with session.get( - url, - allow_redirects=False, - ssl=False, + url, allow_redirects=False, timeout=aiohttp.ClientTimeout(total=self.timeout), ) as response: result["http"] = { @@ -152,121 +162,105 @@ async def _http_enrich( "server": response.headers.get("Server"), "content_type": response.headers.get("Content-Type"), "location": response.headers.get("Location"), + "tls_verified": scheme == "https", "security_headers": { name: response.headers.get(name) for name in ( - "Strict-Transport-Security", - "Content-Security-Policy", - "X-Frame-Options", - "X-Content-Type-Options", - ) - if response.headers.get(name) is not None + "Strict-Transport-Security", "Content-Security-Policy", + "X-Frame-Options", "X-Content-Type-Options", + "Referrer-Policy", "Permissions-Policy", + ) if response.headers.get(name) is not None }, } except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc: - result["http_error"] = str(exc) - + result["http_error"] = f"{type(exc).__name__}: {exc}" return result async def scan_async(self) -> List[Dict[str, Any]]: - """Run the asynchronous scan and return open-port results.""" + """Run a bounded scan; task count is bounded by configured concurrency.""" self.start_time = time.monotonic() + self.results = [] + self.vulnerabilities = [] await self.resolve() - semaphore = asyncio.Semaphore(self.concurrency) - tasks = [ - asyncio.create_task(self._probe_tcp(port, semaphore)) - for port in self.ports - ] - scanned = await asyncio.gather(*tasks) + work_queue: asyncio.Queue[Optional[int]] = asyncio.Queue(maxsize=max(1, self.concurrency * 2)) + worker_count = min(self.concurrency, len(self.ports)) + results: List[Dict[str, Any]] = [] - open_results = [ - result for result in scanned if result["state"] == "open" - ] - - if self.check_http and open_results: - http_results = [r for r in open_results if r["port"] in HTTP_PORTS] + async def worker() -> None: + while True: + port = await work_queue.get() + try: + if port is None: + return + result = await self._probe_tcp(port, semaphore) + if result["state"] == "open": + results.append(result) + finally: + work_queue.task_done() + + async def producer() -> None: + for port in self.ports: + await work_queue.put(port) + for _ in range(worker_count): + await work_queue.put(None) + + workers = [asyncio.create_task(worker()) for _ in range(worker_count)] + producer_task = asyncio.create_task(producer()) + try: + await producer_task + await work_queue.join() + await asyncio.gather(*workers) + except BaseException: + producer_task.cancel() + for task in workers: + task.cancel() + await asyncio.gather(producer_task, *workers, return_exceptions=True) + raise + results.sort(key=lambda item: item["port"]) + if self.check_http and results: + http_results = [item for item in results if item["port"] in HTTP_PORTS] if http_results: - connector = aiohttp.TCPConnector( - limit=self.concurrency, - ssl=False, - ) timeout = aiohttp.ClientTimeout(total=self.timeout) + # aiohttp certificate verification stays enabled by default. + connector = aiohttp.TCPConnector(limit=self.concurrency) async with aiohttp.ClientSession( - connector=connector, - timeout=timeout, + connector=connector, timeout=timeout, headers={"User-Agent": self.user_agent}, ) as session: await asyncio.gather( - *( - self._http_enrich(r, session, semaphore) - for r in http_results - ) + *(self._http_enrich(item, session, semaphore) for item in http_results) ) - - self.results = open_results + self.results = results return self.results def scan(self) -> List[Dict[str, Any]]: - """Synchronous compatibility entry point for scripts.""" + """Synchronous API; call scan_async from an existing event loop.""" return asyncio.run(self.scan_async()) def get_results(self) -> Dict[str, Any]: - duration = ( - time.monotonic() - self.start_time - if self.start_time is not None - else 0.0 - ) return { - "target": self.target, - "resolved_ip": self.resolved_ip, - "scan_type": "async-tcp", - "timestamp": time.time(), - "duration": duration, - "open_ports": self.results, - "vulnerabilities": self.vulnerabilities, + "target": self.target, "resolved_ip": self.resolved_ip, + "scan_type": "async-tcp", "timestamp": time.time(), + "duration": time.monotonic() - self.start_time if self.start_time is not None else 0.0, + "open_ports": self.results, "vulnerabilities": self.vulnerabilities, } @staticmethod def _service_name(port: int) -> str: - services = { - 21: "FTP", - 22: "SSH", - 25: "SMTP", - 53: "DNS", - 80: "HTTP", - 110: "POP3", - 143: "IMAP", - 443: "HTTPS", - 445: "SMB", - 993: "IMAPS", - 995: "POP3S", - 1433: "MSSQL", - 3306: "MySQL", - 3389: "RDP", - 5432: "PostgreSQL", - 5900: "VNC", - 6379: "Redis", - 8080: "HTTP-Alt", - 8443: "HTTPS-Alt", - 27017: "MongoDB", - } - return services.get(port, "unknown") + return { + 21: "FTP", 22: "SSH", 25: "SMTP", 53: "DNS", 80: "HTTP", + 110: "POP3", 143: "IMAP", 443: "HTTPS", 445: "SMB", + 993: "IMAPS", 995: "POP3S", 1433: "MSSQL", 3306: "MySQL", + 3389: "RDP", 5432: "PostgreSQL", 5900: "VNC", 6379: "Redis", + 8080: "HTTP-Alt", 8443: "HTTPS-Alt", 27017: "MongoDB", + }.get(port, "unknown") async def scan( - target: str, - ports: Iterable[int], - timeout: float = 1.0, - concurrency: int = 100, - check_http: bool = True, + target: str, ports: Iterable[int], timeout: float = 1.0, + concurrency: int = 100, check_http: bool = True, ) -> List[Dict[str, Any]]: - """Small functional API for callers that already run an event loop.""" - scanner = AsyncPortScanner( - target, - ports=ports, - timeout=timeout, - concurrency=concurrency, - check_http=check_http, - ) + """Functional API for callers that already run an event loop.""" + scanner = AsyncPortScanner(target, ports, timeout, concurrency, check_http) return await scanner.scan_async() diff --git a/modules/brute_forcer.py b/modules/brute_forcer.py index 6313d2c..287da4c 100644 --- a/modules/brute_forcer.py +++ b/modules/brute_forcer.py @@ -1,82 +1,206 @@ +"""Credential validation for explicitly authorized SSH/FTP audits. + +Never logs or retains plaintext passwords. SSH trusts known-host entries or +an explicitly configured SHA256 host-key fingerprint. +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import logging +import queue import socket import threading -import queue -import logging +from collections import Counter +from pathlib import Path +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + + +def _fingerprint(key: Any) -> str: + digest = hashlib.sha256(key.asbytes()).digest() + return "SHA256:" + base64.b64encode(digest).decode("ascii").rstrip("=") -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class BruteForcer: - def __init__(self, target, port, service, wordlist_path, max_threads=10): - self.target = target - self.port = port - self.service = service - self.wordlist_path = wordlist_path - self.max_threads = max_threads - self.results = [] + """Bounded worker queue with explicit result/error categories.""" + + def __init__( + self, target: str, port: int, service: str, wordlist_path: str, + max_threads: int = 10, known_hosts_path: Optional[str] = None, + expected_host_fingerprint: Optional[str] = None, + max_passwords: int = 100, + allow_insecure_ftp: bool = False, + ) -> None: + self.target = str(target) + self.port = int(port) + self.service = str(service).lower() + self.wordlist_path = str(wordlist_path) + if not 1 <= self.port <= 65535: + raise ValueError("port must be in the range 1..65535") + if self.service not in {"ssh", "ftp"}: + raise ValueError("service must be 'ssh' or 'ftp'") + if isinstance(max_threads, bool) or not 1 <= int(max_threads) <= 32: + raise ValueError("max_threads must be between 1 and 32") + self.max_threads = int(max_threads) + self.known_hosts_path = known_hosts_path + self.expected_host_fingerprint = expected_host_fingerprint + if isinstance(max_passwords, bool) or not 1 <= int(max_passwords) <= 1000: + raise ValueError("max_passwords must be between 1 and 1000") + self.max_passwords = int(max_passwords) + self.allow_insecure_ftp = bool(allow_insecure_ftp) + if self.expected_host_fingerprint and not self.expected_host_fingerprint.startswith("SHA256:"): + raise ValueError("expected_host_fingerprint must use the OpenSSH SHA256: format") + self.results: list[Dict[str, Any]] = [] + self.error_counts: Counter[str] = Counter() self.lock = threading.Lock() - def load_wordlist(self): + def load_wordlist(self) -> list[str]: try: - with open(self.wordlist_path, 'r', errors='ignore') as f: - return [line.strip() for line in f if line.strip()] - except FileNotFoundError: - logging.error(f"Wordlist not found: {self.wordlist_path}") + words = [] + with open(self.wordlist_path, "r", encoding="utf-8", errors="ignore") as stream: + for line in stream: + candidate = line.strip() + if not candidate: + continue + words.append(candidate) + if len(words) >= self.max_passwords: + break + return words + except OSError as exc: + logger.error("Unable to read credential wordlist (%s)", type(exc).__name__) return [] - def check_ssh(self, username, password): + def _record(self, outcome: Dict[str, Any]) -> None: + category = str(outcome.get("category", "unknown_error")) + with self.lock: + if outcome.get("status") == "success": + # Whitelist output fields; never persist a password or exception text. + self.results.append({ + "status": "SUCCESS", "service": self.service, + "target": self.target, "port": self.port, + "username": str(outcome.get("username", "")), + "category": "authenticated", + }) + elif outcome.get("status") in {"error", "failed"}: + self.error_counts[category] += 1 + + def check_ssh(self, username: str, password: str) -> Dict[str, Any]: try: import paramiko - client = paramiko.SSHClient() - client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - client.connect(self.target, port=self.port, username=username, password=password, timeout=2) - with self.lock: - self.results.append({'status': 'SUCCESS', 'user': username, 'pass': password}) - logging.info(f"[+] SSH Success: {username}:{password}") - client.close() - return True + except ImportError: + return {"status": "error", "category": "dependency_missing"} + client = paramiko.SSHClient() + try: + expected = (self.expected_host_fingerprint or "").strip() + # In fingerprint-pinning mode do not load existing known-host keys: + # Paramiko would accept a known key before invoking MissingHostKeyPolicy. + if not expected: + client.load_system_host_keys() + if self.known_hosts_path: + client.load_host_keys(str(Path(self.known_hosts_path).expanduser())) + if expected: + class PinnedHostKeyPolicy(paramiko.MissingHostKeyPolicy): + def missing_host_key(policy_self, ssh_client, hostname, key): + actual = _fingerprint(key) + if not hmac.compare_digest(actual, expected): + raise paramiko.SSHException("Configured host-key fingerprint mismatch") + ssh_client.get_host_keys().add(hostname, key.get_name(), key) + client.set_missing_host_key_policy(PinnedHostKeyPolicy()) + else: + client.set_missing_host_key_policy(paramiko.RejectPolicy()) + client.connect( + self.target, port=self.port, username=username, password=password, + timeout=5, banner_timeout=5, auth_timeout=5, + allow_agent=False, look_for_keys=False, + ) + remote_key = client.get_transport().get_remote_server_key() + if expected and not hmac.compare_digest(_fingerprint(remote_key), expected): + return {"status": "error", "category": "host_key_mismatch"} + return {"status": "success", "category": "authenticated", "username": username} + except paramiko.BadHostKeyException: + return {"status": "error", "category": "host_key_mismatch"} + except paramiko.AuthenticationException: + return {"status": "failed", "category": "invalid_credentials"} + except paramiko.SSHException as exc: + message = str(exc).lower() + category = "host_key_unverified" if "known_hosts" in message or "host key" in message else "ssh_error" + return {"status": "error", "category": category} + except (socket.timeout, TimeoutError): + return {"status": "error", "category": "timeout"} + except OSError: + return {"status": "error", "category": "connection_error"} except Exception: - return False + return {"status": "error", "category": "ssh_error"} + finally: + try: + client.close() + except Exception: + pass - def check_ftp(self, username, password): + def check_ftp(self, username: str, password: str) -> Dict[str, Any]: + # FTP is plaintext unless explicit FTPS is used. The insecure path must be + # deliberately opted into by the caller for isolated legacy test systems. + import ftplib # nosec B402 - FTPS is the default; plain FTP requires explicit opt-in. + + ftp = (ftplib.FTP() if self.allow_insecure_ftp else ftplib.FTP_TLS()) # nosec B321 - plain FTP is explicit opt-in only try: - import ftplib - ftp = ftplib.FTP() - ftp.connect(self.target, self.port, timeout=2) + ftp.connect(self.target, self.port, timeout=5) ftp.login(username, password) - with self.lock: - self.results.append({'status': 'SUCCESS', 'user': username, 'pass': password}) - logging.info(f"[+] FTP Success: {username}:{password}") - ftp.quit() - return True + if not self.allow_insecure_ftp: + ftp.prot_p() + return {"status": "success", "category": "authenticated", "username": username} + except ftplib.error_perm: + return {"status": "failed", "category": "invalid_credentials"} + except (socket.timeout, TimeoutError): + return {"status": "error", "category": "timeout"} + except OSError: + return {"status": "error", "category": "connection_error"} except Exception: - return False - - def run(self): + return {"status": "error", "category": "ftp_error"} + finally: + try: + ftp.quit() + except Exception: + try: + ftp.close() + except Exception: + pass + def run(self) -> list[Dict[str, Any]]: + self.results = [] + self.error_counts.clear() wordlist = self.load_wordlist() if not wordlist: - return - - tasks = queue.Queue() - threads = [] - - def worker(): - while not tasks.empty(): - username, password = tasks.get() - if self.service == 'ssh': - self.check_ssh(username, password) - elif self.service == 'ftp': - self.check_ftp(username, password) - tasks.task_done() - - for _ in range(self.max_threads): - t = threading.Thread(target=worker) - t.daemon = True - t.start() - threads.append(t) + return [] + tasks: queue.Queue[Optional[tuple[str, str]]] = queue.Queue() + for password in wordlist: + tasks.put(("admin", password)) + tasks.put(("root", password)) + worker_count = min(self.max_threads, tasks.qsize()) + for _ in range(worker_count): + tasks.put(None) - for pwd in wordlist: - tasks.put(('admin', pwd)) - tasks.put(('root', pwd)) + def worker() -> None: + while True: + item = tasks.get() + try: + if item is None: + return + username, password = item + try: + outcome = self.check_ssh(username, password) if self.service == "ssh" else self.check_ftp(username, password) + except Exception: + outcome = {"status": "error", "category": "worker_error"} + self._record(outcome) + finally: + tasks.task_done() + threads = [threading.Thread(target=worker, name=f"spectrascan-auth-{i}", daemon=True) for i in range(worker_count)] + for thread in threads: + thread.start() tasks.join() - return self.results \ No newline at end of file + for thread in threads: + thread.join() + return list(self.results) diff --git a/modules/phone_locator.py b/modules/phone_locator.py index 66e152c..e033356 100644 --- a/modules/phone_locator.py +++ b/modules/phone_locator.py @@ -2,7 +2,7 @@ import sys import requests import json -import time +import os class color: PURPLE = '\033[95m' @@ -17,21 +17,35 @@ class color: END = '\033[0m' class config: - # Note: You should replace this with your own API key from numverify.com - key = "83c4959db3119ceb89d4390911a5ce0e" + # Set NUMVERIFY_KEY in the environment; never commit provider credentials. + key = os.environ.get("NUMVERIFY_KEY", "").strip() def main(): if len(sys.argv) != 2: - print(f"Usage: python {sys.argv}