""" JumpServer elFinder 自动化上传脚本 通过 Playwright 浏览器自动化，登录 JumpServer， 打开目标资产的 elFinder 文件管理器，上传部署包。 用法: python jumpserver_elfinder_upload.py --file deploy-server/it-smart-desk-server-deploy.zip --remote /tmp/ python jumpserver_elfinder_upload.py -f deploy-server/it-smart-desk-server-deploy.zip -r /tmp/ --headless """ import sys, os, json, base64, time, pyotp, argparse from pathlib import Path # 路径配置 SCRIPT_DIR = Path(__file__).parent PROJECT_ROOT = SCRIPT_DIR.parent WORKBUDDY_DIR = Path(os.environ.get("WORKBUDDY_DIR", Path.home() / ".workbuddy")) SKILL_DIR = Path(os.environ.get("JP_SKILL_DIR", WORKBUDDY_DIR / "skills" / "jumpserver-automation")) CONFIG_PATH = SKILL_DIR / "config" / "jumpserver_config.json" OTP_SECRET_PATH = SKILL_DIR / "scripts" / "otp_secret.key" OUTPUT_DIR = SKILL_DIR / "scripts" / "webcli_output" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) TARGET_ASSET_NAME = "hz-oa-ai-g-dataquery-90-5-110" TARGET_IP = "10.90.5.110" def load_config(): """加载 JumpServer 配置（用户名 + Base64密码解码）""" with open(CONFIG_PATH, "r", encoding="utf-8") as f: config = json.load(f) password = base64.b64decode(config["password"]).decode("utf-8") return config["url"].rstrip("/"), config.get("username", "sxn"), password def load_otp_secret(): """加载 TOTP 密钥""" if OTP_SECRET_PATH.exists(): return OTP_SECRET_PATH.read_text(encoding="utf-8").strip() return None def screenshot(page, name): """截图到输出目录""" path = OUTPUT_DIR / name page.screenshot(path=str(path)) print(f" 📸 {name}") return str(path) def upload_file(local_file_path, remote_dir="/tmp/", headless=False): """ 通过 JumpServer elFinder Web 界面上传文件到远程服务器 流程: 1. Playwright 浏览器打开 → JumpServer 登录页 2. 用户名 + 密码登录 3. TOTP MFA 验证 4. 进入「我的资产」列表 5. 找到目标主机 → 点击「文件管理」按钮 6. 在新标签页加载 elFinder 文件管理器 7. 导航到目标目录 8. 通过文件上传 input 上传文件 9. 验证上传成功 """ local_file = Path(local_file_path) if not local_file.exists(): print(f"❌ 本地文件不存在: {local_file_path}") return False file_size_mb = local_file.stat().st_size / (1024 * 1024) print(f"📦 本地文件: {local_file_path} ({file_size_mb:.2f} MB)") print(f"📂 目标目录: {remote_dir}") url, username, password = load_config() otp_secret = load_otp_secret() if not otp_secret: print("❌ 找不到 otp_secret.key") return False otp_code = pyotp.TOTP(otp_secret).now() print(f"🔐 OTP: {otp_code}") from playwright.sync_api import sync_playwright with sync_playwright() as p: # 启动 Chrome 浏览器 print("🚀 启动浏览器...") try: browser = p.chromium.launch( channel="chrome", headless=headless, args=["--no-sandbox", "--disable-dev-shm-usage"] ) except Exception: browser = p.chromium.launch(headless=headless) context = browser.new_context( viewport={"width": 1280, "height": 900}, ignore_https_errors=True ) page = context.new_page() try: # ========== 步骤 1: 登录 ========== print("\n[1/7] 登录 JumpServer...") page.goto(f"{url}/users/login/", wait_until="networkidle", timeout=30000) page.wait_for_timeout(1500) page.fill('input[name="username"]', username, timeout=5000) page.fill('input[type="password"]', password, timeout=5000) page.click('button[type="submit"]', timeout=5000) print(" ✅ 已登录") # ========== 步骤 2: MFA 验证 ========== print("[2/7] MFA 验证...") page.wait_for_timeout(3000) content = page.content() if any(kw in content for kw in ["mfa_type", "MFA", "多因子", "验证码"]): print(f" 填写 OTP: {otp_code}") try: sel = page.locator('select[name="mfa_type"], select#mfa-select') if sel.count() > 0 and sel.first.input_value() != 'otp': sel.first.select_option('otp') except: pass code_in = page.locator('input[name="code"]') if code_in.count() > 0: code_in.fill(str(otp_code), timeout=5000) page.locator('#submit_button, button[type="submit"]').first.click(timeout=5000) print(" ✅ OTP 提交成功") # ========== 步骤 3: 进入资产列表 ========== print("[3/7] 进入「我的资产」...") page.wait_for_timeout(4000) page.get_by_text("我的资产").first.click(timeout=10000) page.wait_for_timeout(2000) screenshot(page, "upload_asset_list.png") print(f" 📍 URL: {page.url}") # ========== 步骤 4: 进入文件管理页面 ========== print(f"[4/7] 进入「文件管理」...") # 文件管理是左侧菜单项，不是资产行按钮！ fm_menu = page.get_by_text("文件管理").first if fm_menu.count() > 0: fm_menu.click(timeout=10000) page.wait_for_timeout(3000) print(f" ✅ 点击了左侧菜单「文件管理」") screenshot(page, "upload_file_manager_page.png") print(f" 📍 URL: {page.url}") else: print(" ❌ 未找到左侧菜单「文件管理」") # 备用方案：直接访问 elFinder URL elfinder_url = f"{url}/koko/elfinder/?target={TARGET_ASSET_NAME}&search=" page.goto(elfinder_url, wait_until="domcontentloaded", timeout=15000) page.wait_for_timeout(3000) print(f" ✅ 直接打开 elFinder: {elfinder_url}") # 在文件管理页面中选择目标资产（如果需要） page.wait_for_timeout(2000) page.wait_for_timeout(4000) screenshot(page, "upload_after_fm_click.png") # ========== 步骤 5: 等待 elFinder 加载 ========== print("[5/7] 加载 elFinder 文件管理器...") # 文件管理页面可能直接在当前页面加载 elFinder elfinder_page = page page.wait_for_timeout(5000) screenshot(elfinder_page, "upload_elfinder_loaded.png") # ========== 步骤 6: 导航到目标目录 ========== print(f"[6/7] 导航到目标目录: {remote_dir}") # elFinder 文件树导航 # 尝试通过 JS 直接操作 elFinder 实例 try: result = elfinder_page.evaluate(f""" () => {{ try {{ const elfinder = window.elfFinder || (window.jQuery && window.jQuery('#elfinder').data('elfinder')); if (elfinder) {{ elfinder.exec('open', '{remote_dir}'); return 'navigating to {remote_dir}'; }} return 'elfinder instance not found'; }} catch(e) {{ return 'JS error: ' + e.message; }} }} """) print(f" 📂 导航结果: {result}") except Exception as e: print(f" ⚠️ JS 导航异常: {e}") try: elfinder_page.keyboard.press("Control+l") elfinder_page.wait_for_timeout(500) elfinder_page.keyboard.type(remote_dir) elfinder_page.keyboard.press("Enter") elfinder_page.wait_for_timeout(2000) print(" 📂 键盘导航到目标目录") except Exception as e2: print(f" ⚠️ 键盘导航也失败了: {e2}") page.wait_for_timeout(2000) screenshot(elfinder_page, "upload_before_upload.png") # ========== 步骤 7: 上传文件 ========== print("[7/7] 上传文件...") uploaded = False upload_selectors = [ 'input[type="file"][name="upload[]"]', 'input[type="file"].elfinder-upload-file', '.elfinder-upload-dialog input[type="file"]', '#elfinder input[type="file"]', '.elfinder input[type="file"]', 'input[type="file"]' ] upload_input = None for sel in upload_selectors: try: el = elfinder_page.locator(sel).first if el.count() > 0: upload_input = el print(f" ✅ 找到上传控件: {sel}") break except: continue if upload_input: try: upload_btn = elfinder_page.locator( '.elfinder-toolbar-button:has-text("上传"), ' '.elfinder-buttonset .ui-button:has(.ui-icon-upload), ' 'button:has-text("Upload"), button:has-text("上传")' ).first if upload_btn.count() > 0: upload_btn.click(timeout=3000) elfinder_page.wait_for_timeout(1000) print(" 📤 点击了上传按钮") abs_path = str(local_file.absolute()) upload_input.set_input_files(abs_path, timeout=10000) print(f" ✅ 文件已选择: {abs_path}") wait_time = max(60, int(file_size_mb * 20)) print(f" ⏳ 等待上传完成（预计 {wait_time} 秒）...") for i in range(wait_time // 5): elfinder_page.wait_for_timeout(5000) try: progress_hidden = elfinder_page.evaluate(""" () => { const pbar = document.querySelector('.elfinder-progressbar, .elfinder-upload-progress'); return !pbar || pbar.style.display === 'none' || pbar.offsetParent === null; } """) if progress_hidden: print(f" ✅ 上传进度条已消失，上传可能完成") break except: pass if (i + 1) % 5 == 0: print(f" ⏳ 仍在上传... ({i * 5}s)") elfinder_page.wait_for_timeout(3000) screenshot(elfinder_page, "upload_complete.png") try: file_check = elfinder_page.evaluate(f""" () => {{ try {{ const fm = window.elfFinder || (window.jQuery && window.jQuery('#elfinder').data('elfinder')); if (fm) {{ const files = fm.cwd().files; for (const f of files) {{ if (f.name === '{local_file.name}') {{ return JSON.stringify({{name: f.name, size: f.size, found: true}}); }} }} }} return '{{"found": false}}'; }} catch(e) {{ return '{{"error": "' + e.message + '"}}'; }} }} """) print(f" 📋 文件验证: {file_check}") except Exception as e: print(f" ⚠️ 无法验证文件: {e}") uploaded = True print(" ✅ 上传完成！") except Exception as e: print(f" ❌ 上传失败: {e}") try: screenshot(elfinder_page, "upload_error.png") except: pass else: print(" ❌ 未找到上传控件，尝试 file chooser...") try: with elfinder_page.expect_file_chooser() as fc_info: elfinder_page.locator('body').click(position={"x": 100, "y": 100}) elfinder_page.wait_for_timeout(500) file_chooser = fc_info.value file_chooser.set_files(str(local_file.absolute())) elfinder_page.wait_for_timeout(3000) uploaded = True print(" ✅ 通过 file chooser 上传成功!") except Exception as e: print(f" ❌ 上传失败: {e}") if uploaded: print("\n🎉 上传流程完成！") return True else: print("\n❌ 上传未能完成，请检查截图") return False except Exception as e: print(f"\n💥 异常: {e}") try: screenshot(page, "upload_exception.png") except: pass return False finally: print("\n📸 截图保存在:", OUTPUT_DIR) browser.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description="JumpServer elFinder 文件上传工具") parser.add_argument("-f", "--file", required=True, help="本地文件路径") parser.add_argument("-r", "--remote", default="/tmp/", help="远程目标目录 (默认: /tmp/)") parser.add_argument("--headless", action="store_true", help="无头模式") args = parser.parse_args() success = upload_file(args.file, args.remote, args.headless) sys.exit(0 if success else 1)