|
|
import os
|
|
|
import re
|
|
|
import subprocess
|
|
|
|
|
|
# Global variables for proxy switch count
|
|
|
proxy_switch_count = 0
|
|
|
iface_ipv6_dict = {}
|
|
|
|
|
|
def is_root():
|
|
|
return os.geteuid() == 0
|
|
|
|
|
|
def interface_usable(interface_name, skip_check=False, ipv6_address='2400:3200::1', max_retries=3):
|
|
|
if skip_check:
|
|
|
return True
|
|
|
current_try = 0
|
|
|
while current_try < max_retries:
|
|
|
try:
|
|
|
cmd_result = subprocess.run(["ping", "-c", "1", "-I", interface_name, ipv6_address], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5)
|
|
|
if cmd_result.returncode == 0:
|
|
|
return True # 成功ping通，直接返回True
|
|
|
except subprocess.TimeoutExpired:
|
|
|
print(f"Ping attempt {current_try + 1} of {max_retries} timed out. Retrying...")
|
|
|
except subprocess.SubprocessError as e:
|
|
|
# 捕获其他subprocess相关的异常
|
|
|
print(f"An error occurred while trying to ping: {e}. Retrying...")
|
|
|
current_try += 1
|
|
|
return False # 所有尝试后仍未成功，返回False
|
|
|
|
|
|
def get_existing_interfaces(base_interface='eth0'):
|
|
|
cmd_result = subprocess.run(["ip", "addr", "show"], stdout=subprocess.PIPE)
|
|
|
output = cmd_result.stdout.decode()
|
|
|
|
|
|
# 匹配接口名称
|
|
|
iface_pattern = re.compile(re.escape(base_interface) + r'_([0-9]+)@')
|
|
|
iface_matches = iface_pattern.findall(output)
|
|
|
|
|
|
# 构建完整的接口名称列表
|
|
|
interfaces = [f"{base_interface}_{match}" for match in iface_matches]
|
|
|
|
|
|
# 初始化字典来存储接口名称与其IPv6地址的映射
|
|
|
iface_ipv6_dict = {}
|
|
|
|
|
|
for iface in interfaces:
|
|
|
# 对于每个接口，查找其IPv6地址，这里假设只提取第一个IPv6地址
|
|
|
# 注意：需要确保只匹配特定接口的IPv6地址，因此使用iface作为正则表达式的一部分
|
|
|
cmd_result = subprocess.run(["ip", "addr", "show", iface], stdout=subprocess.PIPE)
|
|
|
output = cmd_result.stdout.decode()
|
|
|
ipv6_pattern = re.compile(r"inet6\s+([0-9a-f:]+)\/\d+")
|
|
|
ipv6_matches = ipv6_pattern.findall(output)
|
|
|
|
|
|
# 过滤掉以"fe80"开头的IPv6地址
|
|
|
ipv6_addresses = [addr for addr in ipv6_matches if not addr.startswith("fe80")]
|
|
|
|
|
|
# 如果存在非链路本地的IPv6地址，只取第一个地址
|
|
|
if ipv6_addresses:
|
|
|
iface_ipv6_dict[iface] = ipv6_addresses[0]
|
|
|
|
|
|
return iface_ipv6_dict
|
|
|
|
|
|
def execute_ip6tables_command(command):
|
|
|
sudo_cmd = ["sudo"] if not is_root() else []
|
|
|
cmd = sudo_cmd + command.split()
|
|
|
subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
|
|
def switch_proxy_server(mode='normal'):
|
|
|
global proxy_switch_count
|
|
|
global iface_ipv6_dict
|
|
|
|
|
|
if mode == 'normal':
|
|
|
if iface_ipv6_dict:
|
|
|
proxy_switch_count += 1
|
|
|
proxy_index = proxy_switch_count % len(iface_ipv6_dict)
|
|
|
selected_interface = list(iface_ipv6_dict.keys())[proxy_index]
|
|
|
ipv6_address = iface_ipv6_dict[selected_interface]
|
|
|
# 清空自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -F FAKE_IPV6_CHAIN')
|
|
|
# 添加SNAT规则
|
|
|
execute_ip6tables_command(f'ip6tables -t nat -A FAKE_IPV6_CHAIN -j SNAT --to-source {ipv6_address}')
|
|
|
|
|
|
print(f"Using interface: {selected_interface}, Connecting to: {ipv6_address}")
|
|
|
|
|
|
def create_ipv6_addresses(n, base_interface='eth0', delete_interface=True):
|
|
|
sudo_cmd = ["sudo"] if not is_root() else []
|
|
|
if delete_interface:
|
|
|
delete_ipv6_addresses(base_interface)
|
|
|
existing_interfaces = list(get_existing_interfaces(base_interface).keys())
|
|
|
interfaces = []
|
|
|
for i in range(1, n + 1):
|
|
|
interface_name = f"{base_interface}_{i}"
|
|
|
|
|
|
# Check if the interface exists, if yes, delete it first
|
|
|
if interface_name in existing_interfaces:
|
|
|
if interface_usable(interface_name):
|
|
|
print(f"Interface {interface_name} already exists. Skipping creation.")
|
|
|
interfaces.append(interface_name)
|
|
|
continue
|
|
|
else:
|
|
|
subprocess.run(sudo_cmd + ["ip", "link", "delete", interface_name])
|
|
|
|
|
|
# Now add the interface
|
|
|
subprocess.run(sudo_cmd + ["ip", "link", "add", "link", base_interface, interface_name, "type", "macvlan", "mode", "bridge"])
|
|
|
subprocess.run(sudo_cmd + ["ip", "link", "set", interface_name, "up"])
|
|
|
#subprocess.run(sudo_cmd + ["dhclient", "-6", "-nw", interface_name])
|
|
|
interfaces.append(interface_name)
|
|
|
return interfaces
|
|
|
|
|
|
def delete_ipv6_addresses(base_interface='eth0'):
|
|
|
sudo_cmd = ["sudo"] if not is_root() else []
|
|
|
existing_interfaces = list(get_existing_interfaces(base_interface).keys())
|
|
|
|
|
|
for interface_name in existing_interfaces:
|
|
|
subprocess.run(sudo_cmd + ["ip", "link", "delete", interface_name])
|
|
|
|
|
|
def stop_proxy_servers(base_interface='eth0', delete_interface=True):
|
|
|
# 删除流量重定向到自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -D POSTROUTING -j FAKE_IPV6_CHAIN')
|
|
|
# 删除自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -X FAKE_IPV6_CHAIN')
|
|
|
|
|
|
if delete_interface:
|
|
|
print("正在关闭代理服务器...")
|
|
|
print("删除IPv6地址...")
|
|
|
delete_ipv6_addresses(base_interface)
|
|
|
print("代理服务器已关闭.")
|
|
|
else:
|
|
|
print("正在关闭代理服务器...")
|
|
|
print("代理服务器已关闭.")
|
|
|
|
|
|
def start_proxy_servers(n, mode='normal', base_interface='eth0', delete_interface=True):
|
|
|
global iface_ipv6_dict
|
|
|
|
|
|
interfaces = create_ipv6_addresses(n, base_interface, delete_interface)
|
|
|
#获取生成的接口及IP
|
|
|
iface_ipv6_dict = get_existing_interfaces(base_interface)
|
|
|
|
|
|
if iface_ipv6_dict:
|
|
|
# 删除流量重定向到自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -D POSTROUTING -j FAKE_IPV6_CHAIN')
|
|
|
# 删除自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -X FAKE_IPV6_CHAIN')
|
|
|
|
|
|
# 创建自定义链
|
|
|
execute_ip6tables_command('ip6tables -t nat -N FAKE_IPV6_CHAIN')
|
|
|
# 流量重定向到自定义链
|
|
|
execute_ip6tables_command(f'ip6tables -t nat -A POSTROUTING -o {base_interface} -j FAKE_IPV6_CHAIN')
|
|
|
|
|
|
if mode == 'normal':
|
|
|
selected_interface = list(iface_ipv6_dict.keys())[0]
|
|
|
ipv6_address = iface_ipv6_dict[selected_interface]
|
|
|
# 添加SNAT规则
|
|
|
execute_ip6tables_command(f'ip6tables -t nat -A FAKE_IPV6_CHAIN -j SNAT --to-source {ipv6_address}')
|
|
|
|
|
|
print(f"Using interface: {selected_interface}, Connecting to: {ipv6_address}")
|
|
|
elif mode == 'random':
|
|
|
for index, (interface, ipv6_address) in enumerate(iface_ipv6_dict.items()):
|
|
|
adjusted_probability = 1/(len(iface_ipv6_dict)-index)
|
|
|
execute_ip6tables_command(f'ip6tables -t nat -A FAKE_IPV6_CHAIN -m statistic --mode random --probability {adjusted_probability} -j SNAT --to-source {ipv6_address}') |
