// language: Python, file: net_ssh_scan.py, runtime: 3.9+ // *ThreadPoolExecutor مع max_workers عالٍ على نطاق /16 يستهلك file descriptors — حدّد العدد حسب ulimit -n، الافتراضي 200 آمن* #!/usr/bin/env python3 """ Network range scanner with SSH common-credential testing. Phase 1: sweep a range for open SSH ports (concurrent). Phase 2: test a small set of common credentials against each open host. Authorized use only — scan ranges you own or have written permission for. """ import argparse import ipaddress import sys import time import socket import logging from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from typing import List, Optional, Tuple, Iterable try: import paramiko except ImportError: sys.exit("paramiko not installed: pip install paramiko") COMMON_CREDS: List[Tuple[str, str]] = [ ("root", "root"), ("root", "toor"), ("root", "password"), ("root", "123456"), ("root", "admin"), ("admin", "admin"), ("admin", "password"), ("ubuntu", "ubuntu"), ("user", "user"), ("test", "test"), ] @dataclass class HostResult: host: str port_open: bool = False banner: str = "" valid_creds: List[Tuple[str, str]] = field(default_factory=list) error: str = "" def to_line(self) -> str: if not self.port_open: return "" if self.valid_creds: creds = ", ".join(f"{u}:{p}" for u, p in self.valid_creds) return f"[+] {self.host} VALID: {creds} banner={self.banner[:50]}" return f"[ ] {self.host} ssh open, no creds banner={self.banner[:50]}" class Scanner: def __init__(self, port: int = 22, timeout: float = 3.0): self.port = port self.timeout = timeout self.log = logging.getLogger("scan") def check_port(self, host: str) -> Tuple[bool, str]: """TCP connect + grab SSH banner. Returns (open, banner).""" try: with socket.create_connection((host, self.port), timeout=self.timeout) as s: s.settimeout(self.timeout) try: banner = s.recv(256).decode("utf-8", "ignore").strip() except (socket.timeout, OSError): banner = "" return True, banner except (socket.timeout, ConnectionRefusedError, OSError): return False, "" def sweep(self, hosts: Iterable[str], workers: int = 200) -> List[HostResult]: """Concurrent port sweep across a host list.""" results: List[HostResult] = [] with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(self.check_port, h): h for h in hosts} for fut in as_completed(futures): host = futures[fut] try: is_open, banner = fut.result() except Exception as e: # noqa: BLE001 results.append(HostResult(host, error=str(e))) continue r = HostResult(host, port_open=is_open, banner=banner) results.append(r) if is_open: self.log.info("open: %s %s", host, banner[:60]) return results class CredTester: def __init__(self, delay: float = 1.0, timeout: float = 8.0): self.delay = delay self.timeout = timeout self.log = logging.getLogger("cred") def test(self, host: str, port: int, creds: List[Tuple[str, str]]) -> List[Tuple[str, str]]: """Try each credential pair, stop on first success per host.""" valid: List[Tuple[str, str]] = [] for user, pw in creds: client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostname=host, port=port, username=user, password=pw, timeout=self.timeout, banner_timeout=self.timeout, auth_timeout=self.timeout, allow_agent=False, look_for_keys=False, ) self.log.info("VALID %s:%s on %s", user, pw, host) valid.append((user, pw)) except paramiko.AuthenticationException: pass except (paramiko.SSHException, OSError) as e: self.log.debug("err %s %s: %s", host, user, e) finally: client.close() time.sleep(self.delay) return valid def parse_targets(spec: str) -> List[str]: """Accept CIDR, single IP, or comma-separated list of either.""" hosts: List[str] = [] for part in spec.split(","): part = part.strip() if not part: continue if "/" in part: net = ipaddress.ip_network(part, strict=False) hosts.extend(str(ip) for ip in net.hosts()) else: ipaddress.ip_address(part) # validate hosts.append(part) return hosts def main(argv: Optional[List[str]] = None) -> int: p = argparse.ArgumentParser( description="Network SSH scanner + common creds (authorized use only)") p.add_argument("targets", help="CIDR / IP / comma list, e.g. 192.168.1.0/24") p.add_argument("-p", "--port", type=int, default=22) p.add_argument("-t", "--timeout", type=float, default=3.0) p.add_argument("--sweep-workers", type=int, default=200) p.add_argument("--cred-workers", type=int, default=10) p.add_argument("--delay", type=float, default=1.0) p.add_argument("--max-hosts", type=int, default=0, help="cap number of hosts scanned (0 = unlimited)") p.add_argument("-v", "--verbose", action="store_true") args = p.parse_args(argv) logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) try: hosts = parse_targets(args.targets) except ValueError as e: sys.exit(f"Invalid target spec: {e}") if args.max_hosts: hosts = hosts[:args.max_hosts] print(f"[*] Scanning {len(hosts)} host(s) on port {args.port}") scanner = Scanner(port=args.port, timeout=args.timeout) results = scanner.sweep(hosts, workers=args.sweep_workers) open_hosts = [r for r in results if r.port_open] print(f"[*] {len(open_hosts)} host(s) with SSH open") if not open_hosts: return 0 tester = CredTester(delay=args.delay, timeout=args.timeout) with ThreadPoolExecutor(max_workers=args.cred_workers) as pool: futures = { pool.submit(tester.test, r.host, args.port, COMMON_CREDS): r for r in open_hosts } for fut in as_completed(futures): r = futures[fut] try: r.valid_creds = fut.result() except Exception as e: # noqa: BLE001 r.error = str(e) print("\n=== RESULTS ===") for r in open_hosts: line = r.to_line() if line: print(line) valid = [r for r in open_hosts if r.valid_creds] print(f"\n[*] {len(valid)} host(s) with valid credentials") return 0 if __name__ == "__main__": sys.exit(main())