| from gevent import monkey |
| |
| monkey.patch_all() |
|
|
| import os |
| import pty |
| import select |
| import struct |
| import fcntl |
| import termios |
| import signal |
| import shlex |
| from flask import Flask, render_template_string, request, redirect, url_for, session |
| from flask_socketio import SocketIO, emit, disconnect |
|
|
| app = Flask(__name__) |
| app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'secret_key_change_this_in_production') |
| app.config['ADMIN_USER'] = 'ikun' |
| app.config['ADMIN_PASS'] = os.environ.get('PASSWORD', 'ikun114514') |
|
|
| |
| socketio = SocketIO(app, cors_allowed_origins='*', async_mode='gevent') |
|
|
| user_sessions = {} |
|
|
| HTML_TEMPLATE = """ |
| <!DOCTYPE html> |
| <html lang="zh-CN"> |
| <head> |
| <meta charset="UTF-8"> |
| <meta name="viewport" content="width=device-width, initial-scale=1.0"> |
| <title>Web Terminal</title> |
| <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/xterm/3.14.5/xterm.min.css" /> |
| <style> |
| body { font-family: sans-serif; background: #2c3e50; display: flex; align-items: center; justify-content: center; height: 100vh; margin: 0; overflow: hidden;} |
| .login-box { background: white; padding: 2rem; border-radius: 8px; box-shadow: 0 4px 6px rgba(0,0,0,0.1); width: 300px; } |
| .terminal-container { width: 100%; height: 100%; background: black; padding: 10px; box-sizing: border-box; } |
| input { display: block; width: 100%; margin-bottom: 1rem; padding: 0.5rem; box-sizing: border-box; } |
| button { background: #3498db; color: white; border: none; padding: 0.5rem 1rem; cursor: pointer; border-radius: 4px; width: 100%; } |
| button:hover { background: #2980b9; } |
| .error { color: red; font-size: 0.9em; margin-bottom: 10px; } |
| |
| /* 状态指示器 */ |
| #status-overlay { |
| position: absolute; |
| top: 10px; |
| right: 10px; |
| background: rgba(255, 255, 255, 0.8); |
| padding: 5px 10px; |
| border-radius: 4px; |
| font-size: 12px; |
| z-index: 1000; |
| pointer-events: none; |
| } |
| </style> |
| </head> |
| <body> |
| |
| {% if page == 'login' %} |
| <div class="login-box"> |
| <h2 style="text-align: center; margin-top: 0;">终端登录</h2> |
| {% if error %} |
| <div class="error">{{ error }}</div> |
| {% endif %} |
| <form method="POST"> |
| <label>用户名</label> |
| <input type="text" name="username" value="ikun" required> |
| <label>密码</label> |
| <input type="password" name="password" required placeholder="请输入密码"> |
| <button type="submit">连接</button> |
| </form> |
| </div> |
| {% else %} |
| <div id="status-overlay">初始化中...</div> |
| <div class="terminal-container" id="terminal"></div> |
| |
| <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.min.js"></script> |
| <script src="https://cdnjs.cloudflare.com/ajax/libs/xterm/3.14.5/xterm.min.js"></script> |
| <script src="https://cdnjs.cloudflare.com/ajax/libs/xterm/3.14.5/addons/fit/fit.min.js"></script> |
| <script> |
| Terminal.applyAddon(fit); |
| const term = new Terminal({ |
| cursorBlink: true, |
| fontSize: 14, |
| fontFamily: 'Menlo, Monaco, "Courier New", monospace', |
| theme: { background: '#1e1e1e' } |
| }); |
| |
| const container = document.getElementById('terminal'); |
| term.open(container); |
| term.fit(); |
| |
| const statusEl = document.getElementById('status-overlay'); |
| |
| const socket = io({ |
| transports: ['websocket', 'polling'], |
| path: '/socket.io' |
| }); |
| |
| socket.on('connect', function() { |
| statusEl.innerText = '🟢 已连接'; |
| statusEl.style.color = 'green'; |
| // 发送初始大小 |
| socket.emit('resize', { cols: term.cols, rows: term.rows }); |
| }); |
| |
| socket.on('connect_error', (error) => { |
| statusEl.innerText = '🔴 连接错误: ' + error; |
| statusEl.style.color = 'red'; |
| console.error('Socket error:', error); |
| }); |
| |
| socket.on('disconnect', (reason) => { |
| statusEl.innerText = '🔴 已断开: ' + reason; |
| statusEl.style.color = 'red'; |
| }); |
| |
| socket.on('pty-output', function(data) { |
| term.write(data); |
| }); |
| |
| term.on('data', function(data) { |
| socket.emit('pty-input', { input: data }); |
| }); |
| |
| window.onresize = function() { |
| term.fit(); |
| socket.emit('resize', { cols: term.cols, rows: term.rows }); |
| }; |
| </script> |
| {% endif %} |
| |
| </body> |
| </html> |
| """ |
|
|
| @app.route('/', methods=['GET', 'POST']) |
| def index(): |
| error = None |
| if request.method == 'POST': |
| username = request.form.get('username') |
| password = request.form.get('password') |
| |
| if username == app.config['ADMIN_USER'] and password == app.config['ADMIN_PASS']: |
| session['logged_in'] = True |
| return redirect(url_for('terminal')) |
| else: |
| error = '密码错误' |
| |
| return render_template_string(HTML_TEMPLATE, page='login', error=error) |
|
|
| @app.route('/terminal') |
| def terminal(): |
| if not session.get('logged_in'): |
| return redirect(url_for('index')) |
| return render_template_string(HTML_TEMPLATE, page='terminal') |
|
|
| |
|
|
| def set_winsize(fd, row, col, xpix=0, ypix=0): |
| try: |
| winsize = struct.pack("HHHH", row, col, xpix, ypix) |
| fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize) |
| except: |
| pass |
|
|
| def read_and_forward_pty_output(fd, sid): |
| """ |
| 2. 关键修复：使用 select 监听 fd，防止 os.read 阻塞 gevent 循环 |
| """ |
| max_read_bytes = 1024 * 20 |
| while True: |
| socketio.sleep(0.01) |
| if fd: |
| try: |
| |
| |
| r, w, x = select.select([fd], [], [], 0.1) |
| if fd in r: |
| output = os.read(fd, max_read_bytes).decode(errors='ignore') |
| if output: |
| socketio.emit('pty-output', output, room=sid) |
| else: |
| |
| break |
| except OSError: |
| break |
| except Exception as e: |
| |
| print(f"Read error: {e}") |
| break |
|
|
| @socketio.on('connect') |
| def connect(): |
| if not session.get('logged_in'): |
| return False |
|
|
| |
| pid, fd = pty.fork() |
| |
| if pid == 0: |
| |
| os.environ['TERM'] = 'xterm-256color' |
| cmd = ["/bin/bash"] |
| os.execvp(cmd[0], cmd) |
| else: |
| |
| user_sessions[request.sid] = {'fd': fd, 'pid': pid} |
| |
| socketio.start_background_task(target=read_and_forward_pty_output, fd=fd, sid=request.sid) |
|
|
| @socketio.on('disconnect') |
| def on_disconnect(): |
| sid = request.sid |
| if sid in user_sessions: |
| fd = user_sessions[sid]['fd'] |
| pid = user_sessions[sid]['pid'] |
| try: |
| os.close(fd) |
| os.kill(pid, signal.SIGKILL) |
| os.waitpid(pid, 0) |
| except OSError: |
| pass |
| del user_sessions[sid] |
|
|
| @socketio.on('pty-input') |
| def on_pty_input(data): |
| sid = request.sid |
| if sid in user_sessions: |
| fd = user_sessions[sid]['fd'] |
| try: |
| |
| os.write(fd, data['input'].encode()) |
| except OSError: |
| pass |
|
|
| @socketio.on('resize') |
| def on_resize(data): |
| sid = request.sid |
| if sid in user_sessions: |
| fd = user_sessions[sid]['fd'] |
| set_winsize(fd, data['rows'], data['cols']) |
|
|
| if __name__ == '__main__': |
| print("Running on port 7860...") |
| socketio.run(app, host='0.0.0.0', port=7860, debug=True) |
