Python安全工具开发实战：编写端口扫描器

import socket
import threading
from queue import Queue
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"[+] {ip}:{port} OPEN")
        sock.close()
    except:
        pass
def main():
    target = "192.168.1.1"
    ports = range(1, 1025)
    
    threads = []
    for port in ports:
        t = threading.Thread(target=scan_port, args=(target, port))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
if __name__ == "__main__":
    main()

import asyncio
import socket
async def scan_port_async(ip, port):
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(ip, port),
            timeout=1
        )
        writer.close()
        await writer.wait_closed()
        return port
    except:
        return None
async def main():
    target = "192.168.1.1"
    ports = range(1, 65536)
    
    tasks = [scan_port_async(target, port) for port in ports]
    results = await asyncio.gather(*tasks)
    
    open_ports = [p for p in results if p is not None]
    print(f"Open ports: {open_ports}")
asyncio.run(main())

def get_banner(ip, port):
    try:
        sock = socket.socket()
        sock.settimeout(2)
        sock.connect((ip, port))
        sock.send(b"HEAD / HTTP/1.1\r\nHost: " + ip.encode() + b"\r\n\r\n")
        banner = sock.recv(1024).decode(errors='ignore')
        sock.close()
        return banner.strip()
    except:
        return ""

port_scanner/
├── scanner.py      # 核心扫描逻辑
├── banner.py       # 服务识别
├── output.py       # 结果输出
├── cli.py          # 命令行接口
└── requirements.txt

python cli.py -t 192.168.1.1 -p 1-1000 -T 100 -o result.json