openclaw/apk-reverse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: apk-reverse Agent Skill

Browse Source 
Progressive-disclosure skill for Android APK reverse engineering:
recon, ad removal, membership limits, dex surgical patching, repack/sign,
runtime data, Frida, environment setup, and a hard-won failure catalogue.
This commit is contained in:
inliver233
2026-09-19 21:25:35 +08:00
commit c66658b6b3
33 changed files with 3713 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/datastore_inject.py
+66
View File
@@ -0,0 +1,66 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""向 app 的 DataStore(video_preferences) 注入 ad_free_expires_at 大值，验证它能否抑制广告 UI。
Preferences 的 protobuf 结构：
PreferenceMap { map<string, Value> preferences = 1; }
entry: 0x0A <len(key)> <key> 0x12 <len(value)> <value>
Value : boolean=1(float wire), float=2, integer=3(varint), long=4(varint), string=5, ...
所以要写一个 longKey："ad_free_expires_at" -> Value{ long = 大值 }
Value 载荷 = 0x20 <varint(long)> (field 4, wire type 0 -> tag 0x20)
"""
import struct
import sys
def varint(n):
out = bytearray()
while True:
b = n & 0x7F
n >>= 7
if n:
out.append(b | 0x80)
else:
out.append(b)
return bytes(out)
def map_entry(key, value_payload):
"""编码 PreferenceMap 里的一个 map entry。
★关键（踩过坑）：map<string,Value> preferences = 1 是一个 map 字段，
编码时必须**两层** 0x0A：
外层: 0x0A <len(entry)> <- 字段1的 tag + 长度
内层: 0x0A <len(key)> <key> 0x12 <len(value)> <value>
只写内层（丢掉外层 tag+长度）会让 DataStore 反序列化失败，
app 直接 uncaughtException 崩溃，而且没有堆栈（被 UCrash 吞掉），极难定位。
"""
kb = key.encode('utf-8')
inner = (b'\x0a' + varint(len(kb)) + kb
+ b'\x12' + varint(len(value_payload)) + value_payload)
return b'\x0a' + varint(len(inner)) + inner
def entry(key, value_payload):
"""兼容旧调用：返回 map_entry。"""
return map_entry(key, value_payload)
def main():
# 保留已有键：player_page_open_count = 1 (int32, field 3 -> tag 0x18)
existing = entry('player_page_open_count', b'\x18\x01')
# 注入一个远超任何现实时间的时间戳（毫秒/秒都覆盖）
BIG = 9999999999999
injected = entry('ad_free_expires_at', b'\x20' + varint(BIG))
data = existing + injected
out = sys.argv[1] if len(sys.argv) > 1 else 'video_preferences.preferences_pb'
with open(out, 'wb') as f:
f.write(data)
print('[ok] wrote %s (%d bytes)' % (out, len(data)))
print('[hex] %s' % data.hex(' '))
print('[note] ad_free_expires_at = %d, varint=%s' % (BIG, varint(BIG).hex(' ')))
if __name__ == '__main__':
main()
scripts/devsh.py
+109
View File
@@ -0,0 +1,109 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Quoting-safe ADB helper.
Why this exists: if you build device commands inside a host shell, the host expands
`$`, `|`, `>`, and quotes *before* adb sees them. Symptoms are host-side errors that
look like device failures -- "Could not find a part of the path", "no closing quote",
"Missing type name after '['", unexpanded variables. Windows PowerShell is the worst
offender (it also eats `$var:`, `[^"...`, `$(...)`).
Calling adb from Python with an argument list avoids all of it.
Configuration (first match wins):
1. CLI flags: --serial / --adb
2. env: ADB_SERIAL / ADB_PATH
3. defaults below
Usage
-----
python devsh.py sh "pm list packages | grep myapp" # as the shell user
python devsh.py su "pm grant <pkg> android.permission.X"
python devsh.py sh "dumpsys window | grep mCurrentFocus"
python devsh.py pull /data/local/tmp/x.apk ./x.apk
python devsh.py push ./x.apk /data/local/tmp/x.apk
python devsh.py dev # list devices
`su` wraps the command in `su -c "..."`; internal double quotes are escaped.
"""
import argparse
import os
import subprocess
import sys
DEFAULT_ADB = 'adb'
DEFAULT_SERIAL = None
def resolve():
ap = argparse.ArgumentParser(add_help=False)
ap.add_argument('--adb')
ap.add_argument('--serial')
known, _ = ap.parse_known_args()
adb = known.adb or os.environ.get('ADB_PATH') or DEFAULT_ADB
serial = known.serial or os.environ.get('ADB_SERIAL') or DEFAULT_SERIAL
return adb, serial
def run(adb, serial, args, timeout=300):
cmd = [adb]
if serial:
cmd += ['-s', serial]
cmd += args
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace', timeout=timeout)
out = r.stdout or ''
if (r.stderr or '').strip():
out += '\n[stderr]\n' + r.stderr
return r.returncode, out
def main():
if len(sys.argv) < 2:
print(__doc__)
return 2
adb, serial = resolve()
argv = [a for a in sys.argv[1:] if not a.startswith('--adb=') and not a.startswith('--serial=')]
# strip flag/value pairs
cleaned = []
skip = False
for a in argv:
if skip:
skip = False
continue
if a in ('--adb', '--serial'):
skip = True
continue
cleaned.append(a)
argv = cleaned
if len(argv) < 2 and argv and argv[0] not in ('dev',):
print(__doc__)
return 2
mode = argv[0]
if mode == 'dev':
rc, out = run(adb, None, ['devices', '-l'])
elif mode in ('sh', 'shell'):
rc, out = run(adb, serial, ['shell', argv[1]])
elif mode == 'su':
cmd = argv[1].replace('"', '\\"')
rc, out = run(adb, serial, ['shell', 'su -c "%s"' % cmd])
elif mode == 'suf': # read a device text file
cmd = argv[1].replace('"', '\\"')
size = argv[2] if len(argv) > 2 else '6000'
rc, out = run(adb, serial, ['shell', 'su -c "head -c %s %s"' % (size, cmd)])
elif mode == 'pull':
rc, out = run(adb, serial, ['pull', argv[1], argv[2]])
elif mode == 'push':
rc, out = run(adb, serial, ['push', argv[1], argv[2]])
else:
print(__doc__)
return 2
sys.stdout.write(out)
return 0 if rc == 0 else 1
if __name__ == '__main__':
sys.exit(main())
scripts/dex_classdiff.py
+101
View File
@@ -0,0 +1,101 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""对比两个 dex 的 class_defs：类名 + access_flags（重点看 ACC_INTERFACE 0x200）。
用途：验证 baksmali -> smali 整树往返是否损伤了 dex 的 interface/class 关系。
ICCE "Found interface X, but class was expected" 就是这类损伤的典型症状。
用法：python dex_classdiff.py <a.dex> <b.dex> [过滤子串]
"""
import struct
import sys
def uleb128(data, off):
result = 0
shift = 0
while True:
b = data[off]
off += 1
result |= (b & 0x7F) << shift
if not (b & 0x80):
break
shift += 7
return result, off
def read_strings(data):
off = struct.unpack('<I', data[0x3C:0x40])[0]
size = struct.unpack('<I', data[0x38:0x40][:4])[0]
size = struct.unpack('<I', data[0x38:0x3C])[0]
out = []
for i in range(size):
sdata_off = struct.unpack('<I', data[off + i * 4: off + i * 4 + 4])[0]
n, p = uleb128(data, sdata_off)
raw = data[p:p + n]
try:
out.append(raw.decode('utf-8', 'replace'))
except Exception:
out.append('')
return out
def class_map(path):
data = open(path, 'rb').read()
strings = read_strings(data)
type_ids_off = struct.unpack('<I', data[0x44:0x48])[0]
n_types = struct.unpack('<I', data[0x40:0x44])[0]
types = []
for i in range(n_types):
idx = struct.unpack('<I', data[type_ids_off + i * 4: type_ids_off + i * 4 + 4])[0]
types.append(strings[idx] if idx < len(strings) else '')
class_defs_size = struct.unpack('<I', data[0x60:0x64])[0]
class_defs_off = struct.unpack('<I', data[0x64:0x68])[0]
out = {}
for i in range(class_defs_size):
base = class_defs_off + i * 32
class_idx, access_flags = struct.unpack('<II', data[base:base + 8])
name = types[class_idx] if class_idx < len(types) else '?'
out[name] = access_flags
return out
def main():
a, b = sys.argv[1], sys.argv[2]
filt = sys.argv[3] if len(sys.argv) > 3 else None
A = class_map(a)
B = class_map(b)
print('A classes=%d B classes=%d' % (len(A), len(B)))
only_a = sorted(set(A) - set(B))
only_b = sorted(set(B) - set(A))
print('only_in_A=%d only_in_B=%d' % (len(only_a), len(only_b)))
for n in only_a[:10]:
print(' A-only:', n)
for n in only_b[:10]:
print(' B-only:', n)
diff_iface = []
diff_flags = []
for name in sorted(set(A) & set(B)):
fa, fb = A[name], B[name]
ia, ib = bool(fa & 0x200), bool(fb & 0x200)
if ia != ib:
diff_iface.append((name, hex(fa), hex(fb)))
elif fa != fb:
diff_flags.append((name, hex(fa), hex(fb)))
print('\n*** ACC_INTERFACE mismatch: %d ***' % len(diff_iface))
for name, fa, fb in (diff_iface if not filt else [x for x in diff_iface if filt in x[0]]):
print(' %-70s A=%s B=%s' % (name, fa, fb))
print('\naccess_flags diff (same interface-ness): %d' % len(diff_flags))
shown = 0
for name, fa, fb in diff_flags:
if filt and filt not in name:
continue
if shown < 25:
print(' %-70s A=%s B=%s' % (name, fa, fb))
shown += 1
if __name__ == '__main__':
main()
scripts/dex_strings.py
+146
View File
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Extract strings, URLs and vendor markers from dex files without a decompiler.
Fast recon: dex strings are stored as plain MUTF-8, so ASCII substrings can be found
by scanning the raw bytes. This answers most classification questions in seconds --
which dex holds the app's code, which ad SDKs ship with it, what endpoints it calls.
Usage
-----
# all URLs across a directory of dex files
python dex_strings.py <dex_dir> --urls
# find which dex files contain a marker (ASCII substring)
python dex_strings.py <dex_dir> --find 'openadsdk|com.qq.e|anythink' --per-file
# dump a class/package inventory per dex (type descriptors)
python dex_strings.py <dex_dir> --classes 'Lcom/example/app/'
# raw string-table dump, length-filtered
python dex_strings.py <dex_dir> --strings --min 8 --max 60
# compare two dex files' string tables (what changed after a patch)
python dex_strings.py --diff a.dex b.dex
Notes
-----
* A marker in the string table means the class/URL is *present*, not that the app
*uses* that feature. Confirm with runtime behavior before acting.
* `--urls` output is the fastest way to see an app's whole API surface and its
third-party endpoints at once.
"""
import argparse
import glob
import os
import re
import struct
import sys
URL_RE = re.compile(rb'https?://[A-Za-z0-9\.\-_:/%\.\?=&~#]{4,200}')
ASCII_RE = re.compile(rb'[\x20-\x7e]{4,200}')
def uleb128(data, off):
r = 0
s = 0
while True:
b = data[off]
off += 1
r |= (b & 0x7F) << s
if not (b & 0x80):
break
s += 7
return r, off
def dex_strings(path):
"""Yield the dex string table in table order."""
data = open(path, 'rb').read()
if data[:4] != b'dex\n':
return
n = struct.unpack('<I', data[0x38:0x3C])[0]
off = struct.unpack('<I', data[0x3C:0x40])[0]
for i in range(n):
sdata = struct.unpack('<I', data[off + i * 4: off + i * 4 + 4])[0]
ln, p = uleb128(data, sdata)
yield bytes(data[p:p + ln])
def iter_dex(target):
if os.path.isdir(target):
return sorted(glob.glob(os.path.join(target, '*.dex')))
return [target]
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument('target')
ap.add_argument('--urls', action='store_true')
ap.add_argument('--strings', action='store_true')
ap.add_argument('--classes', metavar='PREFIX')
ap.add_argument('--find', metavar='REGEX')
ap.add_argument('--per-file', action='store_true')
ap.add_argument('--diff', nargs=2, metavar=('A', 'B'))
ap.add_argument('--min', type=int, default=6)
ap.add_argument('--max', type=int, default=200)
a = ap.parse_args()
if a.diff:
A = set(dex_strings(a.diff[0]))
B = set(dex_strings(a.diff[1]))
only_a = sorted(x for x in A - B if a.min <= len(x) <= a.max)
only_b = sorted(x for x in B - A if a.min <= len(x) <= a.max)
print('A strings=%d B strings=%d' % (len(A), len(B)))
print('only in A: %d' % len(only_a))
for x in only_a[:40]:
print(' -', x.decode('utf-8', 'replace'))
print('only in B: %d' % len(only_b))
for x in only_b[:40]:
print(' +', x.decode('utf-8', 'replace'))
return 0
pat = re.compile(a.find.encode()) if a.find else None
files = iter_dex(a.target)
if a.find and a.per_file:
for fp in files:
data = open(fp, 'rb').read()
hits = len(pat.findall(data))
if hits:
print('%-22s hits=%d' % (os.path.basename(fp), hits))
return 0
seen = set()
for fp in files:
for raw in dex_strings(fp):
s = raw.decode('utf-8', 'replace')
if not (a.min <= len(s) <= a.max):
continue
if pat and not pat.search(s):
continue
if a.classes and not s.startswith(a.classes):
continue
if a.urls and not raw.startswith((b'http://', b'https://')):
continue
if s in seen:
continue
seen.add(s)
if a.urls:
print(s)
elif a.classes:
print('%-18s %s' % (os.path.basename(fp), s))
else:
print('%s' % s)
if a.urls and not seen:
# fall back to raw byte scan (covers URLs stored outside the string table)
print('[note] no URLs in string tables; raw scan:')
for fp in files:
for m in set(URL_RE.findall(open(fp, 'rb').read())):
print(os.path.basename(fp), m.decode('utf-8', 'replace'))
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/dex_strpatch.py
+117
View File
@@ -0,0 +1,117 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""字节级定点 patch dex 里的字符串常量，完全不做 smali 往返。
为什么不用 smali 往返：实测 classes7.dex 经 baksmali->smali 整树重建后，
`io.ktor.client.engine.HttpClientEngine` 的 synthetic access bridge 被破坏，
运行时报 IncompatibleClassChangeError（Found interface, but class was expected）。
class_def 的 access_flags 看不出问题，所以只能改用"只改字节"的方式。
本脚本做法（对 dex 结构零改动）：
1. 把目标字符串替换成**等长**的另一个字符串 —— 长度相同则 string_data_item
的 uleb128 长度前缀不变，dex 的偏移、索引、表全部不变；
2. 重算 dex header 的 signature(SHA-1, 从 offset 32 起) 与 checksum(adler32, 从 offset 12 起)。
用法：python dex_strpatch.py <in.dex> <out.dex> <old_str> <new_str>
要求 len(new) == len(old)，且 old 在 dex 中出现次数恰好为 1。
"""
import hashlib
import struct
import sys
import zlib
def _uleb128(data, off):
result = 0
shift = 0
while True:
b = data[off]
off += 1
result |= (b & 0x7F) << shift
if not (b & 0x80):
break
shift += 7
return result, off
def _read_strings(data):
off = struct.unpack('<I', data[0x3C:0x40])[0]
size = struct.unpack('<I', data[0x38:0x3C])[0]
out = []
for i in range(size):
sdata_off = struct.unpack('<I', data[off + i * 4: off + i * 4 + 4])[0]
n, p = _uleb128(data, sdata_off)
out.append((sdata_off, bytes(data[p:p + n])))
return out
def _order_ok(data, off, ob, nb):
"""检查把 ob 换成 nb 后是否仍满足 string_ids 的字典序约束。
注意：string_ids 里存的是 string_data_item 的偏移，也就是**长度前缀**的位置
（在 UTF-8 数据前若干字节），所以不能直接拿字符串数据偏移去匹配 ——
这里改为按内容匹配目标条目。
"""
entries = _read_strings(data)
idx = None
for i, (_sdata_off, raw) in enumerate(entries):
if raw == ob:
idx = i
break
if idx is None:
print('[order] target string not found in string_ids, skip check')
return True
prev = entries[idx - 1][1] if idx > 0 else None
nxt = entries[idx + 1][1] if idx + 1 < len(entries) else None
if prev is not None and nb <= prev:
print('[order] FAIL: new %r <= prev %r' % (nb, prev))
return False
if nxt is not None and nb >= nxt:
print('[order] FAIL: new %r >= next %r' % (nb, nxt))
return False
print('[order] OK: %r < %r < %r' % (prev, nb, nxt))
return True
def main():
src, dst, old, new = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]
ob, nb = old.encode('utf-8'), new.encode('utf-8')
if len(ob) != len(nb):
print('[FAIL] length mismatch: %d vs %d (must be equal)' % (len(ob), len(nb)))
return 1
data = bytearray(open(src, 'rb').read())
if data[:8] != b'dex\n035\x00' and data[:4] != b'dex\n':
print('[warn] unexpected magic: %r' % bytes(data[:8]))
n = data.count(ob)
print('[info] occurrences of %r = %d' % (old, n))
if n != 1:
print('[FAIL] expected exactly 1 occurrence')
return 1
off = data.find(ob)
# 关键：dex 规范要求 string_ids 表按字符串内容有序。等长替换虽然不动偏移，
# 但会改变该条目在表中的字典序位置 —— 一旦越界，整个 dex 会被 ClassLoader
# 拒绝（表现为 ClassNotFoundException: <Application>）。所以这里先做区间校验。
if not _order_ok(data, off, ob, nb):
print('[FAIL] replacement would break string_ids ordering. '
'Pick a string that stays between its two neighbours.')
return 1
data[off:off + len(ob)] = nb
print('[ok] patched at offset 0x%x: %r -> %r' % (off, old, new))
# signature = SHA-1 over data[32:], stored at 12..32
sig = hashlib.sha1(bytes(data[32:])).digest()
data[12:32] = sig
# checksum = adler32 over data[12:], stored at 8..12 (little endian)
chk = zlib.adler32(bytes(data[12:])) & 0xFFFFFFFF
data[8:12] = struct.pack('<I', chk)
print('[ok] signature=%s checksum=0x%08x' % (sig.hex(), chk))
open(dst, 'wb').write(bytes(data))
print('[ok] wrote %s (%d bytes)' % (dst, len(data)))
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/dexpatch/PatchMethod.java
+97
View File
@@ -0,0 +1,97 @@
import org.jf.dexlib2.DexFileFactory;
import org.jf.dexlib2.Opcode;
import org.jf.dexlib2.Opcodes;
import org.jf.dexlib2.iface.ClassDef;
import org.jf.dexlib2.iface.Method;
import org.jf.dexlib2.iface.MethodImplementation;
import org.jf.dexlib2.builder.MutableMethodImplementation;
import org.jf.dexlib2.immutable.ImmutableMethod;
import org.jf.dexlib2.writer.pool.DexPool;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
/**
* dexlib2 定点 patch：把指定方法的实现整体替换为 return-void（或 return 常量）。
*
* 为什么不用 baksmali/smali 整树往返：实测本项目的 classes7.dex 往返重建后运行时报
* IncompatibleClassChangeError: Found interface io.ktor.client.engine.HttpClientEngine,
* but class was expected
* （R8 生成的 synthetic access bridge 被破坏）。本工具只改目标方法的 code_item，
* 其余类/方法/字符串表原样保留，因此不会有这类损伤。
*
* 用法: java PatchMethod <in.dex> <out.dex> <descriptor> <methodName> <returnType>
* 例: java PatchMethod classes7.dex out.dex Lx6; d V
*/
public class PatchMethod {
public static void main(String[] args) throws Exception {
if (args.length < 5) {
System.err.println("usage: PatchMethod <in.dex> <out.dex> <descriptor> <methodName> <returnType>");
System.exit(2);
}
String inPath = args[0], outPath = args[1], desc = args[2], mName = args[3], ret = args[4];
org.jf.dexlib2.iface.DexFile dex =
DexFileFactory.loadDexFile(new File(inPath), Opcodes.forApi(34));
List<ClassDef> outClasses = new ArrayList<ClassDef>();
int patched = 0;
for (ClassDef cd : dex.getClasses()) {
List<Method> direct = new ArrayList<Method>();
for (Method m : cd.getDirectMethods()) direct.add(m);
List<Method> virtual = new ArrayList<Method>();
for (Method m : cd.getVirtualMethods()) virtual.add(m);
boolean isTargetClass = cd.getType().equals(desc);
boolean changed = false;
if (isTargetClass) {
for (int pass = 0; pass < 2; pass++) {
List<Method> list = (pass == 0) ? direct : virtual;
for (int i = 0; i < list.size(); i++) {
Method m = list.get(i);
if (!m.getName().equals(mName) || !m.getReturnType().equals(ret)) continue;
// 用 return-void 替换整个方法体。
// 寄存器数必须保留原值：Compose/协程方法体里可能仍被其它指令引用同一寄存器窗口，
// 且 dex 校验要求 registers >= 参数寄存器数。
int regs = m.getImplementation() != null ? m.getImplementation().getRegisterCount() : 0;
MethodImplementation impl = new org.jf.dexlib2.immutable.ImmutableMethodImplementation(
regs,
java.util.Collections.singletonList(
new org.jf.dexlib2.immutable.instruction.ImmutableInstruction10x(Opcode.RETURN_VOID)),
null,
null);
Method nm = new ImmutableMethod(
m.getDefiningClass(), m.getName(), m.getParameters(), m.getReturnType(),
m.getAccessFlags(), m.getAnnotations(), m.getHiddenApiRestrictions(),
impl);
list.set(i, nm);
patched++;
changed = true;
System.out.println("[patch] " + desc + "->" + mName + m.getReturnType()
+ " registers=" + regs);
}
}
}
if (!changed) {
outClasses.add(cd);
} else {
outClasses.add(new org.jf.dexlib2.immutable.ImmutableClassDef(
cd.getType(), cd.getAccessFlags(), cd.getSuperclass(),
cd.getInterfaces(), cd.getSourceFile(), cd.getAnnotations(),
cd.getStaticFields(), cd.getInstanceFields(), direct, virtual));
}
}
DexFileFactory.writeDexFile(outPath,
new org.jf.dexlib2.immutable.ImmutableDexFile(Opcodes.forApi(34), outClasses));
System.out.println("[done] patched=" + patched + " -> " + outPath);
if (patched == 0) {
System.err.println("[warn] no method matched!");
System.exit(1);
}
}
}
scripts/dexpatch/PatchMethodExample.java
+108
View File
@@ -0,0 +1,108 @@
import org.jf.dexlib2.DexFileFactory;
import org.jf.dexlib2.Opcode;
import org.jf.dexlib2.Opcodes;
import org.jf.dexlib2.iface.ClassDef;
import org.jf.dexlib2.iface.Method;
import org.jf.dexlib2.iface.instruction.Instruction;
import org.jf.dexlib2.immutable.ImmutableClassDef;
import org.jf.dexlib2.immutable.ImmutableDexFile;
import org.jf.dexlib2.immutable.ImmutableMethod;
import org.jf.dexlib2.immutable.ImmutableMethodImplementation;
import org.jf.dexlib2.immutable.instruction.ImmutableInstruction11x;
import org.jf.dexlib2.immutable.instruction.ImmutableInstruction22c;
import org.jf.dexlib2.immutable.instruction.ImmutableInstruction35c;
import org.jf.dexlib2.immutable.instruction.ImmutableInstruction51l;
import org.jf.dexlib2.immutable.reference.ImmutableFieldReference;
import org.jf.dexlib2.immutable.reference.ImmutableMethodReference;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
/**
* classes7.dex 定点改动（单次读写，绝不要连续重写两次）。
*
* 只做一件事：把免广告到期时间的读取 lambda 改成恒返回"未过期"。
*
* VideoDataStore$currentAdFreeExpiresAt$$inlined$map$1$2.emit(Object; Continuation)Object
* -> 无条件向下游 collector 发一个超大 Long
*
* 效果等价于 DataStore 里 ad_free_expires_at 永远没有过期，
* 「看广告领特权」弹窗因此不再出现，而且这个改动写死在代码里，
* 全新安装同样生效（DataStore 那份是运行时数据，随 APK 分发不了）。
*
* ⚠️ 不要顺手 patch Lx6;->d：那是通用的图片卡片 Composable，
* 内容和广告共用，改成 return-void 会导致正常图片/播放链路一起失效。
*/
public class PatchClasses7v2 {
private static final String AD_CLASS =
"Lcyc/data/datastore/VideoDataStore$currentAdFreeExpiresAt$$inlined$map$1$2;";
private static final long BIG = 4102444800000L; // ~2100-01-01 (ms)
public static void main(String[] args) throws Exception {
String in = args[0], out = args[1];
org.jf.dexlib2.iface.DexFile dex =
DexFileFactory.loadDexFile(new File(in), Opcodes.forApi(34));
List<ClassDef> outClasses = new ArrayList<ClassDef>();
int p = 0;
for (ClassDef cd : dex.getClasses()) {
List<Method> direct = new ArrayList<Method>();
for (Method m : cd.getDirectMethods()) direct.add(m);
List<Method> virtual = new ArrayList<Method>();
for (Method m : cd.getVirtualMethods()) virtual.add(m);
boolean touched = false;
if (cd.getType().equals(AD_CLASS)) {
for (int pass = 0; pass < 2; pass++) {
List<Method> list = (pass == 0) ? direct : virtual;
for (int i = 0; i < list.size(); i++) {
Method m = list.get(i);
if (!m.getName().equals("emit")
|| !m.getReturnType().equals("Ljava/lang/Object;")
|| m.getImplementation() == null) {
continue;
}
List<Instruction> body = new ArrayList<Instruction>();
// v0 = collector, v1:v2 = long, p0=3 p1=4 p2=5
body.add(new ImmutableInstruction22c(Opcode.IGET_OBJECT, 0, 3,
new ImmutableFieldReference(AD_CLASS, "b", "Lkw2;")));
body.add(new ImmutableInstruction51l(Opcode.CONST_WIDE, 1, BIG));
body.add(new ImmutableInstruction35c(Opcode.INVOKE_STATIC, 2, 1, 2, 0, 0, 0,
new ImmutableMethodReference("Ljava/lang/Long;", "valueOf",
Collections.singletonList("J"), "Ljava/lang/Long;")));
body.add(new ImmutableInstruction11x(Opcode.MOVE_RESULT_OBJECT, 1));
body.add(new ImmutableInstruction35c(Opcode.INVOKE_INTERFACE, 3, 0, 1, 5, 0, 0,
new ImmutableMethodReference("Lkw2;", "emit",
Arrays.asList("Ljava/lang/Object;", "Lwj1;"),
"Ljava/lang/Object;")));
body.add(new ImmutableInstruction11x(Opcode.MOVE_RESULT_OBJECT, 0));
body.add(new ImmutableInstruction11x(Opcode.RETURN_OBJECT, 0));
list.set(i, new ImmutableMethod(m.getDefiningClass(), m.getName(),
m.getParameters(), m.getReturnType(), m.getAccessFlags(),
m.getAnnotations(), m.getHiddenApiRestrictions(),
new ImmutableMethodImplementation(6, body, null, null)));
p++;
touched = true;
System.out.println("[patch] emit -> BIG=" + BIG + " (list=" + pass + ")");
}
}
}
if (!touched) outClasses.add(cd);
else outClasses.add(new ImmutableClassDef(
cd.getType(), cd.getAccessFlags(), cd.getSuperclass(), cd.getInterfaces(),
cd.getSourceFile(), cd.getAnnotations(), cd.getStaticFields(),
cd.getInstanceFields(), direct, virtual));
}
DexFileFactory.writeDexFile(out, new ImmutableDexFile(Opcodes.forApi(34), outClasses));
System.out.println("[done] patched=" + p);
if (p == 0) System.exit(1);
}
}
scripts/dexpatch/README.md
+78
View File
@@ -0,0 +1,78 @@
# dexlib2 method-level patcher
The **preferred** way to change app behavior. It replaces only a target method's
implementation and writes a new dex, leaving every other class, method, string and
reference untouched.
Why not whole-tree smali round-trip: `baksmali` → `smali` rebuilds misrepresent R8's
synthetic access bridges. The class table still looks perfect — same class count, zero
`ACC_INTERFACE` mismatches — but at runtime you get
`IncompatibleClassChangeError: Found interface X, but class was expected`.
See `../../references/pitfalls.md` P3.
## Build
Needs the same eight jars `smtool.py` uses:
```
smali-2.5.2.jar antlr-runtime-3.5.2.jar stringtemplate-3.2.1.jar
baksmali-2.5.2.jar util-2.5.2.jar jcommander-1.64.jar
guava-27.1-android.jar dexlib2-2.5.2.jar
```
```bash
# point at your jars (or reuse ../smali_cp.txt)
CP=$(cat ../smali_cp.txt | grep -v '^#' | tr '\n' ';' | sed 's/;$//')
javac -encoding UTF-8 -cp "$CP" PatchMethod.java
```
**Always pass `-encoding UTF-8`** if the source has non-ASCII comments — `javac`
otherwise reads them with the platform default encoding and fails.
## Run
```bash
java -cp "$CP;." PatchMethod <in.dex> <out.dex> <descriptor> <methodName> <returnType>
# example: make Lx6;->d(...)V a no-op
java -cp "$CP;." PatchMethod classes7.dex out.dex 'Lx6;' d V
```
`PatchMethod` replaces the body with `return-void`. It scans **both** `directMethods`
and `virtualMethods` — a method you cannot find is often in the other list
(`static`/`private`/`<init>` are direct; everything else, including `public final`,
is virtual).
Use `PatchMethodExample.java` as the template for anything non-trivial (returning a
constant, emitting a fixed value from a Flow lambda, applying **several** edits to the
same dex in one pass).
## Rules that decide success
1. **One dex, one write pass.** Never chain two patchers over the same dex — the second
serialization drops metadata and ART then refuses to start the process
(`Failure starting process`, no Java stack). Combine all edits into one program.
2. **Preserve the register count.** `registers` must be `>=` the parameter register
count. With `registers = N` and `k` parameters (counting `this` for non-static),
`p0 = N - k`. For `(Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;`
on an instance method, `k = 3`, so with `registers = 6` you get `p0=3, p1=4, p2=5`.
3. **`const-wide` needs the 64-bit instruction form** (`ImmutableInstruction51l`), not
the 32-bit one.
4. **`invoke` operand order in `35c`** is `(opcode, registerCount, regC, regD, regE, regF, regG, ref)`.
`invoke-static {v1, v2}, ...` → `registerCount=2, regC=1, regD=2`.
5. **End every body with a return of the correct type.**
6. **Check the blast radius first** — `../find_refs.py`. A generic helper with dozens of
callers is not your patch point (`../../references/pitfalls.md` P6).
## Verify
```bash
python ../dex_classdiff.py <original.dex> <patched.dex>
python ../smtool.py d <patched.dex> <tmp_tree> # confirm it parses
# then read the target method in tmp_tree to confirm it says what you intended
```
Expect `only_in_A=0`, `only_in_B=0`, `ACC_INTERFACE mismatch: 0`. Then install and
launch — table checks cannot see code-item damage, so runtime verification is
mandatory (`../../references/verification.md`).
scripts/find_refs.py
+112
View File
@@ -0,0 +1,112 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Find every reference to a method/field/class, so you can judge blast radius
BEFORE patching it.
This is the check that prevents the classic mistake: patching something that looked
ad-specific but is actually a general utility with dozens of callers.
Usage
-----
# references to a method, from a smali tree or a directory of dex files
python find_refs.py <tree_or_dex_dir> 'Lcom/pkg/Helper;->showAd(Landroid/app/Activity;)V'
# a whole class (all of its members)
python find_refs.py <tree_or_dex_dir> 'Lcom/pkg/Helper;'
# a field
python find_refs.py <tree_or_dex_dir> 'Lcom/pkg/Helper;->count:I'
# just the counts, no listing
python find_refs.py <tree_or_dex_dir> 'Lcom/pkg/Helper;->showAd' --count-only
Reading the result
------------------
few callers, all inside one feature area -> safe to patch
many callers, or callers in unrelated pkgs -> general utility. DO NOT patch it.
Go one level up and patch the
specific caller instead.
Also inspect the signature. If it mentions Modifier / ContentScale / Shape / View /
ColorScheme or a content-generic model type, it is shared UI, not your target.
"""
import argparse
import os
import re
import sys
TEXT_EXT = ('.smali',)
REF_RE_TMPL = r'invoke[^\n]*%s|sget[^\n]*%s|sput[^\n]*%s|new-instance[^\n]*%s|check-cast[^\n]*%s'
def iter_files(root):
if os.path.isdir(root):
for dp, _, fs in os.walk(root):
for fn in fs:
if fn.endswith(TEXT_EXT):
yield os.path.join(dp, fn)
else:
yield root
def method_owner(text, idx):
"""Map a character offset to the enclosing .method declaration."""
head = text.rfind('.method ', 0, idx)
if head < 0:
return '?'
end = text.find('\n', head)
return text[head:end].strip()
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument('root')
ap.add_argument('needle')
ap.add_argument('--count-only', action='store_true')
ap.add_argument('--max-print', type=int, default=60)
a = ap.parse_args()
esc = re.escape(a.needle)
# match the needle plus one more char so we do not match a longer prefix
pat = re.compile(REF_RE_TMPL % (esc, esc, esc, esc, esc))
total = 0
per_file = []
printed = 0
for fp in iter_files(a.root):
try:
t = open(fp, encoding='utf-8', errors='replace').read()
except Exception:
continue
hits = list(pat.finditer(t))
if not hits:
continue
total += len(hits)
per_file.append((len(hits), fp, hits, t))
per_file.sort(reverse=True)
print('[total refs] %d across %d files' % (total, len(per_file)))
print()
for cnt, fp, hits, t in per_file:
rel = fp
print('%-70s %d' % (rel, cnt))
if a.count_only:
continue
for m in hits:
if printed >= a.max_print:
print(' ... (truncated)')
return 0
owner = method_owner(t, m.start())
print(' in %s' % owner)
print(' %s' % m.group(0).strip())
printed += 1
if total == 0:
print('[note] no references found. Check the descriptor prefix "L...;" and '
'exact obfuscated names, or disassemble the dex first with smtool.py.')
return 1
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/grab_crash.py
+104
View File
@@ -0,0 +1,104 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Capture a crash stack when the app's crash-reporter SDK is hiding it.
Problem this solves
-------------------
Many apps install a global uncaught-exception handler (友盟/UCrash, Bugly, Firebase
Crashlytics in some configs). The Java stack then never reaches `logcat` -- you only
see something like:
W/UCrash.Java: uncaughtException time: <ts>
...and the process dies. Without a stack you are guessing.
Three ways to get it, in order of reliability:
1. Frida: hook Thread.setDefaultUncaughtExceptionHandler (use references/dynamic-frida.md)
2. Race the reporter's own log file (this script): it writes a file under the app's
data dir, then zips, uploads, and deletes it. Poll fast and copy on sight.
3. Use a build with the reporter disabled.
Usage
-----
python grab_crash.py --pkg com.example.app --activity .MainActivity
python grab_crash.py --pkg com.example.app --activity .MainActivity \
--scan-dir /data/data/com.example.app --wait 25
Notes
-----
* `--scan-dir` defaults to /data/data/<pkg>; the script searches a few levels deep for
newly appeared *.log / *.txt / *.stacktrace files, so you do not have to know which
SDK is installed.
* The watcher loop self-terminates after ~80 s, so nothing is left running on device.
* Requires root (the app's data dir is private).
"""
import argparse
import subprocess
import sys
import time
CAPTURE = '/data/local/tmp/_skill_crash_capture.log'
def adb(args, serial=None, timeout=180):
cmd = ['adb']
if serial:
cmd += ['-s', serial]
cmd += args
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace', timeout=timeout)
return (r.stdout or '') + (('\n[stderr]\n' + r.stderr) if (r.stderr or '').strip() else '')
def su(cmd, serial=None):
return adb(['shell', 'su -c "%s"' % cmd.replace('"', '\\"')], serial)
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument('--pkg', required=True)
ap.add_argument('--activity', required=True)
ap.add_argument('--serial')
ap.add_argument('--scan-dir')
ap.add_argument('--wait', type=int, default=25)
a = ap.parse_args()
scan = a.scan_dir or ('/data/data/%s' % a.pkg)
s = a.serial
# On-device watcher: copy any *.log/*.txt that appears under the app dir.
watch = (
"i=0; while [ $i -lt 400 ]; do "
" find {d} -maxdepth 5 -type f \\( -name '*.log' -o -name '*.txt' -o -name '*.stacktrace' "
" -o -name '*.crash' \\) -newermt '-3 minutes' -exec cp -f {{}} {c} \\; 2>/dev/null; "
" i=$((i+1)); sleep 0.2; "
"done"
).format(d=scan, c=CAPTURE)
su('am force-stop %s' % a.pkg, s)
su('rm -f %s' % CAPTURE, s)
print('[grab] arming on-device watcher over %s' % scan)
subprocess.Popen(['adb'] + (['-s', s] if s else []) +
['shell', "su -c '%s'" % watch],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
time.sleep(1.5)
print('[grab] starting app')
print(adb(['shell', 'am start -n %s/%s' % (a.pkg, a.activity)], s))
time.sleep(a.wait)
pid = adb(['shell', 'pidof %s' % a.pkg], s).strip()
print('[grab] pid: %s' % (pid or '(dead -> likely crashed)'))
print('[grab] ==== captured file ====')
out = su('head -c 6000 %s' % CAPTURE, s)
if 'No such file' in out or not out.strip():
print('(nothing captured)')
print('[hint] the reporter may have deleted it faster than 0.2 s, or it writes '
'outside --scan-dir. Use the Frida approach instead.')
else:
print(out)
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/install_test.py
+125
View File
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Install an APK and run a launch health check, extracting the signals that matter.
Automates the verification loop from references/verification.md:
install -> clear logcat -> launch -> sample pid twice -> scan for fatal signatures
-> capture a screenshot.
Why the pid is sampled TWICE: a crash loop shows a live pid at any single instant.
The second sample catches it.
Usage
-----
python install_test.py --apk out/app.apk --pkg com.example.app --activity .MainActivity
python install_test.py --apk out/app.apk --pkg com.example.app --activity .MainActivity \
--shots out/shot.png --uninstall-first --grant READ_PHONE_STATE --grant ACCESS_FINE_LOCATION
Useful flags
------------
--uninstall-first wipe app data (needed when the signing key changed, and the only
way to test "does this survive a fresh install")
--keep-data reinstall over the existing app (requires the same signing key;
preserves login state -- valuable when the feature under test
needs auth)
--grant NAME runtime permission to grant (repeatable). Must be granted as
root: the shell user gets SecurityException.
--wait N seconds between the two pid samples (default 18)
Exit code is 0 only when the app launched AND both pid samples agree.
"""
import argparse
import os
import re
import subprocess
import sys
import time
FATAL = re.compile(
r'FATAL EXCEPTION|VerifyError|IncompatibleClassChangeError|'
r'ClassNotFoundException|NoClassDefFoundError|LinkageError|'
r'Failure starting process|uncaughtException',
re.I)
def adb(args, serial=None, timeout=600):
cmd = ['adb']
if serial:
cmd += ['-s', serial]
cmd += args
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace', timeout=timeout)
return (r.stdout or '') + (('\n[stderr]\n' + r.stderr) if (r.stderr or '').strip() else '')
def sh(cmd, serial=None):
return adb(['shell', cmd], serial)
def su(cmd, serial=None):
return adb(['shell', 'su -c "%s"' % cmd.replace('"', '\\"')], serial)
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument('--apk', required=True)
ap.add_argument('--pkg', required=True)
ap.add_argument('--activity', required=True, help='e.g. .MainActivity or fully qualified')
ap.add_argument('--serial')
ap.add_argument('--uninstall-first', action='store_true')
ap.add_argument('--keep-data', action='store_true')
ap.add_argument('--grant', action='append', default=[])
ap.add_argument('--wait', type=int, default=18)
ap.add_argument('--shots')
a = ap.parse_args()
s = a.serial
print('[0] devices:'); print(adb(['devices', '-l']))
if a.uninstall_first:
print('[1] uninstall'); print(su('am force-stop %s; pm uninstall %s' % (a.pkg, a.pkg), s))
stage = '/data/local/tmp/_skill_stage.apk'
print('[2] push'); print(adb(['push', a.apk, stage], s))
print('[3] install'); print(su('pm install -r -t -d %s' % stage, s))
for perm in a.grant:
p = perm if perm.startswith('android.permission.') else 'android.permission.' + perm
su('pm grant %s %s' % (a.pkg, p), s)
print('[4] clear logcat'); adb(['logcat', '-c'], s)
comp = '%s/%s' % (a.pkg, a.activity if a.activity.startswith('.') or '.' in a.activity
else a.activity)
print('[5] launch'); print(sh('am start -n %s' % comp, s))
time.sleep(6)
p1 = sh('pidof %s' % a.pkg, s).strip()
print('[6] pid @+6s : %s' % (p1 or '(dead)'))
time.sleep(a.wait)
p2 = sh('pidof %s' % a.pkg, s).strip()
print('[7] pid @+%ds : %s' % (6 + a.wait, p2 or '(dead)'))
print('[8] focus : %s' % sh('dumpsys window | grep mCurrentFocus', s).strip())
if a.shots:
su('screencap -p /sdcard/_skill_shot.png', s)
print(adb(['pull', '/sdcard/_skill_shot.png', a.shots], s))
print('[9] fatal signatures:')
log = adb(['logcat', '-d', '-v', 'brief'], s)
hits = [ln for ln in log.splitlines() if FATAL.search(ln)]
if hits:
for ln in hits[-25:]:
print(' ', ln)
else:
print(' none')
ok = bool(p1) and p1 == p2 and not hits
print()
print('[result] %s' % ('OK (launched, stable, no fatal signatures)'
if ok else 'CHECK (see above)'))
print('[note] stability is NOT proof the patch worked. Exercise the changed '
'feature and its neighbours -- see references/verification.md.')
return 0 if ok else 1
if __name__ == '__main__':
sys.exit(main())
scripts/patch_smali.py
+99
View File
@@ -0,0 +1,99 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
方法级 smali 补丁器。
设计要点（踩坑后确定）：
- 只替换「.method 声明行的下一行」到「.end method 之前」的指令体，
保留原 .method / .end method / .annotation 段不动，避免破坏修饰符与注解。
- 补丁清单要求显式给出自洽的 .registers 值：smali 汇编器会校验寄存器不越界，
registers 必须 >= max(参数寄存器数) 且 >= 指令里用到的最大寄存器编号 + 1。
参数寄存器数：static 方法 = 参数个数；非 static = 参数个数 + 1（this）。
- 补丁清单是 JSON 列表，每条：
{
"file": "com/example/Helper.smali", # path relative to smali tree root
"method": ".method public final show(Landroid/app/Activity;)V", # exact .method line
"registers": 3,
"body": ["invoke-interface {p3}, ...;", "return-void"],
"note": "干掉插屏广告" # 可选，仅用于报告
}
用法：
python patch_smali.py <smali_tree> <patch.json> [--dry-run]
"""
import json
import os
import re
import sys
def split_methods(text):
"""把 smali 文本切成 [(method_decl, start_idx, end_idx)]，按行索引。"""
lines = text.split('\n')
out = []
start = None
decl = None
for i, ln in enumerate(lines):
s = ln.strip()
if s.startswith('.method ') and start is None:
start, decl = i, s
elif s == '.end method' and start is not None:
out.append((decl, start, i))
start, decl = None, None
return lines, out
def patch_one(path, decl_wanted, registers, body):
with open(path, 'r', encoding='utf-8') as f:
text = f.read()
lines, methods = split_methods(text)
hits = [m for m in methods if m[0] == decl_wanted.strip()]
if not hits:
return False, 'method not found: %s' % decl_wanted
if len(hits) > 1:
return False, 'ambiguous (%d matches): %s' % (len(hits), decl_wanted)
_, start, end = hits[0]
# 保留 .method 行；重写内部：.registers + body
new_inner = [' .registers %d' % registers, '']
for b in body:
new_inner.append(' ' + b)
new_lines = lines[:start + 1] + new_inner + [''] + lines[end:]
with open(path, 'w', encoding='utf-8', newline='\n') as f:
f.write('\n'.join(new_lines))
return True, 'ok'
def main():
if len(sys.argv) < 3:
print(__doc__)
return 2
tree = sys.argv[1]
spec = sys.argv[2]
dry = '--dry-run' in sys.argv
with open(spec, 'r', encoding='utf-8') as f:
patches = json.load(f)
ok = fail = 0
for p in patches:
full = os.path.join(tree, p['file'].replace('/', os.sep))
if not os.path.isfile(full):
print('[MISS-FILE] %s' % p['file'])
fail += 1
continue
if dry:
with open(full, 'r', encoding='utf-8') as f:
_, methods = split_methods(f.read())
found = any(m[0] == p['method'].strip() for m in methods)
print('[%s] %s :: %s' % ('DRY-OK' if found else 'DRY-MISS', p['file'], p['method']))
ok += 1 if found else 0
fail += 0 if found else 1
continue
okk, msg = patch_one(full, p['method'], p['registers'], p['body'])
print('[%s] %s :: %s (%s)' % ('OK' if okk else 'FAIL', p['file'], p['method'].split('(')[0].replace('.method ', ''), msg))
ok += 1 if okk else 0
fail += 0 if okk else 1
print('\npatched=%d failed=%d%s' % (ok, fail, ' [dry-run]' if dry else ''))
return 0 if fail == 0 else 1
if __name__ == '__main__':
sys.exit(main())
scripts/probe_api.py
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Probe an app's HTTP API the way the app does.
Purpose-built for the common frustration: your request gets 403 while the app
works fine. Almost always the app sends a header you are not sending (frequently
one custom header), or your egress IP differs from the device's.
Usage
-----
python probe_api.py --base https://api.example.com --path /health
python probe_api.py --base https://api.example.com --path /adverts \
--param position=banner --header 'X-App-Name: myapp'
python probe_api.py --base https://api.example.com \
--path "/items?page=1" --raw
Options
-------
--base Base URL (required)
--path Path, may already contain a query string (required)
--param k=v Repeatable; appended as query parameters
--header k=v Repeatable; sent as-is. Always send the app's real headers.
--token T Convenience: adds 'Authorization: Bearer T'
--ua UA User-Agent (default: okhttp/4.12.0)
--no-proxy Ignore environment proxy settings (recommended)
--show Print the first N bytes of the body (default 800)
Interpreting results
--------------------
200 + data -> alive; read the envelope shape
200 + empty list -> valid input, nothing to return. Do NOT assume "ad removed":
it may simply be that no item is configured.
400 + message -> input is validated; the message often lists valid values
401/403, no creds -> auth-gated (server-side ownership)
401/403, forged -> token is verified server-side; client patching cannot mint one
403 + HTML body -> a WAF/CDN refused your *request shape* or egress, not the endpoint
404 -> exact path mismatch
"""
import argparse
import json
import urllib.error
import urllib.parse
import urllib.request
def build_opener(no_proxy):
if no_proxy:
return urllib.request.build_opener(urllib.request.ProxyHandler({}))
return urllib.request.build_opener()
def main():
ap = argparse.ArgumentParser(add_help=True)
ap.add_argument('--base', required=True)
ap.add_argument('--path', required=True)
ap.add_argument('--param', action='append', default=[])
ap.add_argument('--header', action='append', default=[])
ap.add_argument('--token')
ap.add_argument('--ua', default='okhttp/4.12.0')
ap.add_argument('--no-proxy', action='store_true')
ap.add_argument('--show', type=int, default=800)
a = ap.parse_args()
url = a.base.rstrip('/') + a.path
if a.param:
params = []
for p in a.param:
if '=' not in p:
print('[FAIL] --param expects k=v, got %r' % p)
return 2
k, v = p.split('=', 1)
params.append((k, v))
sep = '&' if '?' in url else '?'
url += sep + urllib.parse.urlencode(params)
headers = {
'User-Agent': a.ua,
'Accept': 'application/json',
}
for h in a.header:
if ':' not in h:
print('[FAIL] --header expects "Name: value", got %r' % h)
return 2
k, v = h.split(':', 1)
headers[k.strip()] = v.strip()
if a.token:
headers['Authorization'] = 'Bearer %s' % a.token
print('[req] GET %s' % url)
for k, v in headers.items():
shown = v if k.lower() != 'authorization' else v[:24] + '...'
print(' %s: %s' % (k, shown))
req = urllib.request.Request(url, headers=headers)
op = build_opener(a.no_proxy)
try:
with op.open(req, timeout=30) as r:
body = r.read().decode('utf-8', 'replace')
print('[res] %s %s' % (r.status, r.headers.get('Content-Type', '')))
print(body[:a.show])
if body.strip().startswith(('{', '[')):
try:
parsed = json.loads(body)
if isinstance(parsed, dict) and 'data' in parsed:
d = parsed['data']
if isinstance(d, dict) and 'list' in d:
print('[shape] data.list length=%s' % len(d['list'] or []))
except Exception:
pass
except urllib.error.HTTPError as e:
body = e.read().decode('utf-8', 'replace')
print('[res] HTTP %s' % e.code)
print(body[:a.show])
except Exception as e:
print('[ERR] %s' % e)
return 1
return 0
if __name__ == '__main__':
raise SystemExit(main())
scripts/repack.py
+266
View File
@@ -0,0 +1,266 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Repack + resign pipeline for the ciyuan Android app.
Pipeline: replace dex files -> strip META-INF/ -> zip (AndroidManifest.xml and
resources.arsc kept STORED) -> zipalign + sign via uber-apk-signer -> verify.
Keystore defaults (generated on first use with keytool):
file : <work>/out/cyc.keystore
alias : cyc
storepass / keypass : cyc123456
Usage examples:
# roundtrip: reuse original dexes unchanged, proves the pipeline itself works
python tools/repack.py --apk work/orig.apk --dexdir work/x_orig --out work/out/roundtrip.apk
# real patch: swap a single dex (name=path), other dexes kept from the apk
python tools/repack.py --apk work/orig.apk --dex classes8.dex=work/x_orig/patch/classes8.dex \
--out work/out/patched.apk
Notes:
* All java tooling is invoked through python subprocess (never through the
PowerShell argument passing path, which drops arguments).
* uber-apk-signer silently skips already signed apks (reports 0 processed),
which is why META-INF/ is always stripped before signing.
"""
import argparse
import os
import re
import shutil
import subprocess
import sys
import zipfile
JAVA = r'C:\Program Files\Java\jdk-17.0.4.1\bin\java.exe'
KEYTOOL = r'C:\Program Files\Java\jdk-17.0.4.1\bin\keytool.exe'
JARSIGNER = r'C:\Program Files\Java\jdk-17.0.4.1\bin\jarsigner.exe'
SIGNER_JAR = os.environ.get('APK_SIGNER_JAR', 'uber-apk-signer.jar')
KS_PATH = None # pass --ks on the command line
KS_ALIAS = os.environ.get('APK_KS_ALIAS', 'apkreverse')
KS_PASS = os.environ.get('APK_KS_PASS', '')
STORE_ONLY = ('AndroidManifest.xml', 'resources.arsc')
def log(msg):
print(msg, flush=True)
def run(cmd, **kw):
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace', **kw)
return r
def ensure_keystore(path=KS_PATH, alias=KS_ALIAS, password=KS_PASS):
if os.path.exists(path):
log('[ks] reuse %s' % path)
return path
os.makedirs(os.path.dirname(path), exist_ok=True)
cmd = [KEYTOOL, '-genkeypair', '-keystore', path, '-alias', alias,
'-keyalg', 'RSA', '-keysize', '2048', '-validity', '10000',
'-storepass', password, '-keypass', password,
'-dname', 'CN=apkreverse, OU=dev, O=dev, L=NA, ST=NA, C=NA']
r = run(cmd)
if r.returncode != 0:
log('[ks] keytool failed rc=%d\n%s\n%s' % (r.returncode, r.stdout, r.stderr))
raise SystemExit(1)
log('[ks] generated %s (alias=%s pass=%s)' % (path, alias, password))
return path
def collect_replacement_dex(dexdir=None, dex_pairs=None):
repl = {}
if dexdir:
for name in sorted(os.listdir(dexdir)):
if re.match(r'^classes\d*\.dex$', name):
repl[name] = os.path.join(dexdir, name)
for pair in (dex_pairs or []):
if '=' not in pair:
raise SystemExit('--dex expects name=path, got %r' % pair)
name, p = pair.split('=', 1)
repl[name] = p
return repl
def is_signature_entry(name):
"""True only for JAR/APK signature artifacts directly under META-INF/.
META-INF/services/**, META-INF/androidx/**, META-INF/native-image/** and all
other subdirectories are RUNTIME RESOURCES and must survive repacking.
"""
if not name.upper().startswith('META-INF/'):
return False
rest = name[len('META-INF/'):]
if '/' in rest:
return False
up = rest.upper()
if up == 'MANIFEST.MF':
return True
return up.endswith(('.SF', '.RSA', '.DSA', '.EC'))
def build_unsigned(apk, repl, out_apk, drop_meta=True):
"""Write a new apk with dex replaced and only signature entries dropped.
IMPORTANT (root cause of a crash, do not regress):
Stripping the WHOLE META-INF/ breaks ServiceLoader-based runtime wiring.
This APK ships ServiceLoader registrations that Android reads at runtime:
META-INF/services/kotlinx.coroutines.internal.MainDispatcherFactory -> kc
META-INF/services/io.ktor.client.HttpClientEngineContainer -> OkHttpEngineContainer
META-INF/services/io.ktor.serialization.kotlinx.KotlinxSerializationExtensionProvider
META-INF/services/kotlinx.coroutines.CoroutineExceptionHandler -> wc
META-INF/services/com.arialyy.aria.* , META-INF/services/wt2 , META-INF/services/xx1
Deleting them makes the app die at startup with:
IllegalStateException: Module with the Main dispatcher is missing ...
So only signature artifacts are removed here; META-INF subdirectories are kept.
"""
seen = set()
replaced_names = set(repl)
with zipfile.ZipFile(apk, 'r') as zin, zipfile.ZipFile(out_apk, 'w', zipfile.ZIP_DEFLATED) as zout:
for item in zin.infolist():
name = item.filename
base = os.path.basename(name)
if item.is_dir():
continue
if drop_meta and is_signature_entry(name):
continue
if name in replaced_names:
continue
data = zin.read(name)
keep = name in STORE_ONLY or base in STORE_ONLY
zi = zipfile.ZipInfo(name, date_time=item.date_time)
zi.compress_type = zipfile.ZIP_STORED if keep else zipfile.ZIP_DEFLATED
zi.external_attr = item.external_attr
zi.internal_attr = item.internal_attr
zi.create_system = item.create_system
zout.writestr(zi, data)
seen.add(name)
for name, path in sorted(repl.items()):
with open(path, 'rb') as f:
data = f.read()
zi = zipfile.ZipInfo(name, date_time=(2024, 1, 1, 0, 0, 0))
zi.compress_type = zipfile.ZIP_DEFLATED
zout.writestr(zi, data)
log('[zip] + %s (%d bytes <- %s)' % (name, len(data), path))
seen.add(name)
return seen
def zip_report(apk):
rows = []
with zipfile.ZipFile(apk, 'r') as z:
for i in z.infolist():
if i.filename in STORE_ONLY or os.path.basename(i.filename) in STORE_ONLY:
rows.append('%s method=%d (0=STORED)' % (i.filename, i.compress_type))
if re.match(r'^classes\d*\.dex$', i.filename):
rows.append('%s %d bytes method=%d crc=%08x' % (
i.filename, i.file_size, i.compress_type, i.CRC))
return rows
def sign_apk(unsigned_apk, workdir, ks, alias=KS_ALIAS, password=KS_PASS):
outdir = os.path.join(workdir, 'signed')
if os.path.isdir(outdir):
shutil.rmtree(outdir, ignore_errors=True)
os.makedirs(outdir, exist_ok=True)
cmd = [JAVA, '-jar', SIGNER_JAR,
'--apks', unsigned_apk,
'--ks', ks, '--ksAlias', alias,
'--ksPass', password, '--ksKeyPass', password,
'-o', outdir, '--verbose']
r = run(cmd)
log('[sign] rc=%d' % r.returncode)
log(r.stdout.strip())
if r.stderr.strip():
log('[sign][stderr] ' + r.stderr.strip())
cand = os.path.join(outdir, os.path.basename(unsigned_apk))
if not os.path.exists(cand):
outs = [os.path.join(outdir, f) for f in os.listdir(outdir)] if os.path.isdir(outdir) else []
if not outs:
raise SystemExit('signing produced no output')
cand = outs[0]
return cand, r.stdout
def verify_apk(apk):
log('== verify: %s' % apk)
ok = True
# 1) uber-apk-signer verify (also re-checks zipalign)
r = run([JAVA, '-jar', SIGNER_JAR, '-a', apk, '-y'])
log('[verify:signer] rc=%d\n%s' % (r.returncode, r.stdout.strip()))
if r.returncode != 0 or 'Verified' not in r.stdout:
ok = ok and ('DOES NOT VERIFY' not in r.stdout)
# 2) jarsigner (v1 / jAR signature)
r2 = run([JARSIGNER, '-verify', '-certs', apk])
txt = (r2.stdout or '') + (r2.stderr or '')
log('[verify:jarsigner] rc=%d\n%s' % (r2.returncode, txt.strip()[:2000]))
# 3) keytool cert dump
r3 = run([KEYTOOL, '-printcert', '-jarfile', apk])
log('[verify:keytool] rc=%d\n%s' % (r3.returncode, r3.stdout.strip()[:1500]))
# 4) v2/v3 APK Signing Block presence
with open(apk, 'rb') as f:
blob = f.read()
has_v2 = b'APK Sig Block 42' in blob
log('[verify:v2block] %s' % ('present' if has_v2 else 'MISSING'))
return ok and (r.returncode == 0)
def fingerprint(apk):
r = run([KEYTOOL, '-printcert', '-jarfile', apk])
sha = re.findall(r'(SHA1|SHA256):\s*([0-9A-F:]+)', r.stdout)
return sha
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--apk', required=True)
ap.add_argument('--out', required=True)
ap.add_argument('--dexdir')
ap.add_argument('--dex', action='append', default=[], help='name=path, repeatable')
ap.add_argument('--ks', default=KS_PATH)
ap.add_argument('--no-sign', action='store_true')
ap.add_argument('--workdir', default=None)
args = ap.parse_args()
apk = os.path.abspath(args.apk)
out = os.path.abspath(args.out)
if not os.path.exists(apk):
raise SystemExit('missing apk: %s' % apk)
os.makedirs(os.path.dirname(out), exist_ok=True)
os.makedirs(args.workdir, exist_ok=True)
repl = collect_replacement_dex(args.dexdir, args.dex)
log('[in ] %s (%d bytes)' % (apk, os.path.getsize(apk)))
log('[repl] %s' % ', '.join('%s<-%s' % (k, v) for k, v in sorted(repl.items())))
unsigned = os.path.join(args.workdir, 'unsigned.apk')
if os.path.exists(unsigned):
os.remove(unsigned)
build_unsigned(apk, repl, unsigned)
log('[zip] unsigned written: %d bytes' % os.path.getsize(unsigned))
for row in zip_report(unsigned):
log('[zip] ' + row)
if args.no_sign:
shutil.copy2(unsigned, out)
else:
ks = ensure_keystore(args.ks)
signed, _ = sign_apk(unsigned, args.workdir, ks)
shutil.copy2(signed, out)
log('[out] %s (%d bytes)' % (out, os.path.getsize(out)))
log('== zip layout of final apk')
for row in zip_report(out):
log(' ' + row)
ok = verify_apk(out)
sha = fingerprint(out)
for algo, val in sha:
log('[cert] %s %s' % (algo, val))
log('[result] %s' % ('OK' if ok else 'CHECK-FAILED'))
return 0 if ok else 2
if __name__ == '__main__':
sys.exit(main())
scripts/smali_cp.txt
+10
View File
@@ -0,0 +1,10 @@
# One jar path per line. Adjust the directory to your own tools.
# Needed: smali baksmali dexlib2 util antlr-runtime stringtemplate jcommander guava
smali-2.5.2.jar
antlr-runtime-3.5.2.jar
stringtemplate-3.2.1.jar
baksmali-2.5.2.jar
util-2.5.2.jar
jcommander-1.64.jar
guava-27.1-android.jar
dexlib2-2.5.2.jar
scripts/smtool.py
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""baksmali / smali wrapper.
Assembling and disassembling dex requires smali + baksmali + several runtime jars on
the classpath. Getting the set wrong produces a bare `ClassNotFoundException` that
looks like a broken tool rather than a missing jar, so this wrapper centralizes it.
Two ways to configure the classpath, in priority order:
1. --cp "jar1;jar2" (explicit, per invocation)
2. APK_REVERSE_SMALI_CP env var
3. a `smali_cp.txt` next to this script (one jar path per line, '#' comments ok)
Required jars: smali, baksmali, dexlib2, util, antlr-runtime, stringtemplate,
jcommander, guava.
Usage
-----
python smtool.py d <in.dex> <out_dir> # disassemble to a smali tree
python smtool.py a <in_tree> <out.dex> # assemble a smali tree back to dex
python smtool.py check # print the resolved classpath and exit
Notes
-----
* Always pass the tree directory as its own argument; do not rely on the shell.
* Call java from Python rather than from a shell: PowerShell in particular mangles
arguments containing ';' or '$', which silently drops jars from the classpath.
"""
import os
import subprocess
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
CP_FILE = os.path.join(HERE, 'smali_cp.txt')
JAR_NAMES = [
'smali-2.5.2.jar',
'antlr-runtime-3.5.2.jar',
'stringtemplate-3.2.1.jar',
'baksmali-2.5.2.jar',
'util-2.5.2.jar',
'jcommander-1.64.jar',
'guava-27.1-android.jar',
'dexlib2-2.5.2.jar',
]
def resolve_cp(explicit=None):
if explicit:
return explicit
env = os.environ.get('APK_REVERSE_SMALI_CP')
if env:
return env
if os.path.isfile(CP_FILE):
parts = []
for line in open(CP_FILE, encoding='utf-8'):
line = line.strip()
if line and not line.startswith('#'):
parts.append(line)
if parts:
return os.pathsep.join(parts)
# last resort: look for the jars in a 'tools' dir next to this script
cand = [os.path.join(HERE, 'tools', n) for n in JAR_NAMES]
if all(os.path.isfile(c) for c in cand):
return os.pathsep.join(cand)
return None
def run(cp, main_class, args):
cmd = ['java', '-cp', cp, main_class] + args
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace')
out = (r.stdout or '') + (('\n[stderr]\n' + r.stderr) if (r.stderr or '').strip() else '')
return r.returncode, out
def main():
if len(sys.argv) < 2:
print(__doc__)
return 2
explicit = None
argv = sys.argv[1:]
if '--cp' in argv:
i = argv.index('--cp')
explicit = argv[i + 1]
del argv[i:i + 2]
cp = resolve_cp(explicit)
if not cp:
print('[FAIL] no classpath. Provide --cp, set APK_REVERSE_SMALI_CP, '
'or create smali_cp.txt next to this script.')
return 1
cmd = argv[0]
if cmd == 'check':
print('classpath: %s' % cp)
for j in cp.split(os.pathsep):
print(' %s %s' % ('OK ' if os.path.isfile(j) else 'MISSING', j))
return 0
if cmd in ('d', 'dis', 'disassemble'):
if len(argv) < 3:
print('usage: smtool.py d <in.dex> <out_dir>')
return 2
rc, out = run(cp, 'org.jf.baksmali.Main', ['d', argv[1], '-o', argv[2]])
elif cmd in ('a', 'asm', 'assemble'):
if len(argv) < 3:
print('usage: smtool.py a <in_tree> <out.dex>')
return 2
rc, out = run(cp, 'org.jf.smali.Main', ['a', argv[1], '-o', argv[2]])
else:
print(__doc__)
return 2
print(out.strip()[-4000:])
print('rc=%d' % rc)
return 0 if rc == 0 else 1
if __name__ == '__main__':
sys.exit(main())
scripts/usb_net_proxy.py
+159
View File
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
用于给「无网络」Android 设备供网的极简 HTTP/HTTPS 代理。
场景：手机 WiFi DHCP 失败 -> 无默认网络 -> app 所有请求失败。
解法：手机没有可用网络接口也没关系 —— adb reverse 会在**设备本机的 loopback**
上开监听，不需要任何网络接口。于是：
1) 电脑跑本代理 (0.0.0.0:8080)
2) adb -s <serial> reverse tcp:8080 tcp:8080
3) 设备端 su -c "settings put global http_proxy 127.0.0.1:8080"
4) app 的 HTTP/HTTPS 流量经 USB 到电脑，由电脑代发
支持：
- CONNECT 隧道（HTTPS 主用）
- 明文 HTTP 绝对 URI 转发
仅做转发，不做 MITM（不解密 TLS，也不改流量），因此不干扰证书校验。
用法：python usb_net_proxy.py [listen_port] [logfile]
"""
import socket
import sys
import threading
import time
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
LOGF = sys.argv[2] if len(sys.argv) > 2 else None
_lock = threading.Lock()
_logf = open(LOGF, 'a', encoding='utf-8') if LOGF else None
def log(msg):
line = '[%s] %s' % (time.strftime('%H:%M:%S'), msg)
print(line, flush=True)
if _logf:
with _lock:
_logf.write(line + '\n')
_logf.flush()
def pump(a, b):
"""a -> b 单向搬运，直到任一端关闭。"""
try:
while True:
r, _, _ = select_select([a], [], [], 30)
if not r:
break
data = a.recv(65536)
if not data:
break
b.sendall(data)
except Exception:
pass
finally:
for s in (a, b):
try:
s.shutdown(socket.SHUT_RDWR)
except Exception:
pass
try:
s.close()
except Exception:
pass
def select_select(rlist, wlist, xlist, timeout):
import select
return select.select(rlist, wlist, xlist, timeout)
def read_head(sock, limit=65536):
data = b''
while b'\r\n\r\n' not in data and len(data) < limit:
chunk = sock.recv(4096)
if not chunk:
return data
data += chunk
return data
def handle(client, addr):
client.settimeout(30)
try:
head = read_head(client)
if not head:
client.close()
return
first = head.split(b'\r\n', 1)[0].decode('latin-1')
parts = first.split(' ')
if len(parts) < 3:
client.close()
return
method, target = parts[0].upper(), parts[1]
rest = head.split(b'\r\n\r\n', 1)[1] if b'\r\n\r\n' in head else b''
if method == 'CONNECT':
host, _, port = target.rpartition(':')
port = int(port or 443)
log('CONNECT %s:%d' % (host, port))
remote = socket.create_connection((host, port), timeout=20)
client.sendall(b'HTTP/1.1 200 Connection Established\r\n\r\n')
client.settimeout(None)
remote.settimeout(None)
t = threading.Thread(target=pump, args=(client, remote), daemon=True)
t.start()
pump(remote, client)
else:
# 明文 HTTP：target 一般是绝对 URI
if target.startswith('http://'):
without = target[len('http://'):]
hostport = without.split('/', 1)[0]
path = '/' + without.split('/', 1)[1] if '/' in without else '/'
else:
hostport = None
for ln in head.split(b'\r\n'):
if ln.lower().startswith(b'host:'):
hostport = ln.split(b':', 1)[1].strip().decode()
path = target
if not hostport:
client.close()
return
host, _, port = hostport.rpartition(':')
port = int(port or 80)
log('HTTP %s%s' % (hostport, path))
remote = socket.create_connection((host, port), timeout=20)
req = ('%s %s HTTP/1.1\r\n' % (method, path)).encode()
hdrs = head.split(b'\r\n')[1:]
for h in hdrs:
if h.lower().startswith(b'proxy-connection'):
continue
req += h + b'\r\n'
req += b'\r\n' + rest
remote.sendall(req)
client.settimeout(None)
remote.settimeout(None)
t = threading.Thread(target=pump, args=(client, remote), daemon=True)
t.start()
pump(remote, client)
except Exception as e:
log('ERR %s %s' % (addr, e))
try:
client.close()
except Exception:
pass
def main():
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(('0.0.0.0', PORT))
srv.listen(128)
log('proxy listening on 0.0.0.0:%d' % PORT)
while True:
c, a = srv.accept()
threading.Thread(target=handle, args=(c, a), daemon=True).start()
if __name__ == '__main__':
main()