Spaces:
Paused
Paused
File size: 6,257 Bytes
8d1819a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 |
import os
import platform
import requests
import subprocess
import threading
from python.helpers import files
from python.helpers.print_style import PrintStyle
class CloudflareTunnel:
def __init__(self, port: int):
self.port = port
self.bin_dir = "tmp" # Relative path
self.cloudflared_path = None
self.tunnel_process = None
self.tunnel_url = None
self._stop_event = threading.Event()
def download_cloudflared(self):
"""Downloads the appropriate cloudflared binary for the current system"""
# Create bin directory if it doesn't exist using files helper
os.makedirs(files.get_abs_path(self.bin_dir), exist_ok=True)
# Determine OS and architecture
system = platform.system().lower()
arch = platform.machine().lower()
# Define executable name
executable_name = "cloudflared.exe" if system == "windows" else "cloudflared"
install_path = files.get_abs_path(self.bin_dir, executable_name)
# Return if already exists
if files.exists(self.bin_dir, executable_name):
self.cloudflared_path = install_path
return install_path
# Map platform/arch to download URLs
base_url = "https://github.com/cloudflare/cloudflared/releases/latest/download/"
if system == "darwin": # macOS
# Download and extract .tgz for macOS
download_file = "cloudflared-darwin-amd64.tgz" if arch == "x86_64" else "cloudflared-darwin-arm64.tgz"
download_url = f"{base_url}{download_file}"
download_path = files.get_abs_path(self.bin_dir, download_file)
PrintStyle().print(f"\nDownloading cloudflared from: {download_url}")
response = requests.get(download_url, stream=True)
if response.status_code != 200:
raise RuntimeError(f"Failed to download cloudflared: {response.status_code}")
# Save the .tgz file
with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Extract cloudflared binary from .tgz
import tarfile
with tarfile.open(download_path, "r:gz") as tar:
tar.extract("cloudflared", files.get_abs_path(self.bin_dir))
# Cleanup .tgz file
os.remove(download_path)
else: # Linux and Windows
if system == "linux":
if arch in ["x86_64", "amd64"]:
download_file = "cloudflared-linux-amd64"
elif arch == "arm64" or arch == "aarch64":
download_file = "cloudflared-linux-arm64"
elif arch == "arm":
download_file = "cloudflared-linux-arm"
else:
download_file = "cloudflared-linux-386"
elif system == "windows":
download_file = "cloudflared-windows-amd64.exe"
else:
raise RuntimeError(f"Unsupported platform: {system} {arch}")
download_url = f"{base_url}{download_file}"
download_path = files.get_abs_path(self.bin_dir, download_file)
PrintStyle().print(f"\nDownloading cloudflared from: {download_url}")
response = requests.get(download_url, stream=True)
if response.status_code != 200:
raise RuntimeError(f"Failed to download cloudflared: {response.status_code}")
with open(download_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Rename and set permissions
if os.path.exists(install_path):
os.remove(install_path)
os.rename(download_path, install_path)
# Set executable permissions
if system != "windows":
os.chmod(install_path, 0o755)
self.cloudflared_path = install_path
return install_path
def _extract_tunnel_url(self, process):
"""Extracts the tunnel URL from cloudflared output"""
while not self._stop_event.is_set():
line = process.stdout.readline()
if not line:
break
if isinstance(line, bytes):
line = line.decode('utf-8')
if "trycloudflare.com" in line and "https://" in line:
start = line.find("https://")
end = line.find("trycloudflare.com") + len("trycloudflare.com")
self.tunnel_url = line[start:end].strip()
PrintStyle().print("\n=== Cloudflare Tunnel URL ===")
PrintStyle().print(f"URL: {self.tunnel_url}")
PrintStyle().print("============================\n")
return
def start(self):
"""Starts the cloudflare tunnel"""
if not self.cloudflared_path:
self.download_cloudflared()
PrintStyle().print("\nStarting Cloudflare tunnel...")
# Start tunnel process
self.tunnel_process = subprocess.Popen(
[
str(self.cloudflared_path),
"tunnel",
"--url",
f"http://localhost:{self.port}"
],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=1,
universal_newlines=True
)
# Extract tunnel URL in separate thread
threading.Thread(
target=self._extract_tunnel_url,
args=(self.tunnel_process,),
daemon=True
).start()
def stop(self):
"""Stops the cloudflare tunnel"""
self._stop_event.set()
if self.tunnel_process:
PrintStyle().print("\nStopping Cloudflare tunnel...")
self.tunnel_process.terminate()
self.tunnel_process.wait()
self.tunnel_process = None
self.tunnel_url = None |