""" Hubble 登录同步 — 同设备 mobile → hubble 数据搬运 (v1.3) ========================================================== v1.3 变化: - 修数据膨胀: 只 cp files/keva/ (账号数据 ~50M), 不 cp 整个 files/* (~640M) 每次执行前先 rm -rf 旧 keva 避免累积 - 拆分三个入口: run_hubble (同步+启动), run_hubble_sync (仅同步), run_hubble_launch (仅启动) 对外签名: run_hubble(raw, scratch, log, progress, stop_event) -> (ok, png, info) run_hubble_sync(raw, scratch, log, progress, stop_event) -> (ok, png, info) run_hubble_launch(raw, scratch, log, progress, stop_event) -> (ok, png, info) """ import os import sys import time _HERE = os.path.dirname(os.path.abspath(__file__)) _LIB = os.path.join(_HERE, "lib") if os.path.isdir(_LIB) and _LIB not in sys.path: sys.path.insert(0, _LIB) import paramiko # noqa: F401 from conn import Session SRC_PKG = "com.ss.android.ugc.aweme.mobile" DST_PKG = "com.ss.android.ugc.aweme.hubble" CERT_URL = ( "https%3A%2F%2Fwebcast.amemv.com%2Ffalcon%2Fwebcast_douyin%2F" "page%2Fcertification%2Findex.html" "%3Fenter_from%3Dlive_take%26hide_nav_bar%3D1%26__live_platform__%3Dwebcast" ) INTENT_TEMPLATE = ( "intent:#Intent;launchFlags=0x10000000;" "component={dst}/com.ss.android.ugc.aweme.live.LiveDummyActivity;" "B.hide_status_bar=false;i.intent_type=1;" "B.enable_transparent_background=false;" "i.bundle_web_view_background_color=-1;" "S.url={url};" "S.key_calling_context=com.ss.android.ugc.aweme.shortvideo.ui.scan.ScanNewActivity;" "l.ad_id=0;B.hide_nav_bar=false;B.show_progress=false;" "B.hide_more=true;B.bundle_no_hw_acceleration=false;" "B.disable_open_animate=false;end" ) POLL_INTERVAL = 5 POLL_TIMEOUT = 120 class HubbleError(RuntimeError): pass # ═══════════════════ 内部构建块 ═══════════════════ def _init_session(raw_input_text, log, progress, check_stop): """SSH + adb + root + 双包校验 + 关 multiprocess. 返回 sess.""" log("[1] 建立 SSH 隧道 + adb ready") progress(5) check_stop() try: sess = Session(raw_input_text, log=log, hard_timeout=480) except ValueError as e: raise HubbleError(str(e)) try: sess.start() except paramiko.AuthenticationException: raise HubbleError("SSH 认证失败: 用户名或密码错误") except Exception as e: raise HubbleError(f"建立连接失败: {e}") log(f" serial={sess.serial}") progress(10) who = sess.sh("id", check=False).strip() if "uid=0" not in who: raise HubbleError(f"需要 root, 当前: {who}") brand = sess.sh("getprop ro.product.brand", check=False).strip() model = sess.sh("getprop ro.product.model", check=False).strip() log(f" 设备: {brand} {model} ({who})") progress(12) log("[2] 校验双包 + 关 WebView 多进程") check_stop() src_path = sess.sh(f"pm path {SRC_PKG}", check=False).strip() dst_path = sess.sh(f"pm path {DST_PKG}", check=False).strip() if "package:" not in src_path: raise HubbleError(f"源包未安装: {SRC_PKG}") if "package:" not in dst_path: raise HubbleError(f"目标包未安装: {DST_PKG}") log(" 双包 OK") sess.sh("cmd webviewupdate disable-multiprocess", check=False) mp = sess.sh("dumpsys webviewupdate | grep 'Multiprocess enabled'", check=False).strip() log(f" {mp}") if "false" not in mp: raise HubbleError("cmd webviewupdate disable-multiprocess 未生效") progress(18) return sess def _do_sync(sess, log, progress, check_stop): """force-stop + 清旧 keva + cp keva + 权限 + 授权.""" dst_dir = f"/data/data/{DST_PKG}" src_dir = f"/data/data/{SRC_PKG}" log("[3] force-stop") check_stop() sess.sh(f"am force-stop {DST_PKG}", check=False) time.sleep(1) progress(25) log("[4] overlay cp keva (仅账号数据, 不删 hubble 自有条目)") check_stop() sess.sh(f"mkdir -p {dst_dir}/files/keva", check=False) # overlay: 只覆盖 mobile 有的, 保留 hubble 独有的 (plugin 配置等) sess.sh(f"cp -rf {src_dir}/files/keva/* {dst_dir}/files/keva/ 2>/dev/null; echo cp_done") sz = sess.sh(f"du -s {dst_dir}/files/keva 2>/dev/null | head -1", check=False).strip() log(f" keva overlay done ({sz})") progress(35) log("[5] chown + chcon") check_stop() dst_uidgid = sess.sh(f"stat -c '%u:%g' {dst_dir}").strip() dst_ctx = sess.sh(f"stat -c '%C' {dst_dir}").strip() log(f" uid:gid={dst_uidgid} ctx={dst_ctx}") sess.sh( f"chown {dst_uidgid} {dst_dir}/files 2>/dev/null; " f"chcon {dst_ctx} {dst_dir}/files 2>/dev/null; " f"chown -R {dst_uidgid} {dst_dir}/files/keva; " f"chcon -R {dst_ctx} {dst_dir}/files/keva", check=False, ) progress(40) log("[6] 授权") sess.sh(f"pm grant {DST_PKG} android.permission.POST_NOTIFICATIONS 2>/dev/null", check=False) progress(42) def _do_launch(sess, log, progress, check_stop, pct_start=45, pct_end=97): """暖机 + 轮询 plugin + 启动 intent + 验证. 返回 ok.""" log("[7] 暖机: force-stop → monkey launch") check_stop() sess.sh(f"am force-stop {DST_PKG}", check=False) time.sleep(1) sess.sh(f"monkey -p {DST_PKG} -c android.intent.category.LAUNCHER 1", check=False) time.sleep(5) pid = sess.sh( f"ps -ef | grep {DST_PKG} | grep -v grep | head -1 | awk '{{print $2}}'", check=False, ).strip() log(f" PID={pid}") progress(pct_start) # 轮询: pct_start → pct_start+35 poll_base = pct_start poll_range = 35 log(f"[8] 等待 live plugin 就绪 (最多 {POLL_TIMEOUT}s)") check_stop() elapsed = 0 plugin_ready = False while elapsed < POLL_TIMEOUT: time.sleep(POLL_INTERVAL) elapsed += POLL_INTERVAL check_stop() pct = poll_base + int(elapsed / POLL_TIMEOUT * poll_range) progress(min(pct, poll_base + poll_range - 1)) # .jar jar = sess.sh( f"ls /data/data/{DST_PKG}/files/.patchs/sc_m.l.live.plugin*.jar 2>/dev/null" " | grep -v '\\.tp' | head -1", check=False, timeout=10, ).strip() if jar and "No such" not in jar: log(f" [{elapsed}s] live plugin jar: {jar.split('/')[-1]}") plugin_ready = True break # .tp tp = sess.sh( f"ls /data/data/{DST_PKG}/files/.patchs/sc_m.l.live.plugin*.tp 2>/dev/null | head -1", check=False, timeout=10, ).strip() if tp and "No such" not in tp: if elapsed % 15 == 0: log(f" [{elapsed}s] plugin 下载中...") continue # TTPluginManager if pid: tt = sess.sh( f"logcat -d --pid={pid} 2>/dev/null | grep -c cr_TTPluginManager", check=False, timeout=10, ).strip() try: if int(tt) > 0: log(f" [{elapsed}s] cr_TTPluginManager OK") plugin_ready = True break except ValueError: pass if elapsed % 30 == 0: log(f" [{elapsed}s] 仍在等待...") if not plugin_ready: log(f" [{elapsed}s] 超时, 仍尝试启动") progress(poll_base + poll_range) # 启动 log("[9] 启动实名认证页 (不 force-stop)") check_stop() intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL) out = sess.sh(f'am start "{intent}"', check=False).strip() log(f" {out}") progress(poll_base + poll_range + 2) for i in range(15): check_stop() time.sleep(1) progress(min(poll_base + poll_range + 2 + i, pct_end)) # 验证 top = sess.sh( "dumpsys activity activities | grep topResumedActivity | head -1", check=False, ).strip() log(f" {top}") if "LiveDummyActivity" in top: log(" LiveDummyActivity 在前台 OK") else: log(" [注意] 不在前台, 请看截图确认") return True def _screenshot(sess, scratch_dir, log): ts_stamp = int(time.time()) stamp = f"{os.getpid()}_{ts_stamp}" sess.sh(f"screencap -p /sdcard/hubble_verify_{stamp}.png", check=False) verify_png = None if scratch_dir: verify_png = os.path.join(scratch_dir, f"hubble_verify_{sess.local_port}_{stamp}.png") try: sess.pull(f"/sdcard/hubble_verify_{stamp}.png", verify_png) except Exception as e: log(f" 截图拉取失败: {e}") verify_png = None sess.sh(f"rm -f /sdcard/hubble_verify_{stamp}.png 2>/dev/null; true", check=False) return verify_png # ═══════════════════ 三个公开入口 ═══════════════════ def run_hubble(raw_input_text, scratch_dir, log, progress, stop_event=None): """完整流程: 同步 + 暖机 + 启动.""" def check_stop(): if stop_event and stop_event.is_set(): raise HubbleError("用户已中断") if scratch_dir: try: os.makedirs(scratch_dir, exist_ok=True) except: pass sess = _init_session(raw_input_text, log, progress, check_stop) try: _do_sync(sess, log, progress, check_stop) _do_launch(sess, log, progress, check_stop, pct_start=45, pct_end=97) png = _screenshot(sess, scratch_dir, log) log("\n=== Hubble 同步+启动完成 ===") progress(100) return True, png, None finally: try: sess.close() except: pass def run_hubble_sync(raw_input_text, scratch_dir, log, progress, stop_event=None): """仅同步数据, 不暖机不启动.""" def check_stop(): if stop_event and stop_event.is_set(): raise HubbleError("用户已中断") if scratch_dir: try: os.makedirs(scratch_dir, exist_ok=True) except: pass sess = _init_session(raw_input_text, log, progress, check_stop) try: _do_sync(sess, log, progress, check_stop) log("\n=== Hubble 仅同步完成 ===") log("数据已搬运, 如需验证请单独执行「启动页面」.") progress(100) return True, None, None finally: try: sess.close() except: pass def _quick_launch(sess, log, progress, check_stop): """快速尝试: 直接 am start intent, 不暖机. 返回 True=成功 False=白屏需兜底.""" log("[快速] 尝试直接启动 (不暖机)...") progress(22) check_stop() # 先 force-stop 再启动 sess.sh(f"am force-stop {DST_PKG}", check=False) time.sleep(2) intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL) sess.sh(f'am start "{intent}"', check=False) log(" intent 已发送, 等待 15s...") progress(30) for i in range(15): check_stop() time.sleep(1) progress(30 + i) # 验证: top activity + WebView 日志 top = sess.sh( "dumpsys activity activities | grep topResumedActivity | head -1", check=False, ).strip() if "LiveDummyActivity" not in top: log(" [快速] Activity 不在前台, 需兜底") return False # 检查 WebView 实例是否创建 (cr_AwContents / cr_TTPlugin) pid = sess.sh( f"ps -ef | grep {DST_PKG} | grep -v grep | head -1 | awk '{{print $2}}'", check=False, ).strip() if pid: wv = sess.sh( f"logcat -d --pid={pid} 2>/dev/null | grep -iE 'cr_AwContents|cr_TTPlugin' | tail -1", check=False, timeout=10, ).strip() if wv: log(" [快速] WebView 已创建, 成功!") return True # Activity 在前台但 WebView 未检测到 — 可能白屏 log(" [快速] WebView 未检测到, 可能白屏, 进入兜底...") return False def run_hubble_launch(raw_input_text, scratch_dir, log, progress, stop_event=None): """先快速直接启动, 失败则暖机+轮询兜底.""" def check_stop(): if stop_event and stop_event.is_set(): raise HubbleError("用户已中断") if scratch_dir: try: os.makedirs(scratch_dir, exist_ok=True) except: pass sess = _init_session(raw_input_text, log, progress, check_stop) try: # 第一轮: 快速尝试 if _quick_launch(sess, log, progress, check_stop): progress(95) png = _screenshot(sess, scratch_dir, log) log("\n=== 快速启动成功 ===") progress(100) return True, png, None # 第二轮: 完整暖机兜底 log("[兜底] 进入完整暖机+轮询流程...") _do_launch(sess, log, progress, check_stop, pct_start=50, pct_end=97) png = _screenshot(sess, scratch_dir, log) log("\n=== Hubble 启动页面完成 (兜底) ===") progress(100) return True, png, None finally: try: sess.close() except: pass def run_hubble_quick(raw_input_text, scratch_dir, log, progress, stop_event=None): """快速模式: 只发 intent, 不暖机不轮询不验证.""" def check_stop(): if stop_event and stop_event.is_set(): raise HubbleError("用户已中断") if scratch_dir: try: os.makedirs(scratch_dir, exist_ok=True) except: pass sess = _init_session(raw_input_text, log, progress, check_stop) try: log("[快速] force-stop + am start intent") progress(30) sess.sh(f"am force-stop {DST_PKG}", check=False) time.sleep(1) intent = INTENT_TEMPLATE.format(dst=DST_PKG, url=CERT_URL) sess.sh(f'am start "{intent}"', check=False) log(" intent 已发送") progress(60) log("[快速] 等待 10s...") for i in range(10): check_stop() time.sleep(1) progress(60 + i * 3) png = _screenshot(sess, scratch_dir, log) log("\n=== 快速启动完成 ===") progress(100) return True, png, None finally: try: sess.close() except: pass