import json import shlex import time import hashlib import hmac import docker from flask import Flask, jsonify, request from flask_socketio import SocketIO # 全局签名密钥，前后端统一，生产环境使用环境变量注入 SIGN_SECRET = b"DockerAdminPlatform2026SecretKey123456789" # 签名有效期 300秒（5分钟） SIGN_EXPIRE = 300 # 兼容新旧docker-py客户端 try: cli = docker.DockerClient(base_url='unix:///var/run/docker.sock') api_cli = docker.APIClient(base_url='unix:///var/run/docker.sock') except AttributeError: cli = docker.Client(base_url='unix:///var/run/docker.sock') api_cli = docker.APIClient(base_url='unix:///var/run/docker.sock') app = Flask(__name__) app.config['SECRET_KEY'] = 'docker-terminal-secret-2026' # WebSocket心跳保活，防止Nginx/反向代理断连 socketio = SocketIO( app, cors_allowed_origins="*", async_mode="threading", ping_timeout=60, ping_interval=25 ) # 全局会话存储 session_exec_map = {} # sid => TTY终端流信息 log_task_map = {} # sid => {"cid": "", "running": True} ws_user_map = {} # 新增：sid 映射 WebSocket鉴权用户信息 registry_config = { "repo_name": "", "repo_addr": "", "repo_user": "", "repo_pwd": "" } # 多系统shell兼容列表（Ubuntu/Alpine/标准Linux） SHELL_CANDIDATES = [ "/bin/bash", "/usr/bin/bash", "/bin/sh", "/bin/ash" ] # ===================== 工具函数1：docker socket兼容recv ===================== def docker_recv(sock, size=1024): try: return sock.recv(size) except Exception: try: return sock._sock.recv(size) except Exception: return b"" # ===================== 工具函数2：操作审计日志（终端输入脱敏，不存明文密码） ===================== def write_audit_log(user, role, action, target_name, target_id, result, client_ip): """企业审计记录，可对接数据库docker_audit_log表""" # 终端输入仅标记事件，不记录明文命令，规避密码泄露风险 if action == "terminal_input": result = "command_entered" log_item = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "user": user, "role": role, "action": action, "container_name": target_name, "container_id": target_id, "result": result, "client_ip": client_ip } # 生产环境替换为mysql insert语句 print(json.dumps(log_item, ensure_ascii=False)) # ===================== 工具函数3：HTTP Header HMAC签名鉴权（原有，HTTP接口专用） ===================== def verify_auth_sign(): """ 校验前端PHP生成的HMAC签名，防止伪造X-Role:admin 请求Header要求： X-User: 用户名 X-Role: admin/user/guest X-Time: 当前时间戳（秒） X-Sign: hmac-sha256签名 返回 (user, role, client_ip, True/False) """ user = request.headers.get("X-User", "anonymous") role = request.headers.get("X-Role", "guest") ts_str = request.headers.get("X-Time", "0") sign = request.headers.get("X-Sign", "") client_ip = request.headers.get("X-Client-Ip", request.remote_addr) try: ts = int(ts_str) except ValueError: return user, role, client_ip, False # 校验签名有效期 now = int(time.time()) if abs(now - ts) > SIGN_EXPIRE: return user, role, client_ip, False # 校验HMAC签名 raw_data = f"{user}|{role}|{ts}".encode("utf-8") calc_sign = hmac.new(SIGN_SECRET, raw_data, hashlib.sha256).hexdigest() if calc_sign != sign: return user, role, client_ip, False return user, role, client_ip, True # ===================== 新增工具函数4：WebSocket连接专用鉴权函数 ===================== def verify_socket_auth(auth): user = auth.get("user", "") role = auth.get("role", "") ts_str = auth.get("ts", "0") sign = auth.get("sign", "") try: ts = int(ts_str) except: return user, role, "", False if abs(int(time.time()) - ts) > SIGN_EXPIRE: return user, role, "", False raw_data = f"{user}|{role}|{ts}".encode("utf-8") calc_sign = hmac.new( SIGN_SECRET, raw_data, hashlib.sha256 ).hexdigest() if calc_sign != sign: return user, role, "", False return user, role, "", True # ===================== 工具函数5：环境变量脱敏，屏蔽密码/密钥 ===================== def safe_env_filter(env_list): safe_env = [] sensitive_keywords = ["PASS", "PASSWORD", "TOKEN", "SECRET", "KEY", "CREDENTIAL"] for item in env_list: equal_pos = item.find("=") if equal_pos == -1: safe_env.append(item) continue key = item[:equal_pos] val = item[equal_pos+1:] if any(word in key.upper() for word in sensitive_keywords): safe_env.append(f"{key}=******") else: safe_env.append(item) return safe_env # ===================== 全部HTTP业务接口 ===================== # 1. 容器列表（端口预拼接字符串，前端直接渲染，无需formatPorts） @app.route("/list") def api_list(): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "list_container", "", "", "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) arr = [] for c in cli.containers.list(all=True): img_name = c.image.tags[0] if c.image.tags else "" port_str_arr = [] ports_raw = c.ports for c_port, host_list in ports_raw.items(): if not host_list: continue for host_item in host_list: host_port = host_item.get("HostPort", "") port_str_arr.append(f"{host_port}:{c_port.split('/')[0]}") port_display = ",".join(port_str_arr) if port_str_arr else "-" arr.append({ "id": c.short_id, "full_id": c.id, "name": c.name, "image": img_name, "status": c.status, "ports": port_display }) write_audit_log(user, role, "list_container", "", "", "success", ip) return jsonify({"code":0,"msg":"success","list":arr}) # 2. 镜像列表 @app.route("/image_list") def api_image_list(): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "list_image", "", "", "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) arr = [] for img in cli.images.list(): if not img.tags: continue full_tag = img.tags[0] repo, tag = full_tag.split(":", 1) size_mb = round(img.attrs["Size"] / 1024 / 1024, 1) arr.append({ "repo": repo, "tag": tag, "id": img.short_id, "full_id": img.id, "size": f"{size_mb}MB", "created": img.attrs["Created"] }) write_audit_log(user, role, "list_image", "", "", "success", ip) return jsonify({"code":0,"msg":"success","list":arr}) # 3. 删除镜像 @app.route("/remove_image/") def api_remove_image(image_id): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "remove_image", "", image_id, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "remove_image", "", image_id, "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可删除镜像"}) try: img = cli.images.get(image_id) cli.images.remove(image_id, force=True) write_audit_log(user, role, "remove_image", img.tags[0] if img.tags else "", image_id, "success", ip) return jsonify({"code":0,"msg":"镜像已强制删除"}) except Exception as e: write_audit_log(user, role, "remove_image", "", image_id, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 4. 重启容器 @app.route("/restart/") def api_restart(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "restart", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "restart", "", cid, "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可操作容器"}) try: container = cli.containers.get(cid) container.restart() write_audit_log(user, role, "restart", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"容器已重启"}) except Exception as e: write_audit_log(user, role, "restart", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 5. 停止容器 @app.route("/stop/") def api_stop(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "stop", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "stop", "", cid, "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可操作容器"}) try: container = cli.containers.get(cid) container.stop() write_audit_log(user, role, "stop", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"容器已停止"}) except Exception as e: write_audit_log(user, role, "stop", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 6. 删除容器 @app.route("/rm/") def api_rm(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "rm", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "rm", "", cid, "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可删除容器"}) try: container = cli.containers.get(cid) container.remove(force=True) write_audit_log(user, role, "rm", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"容器已强制删除"}) except Exception as e: write_audit_log(user, role, "rm", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 7. 一次性拉取日志（兼容旧页面单次查看） @app.route("/logs//") def api_logs(cid, tail_lines): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "read_log", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) try: container = cli.containers.get(cid) log_bytes = container.logs(tail=tail_lines) log_str = log_bytes.decode("utf-8", errors="replace") write_audit_log(user, role, "read_log", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"success","data":log_str}) except Exception as e: write_audit_log(user, role, "read_log", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e),"data":""}) # 8. 容器网络详情 @app.route("/network/") def api_network(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "read_network", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) try: container = cli.containers.get(cid) info = container.attrs["NetworkSettings"]["Networks"] write_audit_log(user, role, "read_network", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"success","data":json.dumps(info,indent=2)}) except Exception as e: write_audit_log(user, role, "read_network", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e),"data":""}) # 9. 短连接单次执行命令（旧弹窗兼容，非实时终端） @app.route("/exec/", methods=["POST"]) def api_exec(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "single_exec", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "single_exec", "", cid, "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可执行命令"}) try: data = request.get_json() cmd = data.get("cmd", "") if not cmd: return jsonify({"code":400,"msg":"命令不能为空","data":""}) cmd_list = shlex.split(cmd) container = cli.containers.get(cid) exec_obj = container.exec_run(cmd_list, stdout=True, stderr=True) output = exec_obj.output.decode("utf-8", errors="replace") write_audit_log(user, role, "single_exec", container.name, cid, "command_executed", ip) return jsonify({"code":0,"msg":"success","data":output}) except Exception as e: write_audit_log(user, role, "single_exec", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e),"data":""}) # 10. 创建容器（端口标准格式 + 前端可配置CPU/内存） @app.route("/create", methods=["POST"]) def api_create(): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "create_container", "", "", "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "create_container", "", "", "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可创建容器"}) try: data = request.get_json() name = data.get("name", "") image = data.get("image") cmd = data.get("cmd") ports_input = data.get("ports", {}) # 修复：docker标准端口字典格式 {内部端口:宿主机端口} port_bind = {} for inner_port, host_port in ports_input.items(): port_bind[f"{inner_port}/tcp"] = int(host_port) # 前端可配置资源限制，默认1核512M cpu_core = int(data.get("cpu", 1)) memory_limit = data.get("memory", "512m") nano_cpus = cpu_core * 1000000000 container = cli.containers.run( image=image, name=name if name else None, command=cmd if cmd else None, ports=port_bind, detach=True, restart_policy={"Name": "always"}, mem_limit=memory_limit, nano_cpus=nano_cpus ) write_audit_log(user, role, "create_container", container.name, container.id, "success", ip) return jsonify({"code":0,"msg":f"容器创建成功，ID:{container.short_id}"}) except Exception as e: write_audit_log(user, role, "create_container", "", "", f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 11. 登录私有镜像仓库 @app.route("/bind_repo", methods=["POST"]) def api_bind_repo(): global registry_config user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "repo_login", "", "", "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) if role != "admin": write_audit_log(user, role, "repo_login", "", "", "denied", ip) return jsonify({"code":403,"msg":"权限不足，仅管理员可登录镜像仓库"}) try: data = request.get_json() repo_name = data.get("repo_name","") repo_addr = data.get("repo_addr") repo_user = data.get("repo_user","") repo_pwd = data.get("repo_pwd","") api_cli.login(username=repo_user, password=repo_pwd, registry=repo_addr) registry_config["repo_name"] = repo_name registry_config["repo_addr"] = repo_addr registry_config["repo_user"] = repo_user registry_config["repo_pwd"] = repo_pwd write_audit_log(user, role, "repo_login", repo_addr, "", "success", ip) return jsonify({"code":0,"msg":"镜像仓库登录绑定成功"}) except Exception as e: write_audit_log(user, role, "repo_login", repo_addr, "", f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e)}) # 修复：stats接口online_cpus兜底兼容，无KeyError @app.route("/stats/") def api_stats(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "read_stats", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) try: container = cli.containers.get(cid) raw = container.stats(stream=False) # CPU兼容兜底，不存在online_cpus自动计算 cpu_stats = raw["cpu_stats"] online_cpus = cpu_stats.get("online_cpus", len(cpu_stats["cpu_usage"].get("percpu_usage", [])) or 1) cpu_delta = cpu_stats["cpu_usage"]["total_usage"] - raw["precpu_stats"]["cpu_usage"]["total_usage"] system_delta = cpu_stats["system_cpu_usage"] - raw["precpu_stats"]["system_cpu_usage"] cpu_percent = 0.0 if system_delta > 0 and cpu_delta > 0: cpu_percent = round((cpu_delta / system_delta) * online_cpus * 100, 1) # 内存计算 mem_usage = raw["memory_stats"]["usage"] / 1024 / 1024 mem_limit = raw["memory_stats"]["limit"] / 1024 / 1024 mem_percent = round((mem_usage / mem_limit) * 100, 1) if mem_limit > 0 else 0 # 网络流量 net_rx = 0 net_tx = 0 for net in raw["networks"].values(): net_rx += net["rx_bytes"] / 1024 / 1024 net_tx += net["tx_bytes"] / 1024 / 1024 res = { "cpu": cpu_percent, "mem_mb": round(mem_usage, 1), "mem_limit_mb": round(mem_limit, 1), "mem_percent": mem_percent, "net_rx_mb": round(net_rx, 1), "net_tx_mb": round(net_tx, 1) } write_audit_log(user, role, "read_stats", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"success","data":res}) except Exception as e: write_audit_log(user, role, "read_stats", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e),"data":{}}) # 修复：detail接口ENV脱敏，多网卡IP兼容 @app.route("/detail/") def api_detail(cid): user, role, ip, auth_ok = verify_auth_sign() if not auth_ok: write_audit_log(user, role, "read_detail", "", cid, "auth_failed", ip) return jsonify({"code":403,"msg":"身份校验失败，非法访问"}) try: container = cli.containers.get(cid) attrs = container.attrs nets = attrs["NetworkSettings"]["Networks"] ip_map = {} for net_name, net_info in nets.items(): ip_map[net_name] = net_info.get("IPAddress", "") # 敏感环境变量脱敏 raw_env = attrs["Config"].get("Env", []) safe_env = safe_env_filter(raw_env) res = { "network_ips": ip_map, "mounts": attrs["Mounts"], "env": safe_env, "created": attrs["Created"], "entrypoint": attrs["Config"]["Entrypoint"], "cmd": attrs["Config"]["Cmd"], "restart_policy": attrs["HostConfig"]["RestartPolicy"] } write_audit_log(user, role, "read_detail", container.name, cid, "success", ip) return jsonify({"code":0,"msg":"success","data":res}) except Exception as e: write_audit_log(user, role, "read_detail", "", cid, f"fail:{str(e)}", ip) return jsonify({"code":500,"msg":str(e),"data":{}}) # 根路由 @app.route("/") def index(): return jsonify({"code":400,"msg":"无效操作","list":[],"data":""}) # ===================== WebSocket 连接鉴权事件（新增） ===================== @socketio.on("connect") def handle_connect(auth): user, role, _, auth_ok = verify_socket_auth(auth) if not auth_ok: return False ws_user_map[request.sid] = { "user": user, "role": role } print("WS Connected:", user, role) # ===================== WebSocket 交互式TTY实时终端 ===================== def stream_terminal_output(sid, sock): """后台异步读取docker终端输出，兼容两种socket接收方式""" try: while True: chunk = docker_recv(sock, 1024) if not chunk: break text = chunk.decode("utf-8", errors="replace") socketio.emit("terminal_output", {"data": text}, room=sid) except Exception: pass # 流结束自动清理会话 if sid in session_exec_map: try: session_exec_map[sid]["socket"].close() except Exception: pass del session_exec_map[sid] @socketio.on("terminal_start") def handle_terminal_start(json_data): sid = request.sid # 新增调试输出 socketio.emit( "terminal_output", { "data":"[DEBUG] terminal_start reached\r\n" }, room=sid ) # 替换原verify_auth_sign()，从ws_user_map读取身份 info = ws_user_map.get(request.sid) if not info: socketio.emit( "terminal_output", {"data":"身份认证失效\n"}, room=request.sid ) return user = info["user"] role = info["role"] client_ip = request.remote_addr # 权限拦截：非管理员禁止打开交互式终端 if role != "admin": socketio.emit("terminal_output", {"data": "Permission denied: 仅管理员可访问交互式终端\n"}, room=sid) write_audit_log(user, role, "terminal_start", "", json_data.get("cid"), "denied", client_ip) return cid = json_data.get("cid") if not cid: socketio.emit("terminal_output", {"data": "错误：容器ID不能为空\n"}, room=sid) write_audit_log(user, role, "terminal_start", "", "", "empty_cid", client_ip) return # 自动探测容器可用shell，兼容全版本docker-py（exec_create返回dict取Id） target_shell = None for shell in SHELL_CANDIDATES: try: ret = api_cli.exec_create( container=cid, cmd=[shell, "-c", "echo ok"] ) test_exec_id = ret["Id"] test_stream = api_cli.exec_start(test_exec_id) test_stream.read() target_shell = shell break except Exception: continue if not target_shell: socketio.emit("terminal_output", {"data": "错误：容器内无可用shell(bash/sh/ash)\n"}, room=sid) write_audit_log(user, role, "terminal_start", "", cid, "no_shell", client_ip) return try: # 创建交互式TTY会话，统一提取Id兼容所有SDK版本 exec_ret = api_cli.exec_create( container=cid, cmd=target_shell, stdin=True, stdout=True, stderr=True, tty=True ) exec_id = exec_ret["Id"] sock = api_cli.exec_start(exec_id=exec_id, tty=True, socket=True, detach=False) session_exec_map[sid] = { "cid": cid, "exec_id": exec_id, "socket": sock } socketio.start_background_task(stream_terminal_output, sid, sock) container_name = cli.containers.get(cid).name write_audit_log(user, role, "terminal_start", container_name, cid, "success", client_ip) except Exception as e: socketio.emit("terminal_output", {"data": f"启动终端失败：{str(e)}\n"}, room=sid) write_audit_log(user, role, "terminal_start", "", cid, f"fail:{str(e)}", client_ip) @socketio.on("terminal_input") def handle_terminal_input(json_data): """前端输入字符写入容器标准输入，审计脱敏不记录明文""" sid = request.sid # 替换原verify_auth_sign() info = ws_user_map.get(request.sid) if not info: return user = info["user"] role = info["role"] client_ip = request.remote_addr if sid not in session_exec_map: socketio.emit("terminal_output", {"data": "终端会话已断开，请重新打开\n"}, room=sid) return input_text = json_data.get("data", "") sock = session_exec_map[sid]["socket"] sock.send(input_text.encode("utf-8")) cid = session_exec_map[sid]["cid"] # 审计仅标记事件，不再记录明文命令，规避密码泄露 write_audit_log(user, role, "terminal_input", "", cid, "command_entered", client_ip) @socketio.on("terminal_resize") def handle_terminal_resize(json_data): """前端xterm窗口大小自适应""" sid = request.sid if sid not in session_exec_map: return exec_id = session_exec_map[sid]["exec_id"] height = int(json_data.get("height", 24)) width = int(json_data.get("width", 80)) try: api_cli.exec_resize(exec_id=exec_id, height=height, width=width) except Exception: pass # 新增主动关闭终端事件（解决切换容器残留TTY） @socketio.on("terminal_close") def handle_terminal_close(): sid = request.sid info = ws_user_map.get(request.sid) if not info: return user = info["user"] role = info["role"] client_ip = request.remote_addr if sid in session_exec_map: cid = session_exec_map[sid]["cid"] try: session_exec_map[sid]["socket"].close() except Exception: pass del session_exec_map[sid] write_audit_log(user, role, "terminal_close", "", cid, "manual_close", client_ip) @socketio.on("disconnect") def handle_disconnect(): """前端断开连接，释放docker socket、清理会话，防止句柄泄漏""" sid = request.sid # 清理ws_user_map会话 if sid in ws_user_map: del ws_user_map[sid] info = ws_user_map.get(sid) if not info: user = "" role = "" client_ip = request.remote_addr else: user = info["user"] role = info["role"] client_ip = request.remote_addr # 清理TTY终端 if sid in session_exec_map: cid = session_exec_map[sid]["cid"] try: session_exec_map[sid]["socket"].close() except Exception: pass del session_exec_map[sid] write_audit_log(user, role, "terminal_close", "", cid, "disconnect", client_ip) # 清理实时日志标记，终止日志循环 if sid in log_task_map: del log_task_map[sid] # ===================== WebSocket 实时流式日志 docker logs -f（生成器主动释放） ===================== def stream_follow_log(sid, cid): generator = None try: container = cli.containers.get(cid) generator = container.logs(stream=True, follow=True) log_task_map[sid] = {"cid": cid, "running": True} for line in generator: task_info = log_task_map.get(sid, {}) if not task_info.get("running"): break txt = line.decode("utf-8", errors="replace") socketio.emit("log_output", {"data": txt}, room=sid) except Exception: pass finally: # 主动关闭日志生成器，释放docker daemon长连接 if generator is not None: generator.close() if sid in log_task_map: del log_task_map[sid] @socketio.on("log_start") def handle_log_start(json_data): sid = request.sid # 替换verify_auth_sign，读取ws_user_map info = ws_user_map.get(request.sid) if not info: socketio.emit("log_output", {"data": "身份校验失败"}, room=sid) write_audit_log("", "", "log_start", "", "", "auth_failed", request.remote_addr) return user = info["user"] role = info["role"] client_ip = request.remote_addr cid = json_data.get("cid") if not cid: socketio.emit("log_output", {"data": "容器ID不能为空"}, room=sid) write_audit_log(user, role, "log_start", "", "", "empty_cid", client_ip) return if role != "admin": socketio.emit("log_output", {"data": "权限不足，仅管理员可查看实时日志"}, room=sid) write_audit_log(user, role, "log_start", "", cid, "denied", client_ip) return socketio.start_background_task(stream_follow_log, sid, cid) container_name = cli.containers.get(cid).name write_audit_log(user, role, "log_start", container_name, cid, "success", client_ip) @socketio.on("log_stop") def handle_log_stop(): sid = request.sid info = ws_user_map.get(request.sid) if not info: return user = info["user"] role = info["role"] client_ip = request.remote_addr if sid in log_task_map: log_task_map[sid]["running"] = False cid = log_task_map[sid]["cid"] write_audit_log(user, role, "log_stop", "", cid, "manual_close", client_ip) # 启动入口（eventlet异步，必须socketio.run） if __name__ == "__main__": socketio.run(app, host="0.0.0.0", port=8090, debug=False, allow_unsafe_werkzeug=True)