LinuxServer 配置 V2ray 并测试

## 安装
- 打开 V2ray 的官方GitHub地址：https://raw.githubusercontent.com/v2fly/fhs-install-v2ray/master/install-release.sh
- 浏览器访问，并在 Linux 下创建文件 `install-release.sh` 并复制进去。
- 安装一些重要的工具，不然会安装失败：
```bash
sudo apt update
sudo apt install wget
sudo apt install curl
sudo apt install unzip
```

- 安装 V2ray，一定要 sudo，或者在 root 用户下。-p 参数是使用代理，我局域网里面有一台代理服务器。不用也行，会比较慢。
```bash
sudo bash install-release.sh -p http://192.168.1.xxx:xxxx
```
- 安装后，会显示一些如下信息：
```
installed: /usr/local/bin/v2ray
installed: /usr/local/share/v2ray/geoip.dat
installed: /usr/local/share/v2ray/geosite.dat
installed: /usr/local/etc/v2ray/config.json
installed: /var/log/v2ray/
installed: /var/log/v2ray/access.log
installed: /var/log/v2ray/error.log
installed: /etc/systemd/system/v2ray.service
installed: /etc/systemd/system/v2ray@.service
removed: /tmp/tmp.AgtzdoDa1i
```

- 注意一下 `config.json` 的位置，这是个配置文件。

- 设置 v2ray 开机自启动和启动 v2ray 并查看状态：
```bash
sudo systemctl enable v2ray
sudo systemctl start v2ray
sudo systemctl status v2ray
```

