|
| 1 | +# Proxy Chaining ("WARP on Any") |
| 2 | + |
| 3 | +This is one of the most powerful features of the library, allowing you to route the traffic of any Xray-compatible configuration through a final exit-point proxy. This is commonly used to enhance privacy or bypass complex network restrictions by creating a "WARP on Any" setup, where all your traffic exits through a Cloudflare WARP (WireGuard) endpoint. |
| 4 | + |
| 5 | +The implementation is seamless: once you provide a WARP configuration, the `XrayConfigBuilder` automatically configures all other outbounds to use it as their `dialerProxy`. |
| 6 | + |
| 7 | +## How It Works |
| 8 | + |
| 9 | +Instead of connecting directly to the internet, a primary outbound (like VLESS or Trojan) first routes its traffic through the specified WARP outbound. The traffic flow looks like this: |
| 10 | + |
| 11 | +`Your App` -> `Local SOCKS Inbound` -> `Primary Outbound (e.g., VLESS)` -> `WARP Outbound (WireGuard)` -> `Internet` |
| 12 | + |
| 13 | +This entire chain is managed within a single, merged Xray instance for maximum efficiency. |
| 14 | + |
| 15 | +## Practical Example: Testing Connectivity through WARP |
| 16 | + |
| 17 | +This example demonstrates how to test the connectivity (ping) of several configurations that are all forced to route their traffic through a single WARP profile. |
| 18 | + |
| 19 | +```python |
| 20 | +# examples/07_warp_tester.py |
| 21 | + |
| 22 | +import os |
| 23 | +import sys |
| 24 | +from pathlib import Path |
| 25 | + |
| 26 | +# Add project root to path |
| 27 | +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) |
| 28 | + |
| 29 | +from python_v2ray.downloader import BinaryDownloader |
| 30 | +from python_v2ray.tester import ConnectionTester |
| 31 | +from python_v2ray.config_parser import parse_uri |
| 32 | + |
| 33 | +def main(): |
| 34 | + project_root = Path(__file__).parent.parent |
| 35 | + |
| 36 | + # Ensure all necessary binaries are ready |
| 37 | + downloader = BinaryDownloader(project_root) |
| 38 | + downloader.ensure_all() |
| 39 | + |
| 40 | + # --- 1. Define your WARP config (must be a WireGuard URI) --- |
| 41 | + warp_uri = "wireguard://YOUR_PRIVATE_KEY@engage.cloudflareclient.com:2408?address=172.16.0.2/32&publicKey=...#WARP-Profile" |
| 42 | + |
| 43 | + # --- 2. Define the primary configs to test THROUGH WARP --- |
| 44 | + test_uris = [ |
| 45 | + "vless://YOUR_UUID@your.domain.com:443?type=ws#VLESS-through-WARP", |
| 46 | + "trojan://YOUR_PASSWORD@another.domain.com:443#Trojan-through-WARP", |
| 47 | + # You can even chain another WireGuard config through the main WARP config |
| 48 | + # "wireguard://ANOTHER_PRIVATE_KEY@...#WG-on-WARP" |
| 49 | + ] |
| 50 | + |
| 51 | + # --- 3. Parse all configurations --- |
| 52 | + parsed_warp_config = parse_uri(warp_uri) |
| 53 | + parsed_test_configs = [p for p in (parse_uri(uri) for uri in test_uris) if p] |
| 54 | + |
| 55 | + # --- Input Validation --- |
| 56 | + if not parsed_warp_config or "YOUR_PRIVATE_KEY" in warp_uri: |
| 57 | + print("! Please provide a valid WARP WireGuard URI.") |
| 58 | + return |
| 59 | + if not parsed_test_configs: |
| 60 | + print("! No valid primary configurations found to test.") |
| 61 | + return |
| 62 | + |
| 63 | + tester = ConnectionTester( |
| 64 | + vendor_path=str(project_root / "vendor"), |
| 65 | + core_engine_path=str(project_root / "core_engine") |
| 66 | + ) |
| 67 | + |
| 68 | + # --- 4. Run the test, passing the WARP config to the `warp_config` parameter --- |
| 69 | + print(f"\n--- Running Connectivity Test with WARP-on-Any enabled ---") |
| 70 | + results = tester.test_uris( |
| 71 | + parsed_params=parsed_test_configs, |
| 72 | + warp_config=parsed_warp_config, |
| 73 | + timeout=30 # Allow a longer timeout for chained connections |
| 74 | + ) |
| 75 | + |
| 76 | + # --- 5. Print the results --- |
| 77 | + if results: |
| 78 | + print("\n" + "="*20 + " WARP ON ANY PING TEST RESULTS " + "="*20) |
| 79 | + sorted_results = sorted(results, key=lambda x: x.get('ping_ms', 9999)) |
| 80 | + for result in sorted_results: |
| 81 | + tag = result.get('tag', 'N/A') |
| 82 | + ping = result.get('ping_ms', -1) |
| 83 | + status = result.get('status', 'error') |
| 84 | + |
| 85 | + if status == 'success': |
| 86 | + print(f"* Tag: {tag:<30} | Ping: {ping:>4} ms | Status: SUCCESS") |
| 87 | + else: |
| 88 | + print(f"! Tag: {tag:<30} | Ping: ---- ms | Status: FAILED ({status})") |
| 89 | + else: |
| 90 | + print("! No results were received from the tester.") |
| 91 | + |
| 92 | +if __name__ == "__main__": |
| 93 | + main() |
| 94 | +``` |
| 95 | + |
| 96 | +## API Reference: The `warp_config` Parameter |
| 97 | + |
| 98 | +To enable proxy chaining, a new optional parameter has been added to all primary testing methods: |
| 99 | + |
| 100 | +- `test_uris(..., warp_config: Optional[ConfigParams] = None)` |
| 101 | +- `test_speed(..., warp_config: Optional[ConfigParams] = None)` |
| 102 | +- `test_upload(..., warp_config: Optional[ConfigParams] = None)` |
| 103 | + |
| 104 | +Simply pass a parsed `ConfigParams` object of a valid WireGuard configuration to this parameter, and the `ConnectionTester` will handle the rest. |
| 105 | + |
| 106 | +## Important Considerations |
| 107 | + |
| 108 | +- The `warp_config` **must** be a valid `ConfigParams` object generated from a WireGuard URI. |
| 109 | +- This chaining feature is handled by Xray's `dialerProxy`. Therefore, it only applies to protocols that are managed within the merged Xray instance (VLESS, VMess, Trojan, etc.). It does **not** apply to protocols like Hysteria that are launched in a separate, independent process by the Go test engine. |
0 commit comments