使用Python实现高效的端口扫描器

pip install scapy

from scapy.all import *
import ipaddress
 
def port_scan(ip, ports):
    """
    对指定IP地址的指定端口进行扫描
    :param ip: 目标IP地址
    :param ports: 需要扫描的端口号列表
    :return: 打印开放的端口
    """
    open_ports = []
    for port in ports:
        # 构造SYN包
        packet = IP(dst=ip)/TCP(dport=port, flags="S")
        response = sr1(packet, timeout=1, verbose=0)
        
        if response is not None and TCP in response:
            if response[TCP].flags == 0x12:  # 如果收到SYN-ACK
                # 发送RST包复位连接
                send_rst = sr(IP(dst=ip)/TCP(dport=port, flags="R"), timeout=1, verbose=0)
                open_ports.append(port)
            elif response[TCP].flags == 0x14:  # 如果收到RST
                pass  # 端口关闭
    return open_ports
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 替换为目标IP
    target_ports = [22, 80, 443, 8080]  # 指定需要扫描的端口
    open_ports = port_scan(target_ip, target_ports)
    
    if open_ports:
        print(f"Open ports on {target_ip}: {open_ports}")
    else:
        print(f"No open ports found on {target_ip}")

import socket
from concurrent.futures import ThreadPoolExecutor
 
def scan_port(ip, port):
    try:
        # 创建一个 socket 对象
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(1)  # 设置超时时间
            result = sock.connect_ex((ip, port))
            if result == 0:
                print(f"Port {port} is open")
            else:
                print(f"Port {port} is closed")
    except socket.error as e:
        print(f"Socket error: {e}")
 
def main():
    target_ip = input("Enter the target IP address: ")
    start_port = int(input("Enter the start port: "))
    end_port = int(input("Enter the end port: "))
 
    # 使用 ThreadPoolExecutor 来并行扫描多个端口
    with ThreadPoolExecutor(max_workers=100) as executor:
        for port in range(start_port, end_port + 1):
            executor.submit(scan_port, target_ip, port)
 
if __name__ == "__main__":
    main()

import socket
import threading
 
def scan_port(ip, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)  # 设置超时时间
        result = sock.connect_ex((ip, port))
        if result == 0:
            print(f"Port {port} is open")
        sock.close()
    except Exception as e:
        print(f"Error scanning port {port}: {e}")
 
def main(ip, start_port, end_port):
    threads = []
    for port in range(start_port, end_port + 1):
        thread = threading.Thread(target=scan_port, args=(ip, port))
        threads.append(thread)
        thread.start()
    
    for thread in threads:
        thread.join()
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目标IP地址
    start_port = 1  # 开始端口
    end_port = 1024  # 结束端口
    main(target_ip, start_port, end_port)

import asyncio
import socket
 
async def scan_port(ip, port):
    try:
        conn = asyncio.open_connection(ip, port)
        reader, writer = await asyncio.wait_for(conn, timeout=1)
        print(f"Port {port} is open")
        writer.close()
        await writer.wait_closed()
    except (asyncio.TimeoutError, ConnectionRefusedError):
        pass
 
async def main(ip, start_port, end_port):
    tasks = []
    for port in range(start_port, end_port + 1):
        task = asyncio.create_task(scan_port(ip, port))
        tasks.append(task)
    
    await asyncio.gather(*tasks)
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目标IP地址
    start_port = 1  # 开始端口
    end_port = 1024  # 结束端口
    asyncio.run(main(target_ip, start_port, end_port))

from scapy.all import sr1, IP, TCP
 
def scan_port(ip, port):
    src_port = 1025  # 源端口
    syn_packet = IP(dst=ip) / TCP(sport=src_port, dport=port, flags='S')
    response = sr1(syn_packet, timeout=1, verbose=0)
    
    if response and response.haslayer(TCP) and response.getlayer(TCP).flags & 0x12:  # SYN-ACK
        print(f"Port {port} is open")
        # 发送 RST 包关闭连接
        rst_packet = IP(dst=ip) / TCP(sport=src_port, dport=port, flags='R')
        send(rst_packet, verbose=0)
 
def main(ip, start_port, end_port):
    for port in range(start_port, end_port + 1):
        scan_port(ip, port)
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 目标IP地址
    start_port = 1  # 开始端口
    end_port = 1024  # 结束端口
    main(target_ip, start_port, end_port)