——**Python 高效端口扫描器**网络运维和安全学习都用得上同样是完整代码 逐段讲解。 --- # Python 高效端口扫描器多线程并发、服务识别与结果导出 ## 一、前言 在服务器运维、网络安全检测中我们经常需要知道一台主机开放了哪些端口、运行了什么服务。专业工具如 Nmap 功能强大但如果想**自己理解原理、定制化扫描逻辑**用 Python 写一个端口扫描器是最好的学习方式。 本文用 Python 标准库实现一个**零第三方依赖**的端口扫描器支持多线程并发加速、常见服务自动识别、结果导出 JSON/CSV代码可直接运行。 ⚠️ **合法使用声明**本工具仅用于学习研究和检测**你自己拥有或已获得授权**的网络设备未经授权扫描他人系统属于违法行为。 开发环境Python 3.8 | 仅使用标准库无需 pip install --- ## 二、功能总览 | 功能 | 说明 | |------|------| | TCP 全连接扫描 | 通过 connect_ex 判断端口是否开放结果准确 | | 多线程并发 | 默认 100 线程并发扫描 1000 个端口仅需数秒 | | 服务识别 | 内置 25 常见端口对应服务名80→HTTP、3306→MySQL 等 | | 灵活端口指定 | 支持单个端口、范围 1-1000、逗号列表、common 常用端口、all 全端口 | | 域名/IP 支持 | 自动将域名解析为 IP | | 结果导出 | 支持导出为 JSON 或 CSV 文件 | | 命令行参数 | 用 argparse 实现规范的 CLI 接口 | --- ## 三、完整代码 python # -*- coding: utf-8 -*- 高效端口扫描器 功能TCP全连接扫描、多线程并发、常见服务识别、结果导出JSON/CSV 作者CSDN 博主 日期2026-08-18 import socket import threading import time import json import csv import argparse from queue import Queue from typing import List, Dict, Tuple, Optional # 常见端口-服务对应表 # 用于扫描到开放端口后自动识别是什么服务 COMMON_PORTS { 21: FTP, 22: SSH, 23: Telnet, 25: SMTP, 53: DNS, 80: HTTP, 110: POP3, 135: RPC, 139: NetBIOS, 143: IMAP, 443: HTTPS, 445: SMB, 993: IMAPS, 995: POP3S, 1433: MSSQL, 1521: Oracle, 3306: MySQL, 3389: RDP, 5432: PostgreSQL, 5900: VNC, 6379: Redis, 8080: HTTP-Proxy, 8443: HTTPS-Alt, 8888: HTTP-Alt, 9200: Elasticsearch, 11211: Memcached, 27017: MongoDB, } class PortScanner: 端口扫描器核心类 def __init__(self, timeout: float 1.0, max_threads: int 100): 初始化扫描器 Args: timeout: 单个端口连接超时时间秒超时即认为端口关闭 max_threads: 最大并发线程数越大越快但资源消耗也越大 self.timeout timeout self.max_threads max_threads # 存储扫描结果只存开放端口 self.results: List[Dict] [] # 线程锁保证多线程写 results 时不冲突 self._lock threading.Lock() # 任务队列存放待扫描的端口号 self._queue: Queue Queue() # 域名解析 def resolve_host(self, host: str) - str: 将域名或主机名解析为 IP 地址 Args: host: 域名如 example.com或 IP如 192.168.1.1 Returns: 解析后的 IP 地址字符串 Raises: ValueError: 无法解析时抛出 try: # gethostbyname 会自动处理传IP就直接返回IP传域名就做DNS解析 ip socket.gethostbyname(host) return ip except socket.gaierror: raise ValueError(f无法解析主机名: {host}) # 单个端口扫描核心 def scan_port(self, ip: str, port: int) - Tuple[int, bool, Optional[str]]: 扫描单个端口是否开放TCP全连接扫描 原理尝试与目标 IP:端口 建立 TCP 连接 连接成功说明端口开放连接失败/超时说明关闭。 Args: ip: 目标 IP 地址 port: 端口号 Returns: (端口号, 是否开放, 服务名) 三元组 # 从字典中查服务名查不到则标记为 Unknown service COMMON_PORTS.get(port, Unknown) try: # AF_INET: IPv4, SOCK_STREAM: TCP 协议 sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置超时避免某个端口卡住太久 sock.settimeout(self.timeout) # connect_ex 与 connect 的区别出错时返回错误码而非抛异常 # 返回 0 表示连接成功端口开放非0表示失败 result sock.connect_ex((ip, port)) sock.close() if result 0: return (port, True, service) return (port, False, service) except socket.error: # 网络错误等异常情况统一视为端口关闭 return (port, False, service) # 多线程工作函数 def _worker(self, ip: str): 工作线程函数不断从队列取端口扫描直到取到 None 为止 这是典型的「生产者-消费者」模式 主线程把端口放入队列生产者 多个工作线程从队列取端口扫描消费者。 while True: port self._queue.get() # 取到 None 表示结束信号线程退出 if port is None: break port_num, is_open, service self.scan_port(ip, port) # 只记录开放的端口减少锁竞争和内存占用 if is_open: # 多线程写共享列表必须加锁否则可能数据错乱 with self._lock: self.results.append({ port: port_num, state: open, service: service }) # 通知队列这个任务完成了 self._queue.task_done() # 执行扫描 def scan(self, host: str, ports: List[int]) - List[Dict]: 执行多线程端口扫描 Args: host: 目标主机IP或域名 ports: 待扫描的端口号列表 Returns: 开放端口的结果列表按端口号升序排列 # 第一步解析域名 ip self.resolve_host(host) print(f\n[*] 目标: {host} ({ip})) print(f[*] 扫描端口数: {len(ports)}) print(f[*] 并发线程数: {self.max_threads}) print(f[*] 超时时间: {self.timeout}s) print(- * 50) # 清空上一次的结果 self.results [] start_time time.time() # 第二步把所有端口放入任务队列 for port in ports: self._queue.put(port) # 第三步启动工作线程 # 线程数取 max_threads 和 端口数 的较小值避免端口少时开多余线程 threads [] thread_count min(self.max_threads, len(ports)) for _ in range(thread_count): t threading.Thread(targetself._worker, args(ip,)) t.daemon True # 设为守护线程主线程退出时自动结束 t.start() threads.append(t) # 第四步等待队列中所有任务处理完毕 self._queue.join() # 第五步向队列发送结束信号让工作线程退出 for _ in threads: self._queue.put(None) for t in threads: t.join() elapsed time.time() - start_time # 按端口号排序方便阅读 self.results.sort(keylambda x: x[port]) # 打印结果 print(- * 50) print(f[] 扫描完成耗时 {elapsed:.2f} 秒) print(f[] 发现 {len(self.results)} 个开放端口:\n) if self.results: print(f{端口:10} {状态:10} {服务:20}) print(- * 40) for r in self.results: print(f{r[port]:10} {r[state]:10} {r[service]:20}) else: print([!] 未发现开放端口可能目标不可达或有防火墙) return self.results # 结果导出 def export_json(self, filepath: str): 将扫描结果导出为 JSON 文件 Args: filepath: 输出文件路径如 result.json with open(filepath, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) print(f\n[✓] 结果已导出到 {filepath}) def export_csv(self, filepath: str): 将扫描结果导出为 CSV 文件Excel 可直接打开 Args: filepath: 输出文件路径如 result.csv # utf-8-sig 编码让 Excel 正确识别中文 with open(filepath, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnames[port, state, service]) writer.writeheader() writer.writerows(self.results) print(f\n[✓] 结果已导出到 {filepath}) # 端口字符串解析 def parse_ports(port_str: str) - List[int]: 解析用户输入的端口字符串支持多种格式 - 单个端口: 80 - 端口范围: 1-1000 - 逗号分隔: 22,80,443,8080 - 混合格式: 22,80,1000-2000 - 关键字 common: 扫描内置常用端口 - 关键字 all: 扫描全部 1-65535 端口 Args: port_str: 端口描述字符串 Returns: 排序去重后的端口号列表 if port_str.lower() common: return sorted(COMMON_PORTS.keys()) if port_str.lower() all: return list(range(1, 65536)) ports set() # 用 set 自动去重 for part in port_str.split(,): part part.strip() if - in part: # 范围格式如 1-1000 start, end part.split(-, 1) ports.update(range(int(start), int(end) 1)) else: # 单个端口 ports.add(int(part)) return sorted(ports) # 命令行入口 def main(): 使用 argparse 构建规范的命令行接口 parser argparse.ArgumentParser( descriptionPython 高效端口扫描器 - 仅用于授权检测 ) # 位置参数目标主机必填 parser.add_argument(host, help目标主机IP地址或域名) # 可选参数端口范围 parser.add_argument( -p, --ports, defaultcommon, help端口范围 (默认: common常用端口; 支持 1-1000, 22,80,443, all) ) # 可选参数超时时间 parser.add_argument( -t, --timeout, typefloat, default1.0, help单个端口连接超时秒数 (默认: 1.0) ) # 可选参数线程数 parser.add_argument( -n, --threads, typeint, default100, help最大并发线程数 (默认: 100) ) # 可选参数导出文件 parser.add_argument( -o, --output, help导出结果文件路径 (.json 或 .csv) ) args parser.parse_args() # 解析端口 try: ports parse_ports(args.ports) except ValueError: print([!] 端口格式错误请检查输入示例: 1-1000 或 22,80,443) return # 创建扫描器并执行 scanner PortScanner(timeoutargs.timeout, max_threadsargs.threads) scanner.scan(args.host, ports) # 导出结果 if args.output: if args.output.lower().endswith(.json): scanner.export_json(args.output) elif args.output.lower().endswith(.csv): scanner.export_csv(args.output) else: # 没指定后缀时默认导出 JSON scanner.export_json(args.output .json) if __name__ __main__: main() --- ## 四、代码逐段讲解 ### 4.1 端口扫描原理 scan_port python sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(self.timeout) result sock.connect_ex((ip, port)) **作用** - AF_INET 表示 IPv4SOCK_STREAM 表示 TCP 协议 - connect_ex 是关键它和 connect 功能一样但**连接失败时返回错误码而非抛异常**更适合批量扫描时的性能优化 - 返回 0 连接成功 端口开放非 0 连接失败 端口关闭或被防火墙过滤 这就是 **TCP 全连接扫描Connect Scan**优点是结果准确、不需要管理员权限缺点是会在目标服务器留下完整连接日志。 ### 4.2 多线程模型 _worker 队列 python def _worker(self, ip: str): while True: port self._queue.get() if port is None: break # ... 扫描逻辑 ... self._queue.task_done() **作用** - 采用经典的**生产者-消费者模式**主线程把端口号放入 Queue多个工作线程竞争取出扫描 - Queue 本身是线程安全的多个线程同时 get() 不会冲突 - 取到 None 作为结束信号线程优雅退出 - task_done() 配合 queue.join() 实现「等所有任务做完再继续」 **为什么不用线程池** 手动实现队列模式更轻量且能精确控制结束逻辑适合这种 IO 密集型任务。 ### 4.3 线程锁保护共享数据 python with self._lock: self.results.append({...}) **作用** - 多个线程同时往 self.results 列表 append虽然 Python 的 GIL 保证了单条字节码操作的原子性但复杂操作仍可能出问题 - 加锁是最稳妥的做法with 语法会自动获取和释放锁 ### 4.4 端口解析 parse_ports python if - in part: start, end part.split(-, 1) ports.update(range(int(start), int(end) 1)) **作用** - 用 set 存储端口自动去重比如用户写了 80,80,1-10080 不会重复 - split(-, 1) 的第二个参数 1 表示最多分割一次避免端口范围里出现多个 - 时出错 - 支持 common 和 all 两个关键字方便快速使用 ### 4.5 命令行参数 argparse python parser.add_argument(-p, --ports, defaultcommon, help...) **作用** - argparse 是 Python 标准库中构建 CLI 工具的最佳选择自动生成 --help 帮助信息 - 支持短参数 -p 和长参数 --ports 两种写法 - typefloat / typeint 自动做类型转换和校验 --- ## 五、使用示例 将代码保存为 port_scanner.py在命令行运行 ### 示例1扫描常用端口默认 bash python port_scanner.py 192.168.1.1 输出 [*] 目标: 192.168.1.1 (192.168.1.1) [*] 扫描端口数: 26 [*] 并发线程数: 100 [*] 超时时间: 1.0s -------------------------------------------------- -------------------------------------------------- [] 扫描完成耗时 0.15 秒 [] 发现 3 个开放端口: 端口 状态 服务 ---------------------------------------- 22 open SSH 80 open HTTP 443 open HTTPS ### 示例2扫描指定端口范围并导出结果 bash python port_scanner.py example.com -p 1-1000 -o result.csv ### 示例3扫描特定端口调高线程数和超时 bash python port_scanner.py 10.0.0.5 -p 22,80,443,3306,6379 -n 200 -t 2.0 ### 示例4查看帮助 bash python port_scanner.py --help --- ## 六、性能对比 | 扫描方式 | 1000 个端口耗时 | 说明 | |----------|----------------|------| | 单线程顺序扫描 | ~1000 秒按1秒超时 | 极慢不实用 | | 10 线程并发 | ~100 秒 | 较慢 | | 100 线程并发 | ~10 秒 | 推荐平衡速度与资源 | | 500 线程并发 | ~3-5 秒 | 快但可能触发目标限流 | 端口扫描是 **IO 密集型任务**多线程能极大提升效率。但线程数不是越多越好过高会导致本机资源耗尽或被目标防火墙封禁。 --- ## 七、扩展思路 这个扫描器还可以继续升级 1. **SYN 半开扫描**用 scapy 库发送 SYN 包不建立完整连接更隐蔽需要管理员权限 2. **端口Banner抓取**连接成功后发送请求读取服务返回的版本信息如 SSH 版本、Web Server 类型 3. **IP 段扫描**支持 192.168.1.1-254 或 CIDR 格式 192.168.1.0/24 4. **扫描进度条**用 tqdm 库显示实时扫描进度 5. **结果对比**多次扫描对比端口变化发现新增或关闭的端口 6. **UDP 扫描**目前只支持 TCP可扩展 UDP 端口检测 --- ## 八、总结 本文实现了一个**零依赖、高性能**的 Python 端口扫描器核心知识点包括 - socket 网络编程与 TCP 连接原理 - threading 多线程 Queue 生产者消费者模式 - 线程锁保护共享资源 - argparse 构建专业命令行工具 - JSON/CSV 数据导出 代码约 250 行结构清晰可直接运行。建议先扫描自己的路由器或测试服务器学习使用**切勿扫描未授权的目标**。https://github.com/ramirezolivia00/psoamk/blob/main/2026%E5%AE%98%E6%96%B9%E8%B5%84%E5%8E%86%7C%E9%87%91%E6%B4%8B%E4%BB%A3%E7%90%86.mdhttps://github.com/ramirezolivia00/psoamk/commit/53d38c717b95c2e1519ff265696a46499999f7abhttps://github.com/walterstaylor44/twqxpi/blob/main/2026%E7%AC%AC%E4%B8%80%E7%90%86%E6%83%B3%7CAMD%E8%81%9A%E6%98%9F%E4%BB%A3%E7%90%86.mdhttps://github.com/walterstaylor44/twqxpi/commit/88463ed4f82795d9187368c7c10941e6d0412ea3https://github.com/hollowayronald2/jiskww/blob/main/2026%E7%AC%AC%E4%B8%80%E6%A0%BC%E5%B1%80%7C%E5%AE%89%E4%BF%A113%E4%BB%A3%E7%90%86.mdhttps://github.com/hollowayronald2/jiskww/commit/67ba623aac430d48a131d4e7ac755362acd3ff36https://github.com/martinezmarilyn2345/boqoaz/blob/main/2026%E5%AE%98%E6%96%B9%E9%85%8D%E8%89%B2%7C%E5%8D%8E%E6%B6%A62%E4%BB%A3%E7%90%86.mdhttps://github.com/martinezmarilyn2345/boqoaz/commit/a3e13038e004ed7417c8e1d3a0f1e6a1abaf8a23https://github.com/martinsandra772/iesfzm/blob/main/2026%E5%AE%98%E6%96%B9%E7%9C%9F%E9%A2%98%7C%E5%8D%8E%E6%B6%A6%E4%BB%A3%E7%90%86.mdhttps://github.com/martinsandra772/iesfzm/commit/8ed7ccff3322d44f6a1d6e691365b797fc3d0bbfhttps://github.com/castrojessica1/akwpho/blob/main/2026%E7%AC%AC%E4%B8%80%E7%9B%9B%E4%B8%96%7C%E5%8D%8E%E7%BE%8E%E4%BB%A3%E7%90%86.mdhttps://github.com/castrojessica1/akwpho/commit/6d995dbbc5fcc0cbc6c3df7814cb3f92a95ee73ahttps://github.com/higginsrachel339/ingiql/blob/main/2026%E5%AE%98%E6%96%B9%E5%85%AC%E6%B5%8B%7C%E4%B8%89%E7%89%9B%E4%BB%A3%E7%90%86.mdhttps://github.com/higginsrachel339/ingiql/commit/e408bd1ff3bf8c5df666579db37a2c07b1c80ba4https://github.com/campbellmichael2/nrllkk/blob/main/2026%E5%AE%98%E6%96%B9%E7%94%84%E9%80%89%7Cvs%E5%A8%B1%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/campbellmichael2/nrllkk/commit/1215bf505186b8ad9d205a62b6e14b5073c59ad7https://github.com/wilkinsonandrea9651/trgfxd/blob/main/2026%E5%AE%98%E6%96%B9%E5%A5%97%E8%A3%85%7C%E5%8D%9A%E7%8C%AB%E4%BB%A3%E7%90%86.mdhttps://github.com/wilkinsonandrea9651/trgfxd/commit/6956b9da237d5e6ce023e4b5186f739c1d0e3a99https://github.com/velezkimberly1496/oooavx/blob/main/2026%E5%AE%98%E6%96%B9%E6%9C%9F%E5%88%8A%7C2%E5%8F%B7%E7%AB%99%E4%BB%A3%E7%90%86.mdhttps://github.com/velezkimberly1496/oooavx/commit/c63cc5038bb387f995cc87c05de6229fbf53714dhttps://github.com/mcneildustin18/tmuanf/blob/main/2026%E5%AE%98%E6%96%B9%E5%BC%80%E5%94%AE%7C%E5%82%B2%E4%B8%96%E7%9A%87%E6%9C%9D%E4%BB%A3%E7%90%86.mdhttps://github.com/mcneildustin18/tmuanf/commit/e6dc38ea2288703aead78de0fe5d59dd4a921c08https://github.com/maxwellrachel2/duhdog/blob/main/2026%E7%AC%AC%E4%B8%80%E4%BF%A1%E4%BB%B0%7C%E5%88%A9%E6%BE%B3%E4%BB%A3%E7%90%86.mdhttps://github.com/maxwellrachel2/duhdog/commit/d590c7a83305e1ff82e183ae87cdc843681d22fdhttps://github.com/garciawesley609/zhwonf/blob/main/2026%E5%AE%98%E6%96%B9%E5%90%8D%E7%89%87%7Cv8%E5%A8%B1%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/garciawesley609/zhwonf/commit/6e59fb75f671efe4a9719fe4d3238a986971ce2chttps://github.com/walterstaylor44/twqxpi/blob/main/2026%E5%AE%98%E6%96%B9%E6%AD%A3%E7%89%88%7Cvs%E5%93%88%E5%B8%8C%E4%BB%A3%E7%90%86.mdhttps://github.com/walterstaylor44/twqxpi/commit/cfb8604e1e7fee00e2d677df561593c0256659dchttps://github.com/ramirezolivia00/psoamk/blob/main/2026%E5%AE%98%E6%96%B9%E6%8A%BD%E5%A5%96%7Cv8%E5%93%88%E5%B8%8C%E4%BB%A3%E7%90%86.mdhttps://github.com/ramirezolivia00/psoamk/commit/37f1099297b8f83eff74d1bd35edf79127b2d38dhttps://github.com/lambdaniel5688/xesldc/blob/main/2026%E7%AC%AC%E4%B8%80%E4%BC%98%E5%93%81%7C%E8%93%9D%E5%9B%BE%E4%BB%A3%E7%90%86.mdhttps://github.com/lambdaniel5688/xesldc/commit/ce495adce51353d8ada9bc27f13e5c6373baf225https://github.com/hollowayronald2/jiskww/blob/main/%E3%80%902026%E6%8E%A8%E8%8D%90%2F%E7%AC%AC%E4%B8%80%E3%80%91%7C%E6%B1%87%E4%BC%97%E4%BB%A3%E7%90%86.mdhttps://github.com/hollowayronald2/jiskww/commit/5fbbcdf202b88b4d068dc09edf85d0c19a3f9070https://github.com/andersonjimmy3/eeheee/blob/main/2026%E5%AE%98%E6%96%B9%E5%AE%A3%E4%BC%A0%7C%E4%BC%97%E8%85%BE%E4%BB%A3%E7%90%86.mdhttps://github.com/andersonjimmy3/eeheee/commit/065f1d4c92256ba9c38ae3627c184103fc5c1886https://github.com/martinezmarilyn2345/boqoaz/blob/main/2026%E5%AE%98%E6%96%B9%E6%B0%B4%E5%8D%B0%7C%E5%96%9C%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/martinezmarilyn2345/boqoaz/commit/88ca276023d85846650fe8cf8b91d6e3141bce6chttps://github.com/martinsandra772/iesfzm/blob/main/2026%E5%AE%98%E6%96%B9%E6%9D%A1%E6%96%87%7C%E5%A4%A7%E4%BC%97%E4%BB%A3%E7%90%86.mdhttps://github.com/martinsandra772/iesfzm/commit/c679809e1ac19546417b4132fb34e7226a6b1b6ehttps://github.com/castrojessica1/akwpho/blob/main/2026%E5%AE%98%E6%96%B9%E5%A3%81%E7%BA%B8%7C%E6%81%92%E8%A1%8C8.mdhttps://github.com/castrojessica1/akwpho/commit/a42e7eaf01fdc21e986b9fa95cd320968565b11bhttps://github.com/wilkinsonandrea9651/trgfxd/blob/main/2026%E5%AE%98%E6%96%B9%E5%A4%96%E8%AE%BE%7C%E5%85%AB%E6%96%B9%E4%BB%A3%E7%90%86.mdhttps://github.com/wilkinsonandrea9651/trgfxd/commit/dc6a14ddf4d8c8bf5aad2d31c52a47d55b12a40chttps://github.com/velezkimberly1496/oooavx/blob/main/2026%E5%AE%98%E6%96%B9%E6%9D%83%E9%99%90%7C%E9%AB%98%E5%BE%B7%E4%BB%A3%E7%90%86.md
从零实现 Python 端口扫描器：100 线程并发，3 秒扫完 1000 端口
——**Python 高效端口扫描器**网络运维和安全学习都用得上同样是完整代码 逐段讲解。 --- # Python 高效端口扫描器多线程并发、服务识别与结果导出 ## 一、前言 在服务器运维、网络安全检测中我们经常需要知道一台主机开放了哪些端口、运行了什么服务。专业工具如 Nmap 功能强大但如果想**自己理解原理、定制化扫描逻辑**用 Python 写一个端口扫描器是最好的学习方式。 本文用 Python 标准库实现一个**零第三方依赖**的端口扫描器支持多线程并发加速、常见服务自动识别、结果导出 JSON/CSV代码可直接运行。 ⚠️ **合法使用声明**本工具仅用于学习研究和检测**你自己拥有或已获得授权**的网络设备未经授权扫描他人系统属于违法行为。 开发环境Python 3.8 | 仅使用标准库无需 pip install --- ## 二、功能总览 | 功能 | 说明 | |------|------| | TCP 全连接扫描 | 通过 connect_ex 判断端口是否开放结果准确 | | 多线程并发 | 默认 100 线程并发扫描 1000 个端口仅需数秒 | | 服务识别 | 内置 25 常见端口对应服务名80→HTTP、3306→MySQL 等 | | 灵活端口指定 | 支持单个端口、范围 1-1000、逗号列表、common 常用端口、all 全端口 | | 域名/IP 支持 | 自动将域名解析为 IP | | 结果导出 | 支持导出为 JSON 或 CSV 文件 | | 命令行参数 | 用 argparse 实现规范的 CLI 接口 | --- ## 三、完整代码 python # -*- coding: utf-8 -*- 高效端口扫描器 功能TCP全连接扫描、多线程并发、常见服务识别、结果导出JSON/CSV 作者CSDN 博主 日期2026-08-18 import socket import threading import time import json import csv import argparse from queue import Queue from typing import List, Dict, Tuple, Optional # 常见端口-服务对应表 # 用于扫描到开放端口后自动识别是什么服务 COMMON_PORTS { 21: FTP, 22: SSH, 23: Telnet, 25: SMTP, 53: DNS, 80: HTTP, 110: POP3, 135: RPC, 139: NetBIOS, 143: IMAP, 443: HTTPS, 445: SMB, 993: IMAPS, 995: POP3S, 1433: MSSQL, 1521: Oracle, 3306: MySQL, 3389: RDP, 5432: PostgreSQL, 5900: VNC, 6379: Redis, 8080: HTTP-Proxy, 8443: HTTPS-Alt, 8888: HTTP-Alt, 9200: Elasticsearch, 11211: Memcached, 27017: MongoDB, } class PortScanner: 端口扫描器核心类 def __init__(self, timeout: float 1.0, max_threads: int 100): 初始化扫描器 Args: timeout: 单个端口连接超时时间秒超时即认为端口关闭 max_threads: 最大并发线程数越大越快但资源消耗也越大 self.timeout timeout self.max_threads max_threads # 存储扫描结果只存开放端口 self.results: List[Dict] [] # 线程锁保证多线程写 results 时不冲突 self._lock threading.Lock() # 任务队列存放待扫描的端口号 self._queue: Queue Queue() # 域名解析 def resolve_host(self, host: str) - str: 将域名或主机名解析为 IP 地址 Args: host: 域名如 example.com或 IP如 192.168.1.1 Returns: 解析后的 IP 地址字符串 Raises: ValueError: 无法解析时抛出 try: # gethostbyname 会自动处理传IP就直接返回IP传域名就做DNS解析 ip socket.gethostbyname(host) return ip except socket.gaierror: raise ValueError(f无法解析主机名: {host}) # 单个端口扫描核心 def scan_port(self, ip: str, port: int) - Tuple[int, bool, Optional[str]]: 扫描单个端口是否开放TCP全连接扫描 原理尝试与目标 IP:端口 建立 TCP 连接 连接成功说明端口开放连接失败/超时说明关闭。 Args: ip: 目标 IP 地址 port: 端口号 Returns: (端口号, 是否开放, 服务名) 三元组 # 从字典中查服务名查不到则标记为 Unknown service COMMON_PORTS.get(port, Unknown) try: # AF_INET: IPv4, SOCK_STREAM: TCP 协议 sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置超时避免某个端口卡住太久 sock.settimeout(self.timeout) # connect_ex 与 connect 的区别出错时返回错误码而非抛异常 # 返回 0 表示连接成功端口开放非0表示失败 result sock.connect_ex((ip, port)) sock.close() if result 0: return (port, True, service) return (port, False, service) except socket.error: # 网络错误等异常情况统一视为端口关闭 return (port, False, service) # 多线程工作函数 def _worker(self, ip: str): 工作线程函数不断从队列取端口扫描直到取到 None 为止 这是典型的「生产者-消费者」模式 主线程把端口放入队列生产者 多个工作线程从队列取端口扫描消费者。 while True: port self._queue.get() # 取到 None 表示结束信号线程退出 if port is None: break port_num, is_open, service self.scan_port(ip, port) # 只记录开放的端口减少锁竞争和内存占用 if is_open: # 多线程写共享列表必须加锁否则可能数据错乱 with self._lock: self.results.append({ port: port_num, state: open, service: service }) # 通知队列这个任务完成了 self._queue.task_done() # 执行扫描 def scan(self, host: str, ports: List[int]) - List[Dict]: 执行多线程端口扫描 Args: host: 目标主机IP或域名 ports: 待扫描的端口号列表 Returns: 开放端口的结果列表按端口号升序排列 # 第一步解析域名 ip self.resolve_host(host) print(f\n[*] 目标: {host} ({ip})) print(f[*] 扫描端口数: {len(ports)}) print(f[*] 并发线程数: {self.max_threads}) print(f[*] 超时时间: {self.timeout}s) print(- * 50) # 清空上一次的结果 self.results [] start_time time.time() # 第二步把所有端口放入任务队列 for port in ports: self._queue.put(port) # 第三步启动工作线程 # 线程数取 max_threads 和 端口数 的较小值避免端口少时开多余线程 threads [] thread_count min(self.max_threads, len(ports)) for _ in range(thread_count): t threading.Thread(targetself._worker, args(ip,)) t.daemon True # 设为守护线程主线程退出时自动结束 t.start() threads.append(t) # 第四步等待队列中所有任务处理完毕 self._queue.join() # 第五步向队列发送结束信号让工作线程退出 for _ in threads: self._queue.put(None) for t in threads: t.join() elapsed time.time() - start_time # 按端口号排序方便阅读 self.results.sort(keylambda x: x[port]) # 打印结果 print(- * 50) print(f[] 扫描完成耗时 {elapsed:.2f} 秒) print(f[] 发现 {len(self.results)} 个开放端口:\n) if self.results: print(f{端口:10} {状态:10} {服务:20}) print(- * 40) for r in self.results: print(f{r[port]:10} {r[state]:10} {r[service]:20}) else: print([!] 未发现开放端口可能目标不可达或有防火墙) return self.results # 结果导出 def export_json(self, filepath: str): 将扫描结果导出为 JSON 文件 Args: filepath: 输出文件路径如 result.json with open(filepath, w, encodingutf-8) as f: json.dump(self.results, f, ensure_asciiFalse, indent2) print(f\n[✓] 结果已导出到 {filepath}) def export_csv(self, filepath: str): 将扫描结果导出为 CSV 文件Excel 可直接打开 Args: filepath: 输出文件路径如 result.csv # utf-8-sig 编码让 Excel 正确识别中文 with open(filepath, w, encodingutf-8-sig, newline) as f: writer csv.DictWriter(f, fieldnames[port, state, service]) writer.writeheader() writer.writerows(self.results) print(f\n[✓] 结果已导出到 {filepath}) # 端口字符串解析 def parse_ports(port_str: str) - List[int]: 解析用户输入的端口字符串支持多种格式 - 单个端口: 80 - 端口范围: 1-1000 - 逗号分隔: 22,80,443,8080 - 混合格式: 22,80,1000-2000 - 关键字 common: 扫描内置常用端口 - 关键字 all: 扫描全部 1-65535 端口 Args: port_str: 端口描述字符串 Returns: 排序去重后的端口号列表 if port_str.lower() common: return sorted(COMMON_PORTS.keys()) if port_str.lower() all: return list(range(1, 65536)) ports set() # 用 set 自动去重 for part in port_str.split(,): part part.strip() if - in part: # 范围格式如 1-1000 start, end part.split(-, 1) ports.update(range(int(start), int(end) 1)) else: # 单个端口 ports.add(int(part)) return sorted(ports) # 命令行入口 def main(): 使用 argparse 构建规范的命令行接口 parser argparse.ArgumentParser( descriptionPython 高效端口扫描器 - 仅用于授权检测 ) # 位置参数目标主机必填 parser.add_argument(host, help目标主机IP地址或域名) # 可选参数端口范围 parser.add_argument( -p, --ports, defaultcommon, help端口范围 (默认: common常用端口; 支持 1-1000, 22,80,443, all) ) # 可选参数超时时间 parser.add_argument( -t, --timeout, typefloat, default1.0, help单个端口连接超时秒数 (默认: 1.0) ) # 可选参数线程数 parser.add_argument( -n, --threads, typeint, default100, help最大并发线程数 (默认: 100) ) # 可选参数导出文件 parser.add_argument( -o, --output, help导出结果文件路径 (.json 或 .csv) ) args parser.parse_args() # 解析端口 try: ports parse_ports(args.ports) except ValueError: print([!] 端口格式错误请检查输入示例: 1-1000 或 22,80,443) return # 创建扫描器并执行 scanner PortScanner(timeoutargs.timeout, max_threadsargs.threads) scanner.scan(args.host, ports) # 导出结果 if args.output: if args.output.lower().endswith(.json): scanner.export_json(args.output) elif args.output.lower().endswith(.csv): scanner.export_csv(args.output) else: # 没指定后缀时默认导出 JSON scanner.export_json(args.output .json) if __name__ __main__: main() --- ## 四、代码逐段讲解 ### 4.1 端口扫描原理 scan_port python sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(self.timeout) result sock.connect_ex((ip, port)) **作用** - AF_INET 表示 IPv4SOCK_STREAM 表示 TCP 协议 - connect_ex 是关键它和 connect 功能一样但**连接失败时返回错误码而非抛异常**更适合批量扫描时的性能优化 - 返回 0 连接成功 端口开放非 0 连接失败 端口关闭或被防火墙过滤 这就是 **TCP 全连接扫描Connect Scan**优点是结果准确、不需要管理员权限缺点是会在目标服务器留下完整连接日志。 ### 4.2 多线程模型 _worker 队列 python def _worker(self, ip: str): while True: port self._queue.get() if port is None: break # ... 扫描逻辑 ... self._queue.task_done() **作用** - 采用经典的**生产者-消费者模式**主线程把端口号放入 Queue多个工作线程竞争取出扫描 - Queue 本身是线程安全的多个线程同时 get() 不会冲突 - 取到 None 作为结束信号线程优雅退出 - task_done() 配合 queue.join() 实现「等所有任务做完再继续」 **为什么不用线程池** 手动实现队列模式更轻量且能精确控制结束逻辑适合这种 IO 密集型任务。 ### 4.3 线程锁保护共享数据 python with self._lock: self.results.append({...}) **作用** - 多个线程同时往 self.results 列表 append虽然 Python 的 GIL 保证了单条字节码操作的原子性但复杂操作仍可能出问题 - 加锁是最稳妥的做法with 语法会自动获取和释放锁 ### 4.4 端口解析 parse_ports python if - in part: start, end part.split(-, 1) ports.update(range(int(start), int(end) 1)) **作用** - 用 set 存储端口自动去重比如用户写了 80,80,1-10080 不会重复 - split(-, 1) 的第二个参数 1 表示最多分割一次避免端口范围里出现多个 - 时出错 - 支持 common 和 all 两个关键字方便快速使用 ### 4.5 命令行参数 argparse python parser.add_argument(-p, --ports, defaultcommon, help...) **作用** - argparse 是 Python 标准库中构建 CLI 工具的最佳选择自动生成 --help 帮助信息 - 支持短参数 -p 和长参数 --ports 两种写法 - typefloat / typeint 自动做类型转换和校验 --- ## 五、使用示例 将代码保存为 port_scanner.py在命令行运行 ### 示例1扫描常用端口默认 bash python port_scanner.py 192.168.1.1 输出 [*] 目标: 192.168.1.1 (192.168.1.1) [*] 扫描端口数: 26 [*] 并发线程数: 100 [*] 超时时间: 1.0s -------------------------------------------------- -------------------------------------------------- [] 扫描完成耗时 0.15 秒 [] 发现 3 个开放端口: 端口 状态 服务 ---------------------------------------- 22 open SSH 80 open HTTP 443 open HTTPS ### 示例2扫描指定端口范围并导出结果 bash python port_scanner.py example.com -p 1-1000 -o result.csv ### 示例3扫描特定端口调高线程数和超时 bash python port_scanner.py 10.0.0.5 -p 22,80,443,3306,6379 -n 200 -t 2.0 ### 示例4查看帮助 bash python port_scanner.py --help --- ## 六、性能对比 | 扫描方式 | 1000 个端口耗时 | 说明 | |----------|----------------|------| | 单线程顺序扫描 | ~1000 秒按1秒超时 | 极慢不实用 | | 10 线程并发 | ~100 秒 | 较慢 | | 100 线程并发 | ~10 秒 | 推荐平衡速度与资源 | | 500 线程并发 | ~3-5 秒 | 快但可能触发目标限流 | 端口扫描是 **IO 密集型任务**多线程能极大提升效率。但线程数不是越多越好过高会导致本机资源耗尽或被目标防火墙封禁。 --- ## 七、扩展思路 这个扫描器还可以继续升级 1. **SYN 半开扫描**用 scapy 库发送 SYN 包不建立完整连接更隐蔽需要管理员权限 2. **端口Banner抓取**连接成功后发送请求读取服务返回的版本信息如 SSH 版本、Web Server 类型 3. **IP 段扫描**支持 192.168.1.1-254 或 CIDR 格式 192.168.1.0/24 4. **扫描进度条**用 tqdm 库显示实时扫描进度 5. **结果对比**多次扫描对比端口变化发现新增或关闭的端口 6. **UDP 扫描**目前只支持 TCP可扩展 UDP 端口检测 --- ## 八、总结 本文实现了一个**零依赖、高性能**的 Python 端口扫描器核心知识点包括 - socket 网络编程与 TCP 连接原理 - threading 多线程 Queue 生产者消费者模式 - 线程锁保护共享资源 - argparse 构建专业命令行工具 - JSON/CSV 数据导出 代码约 250 行结构清晰可直接运行。建议先扫描自己的路由器或测试服务器学习使用**切勿扫描未授权的目标**。https://github.com/ramirezolivia00/psoamk/blob/main/2026%E5%AE%98%E6%96%B9%E8%B5%84%E5%8E%86%7C%E9%87%91%E6%B4%8B%E4%BB%A3%E7%90%86.mdhttps://github.com/ramirezolivia00/psoamk/commit/53d38c717b95c2e1519ff265696a46499999f7abhttps://github.com/walterstaylor44/twqxpi/blob/main/2026%E7%AC%AC%E4%B8%80%E7%90%86%E6%83%B3%7CAMD%E8%81%9A%E6%98%9F%E4%BB%A3%E7%90%86.mdhttps://github.com/walterstaylor44/twqxpi/commit/88463ed4f82795d9187368c7c10941e6d0412ea3https://github.com/hollowayronald2/jiskww/blob/main/2026%E7%AC%AC%E4%B8%80%E6%A0%BC%E5%B1%80%7C%E5%AE%89%E4%BF%A113%E4%BB%A3%E7%90%86.mdhttps://github.com/hollowayronald2/jiskww/commit/67ba623aac430d48a131d4e7ac755362acd3ff36https://github.com/martinezmarilyn2345/boqoaz/blob/main/2026%E5%AE%98%E6%96%B9%E9%85%8D%E8%89%B2%7C%E5%8D%8E%E6%B6%A62%E4%BB%A3%E7%90%86.mdhttps://github.com/martinezmarilyn2345/boqoaz/commit/a3e13038e004ed7417c8e1d3a0f1e6a1abaf8a23https://github.com/martinsandra772/iesfzm/blob/main/2026%E5%AE%98%E6%96%B9%E7%9C%9F%E9%A2%98%7C%E5%8D%8E%E6%B6%A6%E4%BB%A3%E7%90%86.mdhttps://github.com/martinsandra772/iesfzm/commit/8ed7ccff3322d44f6a1d6e691365b797fc3d0bbfhttps://github.com/castrojessica1/akwpho/blob/main/2026%E7%AC%AC%E4%B8%80%E7%9B%9B%E4%B8%96%7C%E5%8D%8E%E7%BE%8E%E4%BB%A3%E7%90%86.mdhttps://github.com/castrojessica1/akwpho/commit/6d995dbbc5fcc0cbc6c3df7814cb3f92a95ee73ahttps://github.com/higginsrachel339/ingiql/blob/main/2026%E5%AE%98%E6%96%B9%E5%85%AC%E6%B5%8B%7C%E4%B8%89%E7%89%9B%E4%BB%A3%E7%90%86.mdhttps://github.com/higginsrachel339/ingiql/commit/e408bd1ff3bf8c5df666579db37a2c07b1c80ba4https://github.com/campbellmichael2/nrllkk/blob/main/2026%E5%AE%98%E6%96%B9%E7%94%84%E9%80%89%7Cvs%E5%A8%B1%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/campbellmichael2/nrllkk/commit/1215bf505186b8ad9d205a62b6e14b5073c59ad7https://github.com/wilkinsonandrea9651/trgfxd/blob/main/2026%E5%AE%98%E6%96%B9%E5%A5%97%E8%A3%85%7C%E5%8D%9A%E7%8C%AB%E4%BB%A3%E7%90%86.mdhttps://github.com/wilkinsonandrea9651/trgfxd/commit/6956b9da237d5e6ce023e4b5186f739c1d0e3a99https://github.com/velezkimberly1496/oooavx/blob/main/2026%E5%AE%98%E6%96%B9%E6%9C%9F%E5%88%8A%7C2%E5%8F%B7%E7%AB%99%E4%BB%A3%E7%90%86.mdhttps://github.com/velezkimberly1496/oooavx/commit/c63cc5038bb387f995cc87c05de6229fbf53714dhttps://github.com/mcneildustin18/tmuanf/blob/main/2026%E5%AE%98%E6%96%B9%E5%BC%80%E5%94%AE%7C%E5%82%B2%E4%B8%96%E7%9A%87%E6%9C%9D%E4%BB%A3%E7%90%86.mdhttps://github.com/mcneildustin18/tmuanf/commit/e6dc38ea2288703aead78de0fe5d59dd4a921c08https://github.com/maxwellrachel2/duhdog/blob/main/2026%E7%AC%AC%E4%B8%80%E4%BF%A1%E4%BB%B0%7C%E5%88%A9%E6%BE%B3%E4%BB%A3%E7%90%86.mdhttps://github.com/maxwellrachel2/duhdog/commit/d590c7a83305e1ff82e183ae87cdc843681d22fdhttps://github.com/garciawesley609/zhwonf/blob/main/2026%E5%AE%98%E6%96%B9%E5%90%8D%E7%89%87%7Cv8%E5%A8%B1%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/garciawesley609/zhwonf/commit/6e59fb75f671efe4a9719fe4d3238a986971ce2chttps://github.com/walterstaylor44/twqxpi/blob/main/2026%E5%AE%98%E6%96%B9%E6%AD%A3%E7%89%88%7Cvs%E5%93%88%E5%B8%8C%E4%BB%A3%E7%90%86.mdhttps://github.com/walterstaylor44/twqxpi/commit/cfb8604e1e7fee00e2d677df561593c0256659dchttps://github.com/ramirezolivia00/psoamk/blob/main/2026%E5%AE%98%E6%96%B9%E6%8A%BD%E5%A5%96%7Cv8%E5%93%88%E5%B8%8C%E4%BB%A3%E7%90%86.mdhttps://github.com/ramirezolivia00/psoamk/commit/37f1099297b8f83eff74d1bd35edf79127b2d38dhttps://github.com/lambdaniel5688/xesldc/blob/main/2026%E7%AC%AC%E4%B8%80%E4%BC%98%E5%93%81%7C%E8%93%9D%E5%9B%BE%E4%BB%A3%E7%90%86.mdhttps://github.com/lambdaniel5688/xesldc/commit/ce495adce51353d8ada9bc27f13e5c6373baf225https://github.com/hollowayronald2/jiskww/blob/main/%E3%80%902026%E6%8E%A8%E8%8D%90%2F%E7%AC%AC%E4%B8%80%E3%80%91%7C%E6%B1%87%E4%BC%97%E4%BB%A3%E7%90%86.mdhttps://github.com/hollowayronald2/jiskww/commit/5fbbcdf202b88b4d068dc09edf85d0c19a3f9070https://github.com/andersonjimmy3/eeheee/blob/main/2026%E5%AE%98%E6%96%B9%E5%AE%A3%E4%BC%A0%7C%E4%BC%97%E8%85%BE%E4%BB%A3%E7%90%86.mdhttps://github.com/andersonjimmy3/eeheee/commit/065f1d4c92256ba9c38ae3627c184103fc5c1886https://github.com/martinezmarilyn2345/boqoaz/blob/main/2026%E5%AE%98%E6%96%B9%E6%B0%B4%E5%8D%B0%7C%E5%96%9C%E4%B9%90%E4%BB%A3%E7%90%86.mdhttps://github.com/martinezmarilyn2345/boqoaz/commit/88ca276023d85846650fe8cf8b91d6e3141bce6chttps://github.com/martinsandra772/iesfzm/blob/main/2026%E5%AE%98%E6%96%B9%E6%9D%A1%E6%96%87%7C%E5%A4%A7%E4%BC%97%E4%BB%A3%E7%90%86.mdhttps://github.com/martinsandra772/iesfzm/commit/c679809e1ac19546417b4132fb34e7226a6b1b6ehttps://github.com/castrojessica1/akwpho/blob/main/2026%E5%AE%98%E6%96%B9%E5%A3%81%E7%BA%B8%7C%E6%81%92%E8%A1%8C8.mdhttps://github.com/castrojessica1/akwpho/commit/a42e7eaf01fdc21e986b9fa95cd320968565b11bhttps://github.com/wilkinsonandrea9651/trgfxd/blob/main/2026%E5%AE%98%E6%96%B9%E5%A4%96%E8%AE%BE%7C%E5%85%AB%E6%96%B9%E4%BB%A3%E7%90%86.mdhttps://github.com/wilkinsonandrea9651/trgfxd/commit/dc6a14ddf4d8c8bf5aad2d31c52a47d55b12a40chttps://github.com/velezkimberly1496/oooavx/blob/main/2026%E5%AE%98%E6%96%B9%E6%9D%83%E9%99%90%7C%E9%AB%98%E5%BE%B7%E4%BB%A3%E7%90%86.md