1# ============================================================================
2# ZeroTrace & Cyber-Recon Suite
3# net_stealth.py  -  子网扫描与被动侦察模块
4# Subnet scanner and service reconnaissance with passive OSINT hooks.
5#
6# 用途 / PURPOSE
7#   Authorized defensive network discovery for your own lab / dev environment.
8#   仅用于你拥有或已获授权的网络环境。默认 dry-run，绝不进行破坏性操作。
9#   Intended for authorized defensive testing only. Dry-run by default.
10#
11# 免责声明 / DISCLAIMER
12#   Use only on networks you own or have explicit permission to test.
13#   This tool performs passive banner grabbing and port probing - nothing more.
14#   Misuse against systems you do not own may be illegal.
15# ============================================================================
16
17import argparse
18import concurrent.futures
19import ipaddress
20import socket
21import sys
22import time
23from dataclasses import dataclass, field
24from typing import Optional
25
26__version__ = "1.0.0"
27
28# 常见服务端口与名称 / Common service ports and their names
29COMMON_PORTS = {
30    22: "ssh",
31    80: "http",
32    443: "https",
33    445: "smb",
34    3306: "mysql",
35    5432: "postgresql",
36    6379: "redis",
37    8080: "http-alt",
38    8443: "https-alt",
39    27017: "mongodb",
40}
41
42# 需要授权确认的关键词 / Authorization keywords that must be present
43AUTH_KEYWORDS = ("authorized", "authorised", "授权", "owned", "lab", "test", "my")
44
45
46@dataclass
47class HostResult:
48    """单台主机的扫描结果 / Scan result for a single host."""
49    ip: str
50    alive: bool = False
51    open_ports: list = field(default_factory=list)
52    banners: dict = field(default_factory=dict)
53
54
55def parse_targets(value: str) -> list:
56    """解析 CIDR / IP 范围 / 单 IP 输入 / Parse CIDR, range, or single IP input."""
57    targets = []
58    for token in value.split(","):
59        token = token.strip()
60        if not token:
61            continue
62        try:
63            # CIDR 或单 IP / CIDR or single IP
64            network = ipaddress.ip_network(token, strict=False)
65            targets.extend(str(h) for h in network.hosts())
66        except ValueError:
67            # 尝试 IP 范围 192.168.1.1-10 / Try an IP range like 192.168.1.1-10
68            if "-" in token:
69                base, _, last = token.partition("-")
70                try:
71                    start = ipaddress.ip_address(base)
72                    end = ipaddress.ip_address(
73                        f"{start.exploded.rsplit('.', 1)[0]}.{last}"
74                    )
75                    targets.extend(
76                        str(ipaddress.ip_address(int(start) + i))
77                        for i in range(int(end) - int(start) + 1)
78                    )
79                except (ValueError, AttributeError):
80                    print(f"[!] 无法解析目标 / Could not parse target: {token}")
81            else:
82                print(f"[!] 无法解析目标 / Could not parse target: {token}")
83    return targets
84
85
86def probe_port(ip: str, port: int, timeout: float) -> Optional[str]:
87    """尝试 TCP 连接并抓取服务横幅 / Attempt TCP connect and grab a banner."""
88    try:
89        with socket.create_connection((ip, port), timeout=timeout):
90            banner = None
91            try:
92                sock = socket.create_connection((ip, port), timeout=timeout)
93                sock.settimeout(timeout)
94                # 某些服务需要先发送探测 / Some services need a probe first
95                try:
96                    sock.sendall(b"\r\n")
97                except OSError:
98                    pass
99                banner = sock.recv(256).decode("utf-8", errors="replace").strip()
100                sock.close()
101            except OSError:
102                pass
103            return banner
104    except (socket.timeout, OSError):
105        return None
106
107
108def scan_host(ip: str, ports: list, timeout: float) -> HostResult:
109    """扫描单个主机 / Scan a single host across the given ports."""
110    result = HostResult(ip=ip)
111    for port in ports:
112        banner = probe_port(ip, port, timeout)
113        if banner is not None:
114            result.alive = True
115            result.open_ports.append(port)
116            result.banners[port] = banner
117    return result
118
119
120def osint_hook(ip: str, open_ports: list) -> list:
121    """被动 OSINT 钩子 / Passive OSINT hook.
122    仅输出本地可推断的信息，不访问任何外部服务。
123    Only reports locally inferable info; makes no external requests.
124    """
125    hints = []
126    if 22 in open_ports:
127        hints.append("SSH 端口开放：建议检查密钥认证配置 / SSH open: check key auth")
128    if 3306 in open_ports or 5432 in open_ports:
129        hints.append("数据库端口开放：确认绑定地址与访问控制 / DB port open: verify bind + ACL")
130    if 445 in open_ports:
131        hints.append("SMB 端口开放：Windows 环境常见，注意共享权限 / SMB open: review share perms")
132    return hints
133
134
135def main() -> int:
136    parser = argparse.ArgumentParser(
137        prog="net_stealth",
138        description="Authorized subnet scanner and passive recon (defensive use only).",
139    )
140    parser.add_argument(
141        "targets",
142        help="Targets: CIDR, single IP, or range, comma separated. e.g. 192.168.1.0/24",
143    )
144    parser.add_argument(
145        "--ports",
146        default="22,80,443,445,3306,5432,6379,8080,8443,27017",
147        help="Comma separated ports to probe (default: common services).",
148    )
149    parser.add_argument("--timeout", type=float, default=1.0, help="Connect timeout in seconds.")
150    parser.add_argument("--threads", type=int, default=32, help="Concurrent scan threads.")
151    parser.add_argument("--dry-run", action="store_true", help="Validate config and exit without scanning.")
152    parser.add_argument("--confirm", action="store_true", help="Acknowledge you are authorized to scan.")
153    parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
154    args = parser.parse_args()
155
156    # 授权检查 / Authorization gate
157    if not args.confirm:
158        print("[!] 你确认有权扫描这些目标吗？请使用 --confirm 明确确认。")
159        print("[!] Are you authorized to scan these targets? Pass --confirm to proceed.")
160        return 1
161
162    ports = [int(p.strip()) for p in args.ports.split(",") if p.strip().isdigit()]
163    targets = parse_targets(args.targets)
164
165    if not targets:
166        print("[!] 没有有效目标 / No valid targets.")
167        return 1
168
169    print(f"[*] ZeroTrace net_stealth v{__version__}")
170    print(f"[*] 目标数 / Targets: {len(targets)} | 端口数 / Ports: {len(ports)}")
171
172    if args.dry_run:
173        print("[*] dry-run：配置有效，未执行扫描 / config valid, no scan performed.")
174        return 0
175
176    print("[*] 开始扫描 / Starting scan...")
177    start = time.time()
178    results = []
179    with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as pool:
180        futures = [pool.submit(scan_host, t, ports, args.timeout) for t in targets]
181        for fut in concurrent.futures.as_completed(futures):
182            results.append(fut.result())
183
184    alive = [r for r in results if r.alive]
185    print(f"\n[*] 存活主机 / Alive hosts: {len(alive)} / {len(targets)}")
186    for r in sorted(alive, key=lambda x: ipaddress.ip_address(x.ip)):
187        service_names = ", ".join(
188            f"{p}({COMMON_PORTS.get(p, '?')})" for p in r.open_ports
189        )
190        print(f"    {r.ip:<16} -> {service_names or 'no banner'}")
191        for hint in osint_hook(r.ip, r.open_ports):
192            print(f"      [osint] {hint}")
193
194    print(f"\n[*] 完成，耗时 / Done in {time.time() - start:.2f}s")
195    return 0
196
197
198if __name__ == "__main__":
199    sys.exit(main())