代理池搭建优化-（书接上回，优化改进）

炮台有效炮弹实现

声明
学习视频来自 B 站UP主泷羽sec，如涉及侵权马上删除文章。

笔记的只是方便各位师傅学习知识，以下网站只涉及学习内容，其他的都与本人无关，切莫逾越法律红线，否则后果自负。

✍🏻作者简介：致力于网络安全领域，目前作为一名学习者，很荣幸成为一名分享者，最终目标是成为一名开拓者，很有趣也十分有意义
🤵‍♂️ 个人主页： @One_Blanks
欢迎评论 💬点赞👍🏻 收藏 📂加关注+

关注总部：泷羽Sec

先对Zmap下来的代理IP筛一遍

使用多线程优化了验证效率，使用多个验证URL提高可用性

最新python脚本
经过多线程、线程锁的进一步优化，大大提高了验证效率。

因为我用Zmap使用了80、443端口，所以我使用了替换逗号的一个函数，如果是IP:PORT

的形式直接将 replace_comma_with_colon 函数注释。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading# 指定输入和输出文件路径
raw_input_file = 'raw_ips.txt'  # 原始包含逗号的IP文件
processed_output_file = 'ip.txt'  # 替换逗号后的IP文件
available_proxy_file_path = 'InProxyPool.txt'  # 可用代理输出文件# 多个验证URL，包含国内和国外常见网站，用于更严格地验证代理IP是否可用
test_urls = ['http://www.baidu.com','http://www.jd.com/','http://www.taobao.com/'
]# 线程锁
lock = threading.Lock()
# 标记是否是第一次写入可用代理文件，初始化为True表示第一次
first_write = Truedef replace_comma_with_colon(input_file, output_file):"""替换文件中逗号为冒号，并保存到新文件。"""with open(input_file, 'r') as file:lines = file.readlines()# 替换逗号为冒号modified_lines = [line.replace(',', ':').strip() for line in lines]# 写入新的文件，'w'模式会自动清空原有内容再写入with open(output_file, 'w') as file:for line in modified_lines:file.write(line + '\n')print(f"IP格式处理完成，结果已保存到 {output_file}")def test_proxy(proxy):"""测试单个代理IP是否可用，通过多个URL验证，只有全部验证通过才认定可用，并实时写入文件。"""global first_writeproxy = proxy.strip()  # 移除行尾的换行符if not proxy:  # 跳过空行return None# 设置代理proxies_dict = {'http': f'http://{proxy}','https': f'https://{proxy}'}is_available = Truefor url in test_urls:try:# 发送请求，设置超时时间为5秒response = requests.get(url, proxies=proxies_dict, timeout=5)if response.status_code!= 200:is_available = Falsebreakexcept requests.RequestException as e:is_available = Falsebreakif is_available:print(f'代理IP {proxy} 可用')with lock:# 根据是否第一次写入来决定文件打开模式mode = 'w' if first_write else 'a'with open(available_proxy_file_path, mode) as file:file.write(f'{proxy}\n')if first_write:first_write = False  # 第一次写入后标记设为Falsereturn proxyelse:print(f'代理IP {proxy} 不可用')return Nonedef main():# 第一步：处理IP文件replace_comma_with_colon(raw_input_file, processed_output_file)# 第二步：读取替换后的IP文件，验证代理IPwith open(processed_output_file, 'r') as file:proxies = file.readlines()# 使用ThreadPoolExecutor并发执行with ThreadPoolExecutor(max_workers=20) as executor:future_to_proxy = {executor.submit(test_proxy, proxy): proxy for proxy in proxies}for future in as_completed(future_to_proxy):proxy = future_to_proxy[future]try:future.result()  # 触发验证逻辑except Exception as e:print(f'代理IP {proxy} 验证过程中出现错误: {e}')print(f"\n代理验证完成，可用代理IP已写入文件: {available_proxy_file_path}")if __name__ == "__main__":main()

验证之后形成第一代理池，但是用到目标网站还是会有许多其他的状态码，不能访问的错误，所以我们针对目标指定网站再进行筛选

子弹有效化

使用这个脚本指定目标网站为验证网站，还可以指定其Response数据包的长度，来决定确实是正常访问。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading# 最小响应包长度
MIN_LENGTH = 100
# 最大线程数
MAX_WORKERS = 100
# 输出文件路径
OUTPUT_FILE_PATH = 'FProxies.txt'
# 输入文件路径
input_file_path = 'InProxyPool.txt'def check_proxy(proxy):"""验证代理网址"""test_url = "http://www.google.com"proxies = {"http": f"http://{proxy}", "https": f"http://{proxy}"}try:response = requests.get(test_url, proxies=proxies, timeout=2)if response.status_code == 200 and len(response.content) >= MIN_LENGTH:return proxy  # 返回可用代理except Exception:passreturn None  # 无效代理def validate_proxy(proxy, lock):"""验证单个代理并写入文件"""valid_proxy = check_proxy(proxy)if valid_proxy:with lock:with open(OUTPUT_FILE_PATH, 'a') as outfile:outfile.write(f"{valid_proxy}\n")print(f"有效代理: {valid_proxy}")else:print(f"无效代理: {proxy}")def validate_proxies_from_file(input_file_path):"""从文件中读取代理并验证其有效性，同时将有效代理输出到另一个文件"""lock = threading.Lock()with open(input_file_path, 'r') as infile:proxies = [line.strip() for line in infile]with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:futures = [executor.submit(validate_proxy, proxy, lock) for proxy in proxies]for future in as_completed(futures):future.result()  # 等待所有任务完成validate_proxies_from_file(input_file_path)