- 配置代理，这里写了个脚本：
```python
#!/usr/bin/env python3
"""
V2Ray订阅自动更新脚本
功能：获取订阅地址、解码生成配置文件、重启v2ray服务
"""

import base64
import json
import requests
import os
import sys
import subprocess
import time
from urllib.parse import urlparse
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/tmp/v2ray_update.log'),
        logging.StreamHandler(sys.stdout)
    ]
)

class V2RayConfigUpdater:
    def __init__(self, subscribe_url=None, config_path="/usr/local/etc/v2ray/config.json"):
        self.subscribe_url = subscribe_url
        self.config_path = config_path
        self.temp_config_path = "/tmp/v2ray_config.json"
        
    def get_subscribe_content(self):
        """获取订阅内容"""
        if not self.subscribe_url:
            logging.error("订阅地址未设置")
            return None
            
        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }
            response = requests.get(self.subscribe_url, headers=headers, timeout=30)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            logging.error(f"获取订阅内容失败: {e}")
            return None
    
    def decode_base64_content(self, content):
        """解码base64内容"""
        try:
            # 添加padding（如果需要）
            padding = len(content) % 4
            if padding:
                content += '=' * (4 - padding)
            
            decoded = base64.b64decode(content).decode('utf-8')
            return decoded
        except Exception as e:
            logging.error(f"解码base64内容失败: {e}")
            return None
    
    def parse_vmess_links(self, decoded_content):
        """解析vmess链接"""
        vmess_configs = []
        lines = decoded_content.strip().split('\n')
        
        for line in lines:
            line = line.strip()
            if line.startswith('vmess://'):
                try:
                    # 移除vmess://前缀并解码
                    vmess_base64 = line[8:]
                    vmess_json = self.decode_base64_content(vmess_base64)
                    if vmess_json:
                        config = json.loads(vmess_json)
                        vmess_configs.append(config)
                except Exception as e:
                    logging.warning(f"解析vmess链接失败 {line[:50]}...: {e}")
        
        return vmess_configs
    
    def create_v2ray_config(self, vmess_configs):
        """创建v2ray配置文件"""
        if not vmess_configs:
            logging.error("没有可用的vmess配置")
            return None
        
        # 基础配置模板
        base_config = {
            "log": {
                "access": "/var/log/v2ray/access.log",
                "error": "/var/log/v2ray/error.log",
                "loglevel": "warning"
            },
            "inbounds": [
                {
                    "port": 10808,
                    "protocol": "socks",
                    "settings": {
                        "auth": "noauth",
                        "udp": True,
                        "ip": "127.0.0.1"
                    },
                    "tag": "socks-inbound"
                },
                {
                    "port": 10809,
                    "protocol": "http",
                    "settings": {
                        "timeout": 360
                    },
                    "tag": "http-inbound"
                }
            ],
            "outbounds": [],
            "routing": {
                "domainStrategy": "IPOnDemand",
                "rules": [
                    {
                        "type": "field",
                        "ip": [
                            "geoip:private"
                        ],
                        "outboundTag": "direct"
                    },
                    {
                        "type": "field",
                        "ip": [
                            "geoip:cn"
                        ],
                        "outboundTag": "direct"
                    },
                    {
                        "type": "field",
                        "domain": [
                            "geosite:cn"
                        ],
                        "outboundTag": "direct"
                    }
                ]
            }
        }
        
        # 添加直连规则
        direct_outbound = {
            "tag": "direct",
            "protocol": "freedom",
            "settings": {}
        }
        base_config["outbounds"].append(direct_outbound)
        
        # 添加vmess出站配置
        for i, vmess in enumerate(vmess_configs):
            outbound = {
                "tag": f"proxy-{i}",
                "protocol": "vmess",
                "settings": {
                    "vnext": [
                        {
                            "address": vmess.get("add", ""),
                            "port": int(vmess.get("port", 0)),
                            "users": [
                                {
                                    "id": vmess.get("id", ""),
                                    "alterId": int(vmess.get("aid", 0)),
                                    "security": vmess.get("scy", "auto")
                                }
                            ]
                        }
                    ]
                },
                "streamSettings": {
                    "network": vmess.get("net", "tcp"),
                    "security": vmess.get("tls", "")
                }
            }
            
            # 添加ws设置
            if vmess.get("net") == "ws":
                outbound["streamSettings"]["wsSettings"] = {
                    "path": vmess.get("path", ""),
                    "headers": {
                        "Host": vmess.get("host", "")
                    }
                }
            
            # 添加tcp设置
            if vmess.get("net") == "tcp":
                outbound["streamSettings"]["tcpSettings"] = {
                    "header": {
                        "type": vmess.get("type", "none")
                    }
                }
            
            base_config["outbounds"].append(outbound)
        
        # 设置默认出站规则为第一个代理
        if len(base_config["outbounds"]) > 1:
            base_config["routing"]["rules"].append({
                "type": "field",
                "outboundTag": "proxy-0",
                "network": "tcp,udp"
            })
        
        return base_config
    
    def save_config(self, config):
        """保存配置文件"""
        try:
            with open(self.temp_config_path, 'w', encoding='utf-8') as f:
                json.dump(config, f, indent=2, ensure_ascii=False)
            logging.info(f"配置文件已保存到: {self.temp_config_path}")
            return True
        except Exception as e:
            logging.error(f"保存配置文件失败: {e}")
            return False
    
    def backup_old_config(self):
        """备份旧配置文件"""
        if os.path.exists(self.config_path):
            backup_path = f"{self.config_path}.backup.{int(time.time())}"
            try:
                subprocess.run(['cp', self.config_path, backup_path], check=True)
                logging.info(f"旧配置文件已备份到: {backup_path}")
                return True
            except subprocess.CalledProcessError as e:
                logging.error(f"备份配置文件失败: {e}")
                return False
        return True
    
    def restart_v2ray(self):
        """重启v2ray服务"""
        try:
            # 复制新配置文件
            subprocess.run(['sudo', 'cp', self.temp_config_path, self.config_path], check=True)
            subprocess.run(['sudo', 'chmod', '644', self.config_path], check=True)
            
            # 重启服务
            result = subprocess.run(['sudo', 'systemctl', 'restart', 'v2ray'], 
                                  capture_output=True, text=True)
            
            if result.returncode == 0:
                logging.info("v2ray服务重启成功")
                
                # 检查服务状态
                time.sleep(2)
                status_result = subprocess.run(['sudo', 'systemctl', 'is-active', 'v2ray'], 
                                             capture_output=True, text=True)
                if status_result.stdout.strip() == 'active':
                    logging.info("v2ray服务运行正常")
                    return True
                else:
                    logging.error("v2ray服务启动失败")
                    return False
            else:
                logging.error(f"重启v2ray服务失败: {result.stderr}")
                return False
                
        except subprocess.CalledProcessError as e:
            logging.error(f"执行系统命令失败: {e}")
            return False
    
    def validate_config(self):
        """验证配置文件语法"""
        try:
            with open(self.temp_config_path, 'r', encoding='utf-8') as f:
                json.load(f)  # 尝试解析JSON
            logging.info("配置文件语法验证通过")
            return True
        except json.JSONDecodeError as e:
            logging.error(f"配置文件语法错误: {e}")
            return False
    
    def run(self):
        """执行完整的更新流程"""
        logging.info("开始更新v2ray配置...")
        
        # 1. 获取订阅内容
        logging.info("获取订阅内容...")
        content = self.get_subscribe_content()
        if not content:
            return False
        
        # 2. 解码内容
        logging.info("解码订阅内容...")
        decoded_content = self.decode_base64_content(content)
        if not decoded_content:
            return False
        
        # 3. 解析vmess链接
        logging.info("解析vmess配置...")
        vmess_configs = self.parse_vmess_links(decoded_content)
        if not vmess_configs:
            logging.error("没有找到有效的vmess配置")
            return False
        
        logging.info(f"找到 {len(vmess_configs)} 个vmess配置")
        
        # 4. 创建新配置
        logging.info("生成v2ray配置文件...")
        new_config = self.create_v2ray_config(vmess_configs)
        if not new_config:
            return False
        
        # 5. 保存配置
        if not self.save_config(new_config):
            return False
        
        # 6. 验证配置
        if not self.validate_config():
            return False
        
        # 7. 备份旧配置
        if not self.backup_old_config():
            logging.warning("备份旧配置失败，但继续执行...")
        
        # 8. 重启服务
        logging.info("重启v2ray服务...")
        return self.restart_v2ray()

def main():
    """主函数"""
    # 从环境变量或命令行参数获取订阅地址
    # subscribe_url = os.getenv('V2RAY_SUBSCRIBE_URL')
    subscribe_url="xxxxxx"
    
    if not subscribe_url and len(sys.argv) > 1:
        subscribe_url = sys.argv[1]
    
    if not subscribe_url:
        # 使用您提供的base64编码的订阅内容
        encoded_subscribe = """
xxxxxx
...
"""  # 这里放入您提供的完整base64内容
        
        # 由于内容太长，建议从文件读取或直接使用URL
        print("请设置订阅地址:")
        print("方法1: 设置环境变量 V2RAY_SUBSCRIBE_URL")
        print("方法2: 作为命令行参数传入")
        print("示例: python v2ray_updater.py https://your-subscribe-url")
        return
    
    # 创建更新器并执行
    updater = V2RayConfigUpdater(subscribe_url)
    
    if updater.run():
        logging.info("v2ray配置更新成功!")
    else:
        logging.error("v2ray配置更新失败!")
        sys.exit(1)

if __name__ == "__main__":
    main()
```

