快连VPN用于数字游民与远程工作者的全球节点切换自动化方案

# 示例代码框架 - location_based_switch.py
import requests
import subprocess
import json
import time

# 假设的快连VPN控制命令（需替换为实际命令）
VPN_CONNECT_CMD = "kuailian_cli connect --country {}"
VPN_DISCONNECT_CMD = "kuailian_cli disconnect"
VPN_LIST_SERVERS_CMD = "kuailian_cli list-servers --country {}" # 假设能列出某国所有服务器

def get_current_location():
    """通过IP API获取当前大致地理位置"""
    try:
        resp = requests.get('http://ip-api.com/json/?fields=countryCode,query')
        data = resp.json()
        return data.get('countryCode'), data.get('query')
    except:
        return None, None

def ping_server(server_ip):
    """测试到特定服务器的延迟"""
    # 使用ping3库或系统ping命令，返回平均延迟
    # 示例：使用系统ping（Windows）
    import os
    response = os.popen(f"ping -n 2 {server_ip}").read()
    # 解析response获取平均延迟ms，这里简化处理
    if "平均" in response:
        # 解析逻辑...
        return 50 # 假设值
    return 999 # 超时

def connect_best_server(target_country_code="US"):
    """连接到目标国家延迟最低的服务器"""
    current_country, current_ip = get_current_location()
    print(f"当前位置: {current_country}, IP: {current_ip}")

    if current_country == target_country_code:
        print("已在目标国家，无需切换VPN。")
        # 可选：断开VPN以获得本地最快速度
        # subprocess.run(VPN_DISCONNECT_CMD, shell=True)
        return

    print(f"需要切换至 {target_country_code} 节点。")
    # 1. 获取目标国家服务器列表（假设命令返回JSON）
    # proc = subprocess.run(VPN_LIST_SERVERS_CMD.format(target_country_code), capture_output=True, text=True, shell=True)
    # server_list = json.loads(proc.stdout)
    # 2. 为简化，这里假设我们有一个预定义的服务器IP列表
    server_candidates = ["us-svr1.kuailian.com", "us-svr2.kuailian.com", "us-svr3.kuailian.com"]

    best_server = None
    best_latency = 999
    for server in server_candidates:
        latency = ping_server(server)
        print(f"  服务器 {server} 延迟: {latency}ms")
        if latency < best_latency:
            best_latency = latency
            best_server = server

    if best_server:
        print(f"  选择最佳服务器: {best_server}, 延迟 {best_latency}ms")
        # 3. 执行连接命令（这里需要将server映射为快连VPN内部的节点ID或名称）
        # 假设我们可以通过服务器IP或名称连接
        connect_cmd = f"kuailian_cli connect --server {best_server}"
        # subprocess.run(connect_cmd, shell=True)
        print(f"  执行命令: {connect_cmd}")
        # 4. 验证连接后IP
        time.sleep(3) # 等待连接稳定
        new_country, new_ip = get_current_location()
        if new_country == target_country_code:
            print(f"  切换成功！新IP: {new_ip}")
        else:
            print("  切换失败，请检查。")
    else:
        print("  未找到可用服务器。")

if __name__ == "__main__":
    # 目标国家代码，例如美国为US，日本为JP
    connect_best_server("US")

# 示例代码框架 - app_triggered_switch.py (监控进程)
import psutil
import time
import subprocess

TARGET_APP = "company-erp.exe" # 目标进程名
TARGET_VPN_NODE = "Node-UK-London" # 对应的快连VPN节点名

def is_app_running(app_name):
    """检查目标应用是否在运行"""
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] and app_name.lower() in proc.info['name'].lower():
            return True
    return False

def monitor_and_switch():
    vpn_connected_for_app = False
    last_state = False
    while True:
        current_state = is_app_running(TARGET_APP)
        if current_state and not last_state:
            # 应用刚刚启动
            print(f"检测到 {TARGET_APP} 启动，切换到节点 {TARGET_VPN_NODE}")
            # subprocess.run(f"kuailian_cli connect --node {TARGET_VPN_NODE}", shell=True)
            vpn_connected_for_app = True
        elif not current_state and last_state and vpn_connected_for_app:
            # 应用已关闭
            print(f"{TARGET_APP} 已关闭，恢复原网络设置。")
            # subprocess.run("kuailian_cli disconnect", shell=True)
            vpn_connected_for_app = False
        last_state = current_state
        time.sleep(10) # 每10秒检查一次

if __name__ == "__main__":
    monitor_and_switch()

# 示例代码框架 - wifi_security_auto.py (macOS 示例)
import subprocess
import time

TRUSTED_SSIDS = ["MyHomeWiFi", "SecureOfficeNetwork"]

def get_current_ssid():
    """获取当前连接的Wi-Fi SSID (macOS)"""
    try:
        cmd = "/System/Library/PrivateFrameworks/Apple80211.framework/Versions/Current/Resources/airport -I"
        output = subprocess.check_output(cmd, shell=True, text=True)
        for line in output.split('\n'):
            if ' SSID:' in line:
                return line.split(':')[1].strip()
    except:
        pass
    return None

def enforce_vpn_protection():
    last_ssid = None
    while True:
        current_ssid = get_current_ssid()
        if current_ssid != last_ssid:
            print(f"网络切换: 从 {last_ssid} 到 {current_ssid}")
            if current_ssid and current_ssid not in TRUSTED_SSIDS:
                print(f"  连接到非信任网络 {current_ssid}，强制开启VPN。")
                # 1. 确保Kill Switch功能已启用（可能需要预先在客户端设置）
                # 2. 连接到最优或预设的安全节点
                # subprocess.run("kuailian_cli connect --fastest", shell=True)
            elif current_ssid in TRUSTED_SSIDS:
                print(f"  连接到信任网络 {current_ssid}，VPN策略：保持断开或按需连接。")
                # 可以根据需要断开VPN subprocess.run("kuailian_cli disconnect", shell=True)
            last_ssid = current_ssid
        time.sleep(15)

if __name__ == "__main__":
    enforce_vpn_protection()

# 示例代码框架 - auto_failover.py
import subprocess
import time

PRIMARY_NODE = "Node-US-NYC"
SECONDARY_NODES = ["Node-US-LA", "Node-CA-Toronto", "Node-DE-Frankfurt"]

def check_vpn_connection():
    """检查VPN是否处于连接状态（示例：检查是否存在虚拟网卡或特定IP）"""
    # 方法1: 检查是否有快连VPN进程（不准确）
    # 方法2: 检查是否获得了VPN分配的IP（更可靠）
    try:
        # Linux/macOS示例：检查tun0接口是否存在特定IP段
        # output = subprocess.check_output("ip addr show tun0", shell=True, text=True)
        # 这里简化返回一个状态
        return True # 假设连接正常
    except:
        return False

def connect_with_failover():
    """主重连逻辑，带故障转移"""
    if check_vpn_connection():
        print("VPN连接正常。")
        return True

    print("VPN连接断开，开始重连...")
    nodes_to_try = [PRIMARY_NODE] + SECONDARY_NODES
    for node in nodes_to_try:
        print(f"  尝试连接节点: {node}")
        # subprocess.run(f"kuailian_cli connect --node {node}", shell=True)
        time.sleep(5) # 等待连接建立
        if check_vpn_connection():
            print(f"  成功连接到 {node}")
            return True
        else:
            print(f"  连接 {node} 失败。")
    print("  所有节点尝试均失败。")
    return False

if __name__ == "__main__":
    while True:
        connect_with_failover()
        time.sleep(60) # 每60秒检查一次