1. 为什么我们需要自己写端口扫描器？
端口扫描是网络安全领域最基础也最重要的技能之一。作为一名渗透测试工程师，我几乎每天都会用到端口扫描工具。市面上的确有很多成熟的端口扫描器，比如Nmap、Masscan等，但它们要么功能过于复杂，要么在某些特定场景下不够灵活。
自己动手写一个端口扫描器有几个明显优势：
- 完全可控：你可以根据具体需求定制扫描策略
- 学习价值：深入理解网络通信和端口扫描原理
- 轻量化：针对特定场景可以做得比通用工具更高效
我最近用Python实现了一个轻量级端口扫描器，代码不到200行，但已经具备了基础扫描功能，并且支持一些实用特性。下面分享我的实现思路和关键代码。
2. 端口扫描的核心原理
2.1 TCP三次握手与端口状态
端口扫描的本质是尝试与目标主机的特定端口建立TCP连接。根据RFC 793定义的TCP协议，一个端口可能有以下几种状态：
- 开放(Open)：服务正在监听该端口，会完成三次握手
- 关闭(Closed)：主机响应但无服务监听(RST响应)
- 过滤(Filtered)：无响应(可能被防火墙拦截)
- 未过滤(Unfiltered)：能收到响应但不确定是否开放
最基础的扫描方式就是尝试与目标端口建立完整的TCP连接（全连接扫描），但这种扫描方式会在目标主机留下大量日志记录。
2.2 常见扫描技术对比
| 扫描类型 | 原理 | 优点 | 缺点 | 检测难度 |
|---|---|---|---|---|
| TCP全连接 | 完成三次握手 | 结果准确 | 速度慢，易被发现 | 高 |
| SYN扫描 | 只发SYN包 | 速度快 | 需要root权限 | 中 |
| FIN扫描 | 发送FIN包 | 隐蔽性高 | 结果不够准确 | 低 |
| UDP扫描 | 发送UDP包 | 可检测UDP服务 | 结果不可靠 | 中 |
Python的socket库默认只能实现全连接扫描，要实现更高级的扫描方式需要借助scapy等底层网络库。
3. Python实现基础扫描器
3.1 基础扫描功能实现
我们先实现一个最简单的全连接端口扫描器：
import socket from concurrent.futures import ThreadPoolExecutor def scan_port(host, port): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(1) result = s.connect_ex((host, port)) if result == 0: print(f"Port {port} is open") return port except Exception as e: pass return None def port_scanner(host, ports): open_ports = [] with ThreadPoolExecutor(max_workers=100) as executor: results = executor.map(lambda p: scan_port(host, p), ports) for port in results: if port: open_ports.append(port) return open_ports if __name__ == "__main__": target = "example.com" port_range = range(1, 1025) print(f"Scanning {target}...") open_ports = port_scanner(target, port_range) print(f"Open ports: {open_ports}")这个基础版本已经可以工作，但有几个明显问题：
- 扫描速度较慢（即使使用多线程）
- 容易被防火墙检测到
- 没有实现超时控制
- 结果输出过于简单
3.2 性能优化技巧
提高扫描速度的几个关键点：
- 动态调整超时时间：对响应快的端口使用更短的超时
def dynamic_timeout(port): # 常见服务端口使用更短超时 common_ports = {80:0.5, 443:0.5, 22:0.8, 21:0.8} return common_ports.get(port, 1.5)- 端口扫描顺序优化：先扫描常见服务端口
common_ports = [80, 443, 22, 21, 25, 3389, 3306, 53] other_ports = [p for p in range(1,1025) if p not in common_ports] ports_to_scan = common_ports + other_ports- 智能线程池管理：根据网络延迟动态调整线程数
from threading import Semaphore class DynamicThreadPool: def __init__(self, max_workers): self.semaphore = Semaphore(max_workers) def submit(self, func, *args): self.semaphore.acquire() try: result = func(*args) return result finally: self.semaphore.release()4. 进阶功能实现
4.1 服务指纹识别
检测到开放端口后，我们可以进一步识别运行的服务：
def service_detection(host, port): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(1) s.connect((host, port)) s.send(b"GET / HTTP/1.1\r\n\r\n") banner = s.recv(1024).decode().strip() if "HTTP" in banner: return "HTTP Service" elif "SSH" in banner: return "SSH Service" # 其他服务识别规则... else: return "Unknown Service" except: return "Unable to detect"4.2 结果可视化输出
将扫描结果生成可视化报告：
from prettytable import PrettyTable def generate_report(host, open_ports, services): table = PrettyTable() table.field_names = ["Port", "Status", "Service", "Banner"] for port in open_ports: service = services.get(port, {}).get("service", "Unknown") banner = services.get(port, {}).get("banner", "")[:50] table.add_row([port, "Open", service, banner]) print(f"\nScan Report for {host}") print(table) print(f"\nTotal open ports: {len(open_ports)}")5. 实际应用中的注意事项
5.1 合法性与道德考量
端口扫描是一把双刃剑，使用时必须注意：
- 只扫描你有权限测试的目标
- 避免对关键基础设施进行扫描
- 控制扫描频率，避免造成网络拥塞
- 商业使用前咨询法律顾问
提示：在测试环境中，可以使用localhost或自己搭建的靶机进行练习。
5.2 规避检测的技巧
如果需要在真实环境中使用，可以考虑以下方法降低被发现的风险：
- 随机化扫描顺序和时间间隔
- 使用代理或分布式扫描节点
- 限制扫描速率
- 混合使用多种扫描技术
import random import time def stealth_scan(host, ports): random.shuffle(ports) for port in ports: scan_port(host, port) time.sleep(random.uniform(0.1, 1.5))5.3 常见问题排查
连接超时问题：
- 检查目标主机是否在线
- 确认网络连接正常
- 适当增加超时时间
权限不足问题：
- 某些扫描类型需要root权限
- Linux/Mac下使用sudo运行
- Windows下以管理员身份运行
防火墙干扰：
- 尝试不同的扫描技术
- 调整扫描速率
- 使用分散的源IP地址
6. 扩展思路与进阶方向
这个基础扫描器还可以进一步扩展：
集成更多扫描技术：
- 使用scapy实现SYN扫描
- 添加UDP端口扫描功能
- 实现操作系统指纹识别
分布式扫描架构：
- 使用Redis作为任务队列
- 多台机器协同扫描
- 结果集中存储分析
自动化漏洞检测：
- 常见服务漏洞检测
- 弱口令爆破
- CVE漏洞匹配
Web界面：
- Flask/Django开发管理界面
- 可视化扫描结果
- 历史记录查询
# 使用scapy实现SYN扫描示例 from scapy.all import * def syn_scan(host, port): ans = sr1(IP(dst=host)/TCP(dport=port,flags="S"), timeout=1, verbose=0) if ans and ans.haslayer(TCP): if ans.getlayer(TCP).flags == 0x12: # SYN-ACK return True return False实现一个完整的端口扫描器需要考虑的细节很多，但核心原理并不复杂。我建议先从基础版本开始，然后逐步添加你需要的功能。在实际项目中，我会根据目标环境的特点调整扫描策略，这也是自己编写扫描器的优势所在。
