Add files via upload

Author: arshiacomplus

python_v2ray/config_parser.py

@@ -178,10 +178,8 @@ def _parse_wireguard(uri: str, common: dict) -> ConfigParams:
 def _parse_hysteria(uri: str, common: dict) -> ConfigParams:
     params = _parse_query_params(urllib.parse.urlparse(uri).query)
     password = urllib.parse.urlparse(uri).username
-
     return ConfigParams(
         **common,
-        protocol="hysteria2",
         hy2_password=password,
         security="tls",
         sni=params.get("sni", common['address']),
@@ -231,6 +229,8 @@ def build_outbound_from_params(self, params: ConfigParams, fragment_config: Opti
         * The main engine. Converts ConfigParams into a complete Xray outbound dictionary.
         * Now with added support for TLS fragmentation.
         """
+        if params.protocol in ["hysteria", "hysteria2","hy2"]:
+            params.protocol = "socks"
         use_fragment = fragment_config is not None
         stream_settings = self._build_stream_settings(params, fragment=use_fragment, **kwargs)
 

python_v2ray/downloader.py

@@ -1,20 +1,14 @@
-import requests
-import sys
-import os
-import zipfile
-import io
-import platform
+# python_v2ray/downloader.py
+
+import requests, sys, os, zipfile, io, platform
 from pathlib import Path
 from typing import Optional
 
 XRAY_REPO = "GFW-knocker/Xray-core"
 OWN_REPO = "arshiacomplus/python_v2ray"
+HYSTERIA_REPO = "apernet/hysteria"
 
 class BinaryDownloader:
-    """
-    * Handles the logic of downloading and extracting necessary binaries
-    * from GitHub Releases for the current operating system and architecture.
-    """
     def __init__(self, project_root: Path):
         self.project_root = project_root
         self.vendor_path = self.project_root / "vendor"
@@ -23,52 +17,54 @@ def __init__(self, project_root: Path):
         self.arch = self._get_arch_name()
 
     def _get_os_name(self) -> str:
-        """* Determines the OS name used in release assets."""
         if sys.platform == "win32": return "windows"
-        if sys.platform == "darwin": return "macos"
+        if sys.platform == "darwin": return "darwin"
         return "linux"
 
     def _get_arch_name(self) -> str:
-        """* Determines the architecture name (e.g., 64, 32, arm64-v8a)."""
         machine = platform.machine().lower()
-        if "amd64" in machine or "x86_64" in machine:
-            return "64"
-        if "arm64" in machine or "aarch64" in machine:
-            return "arm64-v8a"
-        if "386" in machine or "x86" in machine:
-            return "32"
+        if "amd64" in machine or "x86_64" in machine: return "amd64"
+        if "arm64" in machine or "aarch64" in machine: return "arm64"
+        if "386" in machine or "x86" in machine: return "386"
         return "unsupported"
 
     def _get_asset_url(self, assets: list, name_prefix: str) -> Optional[str]:
-        """
-        * It finds the download URL for a specific asset based on OS and arch.
-        """
-        asset_name = f"{name_prefix}-{self.os_name}-{self.arch}.zip"
-        print(f"note: Searching for asset: {asset_name}")
+        # * This logic is now smarter to handle different naming conventions
+
+        if name_prefix == "hysteria":
+            asset_name = f"{name_prefix}-{self.os_name}-{self.arch}"
+            if self.os_name == "windows":
+                asset_name += ".exe"
+        elif name_prefix == "Xray":
+            arch_name = "64" if self.arch == "amd64" else self.arch
+            os_name = "macos" if self.os_name == "darwin" else self.os_name # Xray uses macos
+            asset_name = f"{name_prefix}-{os_name}-{arch_name}.zip"
+        else:
+            arch_name = "64" if self.arch == "amd64" else self.arch
+            os_name = "macos" if self.os_name == "darwin" else self.os_name
+            asset_name = f"{name_prefix}-{os_name}-{arch_name}.zip"
 
+        print(f"note: Searching for asset: {asset_name}")
         for asset in assets:
             if asset['name'].lower() == asset_name.lower():
                 return asset['browser_download_url']
         return None
 
     def ensure_binary(self, name: str, target_dir: Path, repo: str) -> bool:
-        """
-        * Checks for a binary, and if not found, downloads and extracts the
-        * entire contents of the zip file (including .dat files).
-        """
         exe_name = f"{name}.exe" if sys.platform == "win32" else name
         target_file = target_dir / exe_name
 
-        geoip_file = target_dir / "geoip.dat"
-
-        if target_file.is_file() and (name != "xray" or geoip_file.is_file()):
-            print(f"* Binary '{exe_name}' and necessary assets already exist.")
+        if target_file.is_file():
+            print(f"* Binary '{exe_name}' already exists.")
             return True
 
-        print(f"! Binary '{exe_name}' not found. Attempting to download from '{repo}'...")
-
+        print(f"! Binary '{exe_name}' not found. Downloading from '{repo}'...")
         try:
-            release_url = f"https://api.github.com/repos/{repo}/releases/latest"
+            if name == "hysteria":
+                release_url = f"https://api.github.com/repos/{repo}/releases/tags/app%2Fv2.6.2" # ! Hardcoded to a specific stable version
+            else:
+                release_url = f"https://api.github.com/repos/{repo}/releases/latest"
+
             response = requests.get(release_url, timeout=10)
             response.raise_for_status()
             assets = response.json().get("assets", [])
@@ -77,36 +73,46 @@ def ensure_binary(self, name: str, target_dir: Path, repo: str) -> bool:
             download_url = self._get_asset_url(assets, asset_prefix)
 
             if not download_url:
-                print(f"! ERROR: Could not find downloadable asset for '{name}' matching '{self.os_name}-{self.arch}' in repo '{repo}'.")
+                print(f"! ERROR: Could not find downloadable asset for '{name}'.")
                 return False
 
-            print(f"* Downloading from: {download_url}")
+            print(f"* Downloading: {download_url}")
             asset_response = requests.get(download_url, timeout=120, stream=True)
             asset_response.raise_for_status()
 
-            print(f"* Extracting all files to '{target_dir}'...")
-            with zipfile.ZipFile(io.BytesIO(asset_response.content)) as z:
-                z.extractall(path=target_dir)
-            print(f"* Successfully downloaded and extracted all assets for '{name}'.")
-
-            if sys.platform != "win32" and target_file.is_file():
+            if not download_url.endswith('.zip'):
+                with open(target_file, "wb") as f:
+                    f.write(asset_response.content)
+            else:
+                with zipfile.ZipFile(io.BytesIO(asset_response.content)) as z:
+                    for member_name in z.namelist():
+                        if Path(member_name).name.lower() == exe_name.lower():
+                            source = z.open(member_name)
+                            target = open(target_file, "wb")
+                            with source, target: target.write(source.read())
+
+                            for dat_file in ["geoip.dat", "geosite.dat"]:
+                                if dat_file in z.namelist():
+                                    with z.open(dat_file) as source_dat, open(target_dir / dat_file, "wb") as target_dat:
+                                        target_dat.write(source_dat.read())
+
+            print(f"* Successfully downloaded '{exe_name}'.")
+            if sys.platform != "win32":
                 os.chmod(target_file, 0o755)
-
             return True
 
         except Exception as e:
-            print(f"! ERROR during download/extraction: {e}")
+            print(f"! ERROR during download/extraction for '{name}': {e}")
             return False
 
     def ensure_all(self):
-        """* Ensures all necessary binaries are present."""
         print("--- Checking for necessary binaries & databases ---")
         self.vendor_path.mkdir(exist_ok=True)
         self.core_engine_path.mkdir(exist_ok=True)
-
         xray_ok = self.ensure_binary("xray", self.vendor_path, XRAY_REPO)
         engine_ok = self.ensure_binary("core_engine", self.core_engine_path, OWN_REPO)
+        hysteria_ok = self.ensure_binary("hysteria", self.vendor_path, HYSTERIA_REPO)
 
         print("-------------------------------------------------")
-        if not (xray_ok and engine_ok):
-            raise RuntimeError("Could not obtain all necessary files.")
+        if not (xray_ok and engine_ok and hysteria_ok):
+            raise RuntimeError("Could not obtain all necessary binaries.")

python_v2ray/tester.py

@@ -1,105 +1,66 @@
-
-import subprocess
-import json
-import os
-from typing import List, Dict, Any, Optional
+import subprocess, json, os, sys
 from pathlib import Path
-import sys
+from typing import List, Dict, Any, Optional
+from .config_parser import XrayConfigBuilder, ConfigParams
 
 class ConnectionTester:
-    """
-    * This class acts as a bridge between Python and the high-performance Go tester engine.
-    """
-
     def __init__(self, vendor_path: str, core_engine_path: str):
-        """
-        Args:
-            vendor_path (str): Path to the 'vendor' directory (for xray executable).
-            core_engine_path (str): Path to the 'core_engine' directory (for go engine).
-        """
+        self.vendor_path = Path(vendor_path)
+        self.core_engine_path = Path(core_engine_path)
+
         if sys.platform == "win32":
-            tester_exe_name = "core_engine.exe"
-            xray_exe_name = "xray.exe"
+            self.tester_exe = "core_engine.exe"; self.xray_exe = "xray.exe"; self.hysteria_exe = "hysteria.exe"
         elif sys.platform == "darwin":
-            tester_exe_name = "core_engine_macos"
-            xray_exe_name = "xray_macos"
-        else: # Linux and others
-            tester_exe_name = "core_engine_linux"
-            xray_exe_name = "xray_linux"
-
-        # * Use pathlib for safe and cross-platform path construction
-        self.tester_path = str(Path(core_engine_path) / tester_exe_name)
-        self.xray_path = str(Path(vendor_path) / xray_exe_name)
-
-        if not os.path.exists(self.tester_path):
-            raise FileNotFoundError(f"Tester executable not found at: {self.tester_path}")
-        if not os.path.exists(self.xray_path):
-            raise FileNotFoundError(f"Xray executable not found at: {self.xray_path}")
-    def test_outbounds(self, outbounds: List[Dict[str, Any]], fragment_config: Optional[Dict[str, Any]] = None, timeout: int = 60) -> List[Dict[str, Any]]:
-        """
-        * Tests a list of outbound configs concurrently using the Go engine, with optional fragmentation.
+            self.tester_exe = "core_engine_macos"; self.xray_exe = "xray_macos"; self.hysteria_exe = "hysteria_macos"
+        else:
+            self.tester_exe = "core_engine_linux"; self.xray_exe = "xray_linux"; self.hysteria_exe = "hysteria_linux"
 
-        Args:
-            outbounds (List[Dict[str, Any]]): A list of Xray outbound config dictionaries.
-            fragment_config (Optional[Dict[str, Any]]): Settings for TLS fragmentation.
-            timeout (int): Total time in seconds to wait for all tests to complete.
+        if not (self.core_engine_path / self.tester_exe).is_file(): raise FileNotFoundError(f"Tester executable not found")
 
-        Returns:
-            A list of result dictionaries.
-        """
-        if not outbounds:
-            return []
-        fragment_json_bytes = json.dumps(fragment_config).encode('utf-8') if fragment_config else b'null'
+    def test_outbounds(self, parsed_params: List[ConfigParams], fragment_config: Optional[Dict[str, Any]] = None, timeout: int = 60) -> List[Dict[str, Any]]:
+        if not parsed_params: return []
 
         test_configs = []
         base_port = 20800
-        for i, outbound in enumerate(outbounds):
-            if fragment_config and outbound.get("protocol") not in ["freedom", "blackhole"]:
-                if "streamSettings" not in outbound:
-                    outbound["streamSettings"] = {}
-                outbound["streamSettings"]["sockopt"] = {"dialerProxy": "fragment"}
+        builder = XrayConfigBuilder()
+
+        for i, params in enumerate(parsed_params):
+            config_dict = {}
+            client_path = ""
+            protocol = params.protocol
+
+            if protocol in ["hysteria", "hysteria2"]:
+                protocol = "hysteria2"
+                client_path = str(self.vendor_path / self.hysteria_exe)
+                config_dict = {
+                    "server": f"{params.address}:{params.port}",
+                    "auth": params.hy2_password,
+                    "socks5": {"listen": f"127.0.0.1:{base_port + i}"},
+                    "tls": {"sni": params.sni, "insecure": True}
+                }
+            else:
+                client_path = str(self.vendor_path / self.xray_exe)
+                outbound = builder.build_outbound_from_params(params, fragment_config=fragment_config)
+                if fragment_config:
+                    if "streamSettings" not in outbound: outbound["streamSettings"] = {}
+                    outbound["streamSettings"]["sockopt"] = {"dialerProxy": "fragment"}
+                config_dict = outbound
 
             test_configs.append({
-                "tag": outbound.get("tag", f"outbound_{i}"),
-                "config": outbound,
+                "tag": params.tag,
+                "protocol": protocol,
+                "config": config_dict,
                 "test_port": base_port + i,
-                "xray_path": self.xray_path,
-                "fragment_config": json.loads(fragment_json_bytes),
+                "client_path": client_path,
+                "fragment_config": fragment_config,
             })
 
-        input_json = json.dumps(test_configs)
+        input_json = json.dumps(test_configs, default=lambda o: o.__dict__)
 
         try:
-            with subprocess.Popen(
-                [self.tester_path],
-                stdin=subprocess.PIPE,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-                encoding='utf-8'
-            ) as process:
+            with subprocess.Popen([str(self.core_engine_path / self.tester_exe)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding='utf-8') as process:
                 stdout, stderr = process.communicate(input=input_json, timeout=timeout)
-
-                if process.returncode != 0:
-                    print(f"! Go engine exited with an error (return code {process.returncode}):")
-                    print(f"--- STDERR ---:\n{stderr}\n--------------")
-                    return []
-
-                if not stdout:
-                    print("! Go engine returned no output.")
-                    return []
-
-                results = json.loads(stdout)
-                return results
-
-        except subprocess.TimeoutExpired:
-            process.kill()
-            print(f"! Testing process timed out after {timeout} seconds.")
-            return []
-        except json.JSONDecodeError:
-            print("! Failed to decode JSON from Go engine. Raw output:")
-            print(f"--- STDOUT ---:\n{stdout}\n--------------")
-            return []
+                if process.returncode != 0: print(f"! Go engine error:\n{stderr}"); return []
+                return json.loads(stdout) if stdout else []
         except Exception as e:
-            print(f"! An unexpected error occurred while running the tester: {e}")
-            return []
+            print(f"! Tester execution error: {e}"); return []