from gevent import monkey
# 1. 关键修复：必须在导入 socket, os 等其他库之前打补丁
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')
# 切换为 gevent 模式
socketio = SocketIO(app, cors_allowed_origins='*', async_mode='gevent')
user_sessions = {}
HTML_TEMPLATE = """
Web Terminal
{% if page == 'login' %}
终端登录
{% if error %}
{{ error }}
{% endif %}
{% else %}
初始化中...
{% endif %}
"""
@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')
# --- SocketIO 逻辑 ---
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) # 让出控制权给其他 gevent 协程
if fd:
try:
# select 会等待 fd 变为可读，超时时间 0.1 秒
# 这样不会一直卡在 read 上，允许 loop 处理其他事件
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:
# EOF
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:
# 写入输入到 PTY
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)