编写的ipscan程序，可以扫描某个IP地址段的端口，程序如下：
import asyncio
import ipaddress
import time
import sys
import gc
import os

==========================================

⚙️ 常用修改参数配置区

==========================================

BASE_DIR = r"C:\Users\Administrator\Desktop\pyscan" IP_FILE = os.path.join(BASE_DIR, "scanip.txt") FOUND_FILE = os.path.join(BASE_DIR, "rosip.txt")

TARGET_PORT = 1723
CONCURRENCY = 400
TIMEOUT = 2
BATCH_SIZE = 10000
PROGRESS_INTERVAL = 1000 # 改为1000，减少进度更新频率
SAVE_INTERVAL = 60

==========================================

semaphore = asyncio.Semaphore(CONCURRENCY)
found_ips_buffer = []
lock = asyncio.Lock()
total_found = 0

def read_ip_ranges(file_path):

CODE

ip_ranges = []
if not os.path.exists(file_path):
    print(f"[!] 警告：找不到文件 {file_path}，请先创建该文件并填入要扫描的IP段！")
    return ip_ranges
with open(file_path, 'r', encoding='utf-8') as f:
    for line in f:
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        ip_ranges.append(line)
print(f"[*] 从 {file_path} 成功读取到 {len(ip_ranges)} 个IP段")
return ip_ranges

async def periodic_writer():

CODE

global found_ips_buffer
while True:
    await asyncio.sleep(SAVE_INTERVAL)
    async with lock:
        if found_ips_buffer:
            with open(FOUND_FILE, "a", encoding="utf-8") as f:
                f.write("\n".join(found_ips_buffer) + "\n")
            count = len(found_ips_buffer)
            found_ips_buffer = []
            # 为了不打断进度条，这里不打印任何提示，数据静默写入
            # print(f"\n[系统提示] 已自动保存 {count} 个新发现的IP到 {FOUND_FILE}\n")

async def scan_port(ip, port):

CODE

global total_found
async with semaphore:
    try:
        reader, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=TIMEOUT)
        peer = writer.get_extra_info('peername')
        async with lock:
            found_ips_buffer.append(ip)
            total_found += 1
        # 【重点修改】去掉了这行输出日志，进度条不会被打断了
        # print(f"\n[+] 发现目标! {ip}:{port} -> {peer}")
        
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass
    except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
        pass
    except Exception:
        pass

def show_progress(current, total, ip_range):

CODE

# 只在交互式终端显示进度条
if not sys.stdout.isatty():
    return
if total == 0:
    pct = 100.0
else:
    pct = (current / total) * 100
msg = f"\r[当前网段: {ip_range}] 进度: {pct:.2f}% | 已扫描: {current} | 已发现: {total_found}"
sys.stdout.write(f"{msg:<120}")
sys.stdout.flush()

async def scan_ip_range(ip_range_str):

CODE

if "-" in ip_range_str:
    start_ip, end_ip = ip_range_str.split("-")
    network = list(ipaddress.summarize_address_range(ipaddress.IPv4Address(start_ip.strip()), ipaddress.IPv4Address(end_ip.strip())))
    total_ips = sum(net.num_addresses for net in network)
    def ip_generator():
        for net in network:
            for ip in net:
                yield str(ip)
else:
    net = ipaddress.ip_network(ip_range_str, strict=False)
    total_ips = net.num_addresses
    def ip_generator():
        for ip in net:
            yield str(ip)

CODE

batch_tasks = []
current_count = 0
last_print_pct = -1  # 记录上次打印的百分比

CODE

print(f"\n[*] 开始扫描网段: {ip_range_str} (共 {total_ips} 个IP)")
show_progress(0, total_ips, ip_range_str)

CODE

for ip in ip_generator():
    batch_tasks.append(asyncio.create_task(scan_port(ip, TARGET_PORT)))
    current_count += 1

CODE

    # 每 PROGRESS_INTERVAL 个IP检查一次是否需要更新进度
    if current_count % PROGRESS_INTERVAL == 0 or current_count == total_ips:
        current_pct = int((current_count / total_ips) * 100) if total_ips > 0 else 0
        # 只有百分比变化时才更新进度
        if current_pct != last_print_pct:
            show_progress(current_count, total_ips, ip_range_str)
            last_print_pct = current_pct
        # 让出控制权，但不会阻塞
        await asyncio.sleep(0)

CODE

    # 批量处理
    if current_count % BATCH_SIZE == 0:
        await asyncio.gather(*batch_tasks, return_exceptions=True)
        batch_tasks.clear()
        gc.collect()

CODE

if batch_tasks:
    await asyncio.gather(*batch_tasks, return_exceptions=True)
    batch_tasks.clear()
    gc.collect()

CODE

show_progress(total_ips, total_ips, ip_range_str)
print()

async def main():

CODE

global found_ips_buffer, total_found
ip_ranges = read_ip_ranges(IP_FILE)
if not ip_ranges:
    return
print(f"[*] 开始扫描端口 {TARGET_PORT}，发现的目标将每隔 {SAVE_INTERVAL} 秒保存到 {FOUND_FILE}")

CODE

writer_task = asyncio.create_task(periodic_writer())
start_time = time.time()

CODE

try:
    for ip_range in ip_ranges:
        await scan_ip_range(ip_range)
except KeyboardInterrupt:
    print("\n[!] 收到中断，正在优雅退出...")
finally:
    async with lock:
        if found_ips_buffer:
            with open(FOUND_FILE, "a", encoding="utf-8") as f:
                f.write("\n".join(found_ips_buffer) + "\n")
            found_ips_buffer = []
    writer_task.cancel()
    
end_time = time.time()
print(f"\n[*] 扫描结束，总耗时: {end_time - start_time:.2f} 秒")
print(f"[*] 共发现 {total_found} 个开放端口，结果已保存在 {FOUND_FILE}")

if __name__ == "__main__":

CODE

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

IPSCAN程序

==========================================

⚙️ 常用修改参数配置区

==========================================

==========================================

💬 评论 (0)