- 注意，代码里面的`subscribe_url=`后面要换成自己的订阅地址，`config_path=`后面要换成安装完成后看到的 `config.json` 地址。
- 运行脚本
```bash
sudo python3 v2ray-config.py
```

## 测试代理

```python
#!/usr/bin/env python3
"""
V2Ray连接测试脚本
"""

import requests
import socket
import subprocess
import json
import time

def test_v2ray_service():
    """测试V2Ray服务状态"""
    print("=" * 50)
    print("1. 测试V2Ray服务状态")
    print("=" * 50)
    
    try:
        # 检查服务状态
        result = subprocess.run(['systemctl', 'is-active', 'v2ray'], 
                               capture_output=True, text=True)
        if result.stdout.strip() == 'active':
            print("✅ V2Ray服务正在运行")
        else:
            print("❌ V2Ray服务未运行")
            return False
        
        # 检查端口监听
        result = subprocess.run(['ss', '-tlnp'], capture_output=True, text=True)
        if '10808' in result.stdout:
            print("✅ SOCKS5端口(10808)正在监听")
        else:
            print("❌ SOCKS5端口(10808)未监听")
            
        if '10809' in result.stdout:
            print("✅ HTTP端口(10809)正在监听")
        else:
            print("❌ HTTP端口(10809)未监听")
            
        return True
    except Exception as e:
        print(f"❌ 服务状态检查失败: {e}")
        return False

def test_socks5_proxy():
    """测试SOCKS5代理"""
    print("\n" + "=" * 50)
    print("2. 测试SOCKS5代理连接")
    print("=" * 50)
    
    try:
        # 使用curl测试SOCKS5代理
        result = subprocess.run([
            'curl', '-x', 'socks5://127.0.0.1:10808',
            '--connect-timeout', '10',
            'http://httpbin.org/ip'
        ], capture_output=True, text=True, timeout=30)
        
        if result.returncode == 0:
            print("✅ SOCKS5代理连接成功")
            try:
                ip_info = json.loads(result.stdout)
                print(f"📍 当前IP: {ip_info.get('origin', '未知')}")
            except:
                print("📍 响应内容:", result.stdout[:100])
            return True
        else:
            print("❌ SOCKS5代理连接失败")
            print("错误信息:", result.stderr)
            return False
    except subprocess.TimeoutExpired:
        print("❌ SOCKS5代理连接超时")
        return False
    except Exception as e:
        print(f"❌ SOCKS5代理测试异常: {e}")
        return False

def test_http_proxy():
    """测试HTTP代理"""
    print("\n" + "=" * 50)
    print("3. 测试HTTP代理连接")
    print("=" * 50)
    
    try:
        # 使用curl测试HTTP代理
        result = subprocess.run([
            'curl', '-x', 'http://127.0.0.1:10809',
            '--connect-timeout', '10',
            'http://httpbin.org/ip'
        ], capture_output=True, text=True, timeout=30)
        
        if result.returncode == 0:
            print("✅ HTTP代理连接成功")
            try:
                ip_info = json.loads(result.stdout)
                print(f"📍 当前IP: {ip_info.get('origin', '未知')}")
            except:
                print("📍 响应内容:", result.stdout[:100])
            return True
        else:
            print("❌ HTTP代理连接失败")
            print("错误信息:", result.stderr)
            return False
    except subprocess.TimeoutExpired:
        print("❌ HTTP代理连接超时")
        return False
    except Exception as e:
        print(f"❌ HTTP代理测试异常: {e}")
        return False

def test_with_requests():
    """使用Python requests库测试"""
    print("\n" + "=" * 50)
    print("4. 使用Python requests测试")
    print("=" * 50)
    
    try:
        proxies = {
            'http': 'http://127.0.0.1:10809',
            'https': 'http://127.0.0.1:10809',
            'socks5': 'socks5://127.0.0.1:10808'
        }
        
        # 测试HTTP代理
        response = requests.get('http://httpbin.org/ip', 
                               proxies={'http': 'http://127.0.0.1:10809'}, 
                               timeout=10)
        if response.status_code == 200:
            ip_info = response.json()
            print("✅ HTTP代理(requests)连接成功")
            print(f"📍 当前IP: {ip_info.get('origin', '未知')}")
        else:
            print("❌ HTTP代理(requests)连接失败")
            return False
        
        # 测试SOCKS5代理
        try:
            response = requests.get('http://httpbin.org/ip', 
                                  proxies={'http': 'socks5://127.0.0.1:10808'}, 
                                  timeout=10)
            if response.status_code == 200:
                ip_info = response.json()
                print("✅ SOCKS5代理(requests)连接成功")
                print(f"📍 当前IP: {ip_info.get('origin', '未知')}")
            else:
                print("❌ SOCKS5代理(requests)连接失败")
        except Exception as e:
            print(f"⚠️ SOCKS5代理(requests)测试失败: {e}")
            
        return True
    except Exception as e:
        print(f"❌ Python requests测试失败: {e}")
        return False

def test_google_access():
    """测试Google访问（需要代理）"""
    print("\n" + "=" * 50)
    print("5. 测试Google访问")
    print("=" * 50)
    
    try:
        result = subprocess.run([
            'curl', '-x', 'socks5://127.0.0.1:10808',
            '--connect-timeout', '15',
            'https://www.google.com'
        ], capture_output=True, text=True, timeout=30)
        
        if result.returncode == 0 and 'google' in result.stdout.lower():
            print("✅ Google访问成功（代理工作正常）")
            return True
        else:
            print("❌ Google访问失败")
            if "resolve" in result.stderr.lower():
                print("⚠️ 可能是DNS解析问题，尝试检查DNS设置")
            return False
    except subprocess.TimeoutExpired:
        print("❌ Google访问超时")
        return False
    except Exception as e:
        print(f"❌ Google访问测试异常: {e}")
        return False

def check_config_file():
    """检查配置文件"""
    print("\n" + "=" * 50)
    print("6. 检查配置文件")
    print("=" * 50)
    
    config_path = "/usr/local/etc/v2ray/config.json"
    
    try:
        with open(config_path, 'r') as f:
            config = json.load(f)
        
        print("✅ 配置文件格式正确")
        print(f"📁 配置文件路径: {config_path}")
        
        # 检查基本配置项
        if 'outbounds' in config and len(config['outbounds']) > 0:
            outbound = config['outbounds'][0]
            if outbound.get('protocol') == 'vmess':
                print("✅ 使用VMess协议")
                if 'vnext' in outbound.get('settings', {}):
                    server = outbound['settings']['vnext'][0]
                    print(f"📍 服务器: {server.get('address')}:{server.get('port')}")
        
        if 'inbounds' in config:
            for inbound in config['inbounds']:
                if inbound.get('port') == 10808:
                    print("✅ SOCKS5代理端口: 10808")
                if inbound.get('port') == 10809:
                    print("✅ HTTP代理端口: 10809")
                    
        return True
    except FileNotFoundError:
        print("❌ 配置文件不存在")
        return False
    except json.JSONDecodeError as e:
        print(f"❌ 配置文件格式错误: {e}")
        return False
    except Exception as e:
        print(f"❌ 配置文件检查失败: {e}")
        return False

def main():
    """主测试函数"""
    print("🚀 开始V2Ray连接测试")
    print("=" * 60)
    
    tests = [
        test_v2ray_service,
        check_config_file,
        test_socks5_proxy,
        test_http_proxy,
        test_with_requests,
        test_google_access
    ]
    
    passed_tests = 0
    total_tests = len(tests)
    
    for test in tests:
        if test():
            passed_tests += 1
        time.sleep(1)  # 短暂延迟
    
    print("\n" + "=" * 60)
    print("📊 测试结果汇总")
    print("=" * 60)
    print(f"✅ 通过测试: {passed_tests}/{total_tests}")
    
    if passed_tests == total_tests:
        print("🎉 所有测试通过！V2Ray配置成功！")
        print("\n📋 使用说明:")
        print("SOCKS5代理: socks5://127.0.0.1:10808")
        print("HTTP代理: http://127.0.0.1:10809")
    else:
        print("❌ 部分测试失败，请检查V2Ray配置")
        print("\n🔧 故障排除:")
        print("1. 检查V2Ray服务状态: sudo systemctl status v2ray")
        print("2. 查看V2Ray日志: sudo journalctl -u v2ray -f")
        print("3. 检查配置文件: /usr/local/etc/v2ray/config.json")

if __name__ == "__main__":
    main()

```