mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2025-12-14 16:16:21 +00:00
466 lines
15 KiB
Python
466 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
ProxMenux Terminal WebSocket Routes
|
|
Provides a WebSocket endpoint for interactive terminal sessions
|
|
"""
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
from flask_sock import Sock
|
|
import subprocess
|
|
import os
|
|
import pty
|
|
import select
|
|
import struct
|
|
import fcntl
|
|
import termios
|
|
import threading
|
|
import time
|
|
import requests
|
|
import json
|
|
import tempfile
|
|
import base64
|
|
|
|
terminal_bp = Blueprint('terminal', __name__)
|
|
sock = Sock()
|
|
|
|
# Active terminal sessions
|
|
active_sessions = {}
|
|
|
|
@terminal_bp.route('/api/terminal/health', methods=['GET'])
|
|
def terminal_health():
|
|
"""Health check for terminal service"""
|
|
return {'success': True, 'active_sessions': len(active_sessions)}
|
|
|
|
@terminal_bp.route('/api/terminal/search-command', methods=['GET'])
|
|
def search_command():
|
|
"""Proxy endpoint for cheat.sh API to avoid CORS issues"""
|
|
query = request.args.get('q', '')
|
|
|
|
if not query or len(query) < 2:
|
|
return jsonify({'error': 'Query too short'}), 400
|
|
|
|
try:
|
|
url = f'https://cht.sh/{query.replace(" ", "+")}?QT'
|
|
headers = {
|
|
'User-Agent': 'curl/7.68.0'
|
|
}
|
|
|
|
response = requests.get(url, headers=headers, timeout=10)
|
|
|
|
if response.status_code == 200:
|
|
content = response.text
|
|
examples = []
|
|
current_description = []
|
|
|
|
for line in content.split('\n'):
|
|
stripped = line.strip()
|
|
|
|
# Ignorar líneas vacías
|
|
if not stripped:
|
|
continue
|
|
|
|
# Si es un comentario
|
|
if stripped.startswith('#'):
|
|
# Acumular descripciones
|
|
current_description.append(stripped[1:].strip())
|
|
# Si no es comentario, es un comando
|
|
elif stripped and not stripped.startswith('http'):
|
|
# Unir las descripciones acumuladas
|
|
description = ' '.join(current_description) if current_description else ''
|
|
|
|
examples.append({
|
|
'description': description,
|
|
'command': stripped
|
|
})
|
|
|
|
# Resetear descripciones para el siguiente comando
|
|
current_description = []
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'examples': examples
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'API returned status {response.status_code}'
|
|
}), response.status_code
|
|
|
|
except requests.Timeout:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Request timeout'
|
|
}), 504
|
|
except Exception as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
def set_winsize(fd, rows, cols):
|
|
"""Set terminal window size"""
|
|
try:
|
|
winsize = struct.pack('HHHH', rows, cols, 0, 0)
|
|
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
|
|
except Exception as e:
|
|
print(f"Error setting window size: {e}")
|
|
|
|
def read_and_forward_output(master_fd, ws):
|
|
"""Read from PTY and send to WebSocket"""
|
|
while True:
|
|
try:
|
|
# Use select with timeout to check if data is available
|
|
r, _, _ = select.select([master_fd], [], [], 0.01)
|
|
if master_fd in r:
|
|
try:
|
|
data = os.read(master_fd, 4096)
|
|
if data:
|
|
ws.send(data.decode('utf-8', errors='ignore'))
|
|
else:
|
|
break
|
|
except OSError:
|
|
break
|
|
except Exception as e:
|
|
print(f"Error reading from PTY: {e}")
|
|
break
|
|
|
|
@sock.route('/ws/terminal')
|
|
def terminal_websocket(ws):
|
|
"""WebSocket endpoint for terminal sessions"""
|
|
|
|
# Create pseudo-terminal
|
|
master_fd, slave_fd = pty.openpty()
|
|
|
|
# Start bash process
|
|
shell_process = subprocess.Popen(
|
|
['/bin/bash', '-i'],
|
|
stdin=slave_fd,
|
|
stdout=slave_fd,
|
|
stderr=slave_fd,
|
|
preexec_fn=os.setsid,
|
|
cwd='/',
|
|
env=dict(os.environ, TERM='xterm-256color', PS1='\\u@\\h:\\w\\$ ')
|
|
)
|
|
|
|
session_id = id(ws)
|
|
active_sessions[session_id] = {
|
|
'process': shell_process,
|
|
'master_fd': master_fd
|
|
}
|
|
|
|
# Set non-blocking mode for master_fd
|
|
flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
|
|
fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
|
|
|
|
# Set initial terminal size
|
|
set_winsize(master_fd, 30, 120)
|
|
|
|
# Start thread to read PTY output and forward to WebSocket
|
|
output_thread = threading.Thread(
|
|
target=read_and_forward_output,
|
|
args=(master_fd, ws),
|
|
daemon=True
|
|
)
|
|
output_thread.start()
|
|
|
|
try:
|
|
while True:
|
|
# Receive data from WebSocket (blocking)
|
|
data = ws.receive(timeout=None)
|
|
|
|
if data is None:
|
|
# Client closed connection
|
|
break
|
|
|
|
handled = False
|
|
|
|
# Try to handle JSON control messages (e.g. resize)
|
|
if isinstance(data, str):
|
|
try:
|
|
msg = json.loads(data)
|
|
except Exception:
|
|
msg = None
|
|
|
|
if isinstance(msg, dict) and msg.get('type') == 'resize':
|
|
cols = int(msg.get('cols', 120))
|
|
rows = int(msg.get('rows', 30))
|
|
set_winsize(master_fd, rows, cols)
|
|
handled = True
|
|
|
|
if handled:
|
|
# Control message processed, do not send to bash
|
|
continue
|
|
|
|
# Optional: legacy resize escape sequence support
|
|
if isinstance(data, str) and data.startswith('\x1b[8;'):
|
|
try:
|
|
parts = data[4:-1].split(';')
|
|
rows, cols = int(parts[0]), int(parts[1])
|
|
set_winsize(master_fd, rows, cols)
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# Send input to bash
|
|
try:
|
|
os.write(master_fd, data.encode('utf-8'))
|
|
except OSError as e:
|
|
print(f"Error writing to PTY: {e}")
|
|
break
|
|
|
|
# Check if process is still alive
|
|
if shell_process.poll() is not None:
|
|
break
|
|
|
|
except Exception as e:
|
|
print(f"Terminal session error: {e}")
|
|
finally:
|
|
# Cleanup
|
|
try:
|
|
shell_process.terminate()
|
|
shell_process.wait(timeout=1)
|
|
except:
|
|
try:
|
|
shell_process.kill()
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
os.close(master_fd)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
os.close(slave_fd)
|
|
except:
|
|
pass
|
|
|
|
if session_id in active_sessions:
|
|
del active_sessions[session_id]
|
|
|
|
@sock.route('/ws/script/<session_id>')
|
|
def script_websocket(ws, session_id):
|
|
"""WebSocket endpoint for executing scripts with hybrid web mode"""
|
|
|
|
try:
|
|
init_data = ws.receive(timeout=10)
|
|
|
|
if not init_data:
|
|
error_msg = '{"type": "error", "message": "No script data received"}\r\n'
|
|
ws.send(error_msg)
|
|
return
|
|
|
|
script_data = json.loads(init_data)
|
|
|
|
script_path = script_data.get('script_path')
|
|
params = script_data.get('params', {})
|
|
|
|
if not script_path:
|
|
error_msg = '{"type": "error", "message": "No script_path provided"}\r\n'
|
|
ws.send(error_msg)
|
|
return
|
|
|
|
if not os.path.exists(script_path):
|
|
error_msg = f'{{"type": "error", "message": "Script not found: {script_path}"}}\r\n'
|
|
ws.send(error_msg)
|
|
return
|
|
|
|
except Exception as e:
|
|
error_msg = f'{{"type": "error", "message": "Invalid init data: {str(e)}"}}\r\n'
|
|
ws.send(error_msg)
|
|
return
|
|
|
|
web_log_fd, web_log_path = tempfile.mkstemp(suffix='.log', prefix='proxmenux_web_')
|
|
|
|
# Create pseudo-terminal for script execution
|
|
master_fd, slave_fd = pty.openpty()
|
|
|
|
env = os.environ.copy()
|
|
env['EXECUTION_MODE'] = 'web'
|
|
env['WEB_LOG'] = web_log_path
|
|
for key, value in params.items():
|
|
env[key] = str(value)
|
|
env['PYTHONUNBUFFERED'] = '1'
|
|
env['TERM'] = 'xterm-256color'
|
|
|
|
script_process = subprocess.Popen(
|
|
['/bin/bash', script_path],
|
|
stdin=slave_fd,
|
|
stdout=slave_fd,
|
|
stderr=slave_fd,
|
|
preexec_fn=os.setsid,
|
|
env=env
|
|
)
|
|
|
|
# Set non-blocking mode for master_fd
|
|
flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
|
|
fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
|
|
|
|
# Set terminal size
|
|
set_winsize(master_fd, 30, 120)
|
|
|
|
def monitor_web_log():
|
|
last_position = 0
|
|
|
|
while script_process.poll() is None:
|
|
try:
|
|
if os.path.exists(web_log_path):
|
|
with open(web_log_path, 'r') as f:
|
|
f.seek(last_position)
|
|
new_lines = f.readlines()
|
|
last_position = f.tell()
|
|
|
|
for line in new_lines:
|
|
line = line.strip()
|
|
if line.startswith('WEB_INTERACTION:'):
|
|
try:
|
|
# Parse: WEB_INTERACTION:type:id:title_b64:message_b64[:options_json]
|
|
parts = line[16:].split(':', 4)
|
|
interaction_type = parts[0]
|
|
interaction_id = parts[1]
|
|
title_b64 = parts[2]
|
|
message_b64 = parts[3]
|
|
|
|
title = base64.b64decode(title_b64).decode('utf-8')
|
|
message = base64.b64decode(message_b64).decode('utf-8')
|
|
|
|
interaction_data = {
|
|
'type': 'web_interaction',
|
|
'interaction': {
|
|
'type': interaction_type,
|
|
'id': interaction_id,
|
|
'title': title,
|
|
'message': message
|
|
}
|
|
}
|
|
|
|
# Parse options for menu
|
|
if interaction_type == 'menu' and len(parts) > 4:
|
|
options_json = parts[4]
|
|
interaction_data['interaction']['options'] = json.loads(options_json)
|
|
|
|
# Parse default for inputbox
|
|
if interaction_type == 'inputbox' and len(parts) > 4:
|
|
default_b64 = parts[4]
|
|
interaction_data['interaction']['default'] = base64.b64decode(default_b64).decode('utf-8')
|
|
|
|
# Send interaction to WebSocket
|
|
ws.send(json.dumps(interaction_data))
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
time.sleep(0.01)
|
|
except Exception as e:
|
|
break
|
|
|
|
web_log_thread = threading.Thread(target=monitor_web_log, daemon=True)
|
|
web_log_thread.start()
|
|
|
|
# Thread to read script output and forward to WebSocket
|
|
def read_script_output():
|
|
while True:
|
|
try:
|
|
r, _, _ = select.select([master_fd], [], [], 0.01)
|
|
if master_fd in r:
|
|
try:
|
|
data = os.read(master_fd, 4096)
|
|
if not data:
|
|
break
|
|
|
|
text = data.decode('utf-8', errors='ignore')
|
|
|
|
# Send raw text to terminal
|
|
try:
|
|
ws.send(text)
|
|
except Exception as e:
|
|
break
|
|
|
|
except OSError as e:
|
|
break
|
|
except Exception as e:
|
|
break
|
|
|
|
script_process.wait()
|
|
exit_code = script_process.returncode if script_process.returncode is not None else 0
|
|
|
|
try:
|
|
ws.send(f'\r\n[Script exited with code {exit_code}]\r\n')
|
|
except Exception as e:
|
|
pass
|
|
|
|
output_thread = threading.Thread(target=read_script_output, daemon=True)
|
|
output_thread.start()
|
|
|
|
try:
|
|
while True:
|
|
data = ws.receive(timeout=None)
|
|
|
|
if data is None:
|
|
break
|
|
|
|
try:
|
|
msg = json.loads(data)
|
|
|
|
if msg.get('type') == 'interaction_response':
|
|
interaction_id = msg.get('id')
|
|
value = msg.get('value')
|
|
|
|
# Write response to the file the script is waiting for
|
|
response_file = f"/tmp/proxmenux_response_{interaction_id}"
|
|
|
|
with open(response_file, 'w') as f:
|
|
f.write(value)
|
|
|
|
continue
|
|
|
|
# Handle resize
|
|
if msg.get('type') == 'resize':
|
|
cols = int(msg.get('cols', 120))
|
|
rows = int(msg.get('rows', 30))
|
|
set_winsize(master_fd, rows, cols)
|
|
continue
|
|
|
|
except json.JSONDecodeError:
|
|
# Raw text input, send to script
|
|
try:
|
|
os.write(master_fd, data.encode('utf-8'))
|
|
except OSError as e:
|
|
break
|
|
|
|
if script_process.poll() is not None:
|
|
break
|
|
|
|
except Exception as e:
|
|
pass
|
|
finally:
|
|
try:
|
|
script_process.terminate()
|
|
script_process.wait(timeout=1)
|
|
except:
|
|
try:
|
|
script_process.kill()
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
os.close(master_fd)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
os.close(slave_fd)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
os.close(web_log_fd)
|
|
os.unlink(web_log_path)
|
|
except:
|
|
pass
|
|
|
|
def init_terminal_routes(app):
|
|
"""Initialize terminal routes with Flask app"""
|
|
sock.init_app(app)
|
|
app.register_blueprint(terminal_bp)
|