并发扫描TCP、UDP端口

## 扫描主机tcp端口

```
# 扫描主机tcp端口

import socket
import concurrent.futures

# 目标主机
# host = '192.168.1.1'
host = '172.16.1.100'

# 端口范围
port_start = 1
port_end = 65535

# 用于保存开放端口的列表
open_ports = []

def scan_port(port):
    try:
        # 创建一个socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 设置超时时间，防止等待太久
        sock.settimeout(1)
        # 尝试连接目标端口
        result = sock.connect_ex((host, port))
        sock.close()
        # 如果连接成功，说明端口是开放的
        if result == 0:
            open_ports.append(port)
            print(f'Port {port} is open\n')
    except socket.error as e:
        pass  # 忽略错误，继续扫描下一个端口

def main():
    # 使用ThreadPoolExecutor实现并发
    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
        # 将端口扫描任务分配给线程池
        futures = [executor.submit(scan_port, port) for port in range(port_start, port_end + 1)]
        # 等待所有任务完成
        concurrent.futures.wait(futures)
    
    # 将开放的端口写入文件
    with open('D:/acl/port_scan_results.txt', 'w') as file:
        for port in sorted(open_ports):
            
            file.write(f'Port {port} is open\n')

if __name__ == '__main__':
    main()
```

## 扫描主机udp端口
```
# 扫描主机udp端口

import socket
import concurrent.futures

# 请注意，这个脚本可能不会检测到所有开放的UDP端口，因为有些服务可能不会响应空的数据包。
# 此外，由于UDP是无连接的，即使没有服务监听某个端口，发送到该端口的数据包也可能因为网络设备（如路由器）的响应而被错误地认为是开放的。

# 目标主机
host = '192.168.1.1'

# 端口范围
port_start = 1
port_end = 65535

# 用于保存开放端口的列表
open_ports = []

def scan_udp_port(port):
    try:
        # 创建一个UDP socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # 设置超时时间
        sock.settimeout(1)
        # 向目标端口发送一个空的数据包
        sock.sendto(b'', (host, port))
        # 尝试接收响应
        sock.recvfrom(1024)
        # 如果收到响应，说明端口是开放的
        print(f'UDP Port {port} is open\n')
        open_ports.append(port)
    except socket.error:
        pass  # 忽略错误，继续扫描下一个端口
    finally:
        # 关闭socket
        sock.close()

def main():
    # 使用ThreadPoolExecutor实现并发
    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
        # 将端口扫描任务分配给线程池
        futures = [executor.submit(scan_udp_port, port) for port in range(port_start, port_end + 1)]
        # 等待所有任务完成
        concurrent.futures.wait(futures)
    
    # 将开放的端口写入文件
    with open('udp_port_scan_results.txt', 'w') as file:
        for port in sorted(open_ports):
            file.write(f'UDP Port {port} is open\n')

if __name__ == '__main__':
    main()

```

## 优化

```

# 让结果实时保存到udp_port_scan_results.txt文件中，频繁地打开和关闭文件可能会影响性能。
# 在这个脚本中，我们对每个端口都进行了扫描，无论端口是开放的还是关闭的，都会将结果写入文件。如果端口开放，我们会写入Port is open，如果端口关闭，我们会写入Port is closed。

# 请注意，这个脚本会覆盖udp_port_scan_results.txt文件中的现有内容，因为它是以'w'模式打开文件的。如果你想要追加到现有文件，可以改用'a'模式。

# 此外，由于UDP是无连接的，即使没有服务监听某个端口，发送到该端口的数据包也可能因为网络设备（如路由器）的响应而被错误地认为是开放的。因此，这个脚本的结果可能不是100%准确。

import socket
import concurrent.futures

# 目标主机
host = '192.168.1.1'

# 端口范围
port_start = 1
port_end = 65535

def scan_udp_port(port, file):
    try:
        # 创建一个UDP socket对象
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # 设置超时时间
        sock.settimeout(1)
        # 向目标端口发送一个空的数据包
        sock.sendto(b'', (host, port))
        # 尝试接收响应
        data, _ = sock.recvfrom(1024)
        # 如果收到响应，说明端口是开放的
        if data:
            status = 'is open'
        else:
            status = 'is closed'
        file.write(f'UDP Port {port} {status}\n')
        file.flush()  # 确保数据被写入文件
    except socket.error as e:
        status = 'is closed'
        file.write(f'UDP Port {port} {status}\n')
        file.flush()  # 确保数据被写入文件
    finally:
        # 关闭socket
        sock.close()

def main():
    # 打开文件用于追加
    with open('udp_port_scan_results.txt', 'w') as file:
        # 使用ThreadPoolExecutor实现并发
        with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
            # 将端口扫描任务分配给线程池
            futures = [executor.submit(scan_udp_port, port, file) for port in range(port_start, port_end + 1)]
            # 等待所有任务完成
            concurrent.futures.wait(futures)

if __name__ == '__main__':
    main()
```