pip install scapy

from scapy.all import *
import ipaddress
 
def port_scan(ip, ports):
    """
    對(duì)指定IP地址的指定端口進(jìn)行掃描
    :param ip: 目標(biāo)IP地址
    :param ports: 需要掃描的端口號(hào)列表
    :return: 打印開(kāi)放的端口
    """
    open_ports = []
    for port in ports:
        # 構(gòu)造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
                # 發(fā)送RST包復(fù)位連接
                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  # 端口關(guān)閉
    return open_ports
 
if __name__ == "__main__":
    target_ip = "192.168.1.1"  # 替換為目標(biāo)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:
        # 創(chuàng)建一個(gè) socket 對(duì)象
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(1)  # 設(shè)置超時(shí)時(shí)間
            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 來(lái)并行掃描多個(gè)端口
    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)  # 設(shè)置超時(shí)時(shí)間
        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"  # 目標(biāo)IP地址
    start_port = 1  # 開(kāi)始端口
    end_port = 1024  # 結(jié)束端口
    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"  # 目標(biāo)IP地址
    start_port = 1  # 開(kāi)始端口
    end_port = 1024  # 結(jié)束端口
    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")
        # 發(fā)送 RST 包關(guān)閉連接
        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"  # 目標(biāo)IP地址
    start_port = 1  # 開(kāi)始端口
    end_port = 1024  # 結(jié)束端口
    main(target_ip, start_port, end_port)