|
|
import gradio as gr |
|
|
import subprocess |
|
|
import os |
|
|
import time |
|
|
import threading |
|
|
import socket |
|
|
from pathlib import Path |
|
|
|
|
|
class VPNManager: |
|
|
def __init__(self): |
|
|
self.vpn_process = None |
|
|
self.status = "Stopped" |
|
|
self.setup_complete = False |
|
|
|
|
|
def setup_vpn(self): |
|
|
"""Setup OpenVPN server configuration""" |
|
|
try: |
|
|
|
|
|
result = subprocess.run(['/app/setup_vpn.sh'], |
|
|
capture_output=True, text=True) |
|
|
|
|
|
if result.returncode == 0: |
|
|
self.setup_complete = True |
|
|
return "VPN setup completed successfully" |
|
|
else: |
|
|
return f"Setup failed: {result.stderr}" |
|
|
except Exception as e: |
|
|
return f"Setup error: {str(e)}" |
|
|
|
|
|
def start_vpn(self): |
|
|
"""Start OpenVPN server""" |
|
|
if not self.setup_complete: |
|
|
return "Please complete setup first" |
|
|
|
|
|
try: |
|
|
|
|
|
self.vpn_process = subprocess.Popen([ |
|
|
'openvpn', |
|
|
'--config', '/etc/openvpn/server/server.conf', |
|
|
'--daemon' |
|
|
]) |
|
|
|
|
|
time.sleep(2) |
|
|
|
|
|
if self.vpn_process.poll() is None: |
|
|
self.status = "Running" |
|
|
return "VPN server started successfully" |
|
|
else: |
|
|
self.status = "Failed" |
|
|
return "Failed to start VPN server" |
|
|
|
|
|
except Exception as e: |
|
|
return f"Start error: {str(e)}" |
|
|
|
|
|
def stop_vpn(self): |
|
|
"""Stop OpenVPN server""" |
|
|
try: |
|
|
if self.vpn_process: |
|
|
self.vpn_process.terminate() |
|
|
self.vpn_process = None |
|
|
|
|
|
|
|
|
subprocess.run(['pkill', '-f', 'openvpn'], |
|
|
capture_output=True) |
|
|
|
|
|
self.status = "Stopped" |
|
|
return "VPN server stopped" |
|
|
except Exception as e: |
|
|
return f"Stop error: {str(e)}" |
|
|
|
|
|
def get_status(self): |
|
|
"""Get current VPN status""" |
|
|
return self.status |
|
|
|
|
|
def get_client_config(self): |
|
|
"""Generate client configuration""" |
|
|
if not self.setup_complete: |
|
|
return "Setup not complete" |
|
|
|
|
|
try: |
|
|
|
|
|
config_path = "/etc/openvpn/server/client.ovpn" |
|
|
if os.path.exists(config_path): |
|
|
with open(config_path, 'r') as f: |
|
|
return f.read() |
|
|
else: |
|
|
return "Client config not found" |
|
|
except Exception as e: |
|
|
return f"Config error: {str(e)}" |
|
|
|
|
|
|
|
|
vpn_manager = VPNManager() |
|
|
|
|
|
def setup_interface(): |
|
|
return vpn_manager.setup_vpn() |
|
|
|
|
|
def start_interface(): |
|
|
return vpn_manager.start_vpn() |
|
|
|
|
|
def stop_interface(): |
|
|
return vpn_manager.stop_vpn() |
|
|
|
|
|
def status_interface(): |
|
|
return vpn_manager.get_status() |
|
|
|
|
|
def config_interface(): |
|
|
return vpn_manager.get_client_config() |
|
|
|
|
|
|
|
|
with gr.Blocks(title="VPN Server - Hugging Face") as demo: |
|
|
gr.Markdown("# VPN Server on Hugging Face") |
|
|
gr.Markdown("**Port 7860 Access Only** - Manage your VPN server") |
|
|
|
|
|
with gr.Row(): |
|
|
with gr.Column(): |
|
|
gr.Markdown("### Server Controls") |
|
|
setup_btn = gr.Button("Setup VPN", variant="primary") |
|
|
start_btn = gr.Button("Start VPN", variant="secondary") |
|
|
stop_btn = gr.Button("Stop VPN", variant="stop") |
|
|
status_btn = gr.Button("Check Status") |
|
|
|
|
|
with gr.Column(): |
|
|
gr.Markdown("### Output") |
|
|
output = gr.Textbox(label="Status", lines=3, interactive=False) |
|
|
|
|
|
with gr.Row(): |
|
|
gr.Markdown("### Client Configuration") |
|
|
config_btn = gr.Button("Get Client Config") |
|
|
config_output = gr.Textbox(label="Client Config", lines=10, interactive=False) |
|
|
|
|
|
|
|
|
setup_btn.click(setup_interface, outputs=output) |
|
|
start_btn.click(start_interface, outputs=output) |
|
|
stop_btn.click(stop_interface, outputs=output) |
|
|
status_btn.click(status_interface, outputs=output) |
|
|
config_btn.click(config_interface, outputs=config_output) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
demo.launch( |
|
|
server_name="0.0.0.0", |
|
|
server_port=7860, |
|
|
share=False |
|
|
) |