's'
This commit is contained in:
156
_codex_full_pipeline.py
Normal file
156
_codex_full_pipeline.py
Normal file
@@ -0,0 +1,156 @@
|
||||
"""
|
||||
Windows 本机 -> SSH Ubuntu(接真机 adb):
|
||||
编译 gps 双包 -> 卸旧装新 -> adb 自动授权/启动 -> jadx 反编译重建主包+Xposed+(若有)原版
|
||||
-> 运行时截图、dumpsys、meminfo、logcat、su 下 Magisk/LSPosed 路径与应用私有文件枚举
|
||||
-> 可选 SFTP 拉回 analysis_pull/<时间戳>/ 到本仓库目录
|
||||
|
||||
做不到且未冒充能做的:在图形界面里替你点 LSPosed 勾选作用域(无稳定 adb API)。
|
||||
|
||||
环境变量:
|
||||
FL_SSH_PASS 或 fl_ssh_pass.txt
|
||||
FL_CODEX_PULL=0 关闭拉回 Windows
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import paramiko
|
||||
|
||||
from _gps_ssh_auth import ssh_password
|
||||
|
||||
HOST = "192.168.2.179"
|
||||
USER = "eric"
|
||||
GPS_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
REBUILT = os.path.join(GPS_ROOT, "rebuilt_app")
|
||||
TGZ = os.path.join(GPS_ROOT, "rebuilt_sync.tgz")
|
||||
REMOTE_TGZ = "/home/eric/rebuilt_sync.tgz"
|
||||
REMOTE_PROJ = "/home/eric/codex_apk_recovery/rebuilt_app"
|
||||
REMOTE_OUT = "/home/eric/gps_build_artifacts"
|
||||
|
||||
REMOTE_SCRIPT = f"""set -e
|
||||
export ANDROID_HOME="${{ANDROID_HOME:-$HOME/android-sdk}}"
|
||||
export CODEX_PIPELINE=1
|
||||
if [[ -d /usr/lib/jvm/java-17-openjdk-amd64 ]]; then export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64; fi
|
||||
cd /tmp
|
||||
rm -rf rebuilt_app
|
||||
tar xzf {REMOTE_TGZ}
|
||||
rm -rf {REMOTE_PROJ}
|
||||
mv rebuilt_app {REMOTE_PROJ}
|
||||
chmod +x {REMOTE_PROJ}/gradlew
|
||||
chmod +x {REMOTE_PROJ}/scripts/*.sh
|
||||
bash {REMOTE_PROJ}/scripts/ubuntu_gps_pipeline.sh
|
||||
"""
|
||||
|
||||
|
||||
def _is_remote_dir(sftp: paramiko.SFTPClient, rpath: str, st_mode: int) -> bool:
|
||||
if st_mode and stat.S_ISDIR(st_mode):
|
||||
return True
|
||||
try:
|
||||
sftp.listdir(rpath)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def sftp_get_dir(sftp: paramiko.SFTPClient, remote: str, local: str) -> None:
|
||||
try:
|
||||
os.makedirs(local, exist_ok=True)
|
||||
for item in sftp.listdir_attr(remote):
|
||||
name = item.filename
|
||||
if name in (".", ".."):
|
||||
continue
|
||||
rpath = f"{remote}/{name}"
|
||||
lpath = os.path.join(local, name)
|
||||
mode = int(item.st_mode or 0)
|
||||
if _is_remote_dir(sftp, rpath, mode):
|
||||
sftp_get_dir(sftp, rpath, lpath)
|
||||
else:
|
||||
try:
|
||||
sftp.get(rpath, lpath)
|
||||
except OSError:
|
||||
pass
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
pwd = ssh_password(GPS_ROOT)
|
||||
if not pwd:
|
||||
print("Set FL_SSH_PASS or fl_ssh_pass.txt in", GPS_ROOT, file=sys.stderr)
|
||||
sys.exit(2)
|
||||
if not os.path.isdir(REBUILT):
|
||||
print("Missing rebuilt_app", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
subprocess.run(
|
||||
[
|
||||
"tar",
|
||||
"-czf",
|
||||
TGZ,
|
||||
"--exclude=rebuilt_app/.gradle",
|
||||
"--exclude=rebuilt_app/app/build",
|
||||
"--exclude=rebuilt_app/build",
|
||||
"--exclude=rebuilt_app/fl_xposed/build",
|
||||
"rebuilt_app",
|
||||
],
|
||||
cwd=GPS_ROOT,
|
||||
check=True,
|
||||
)
|
||||
|
||||
transport = paramiko.Transport((HOST, 22))
|
||||
transport.connect(username=USER, password=pwd)
|
||||
sftp = paramiko.SFTPClient.from_transport(transport)
|
||||
assert sftp is not None
|
||||
sftp.put(TGZ, REMOTE_TGZ)
|
||||
sftp.close()
|
||||
|
||||
client = paramiko.SSHClient()
|
||||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||||
client.connect(HOST, username=USER, password=pwd, timeout=90)
|
||||
_stdin, stdout, stderr = client.exec_command(REMOTE_SCRIPT, get_pty=True, timeout=1800)
|
||||
for line in stdout:
|
||||
sys.stdout.write(line)
|
||||
err = stderr.read().decode(errors="replace")
|
||||
if err.strip():
|
||||
sys.stderr.write(err)
|
||||
code = stdout.channel.recv_exit_status()
|
||||
|
||||
pull = os.environ.get("FL_CODEX_PULL", "1").strip().lower() not in ("0", "false", "no", "off")
|
||||
if pull and code == 0:
|
||||
sftp = paramiko.SFTPClient.from_transport(transport)
|
||||
assert sftp is not None
|
||||
stamp = time.strftime("%Y%m%d_%H%M%S")
|
||||
pull_root = os.path.join(GPS_ROOT, "analysis_pull", stamp)
|
||||
os.makedirs(pull_root, exist_ok=True)
|
||||
sftp_get_dir(sftp, f"{REMOTE_OUT}/jadx", os.path.join(pull_root, "jadx"))
|
||||
try:
|
||||
names = sftp.listdir(REMOTE_OUT)
|
||||
except OSError:
|
||||
names = []
|
||||
dumps = sorted(x for x in names if x.startswith("runtime_dump_"))
|
||||
if dumps:
|
||||
sftp_get_dir(sftp, f"{REMOTE_OUT}/{dumps[-1]}", os.path.join(pull_root, dumps[-1]))
|
||||
for fn in ("fakelocation-gps-debug.apk", "FL-Xposed-gps-debug.apk"):
|
||||
try:
|
||||
sftp.get(f"{REMOTE_OUT}/{fn}", os.path.join(pull_root, fn))
|
||||
except OSError:
|
||||
pass
|
||||
sftp.close()
|
||||
print("Pulled ->", pull_root)
|
||||
|
||||
client.close()
|
||||
transport.close()
|
||||
sys.exit(code)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user