炮台有效炮弹实现
声明
学习视频来自 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 = True
def 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_write
proxy = proxy.strip() # 移除行尾的换行符
if not proxy: # 跳过空行
return None
# 设置代理
proxies_dict = {
'http': f'http://{proxy}',
'https': f'https://{proxy}'
}
is_available = True
for url in test_urls:
try:
# 发送请求，设置超时时间为5秒
response = requests.get(url, proxies=proxies_dict, timeout=5)
if response.status_code!= 200:
is_available = False
break
except requests.RequestException as e:
is_available = False
break
if 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 # 第一次写入后标记设为False
return proxy
else:
print(f'代理IP {proxy} 不可用')
return None
def main():
# 第一步：处理IP文件
replace_comma_with_colon(raw_input_file, processed_output_file)
# 第二步：读取替换后的IP文件，验证代理IP
with 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:
pass
return 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)
- 筛选出来的就全是有效的代理了，然后直接字典开轰
