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

Broaden skill with unpacking, TLS triage, third-party build auditing, runtime probe

Browse Source 
references: recon gains packer identification + unpacking workflow (payload/plaintext size mapping, dump validation, hash de-duplication, business-dex heuristic); repack-and-sign gains the de-shelled repack pipeline, why apktool over binary AXML, and the apksigner scheme-selection caveat; new tls-and-cert documents feature-scoped network failures, dual trust chains and the fix template; new third-party-builds covers auditing a cracked/modded APK before trusting it; dynamic-frida gains version pinning, remote-device path, the four-layer probe and hook-never-fires triage; environment gains UI automation pitfalls; ad-removal gains the single convergence point, dynamic-plugin SDKs and the global timestamp gate; pitfalls gains P13-P17; verification gains rung 7 and the three-signal ad check.

scripts: new frida_probe.js (four-layer runtime probe), run_probe.py and tls_check.py; repack.py fixes two crash-on-default bugs and a misleading META-INF note; datastore_inject.py, PatchMethodExample.java and build_scripts.py lose hardcoded values - the example class name did not match its file and could not compile.
This commit is contained in:
inliver233
2026-09-19 21:44:00 +08:00
parent c66658b6b3
commit 1a5cb6245f
18 changed files with 3089 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/datastore_inject.py
+314 -33
View File
@@ -1,20 +1,82 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""向 app 的 DataStore(video_preferences) 注入 ad_free_expires_at 大值，验证它能否抑制广告 UI。
"""Encode or inject AndroidX DataStore (Preferences) protobuf entries.
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, ...
WHY THIS EXISTS
---------------
Some feature gates are not decided in code but in a preferences file that the app
reads at runtime, e.g. a "free until <timestamp>" value. Changing the code alone can
then be pointless, while writing the state directly proves the hypothesis in one
shot: if the behavior changes, the key really is the gate; if it does not, the gate
is elsewhere (usually server-side).
所以要写一个 longKey："ad_free_expires_at" -> Value{ long = 大值 }
Value 载荷 = 0x20 <varint(long)> (field 4, wire type 0 -> tag 0x20)
WHAT THE FILE IS
----------------
DataStore/Preferences serializes to a protobuf on disk. On a device it lives under
the app's private directory, typically:
<app_files_dir>/datastore/<name>.preferences_pb
and the container is:
message PreferenceMap { map<string, Value> preferences = 1; }
message Value { oneof { bool boolean = 1; float float = 2; int32 integer = 3;
int64 long = 4; string string = 5; double double = 6;
bytes bytes = 8; } }
The map field is the part people get wrong: every entry needs BOTH the outer
0x0A <len> (the map field itself) AND the inner 0x0A <len key> <key> +
0x12 <len value> <value>. Writing only the inner form produces a file that
deserializes into nothing, and some apps then die on the next read with an
exception that is swallowed by their crash reporter -- no stack, no clue.
Bytes writes here are the exact encoding, so this script doubles as a reference.
USAGE
-----
# write a fresh file with one big "never expires" timestamp
python datastore_inject.py --file example.preferences_pb \
--key example_expiry_epoch_ms --type long --value 4102444800000
# keep every other key from an existing file and replace/add just this one
python datastore_inject.py --in pulled.preferences_pb --file patched.preferences_pb \
--key example_expiry_epoch_ms --type long --value 4102444800000
# add more than one entry
python datastore_inject.py --file out.pb --key example_flag --type bool --value true \
--extra example_name=hello:string
# inspect a pulled file (what keys exist, and what they decode to)
python datastore_inject.py --in pulled.preferences_pb --list
Deploying the result: the app must not be running while you overwrite the file
(it will rewrite the file on exit), so push, fix ownership/selinux context if
needed, then start the app. Verify by reading the key back with --list.
Pure standard library, Python 3.9+.
"""
import argparse
import struct
import sys
DEFAULT_FILE = 'example.preferences_pb'
DEFAULT_KEY = 'example_expiry_epoch_ms'
DEFAULT_VALUE = '4102444800000' # ~2100-01-01 in ms: "far future" in either unit
TYPES = ('long', 'int', 'bool', 'float', 'double', 'string', 'bytes')
# tag = (field_number << 3) | wire_type, per the Value oneof above.
VALUE_TAG = {
'bool': (1, 0),
'float': (2, 5),
'integer': (3, 0),
'long': (4, 0),
'string': (5, 2),
'double': (6, 1),
'bytes': (8, 2),
}
def varint(n):
"""Unsigned LEB128. Negative int32/int64 are encoded as 64-bit twos complement."""
if n < 0:
n += 1 << 64
out = bytearray()
while True:
b = n & 0x7F
@@ -26,41 +88,260 @@ def varint(n):
return bytes(out)
def map_entry(key, value_payload):
"""编码 PreferenceMap 里的一个 map entry。
def read_varint(buf, pos):
result = 0
shift = 0
while True:
if pos >= len(buf):
raise ValueError('truncated varint at offset %d' % pos)
b = buf[pos]
pos += 1
result |= (b & 0x7F) << shift
if not (b & 0x80):
return result, pos
shift += 7
if shift > 70:
raise ValueError('varint too long at offset %d' % pos)
★关键（踩过坑）：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 吞掉），极难定位。
def tag_bytes(field, wire):
return varint((field << 3) | wire)
def encode_value(kind, raw):
"""Encode the Value payload (the bytes after the inner 0x12 <len>)."""
if kind == 'bool':
field, wire = VALUE_TAG['bool']
truthy = str(raw).strip().lower() in ('1', 'true', 'yes', 'on')
return tag_bytes(field, wire) + varint(1 if truthy else 0)
if kind in ('long', 'int'):
field, wire = VALUE_TAG['long' if kind == 'long' else 'integer']
return tag_bytes(field, wire) + varint(int(str(raw), 0))
if kind == 'float':
field, wire = VALUE_TAG['float']
return tag_bytes(field, wire) + struct.pack('<f', float(raw))
if kind == 'double':
field, wire = VALUE_TAG['double']
return tag_bytes(field, wire) + struct.pack('<d', float(raw))
if kind == 'string':
field, wire = VALUE_TAG['string']
data = str(raw).encode('utf-8')
return tag_bytes(field, wire) + varint(len(data)) + data
if kind == 'bytes':
field, wire = VALUE_TAG['bytes']
data = bytes.fromhex(str(raw))
return tag_bytes(field, wire) + varint(len(data)) + data
raise ValueError('unknown type: %s' % kind)
def encode_entry(key, value_payload):
"""Encode one PreferenceMap entry.
Two levels of 0x0A on purpose:
outer: 0x0A <len(inner)> <- the map field (field 1)
inner: 0x0A <len(key)> <key> 0x12 <len(value)> <value>
Dropping the outer tag+length is the classic mistake: DataStore then
deserializes an empty map, and the failure surfaces far away from here.
"""
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
inner = (tag_bytes(1, 2) + varint(len(kb)) + kb
+ tag_bytes(2, 2) + varint(len(value_payload)) + value_payload)
return tag_bytes(1, 2) + varint(len(inner)) + inner
def entry(key, value_payload):
"""兼容旧调用：返回 map_entry。"""
return map_entry(key, value_payload)
def split_entries(data):
"""Split a PreferenceMap into raw top-level field slices."""
pos = 0
entries = []
while pos < len(data):
start = pos
tag, pos = read_varint(data, pos)
wire = tag & 0x07
if wire == 2:
length, pos = read_varint(data, pos)
pos += length
elif wire == 0:
_, pos = read_varint(data, pos)
elif wire == 5:
pos += 4
elif wire == 1:
pos += 8
else:
raise ValueError('unsupported wire type %d at offset %d' % (wire, start))
if pos > len(data):
raise ValueError('truncated field at offset %d' % start)
entries.append((tag, data[start:pos]))
return entries
def decode_entry(raw):
"""Return (key, value_payload) of one entry, or (None, None) if malformed."""
try:
pos = 0
tag, pos = read_varint(raw, pos)
if (tag >> 3) != 1 or (tag & 7) != 2:
return None, None
length, pos = read_varint(raw, pos)
inner = raw[pos:pos + length]
p = 0
t1, p = read_varint(inner, p)
if (t1 >> 3) != 1:
return None, None
klen, p = read_varint(inner, p)
key = inner[p:p + klen].decode('utf-8', 'replace')
p += klen
t2, p = read_varint(inner, p)
if (t2 >> 3) != 2:
return None, None
vlen, p = read_varint(inner, p)
return key, inner[p:p + vlen]
except Exception:
return None, None
def describe_value(payload):
"""Best-effort decode of a Value payload for the --list output."""
try:
pos = 0
tag, pos = read_varint(payload, pos)
field, wire = tag >> 3, tag & 7
if field == 1 and wire == 0:
val, _ = read_varint(payload, pos)
return 'bool=%s' % bool(val)
if field == 3 and wire == 0:
val, _ = read_varint(payload, pos)
return 'integer=%d' % val
if field == 4 and wire == 0:
val, _ = read_varint(payload, pos)
return 'long=%d' % val
if field == 2 and wire == 5:
return 'float=%r' % struct.unpack('<f', payload[pos:pos + 4])[0]
if field == 6 and wire == 1:
return 'double=%r' % struct.unpack('<d', payload[pos:pos + 8])[0]
if field == 5 and wire == 2:
length, pos = read_varint(payload, pos)
return 'string=%r' % payload[pos:pos + length].decode('utf-8', 'replace')
if field == 8 and wire == 2:
length, pos = read_varint(payload, pos)
return 'bytes=%s' % payload[pos:pos + length].hex()
except Exception:
pass
return 'raw=%s' % payload.hex()
def parse_extra(items):
"""--extra key=value[:type] -> [(key, type, value)]"""
out = []
for item in items:
if '=' not in item:
raise SystemExit('--extra expects key=value[:type], got %r' % item)
key, rest = item.split('=', 1)
kind = 'string'
if ':' in rest:
rest, kind = rest.rsplit(':', 1)
if kind not in TYPES:
raise SystemExit('--extra %s: unknown type %r (choose from %s)'
% (key, kind, ', '.join(TYPES)))
out.append((key, kind, rest))
return out
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)))
ap = argparse.ArgumentParser(
description='Encode or inject AndroidX DataStore (Preferences) protobuf '
'entries, with the correct two-level map encoding.')
ap.add_argument('--file', default=DEFAULT_FILE,
help='output .pb path (default: %s)' % DEFAULT_FILE)
ap.add_argument('--key', default=DEFAULT_KEY,
help='preference key to write (default: %s)' % DEFAULT_KEY)
ap.add_argument('--value', default=DEFAULT_VALUE,
help='value for --key (default: %s)' % DEFAULT_VALUE)
ap.add_argument('--type', choices=TYPES, default='long',
help='value type (default: long)')
ap.add_argument('--in', dest='existing', default=None,
help='existing .pb to read: preserves all other keys')
ap.add_argument('--extra', action='append', default=[],
help='key=value[:type], repeatable, added alongside --key')
ap.add_argument('--list', action='store_true',
help='list the entries of --in and exit (no writing)')
ap.add_argument('--dry-run', action='store_true',
help='print what would be written, write nothing')
args = ap.parse_args()
if args.list:
if not args.existing:
raise SystemExit('--list needs --in <file.preferences_pb>')
with open(args.existing, 'rb') as fh:
data = fh.read()
print('[in] %s (%d bytes)' % (args.existing, len(data)))
shown = 0
for tag, raw in split_entries(data):
key, payload = decode_entry(raw)
if key is None:
print(' <unparsed field tag=0x%02x, %d bytes>' % (tag, len(raw)))
continue
print(' %-32s %s (value=%d bytes)' % (key, describe_value(payload), len(payload)))
shown += 1
print('[ok] %d entry(ies)' % shown)
return 0
# Build the entry list: preserved entries first, then the injected ones.
chunks = []
if args.existing:
with open(args.existing, 'rb') as fh:
original = fh.read()
replaced = {args.key}
replaced.update(k for k, _, _ in parse_extra(args.extra))
kept = 0
for _tag, raw in split_entries(original):
key, _payload = decode_entry(raw)
if key is None:
chunks.append(raw) # unknown field: keep it verbatim
continue
if key in replaced:
print('[keep-skip] %s is being replaced' % key)
continue
chunks.append(raw)
kept += 1
print('[in] preserved %d existing entry(ies) from %s' % (kept, args.existing))
target = encode_entry(args.key, encode_value(args.type, args.value))
chunks.append(target)
print('[inject] %s (%s) = %s -> %d bytes' % (args.key, args.type, args.value, len(target)))
for key, kind, value in parse_extra(args.extra):
entry = encode_entry(key, encode_value(kind, value))
chunks.append(entry)
print('[inject] %s (%s) = %s -> %d bytes' % (key, kind, value, len(entry)))
data = b''.join(chunks)
# Self-check: re-parse what we just built. A file that does not parse back is
# exactly the failure mode that kills the app with no usable stack.
found = {}
for _tag, raw in split_entries(data):
key, payload = decode_entry(raw)
if key is not None:
found[key] = describe_value(payload)
for key in [args.key] + [k for k, _t, _v in parse_extra(args.extra)]:
if key not in found:
raise SystemExit('internal error: %r does not parse back; refusing to write' % key)
print('[check] re-parsed %d entry(ies); %s = %s' % (len(found), args.key, found[args.key]))
print('[hex] %s' % data.hex(' '))
print('[note] ad_free_expires_at = %d, varint=%s' % (BIG, varint(BIG).hex(' ')))
if args.dry_run:
print('[dry-run] nothing written')
return 0
with open(args.file, 'wb') as fh:
fh.write(data)
print('[ok] wrote %s (%d bytes)' % (args.file, len(data)))
print('[note] push it to <app_files_dir>/datastore/<name>.preferences_pb while the app '
'is NOT running, then start the app and read the key back with --list')
return 0
if __name__ == '__main__':
main()
sys.exit(main())
scripts/dexpatch/PatchMethodExample.java
+123 -29
View File
@@ -22,33 +22,111 @@ import java.util.Collections;
import java.util.List;
/**
* classes7.dex 定点改动（单次读写，绝不要连续重写两次）。
* dexlib2 surgical rewrites -- worked example: force one lambda's emit() to hand a
* constant downstream instead of the value it read from its source.
*
* 只做一件事：把免广告到期时间的读取 lambda 改成恒返回"未过期"。
* WHAT IT PATCHES
* <target>$$inlined$map$1$2.emit(Object, Continuation)Object
* -> send a fixed wide (long) constant to the downstream collector, unconditionally
*
* VideoDataStore$currentAdFreeExpiresAt$$inlined$map$1$2.emit(Object; Continuation)Object
* -> 无条件向下游 collector 发一个超大 Long
* WHY THIS SHAPE
* Kotlin's `map { ... }` on a Flow compiles to a synthetic class whose emit()
* reads the source value from a field and forwards it to the collector. Replacing
* that single forwarding body turns "value read from the data source" into
* "constant", without touching the data source itself. Doing it in dex keeps the
* change inside the APK, so a fresh install behaves the same way -- writing the
* value into the app's runtime state instead would not survive a reinstall.
*
* 效果等价于 DataStore 里 ad_free_expires_at 永远没有过期，
* 「看广告领特权」弹窗因此不再出现，而且这个改动写死在代码里，
* 全新安装同样生效（DataStore 那份是运行时数据，随 APK 分发不了）。
* ONE READ, ONE WRITE. Never rewrite the same dex twice in the same run: dexlib2
* round-trips degrade an R8-optimized dex, and the second pass is where it shows.
*
* ⚠️ 不要顺手 patch Lx6;->d：那是通用的图片卡片 Composable，
* 内容和广告共用，改成 return-void 会导致正常图片/播放链路一起失效。
* ADAPT TO YOUR TARGET
* java PatchMethodExample <in.dex> <out.dex> [targetClass] [fieldOwner:fieldName:fieldType]
* [collectorIface] [continuationType]
* Every optional argument defaults to an obviously-fake example value below, and
* the effective values are printed at startup, so a run is never ambiguous.
*
* HOW TO FIND THE REAL VALUES
* * target class: search the dex strings for the source's field/data key, then read
* the call site with baksmali -- see scripts/dex_strings.py and scripts/find_refs.py.
* * collector iface / continuation type: read them off the emit() signature in smali.
* After R8 they are usually single letters; do not guess, copy them.
*
* DO NOT PATCH SHARED CODE BY ACCIDENT
* A generic card/image composable is commonly shared between ordinary content and
* ad slots. Turning such a method into `return-void` breaks normal rendering too.
* Before patching anything, count its callers (scripts/find_refs.py).
*/
public class PatchClasses7v2 {
public class PatchMethodExample {
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)
/** Synthetic class that owns the lambda: <Owner>$<member>$$inlined$map$1$2. */
private static String TARGET_CLASS =
"Lcom/example/app/data/ExampleStore$currentValue$$inlined$map$1$2;";
/** The field on the lambda that holds the downstream collector. */
private static String FIELD_OWNER = "Lcom/example/app/data/ExampleStore$currentValue$$inlined$map$1$2;";
private static String FIELD_NAME = "b";
private static String FIELD_TYPE = "Lcom/example/app/Collector;";
/** The collector type and the Continuation type used by emit(). */
private static String COLLECTOR_IFACE = "Lcom/example/app/Collector;";
private static String CONTINUATION_TYPE = "Lkotlin/coroutines/Continuation;";
private static final long CONSTANT = 4102444800000L; // ~2100-01-01 (ms)
private static void usage() {
System.out.println("usage: PatchMethodExample <in.dex> <out.dex> "
+ "[targetClass] [fieldOwner:fieldName:fieldType] "
+ "[collectorIface] [continuationType]");
System.out.println(" defaults (example values, replace for your target):");
System.out.println(" targetClass " + TARGET_CLASS);
System.out.println(" fieldRef " + FIELD_OWNER + ":" + FIELD_NAME + ":" + FIELD_TYPE);
System.out.println(" collectorIface " + COLLECTOR_IFACE);
System.out.println(" continuationType " + CONTINUATION_TYPE);
}
private static void applyArgs(String[] args) {
if (args.length > 2 && !args[2].isEmpty()) {
TARGET_CLASS = args[2];
}
if (args.length > 3 && !args[3].isEmpty()) {
String[] parts = args[3].split(":");
if (parts.length != 3) {
throw new IllegalArgumentException(
"fieldRef must be owner:name:type, got " + args[3]);
}
FIELD_OWNER = parts[0];
FIELD_NAME = parts[1];
FIELD_TYPE = parts[2];
}
if (args.length > 4 && !args[4].isEmpty()) {
COLLECTOR_IFACE = args[4];
}
if (args.length > 5 && !args[5].isEmpty()) {
CONTINUATION_TYPE = args[5];
}
}
public static void main(String[] args) throws Exception {
if (args.length < 2) {
usage();
System.exit(2);
}
applyArgs(args);
String in = args[0], out = args[1];
System.out.println("[in ] " + in);
System.out.println("[cfg] targetClass=" + TARGET_CLASS);
System.out.println("[cfg] fieldRef=" + FIELD_OWNER + ":" + FIELD_NAME + ":" + FIELD_TYPE);
System.out.println("[cfg] collectorIface=" + COLLECTOR_IFACE);
System.out.println("[cfg] continuationType=" + CONTINUATION_TYPE);
System.out.println("[cfg] constant=" + CONSTANT);
org.jf.dexlib2.iface.DexFile dex =
DexFileFactory.loadDexFile(new File(in), Opcodes.forApi(34));
List<ClassDef> outClasses = new ArrayList<ClassDef>();
int p = 0;
int patched = 0;
for (ClassDef cd : dex.getClasses()) {
List<Method> direct = new ArrayList<Method>();
@@ -57,7 +135,7 @@ public class PatchClasses7v2 {
for (Method m : cd.getVirtualMethods()) virtual.add(m);
boolean touched = false;
if (cd.getType().equals(AD_CLASS)) {
if (cd.getType().equals(TARGET_CLASS)) {
for (int pass = 0; pass < 2; pass++) {
List<Method> list = (pass == 0) ? direct : virtual;
for (int i = 0; i < list.size(); i++) {
@@ -67,18 +145,23 @@ public class PatchClasses7v2 {
|| m.getImplementation() == null) {
continue;
}
// Register layout for emit(Object, Continuation):
// v0 scratch (return value)
// v1:v2 the wide constant we build
// v3 = p0 this (the lambda)
// v4 = p1 the value being forwarded (ignored)
// v5 = p2 the downstream Continuation
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));
new ImmutableFieldReference(FIELD_OWNER, FIELD_NAME, FIELD_TYPE)));
body.add(new ImmutableInstruction51l(Opcode.CONST_WIDE, 1, CONSTANT));
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;"),
new ImmutableMethodReference(COLLECTOR_IFACE, "emit",
Arrays.asList("Ljava/lang/Object;", CONTINUATION_TYPE),
"Ljava/lang/Object;")));
body.add(new ImmutableInstruction11x(Opcode.MOVE_RESULT_OBJECT, 0));
body.add(new ImmutableInstruction11x(Opcode.RETURN_OBJECT, 0));
@@ -87,22 +170,33 @@ public class PatchClasses7v2 {
m.getParameters(), m.getReturnType(), m.getAccessFlags(),
m.getAnnotations(), m.getHiddenApiRestrictions(),
new ImmutableMethodImplementation(6, body, null, null)));
p++;
patched++;
touched = true;
System.out.println("[patch] emit -> BIG=" + BIG + " (list=" + pass + ")");
System.out.println("[patch] emit -> constant=" + CONSTANT
+ " (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));
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);
System.out.println("[out ] " + out + " patched=" + patched);
if (patched == 0) {
// The usual causes: wrong target class (R8 renamed it), the method is not
// emit(Object,Continuation), or the class is not in THIS dex.
System.err.println("[fail] target not found in this dex. Check: the exact "
+ "TARGET_CLASS descriptor, that emit(Object,Continuation) exists there, "
+ "and that you loaded the dex that actually contains this class.");
System.exit(1);
}
}
}
scripts/frida_probe.js
+572
View File
@@ -0,0 +1,572 @@
/*
* frida_probe.js -- four-layer full-chain network probe for a hardened Android app.
*
* WHY THIS SHAPE
* --------------
* When you cannot tell why a request never reaches the server (or why it fails
* silently), a single-layer hook always leaves you guessing. This script hooks
* FOUR independent layers at once, so *whichever* layer the app actually uses
* produces evidence:
*
* [APP] the app's own network wrapper class(es) -> what business method asked
* [OKHTTP] OkHttp full chain -> newCall/build/execute/async/proceed
* [RAW-URL] java.net.URL.openConnection -> HttpsURLConnection, which uses the
* SYSTEM trust store and therefore
* fails differently from OkHttp
* [THROW] java.lang.Throwable.getMessage -> recovers the exception text that an
* upper layer swallowed with catch {}
*
* The RAW-URL vs OKHTTP distinction is the single most valuable signal here: an app
* commonly has two unrelated trust chains, so "login fails but browsing works" is
* explained by which of the two fired. See LESSONS G3.
*
* Every hook is installed inside its own try/catch. One missing class (a different
* OkHttp major version, a renamed obfuscated class) never disables the other three
* layers. Failures are reported as HOOK-FAIL lines instead of killing the script.
*
* ADAPT TO YOUR TARGET (only this block matters)
* ---------------------------------------------
* 1) APP_NET_CLASSES -> replace with the app's own network wrapper class name(s)
* 2) APP_PKG_PREFIX -> the app's package prefix, used only to filter noise
* 3) ENABLE_* flags -> turn layers off if they are too noisy
* Everything else is target-independent.
*
* HOW TO RUN
* ----------
* Prefer the bundled injector, which also writes a log file and stays resident:
* python run_probe.py frida_probe.js 10 --pkg com.example.app
* Or by hand against a remote device:
* frida -H 127.0.0.1:27042 -p <pid> -l frida_probe.js
*
* OUTPUT FORMAT
* -------------
* Every line is <ISO-ish local timestamp> <TAG> <key=value ...>
* TAGS: READY | HOOK-FAIL | FATAL | APP | OKHTTP-NEWCALL | OKHTTP-BUILD |
* OKHTTP-EXEC | OKHTTP-ASYNC | OKHTTP-PROCEED | RAW-URL | DNS | THROW
* Add STACK to a line only where a call chain is genuinely useful (APP, RAW-URL),
* because it is the expensive part.
*/
/* ===================================================================
* 1. ADAPT THESE LINES
* =================================================================== */
// The app's own network wrapper / API helper class(es). Enumerate ALL declared
// methods at runtime, so you do not have to know the obfuscated method names.
// Example: 'com.example.app.net.HttpHelper', 'com.example.app.api.ApiClient'
var APP_NET_CLASSES = ['com.example.app.net.HttpHelper'];
// Used only to filter THROW noise: messages from classes whose name contains this
// prefix are always kept. Leave as-is if you do not care.
var APP_PKG_PREFIX = 'com.example.app';
// Layer switches. Turn a layer off if it produces more noise than signal.
var ENABLE_APP = true; // [APP] app's own wrapper classes (needs APP_NET_CLASSES)
var ENABLE_OKHTTP = true; // [OKHTTP] OkHttp newCall/build/execute/AsyncCall/proceed
var ENABLE_RAW_URL = true; // [RAW-URL] java.net.URL.openConnection (HttpsURLConnection)
var ENABLE_THROWABLE = true; // [THROW] Throwable.getMessage, deduplicated
var ENABLE_DNS = true; // [DNS] InetAddress.getAllByName, one line per unique host
var STACK_ON_APP = true; // attach a stack to [APP] hits
var STACK_ON_RAW_URL = true; // attach a stack to [RAW-URL] hits
var STACK_LINES = 12; // how many stack frames to keep
// Safety valves: this probe must never become the thing that kills the app.
var MAX_STRING = 3000; // truncate any single value
var THROWABLE_MAX_LINES = 400;// hard cap on THROW lines
var THROWABLE_STACK = false; // stacks on every Throwable line are usually too much
/* ===================================================================
* 2. Plumbing (target-independent)
* =================================================================== */
function pad(n, w) {
var s = '' + n;
while (s.length < w) s = '0' + s;
return s;
}
// Local wall-clock with milliseconds. Local, not UTC, because you will be
// comparing these lines against logcat and your own shell history.
function stamp() {
var d = new Date();
return d.getFullYear() + '-' + pad(d.getMonth() + 1, 2) + '-' + pad(d.getDate(), 2) + ' ' +
pad(d.getHours(), 2) + ':' + pad(d.getMinutes(), 2) + ':' + pad(d.getSeconds(), 2) + '.' +
pad(d.getMilliseconds(), 3);
}
function str(v, max) {
if (v === null) return 'null';
if (v === undefined) return 'undefined';
var s;
try { s = '' + v; } catch (e) { s = '<unprintable>'; }
max = max || MAX_STRING;
if (s.length > max) s = s.substring(0, max) + '...<+' + (s.length - max) + ' chars>';
return s;
}
function sendLine(tag, fields) {
var parts = [];
if (fields) {
for (var k in fields) {
if (!fields.hasOwnProperty(k)) continue;
if (fields[k] === null || fields[k] === undefined || fields[k] === '') continue;
parts.push(k + '=' + str(fields[k]));
}
}
var line = { t: stamp(), tag: tag, msg: parts.join(' ') };
try { send(line); } catch (e) { /* transport gone; nothing sane to do */ }
}
// Call-chain capture. Expensive, so it is opt-in per hook.
function stackText(lines) {
try {
var Log = Java.use('android.util.Log');
var Throwable = Java.use('java.lang.Throwable');
var full = Log.getStackTraceString(Throwable.$new());
var rows = ('' + full).split('\n');
var kept = [];
// Drop the probe's own frames at the top; they are never the interesting part.
for (var i = 0; i < rows.length && kept.length < (lines || STACK_LINES); i++) {
if (rows[i].indexOf('frida') >= 0) continue;
if (rows[i].indexOf('java.lang.Throwable') >= 0) continue;
if (rows[i].indexOf('android.util.Log') >= 0) continue;
kept.push(rows[i].replace(/^\s+/, ''));
}
return kept.join(' <- ');
} catch (e) {
return '<stack unavailable: ' + e + '>';
}
}
// The whole point: a failure to install one hook must not affect the others.
function guard(label, fn) {
try {
fn();
return true;
} catch (e) {
sendLine('HOOK-FAIL', { where: label, err: e });
return false;
}
}
// Run one whole layer and return how many hooks it really installed.
// > 0 that many hooks are live
// = 0 the layer installed nothing (its classes were not found); NOT a success
// =-1 the layer is switched off in the config header
// Returning a count instead of a boolean matters: a layer whose classes are absent
// still returns normally (each inner hook catches its own error), so a boolean
// "did not throw" would report a layer as installed when it hooked nothing at all.
function runLayer(label, fn) {
try {
return fn();
} catch (e) {
sendLine('HOOK-FAIL', { where: 'install ' + label, err: e });
return 0;
}
}
/* ===================================================================
* 3. Layer [APP]: the app's own network wrapper
* =================================================================== */
// Why getDeclaredMethods: obfuscated wrappers routinely expose the same name with
// several overloads, and a decompiler's short names (a, b, c) are not reliable
// enough to enumerate by hand. Enumerate at runtime, wrap each overload, and let
// the stack tell you which one the business code called.
// Build a replacement of a fixed arity that reports the hit and then delegates.
//
// Two deliberate choices here, both about not breaking the app:
// * The original is NOT cached in a variable. Frida does not hand out the pre-hook
// implementation through `overload.implementation` (it is undefined until you
// replace it), so `var orig = ov.implementation; ... orig.apply(...)` would
// throw a TypeError on the first real call and propagate straight into the app's
// network path. Calling the same name on `this` from inside the replacement is
// the documented way to reach the original, and it re-resolves the overload from
// the runtime argument types.
// * The delegate call is written out per arity instead of spread/apply, so no
// engine-specific behaviour is involved. Methods with more than 8 parameters are
// reported and left alone rather than bound incorrectly.
function appReplacement(className, methodName, arity) {
var label = className + '.' + methodName;
var note = function (argc) {
sendLine('APP', {
call: label,
argc: argc,
stack: STACK_ON_APP ? stackText() : null
});
};
switch (arity) {
case 0: return function () {
note(arguments.length); return this[methodName]();
};
case 1: return function (a0) {
note(arguments.length); return this[methodName](a0);
};
case 2: return function (a0, a1) {
note(arguments.length); return this[methodName](a0, a1);
};
case 3: return function (a0, a1, a2) {
note(arguments.length); return this[methodName](a0, a1, a2);
};
case 4: return function (a0, a1, a2, a3) {
note(arguments.length); return this[methodName](a0, a1, a2, a3);
};
case 5: return function (a0, a1, a2, a3, a4) {
note(arguments.length); return this[methodName](a0, a1, a2, a3, a4);
};
case 6: return function (a0, a1, a2, a3, a4, a5) {
note(arguments.length); return this[methodName](a0, a1, a2, a3, a4, a5);
};
case 7: return function (a0, a1, a2, a3, a4, a5, a6) {
note(arguments.length); return this[methodName](a0, a1, a2, a3, a4, a5, a6);
};
case 8: return function (a0, a1, a2, a3, a4, a5, a6, a7) {
note(arguments.length); return this[methodName](a0, a1, a2, a3, a4, a5, a6, a7);
};
default: return null;
}
}
function installAppLayer() {
if (!ENABLE_APP) return -1;
if (!APP_NET_CLASSES.length) {
sendLine('HOOK-FAIL', { where: 'APP', err: 'APP_NET_CLASSES is empty; edit the header' });
return 0;
}
var wrappedTotal = 0;
APP_NET_CLASSES.forEach(function (className) {
guard('APP ' + className, function () {
var Cls = Java.use(className);
var declared = Cls.class.getDeclaredMethods();
var total = declared.length;
var wrapped = 0;
var skipped = 0;
for (var i = 0; i < total; i++) {
(function (idx) {
var methodName;
try { methodName = '' + declared[idx].getName(); } catch (e) { return; }
guard('APP ' + className + '.' + methodName, function () {
var overloads = Cls[methodName];
if (!overloads || !overloads.overloads) return;
overloads.overloads.forEach(function (ov) {
var argTypes;
try { argTypes = ov.argumentTypes; } catch (e) { argTypes = null; }
var arity = (argTypes && typeof argTypes.length === 'number') ? argTypes.length : -1;
var label = className + '.' + methodName + '(' + (argTypes ? argTypes.join(',') : '?') + ')';
guard('APP bind ' + label, function () {
var replacement = appReplacement(className, methodName, arity);
if (replacement === null) {
skipped++;
sendLine('APP', {
call: label,
skipped: 'arity ' + arity + ' exceeds 8; left unwrapped on purpose'
});
return;
}
ov.implementation = replacement;
wrapped++;
});
});
});
})(i);
}
wrappedTotal += wrapped;
sendLine('APP', {
call: className,
declared_methods: total,
overloads_wrapped: wrapped,
overloads_skipped: skipped
});
});
});
return wrappedTotal;
}
/* ===================================================================
* 4. Layer [OKHTTP]: the whole OkHttp chain
* =================================================================== */
// Five independent anchors on purpose. Any single one can be missing depending on
// the OkHttp major version and on R8 inlining, and each still answers a different
// question:
// newCall -> a request object reached OkHttp (earliest reliable URL)
// build -> the URL existed even if the call was never executed
// execute -> synchronous call actually ran, with status code and duration
// AsyncCall.run-> asynchronous call actually ran (enqueue path)
// proceed -> every actual network step, including redirects and retries
function installOkHttpLayer() {
if (!ENABLE_OKHTTP) return -1;
var installed = 0;
var can = function (label, fn) { if (guard(label, fn)) installed++; };
can('OKHTTP newCall', function () {
var Client = Java.use('okhttp3.OkHttpClient');
Client.newCall.overload('okhttp3.Request').implementation = function (req) {
var url = '<unknown>';
try { url = '' + req.url().toString(); } catch (e) { url = '<url error: ' + e + '>'; }
sendLine('OKHTTP-NEWCALL', { url: url });
return this.newCall(req);
};
});
can('OKHTTP Request$Builder.build', function () {
var Builder = Java.use('okhttp3.Request$Builder');
Builder.build.implementation = function () {
var req = this.build();
var url = '<unknown>';
try { url = '' + req.url().toString(); } catch (e) { url = '<url error: ' + e + '>'; }
sendLine('OKHTTP-BUILD', { url: url });
return req;
};
});
can('OKHTTP RealCall.execute', function () {
var RealCall = Java.use('okhttp3.internal.connection.RealCall');
RealCall.execute.implementation = function () {
var url = '<unknown>';
try { url = '' + this.request().url().toString(); } catch (e) { url = '<url error: ' + e + '>'; }
var t0 = Date.now();
var resp = this.execute();
var code = '?';
try { code = '' + resp.code(); } catch (e) { code = '<code error>'; }
sendLine('OKHTTP-EXEC', { url: url, ms: Date.now() - t0, code: code });
return resp;
};
});
can('OKHTTP RealCall$AsyncCall.run', function () {
var AsyncCall = Java.use('okhttp3.internal.connection.RealCall$AsyncCall');
AsyncCall.run.implementation = function () {
// The URL is not reliably reachable from here across OkHttp versions; the
// newCall/build hooks above supply it. This hook proves *when* the async
// path actually started and finished, which is what matters for timing.
var t0 = Date.now();
this.run();
sendLine('OKHTTP-ASYNC', { phase: 'done', ms: Date.now() - t0 });
};
});
can('OKHTTP RealInterceptorChain.proceed', function () {
var Chain = Java.use('okhttp3.internal.http.RealInterceptorChain');
Chain.proceed.overload('okhttp3.Request').implementation = function (req) {
var url = '<unknown>';
try { url = '' + req.url().toString(); } catch (e) { url = '<url error: ' + e + '>'; }
var t0 = Date.now();
var resp = this.proceed(req);
var code = '?';
try { code = '' + resp.code(); } catch (e) { code = '<code error>'; }
sendLine('OKHTTP-PROCEED', { url: url, ms: Date.now() - t0, code: code });
return resp;
};
});
return installed;
}
/* ===================================================================
* 5. Layer [RAW-URL]: HttpsURLConnection -> the SYSTEM trust store
* =================================================================== */
// Many apps route login/registration through java.net.URL.openConnection instead
// of OkHttp. That path validates against the system trust store, so an expired or
// untrusted server certificate breaks it while the OkHttp path keeps working.
// If this layer fires but the OkHttp layer stays silent, you are on the other
// trust chain -- check the certificate out-of-band with tls_check.py.
function installRawUrlLayer() {
if (!ENABLE_RAW_URL) return -1;
var URL;
try {
URL = Java.use('java.net.URL');
} catch (e) {
sendLine('HOOK-FAIL', { where: 'RAW-URL java.net.URL', err: e });
return 0;
}
var installed = 0;
var can = function (label, fn) { if (guard(label, fn)) installed++; };
can('RAW-URL openConnection()', function () {
// Report AFTER the original returns, so we can also name the concrete connection
// class (HttpURLConnection vs HttpsURLConnection) -- that distinction is what
// tells you which trust chain the request is about to use.
// As in the APP layer, the original is reached via `this.openConnection(...)`
// rather than a cached function reference.
URL.openConnection.overload().implementation = function () {
var url = '<unknown>';
try { url = '' + this.toString(); } catch (e) { url = '<url error: ' + e + '>'; }
var conn = this.openConnection();
var connCls = '?';
try { connCls = '' + conn.getClass().getName(); } catch (e) { connCls = '<class error>'; }
sendLine('RAW-URL', {
url: url,
kind: 'no-arg',
conn: connCls,
stack: STACK_ON_RAW_URL ? stackText() : null
});
return conn;
};
});
can('RAW-URL openConnection(Proxy)', function () {
URL.openConnection.overload('java.net.Proxy').implementation = function (proxy) {
var url = '<unknown>';
try { url = '' + this.toString(); } catch (e) { url = '<url error: ' + e + '>'; }
var conn = this.openConnection(proxy);
var connCls = '?';
try { connCls = '' + conn.getClass().getName(); } catch (e) { connCls = '<class error>'; }
sendLine('RAW-URL', {
url: url,
kind: 'proxy',
conn: connCls,
stack: STACK_ON_RAW_URL ? stackText() : null
});
return conn;
};
});
return installed;
}
/* ===================================================================
* 6. Layer [THROW]: recover swallowed exception text
* =================================================================== */
// An upper-layer `catch (Exception e) { }` hides exactly the message you need.
// Hooking Throwable.getMessage is the cheapest way to get it back, but it is a
// very hot method, so: only concrete Exception/Error subclasses, deduplicated by
// class+message, hard-capped, and stacks off by default.
function installThrowableLayer() {
if (!ENABLE_THROWABLE) return -1;
// Returns 1 only when the hook is really bound to the runtime.
return guard('THROWABLE getMessage', function () {
var seen = {};
var printed = 0;
var suppressed = 0;
var Throwable = Java.use('java.lang.Throwable');
Throwable.getMessage.implementation = function () {
var msg;
try {
msg = this.getMessage();
} catch (e) {
return null;
}
try {
var cls = '' + this.getClass().getName();
var looksLikeError = (cls.indexOf('Exception') >= 0 || cls.indexOf('Error') >= 0);
var isAppClass = (cls.indexOf(APP_PKG_PREFIX) === 0);
if (looksLikeError || isAppClass) {
var key = cls + '|' + msg;
if (seen[key]) {
seen[key]++;
} else if (printed < THROWABLE_MAX_LINES) {
seen[key] = 1;
printed++;
sendLine('THROW', {
cls: cls,
msg: msg,
stack: THROWABLE_STACK ? stackText() : null
});
} else {
suppressed++;
if (suppressed === 1 || suppressed % 200 === 0) {
sendLine('THROW', { note: 'cap reached, further unique messages suppressed', printed: printed });
}
}
}
} catch (e) {
// Never let the reporting path throw: this hook is on a very hot path.
}
return msg;
};
}) ? 1 : 0;
}
/* ===================================================================
* 7. Optional layer [DNS]: which hosts the app really resolves
* =================================================================== */
// Cheap, high-signal: if an SDK's domains are never resolved, that subsystem never
// started -- much stronger evidence than "logcat was quiet". Deduplicated per host
// so a retry loop cannot flood the log.
function installDnsLayer() {
if (!ENABLE_DNS) return -1;
return guard('DNS getAllByName', function () {
var seen = {};
var Inet = Java.use('java.net.InetAddress');
Inet.getAllByName.overload('java.lang.String').implementation = function (host) {
var first = false;
try {
var key = '' + host;
if (!seen[key]) { seen[key] = 1; first = true; }
} catch (e) { /* ignore */ }
if (first) sendLine('DNS', { host: host });
return this.getAllByName(host);
};
}) ? 1 : 0;
}
/* ===================================================================
* 8. Boot
* =================================================================== */
var JAVA_BRIDGE_HINT =
'align the host frida package and the on-device frida-server to the same 16.x version; ' +
'17.x can fail to locate the Android dynamic linker and removes the built-in Java bridge';
function main() {
if (typeof Java === 'undefined' || Java === null) {
// frida 17+ removed the built-in Java bridge.
sendLine('FATAL', {
err: 'Java bridge is not available in this frida runtime',
fix: JAVA_BRIDGE_HINT
});
return;
}
// Java.perform can fail BEFORE the callback ever runs: on a non-Android host, and
// on frida 17.x, where it throws "Java API not available". Without this catch the
// whole script dies as an opaque `type: error` message with a stack pointing into
// frida-java-bridge, which tells you nothing about the version mismatch that
// actually caused it.
try {
Java.perform(function () {
var installed = [];
var failed = [];
var disabled = [];
[
['APP', installAppLayer],
['OKHTTP', installOkHttpLayer],
['RAW-URL', installRawUrlLayer],
['THROW', installThrowableLayer],
['DNS', installDnsLayer]
].forEach(function (pair) {
var count = runLayer(pair[0], pair[1]);
if (count === -1) disabled.push(pair[0]);
else if (count > 0) installed.push(pair[0] + ':' + count);
else failed.push(pair[0]);
});
sendLine('READY', {
installed: installed.join(',') || 'none',
failed: failed.join(',') || 'none',
disabled: disabled.join(',') || 'none',
note: 'hooks are live; now drive the app UI'
});
if (failed.length) {
sendLine('READY-HINT', {
failed: failed.join(','),
fix: 'a layer reporting 0 hooks means its classes were not found: the app may '
+ 'not use that library, R8 may have renamed the class, or the classloader '
+ 'differs. Layers that installed keep working -- debug the failed ones by '
+ 'their HOOK-FAIL line, not the working ones.'
});
}
});
} catch (e) {
sendLine('FATAL', {
err: 'Java.perform failed: ' + e,
fix: JAVA_BRIDGE_HINT
});
}
}
// Last-resort net: any unexpected top-level error still leaves a readable line in
// the log instead of only a raw frida script error.
try {
main();
} catch (e) {
sendLine('FATAL', { err: 'probe aborted: ' + e, note: 'no hooks were installed' });
}
scripts/repack.py
+298 -99
View File
@@ -1,46 +1,47 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Repack + resign pipeline for the ciyuan Android app.
"""Repack an APK with replaced dex files, drop ONLY the signature entries, and re-sign.
Pipeline: replace dex files -> strip META-INF/ -> zip (AndroidManifest.xml and
resources.arsc kept STORED) -> zipalign + sign via uber-apk-signer -> verify.
Pipeline: replace dex files -> drop signature artifacts -> zip, keeping
AndroidManifest.xml and resources.arsc STORED -> zipalign + sign -> verify.
Keystore defaults (generated on first use with keytool):
file : <work>/out/cyc.keystore
alias : cyc
storepass / keypass : cyc123456
Everything machine-specific is a parameter or resolved from PATH. No JDK path is
baked in, no keystore password is baked in, no target app is assumed.
Keystore handling:
--ks <path> is where the keystore lives. If it does not exist, one is generated
with keytool and its password is stored next to it as <ks>.pass.txt, so the next
run reuses it. Passwords are never hardcoded in this script. That file is a local
artifact -- do not commit it.
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
# roundtrip: reuse the original dexes unchanged, proves the pipeline itself works
python 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 \
python repack.py --apk work/orig.apk --dex classes8.dex=work/patch/classes8.dex \
--out work/out/patched.apk
# no signing, no java tooling required at all
python repack.py --apk work/orig.apk --dexdir work/x_orig --out work/unsigned.apk --no-sign
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.
* All java tooling is invoked through python subprocess, never through a host
shell (PowerShell drops arguments in ways that look like tool failures).
* uber-apk-signer silently skips already-signed apks (reports 0 processed), which
is why signature entries are always dropped before signing.
* uber-apk-signer is used for SIGNING to keep the verified path unchanged;
apksigner is used for VERIFYING when present, with an explicit sdk range.
"""
import argparse
import os
import re
import secrets
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')
@@ -49,27 +50,118 @@ def log(msg):
def run(cmd, **kw):
r = subprocess.run(cmd, capture_output=True, text=True, errors='replace', **kw)
return r
return subprocess.run(cmd, capture_output=True, text=True, errors='replace', **kw)
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,
# ---------------------------------------------------------------------------
# Tool resolution: PATH first, explicit flag second, never a baked-in path.
# ---------------------------------------------------------------------------
def sibling_tool(exe, name):
"""Look for `name` next to `exe`, for JDKs whose bin dir is not on PATH."""
if not exe:
return None
folder = os.path.dirname(exe)
for cand in (name, name + '.exe', name + '.bat'):
path = os.path.join(folder, cand)
if os.path.exists(path):
return path
return None
def resolve_tool(name, explicit=None, hint_exe=None):
if explicit:
return explicit
found = shutil.which(name)
if found:
return found
return sibling_tool(hint_exe, name)
def resolve_tools(args):
"""Locate java/keytool/jarsigner/zipalign/apksigner. Missing ones stay None."""
java = resolve_tool('java', args.java)
tools = {
'java': java,
'keytool': resolve_tool('keytool', args.keytool, java),
'jarsigner': resolve_tool('jarsigner', args.jarsigner, java),
'zipalign': resolve_tool('zipalign', args.zipalign),
'apksigner': resolve_tool('apksigner', args.apksigner),
}
return tools
def require(tools, key, why):
if tools.get(key):
return tools[key]
raise SystemExit(
"error: '%s' was not found on PATH, and it is needed to %s.\n"
" install a JDK (17+ recommended) / Android build-tools and put them on PATH,\n"
" or pass the path explicitly (--%s <path>).\n"
" Path lookup is deliberate: a machine-specific absolute path must never be\n"
" baked into this script." % (key, why, key))
# ---------------------------------------------------------------------------
# Keystore
# ---------------------------------------------------------------------------
def read_pwd_file(path):
if not os.path.exists(path):
return None
try:
with open(path, encoding='utf-8') as fh:
txt = fh.read().strip()
return txt or None
except OSError:
return None
def ensure_keystore(ks_path, alias, password, keytool):
"""Return the keystore password, generating the keystore on first use."""
pwd_file = ks_path + '.pass.txt'
if os.path.exists(ks_path):
pwd = password or read_pwd_file(pwd_file)
if not pwd:
raise SystemExit(
'error: keystore %s exists but its password is unknown.\n'
' pass --ks-pass <pw>, or write the password into %s' % (ks_path, pwd_file))
log('[ks] reuse %s (alias=%s)' % (ks_path, alias))
return pwd
pwd = password or read_pwd_file(pwd_file)
generated = pwd is None
if generated:
# keytool rejects passwords shorter than 6 characters.
pwd = secrets.token_urlsafe(12)
folder = os.path.dirname(ks_path)
if folder:
os.makedirs(folder, exist_ok=True)
cmd = [keytool, '-genkeypair', '-keystore', ks_path, '-alias', alias,
'-keyalg', 'RSA', '-keysize', '2048', '-validity', '10000',
'-storepass', password, '-keypass', password,
'-storepass', pwd, '-keypass', pwd,
'-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
try:
with open(pwd_file, 'w', encoding='utf-8') as fh:
fh.write(pwd + '\n')
os.chmod(pwd_file, 0o600)
except OSError as exc:
log('[ks] could not persist the password: %s' % exc)
log('[ks] generated %s (alias=%s)' % (ks_path, alias))
if generated:
log('[ks] password stored in %s -- local artifact, do not commit it' % pwd_file)
return pwd
# ---------------------------------------------------------------------------
# Packing
# ---------------------------------------------------------------------------
def collect_replacement_dex(dexdir=None, dex_pairs=None):
repl = {}
if dexdir:
@@ -79,16 +171,17 @@ def collect_replacement_dex(dexdir=None, dex_pairs=None):
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
name, path = pair.split('=', 1)
repl[name] = path
return repl
def is_signature_entry(name):
"""True only for JAR/APK signature artifacts directly under META-INF/.
"""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.
META-INF/services/**, META-INF/androidx/**, META-INF/native-image/** and every
other subdirectory are RUNTIME RESOURCES and must survive repacking. Anything
that is not directly under META-INF/ is kept by construction.
"""
if not name.upper().startswith('META-INF/'):
return False
@@ -101,20 +194,20 @@ def is_signature_entry(name):
return up.endswith(('.SF', '.RSA', '.DSA', '.EC'))
def build_unsigned(apk, repl, out_apk, drop_meta=True):
def build_unsigned(apk, repl, out_apk, drop_signatures=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:
IMPORTANT (root cause of a startup crash, do not regress):
Stripping the WHOLE META-INF/ breaks ServiceLoader-based runtime wiring. Android
reads these registries at runtime, and they live in the APK:
META-INF/services/kotlinx.coroutines.internal.MainDispatcherFactory
META-INF/services/io.ktor.client.HttpClientEngineContainer
META-INF/services/<lib>.core.* (third-party SDK registries)
META-INF/services/<obfuscated-class-name> (R8-renamed providers)
Deleting them makes the app die at startup with something like:
IllegalStateException: Module with the Main dispatcher is missing ...
So only signature artifacts are removed here; META-INF subdirectories are kept.
and the message never points at META-INF, so the cause is very hard to find.
Therefore: drop signature artifacts only, keep every META-INF subdirectory.
"""
seen = set()
replaced_names = set(repl)
@@ -124,7 +217,8 @@ def build_unsigned(apk, repl, out_apk, drop_meta=True):
base = os.path.basename(name)
if item.is_dir():
continue
if drop_meta and is_signature_entry(name):
if drop_signatures and is_signature_entry(name):
log('[zip] - %s (signature artifact)' % name)
continue
if name in replaced_names:
continue
@@ -138,8 +232,8 @@ def build_unsigned(apk, repl, out_apk, drop_meta=True):
zout.writestr(zi, data)
seen.add(name)
for name, path in sorted(repl.items()):
with open(path, 'rb') as f:
data = f.read()
with open(path, 'rb') as fh:
data = fh.read()
zi = zipfile.ZipInfo(name, date_time=(2024, 1, 1, 0, 0, 0))
zi.compress_type = zipfile.ZIP_DEFLATED
zout.writestr(zi, data)
@@ -160,12 +254,23 @@ def zip_report(apk):
return rows
def sign_apk(unsigned_apk, workdir, ks, alias=KS_ALIAS, password=KS_PASS):
# ---------------------------------------------------------------------------
# Sign / verify
# ---------------------------------------------------------------------------
def sign_apk(unsigned_apk, workdir, ks, alias, password, tools, signer_jar):
java = require(tools, 'java', 'run the signer jar')
if not os.path.exists(signer_jar):
raise SystemExit(
'error: signer jar not found: %s\n'
' pass --signer-jar <path>, or set APK_SIGNER_JAR.\n'
' Any zipalign+apksigner based signer works here.' % signer_jar)
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,
cmd = [java, '-jar', signer_jar,
'--apks', unsigned_apk,
'--ks', ks, '--ksAlias', alias,
'--ksPass', password, '--ksKeyPass', password,
@@ -175,67 +280,149 @@ def sign_apk(unsigned_apk, workdir, ks, alias=KS_ALIAS, password=KS_PASS):
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 []
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
return cand
def verify_apk(apk):
def verify_apk(apk, tools, signer_jar=None, expect_signed=True):
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)
sig_checks = 0 # real signature-verification tools that actually ran
# 1) apksigner -- the authoritative check, but ONLY with an explicit sdk range.
#
# READ THIS BEFORE CONCLUDING THE SIGNATURE IS BROKEN:
# `apksigner verify` with no range checks the signature schemes implied by the
# APK's own minSdkVersion. With minSdk >= 24 it prints v1/v2 as false while the
# files in META-INF are perfectly fine, which looks exactly like a failed signing
# step. Always pass --min-sdk-version / --max-sdk-version so the v1/v2/v3 results
# are meaningful.
if tools.get('apksigner'):
sig_checks += 1
r = run([tools['apksigner'], 'verify', '--print-certs', '--verbose',
'--min-sdk-version', '21', '--max-sdk-version', '34', apk])
log('[verify:apksigner] rc=%d\n%s' % (r.returncode, ((r.stdout or '') +
(r.stderr or '')).strip()[:3000]))
if r.returncode != 0:
ok = False
else:
log('[verify:apksigner] skipped: apksigner not on PATH.\n'
' Do NOT judge the signature from a bare `apksigner verify`: with\n'
' minSdk >= 24 its default range reports v1/v2 as false even when the\n'
' signature is valid. The correct invocation is:\n'
' apksigner verify --print-certs --verbose --min-sdk-version 21 '
'--max-sdk-version 34 <apk>')
# 2) signer jar's own verify (also re-checks alignment)
if signer_jar and os.path.exists(signer_jar) and tools.get('java'):
sig_checks += 1
r = run([tools['java'], '-jar', signer_jar, '-a', apk, '-y'])
log('[verify:signer] rc=%d\n%s' % (r.returncode, (r.stdout or '').strip()))
if r.returncode != 0 and 'DOES NOT VERIFY' in (r.stdout or ''):
ok = False
# 3) jarsigner (v1 / JAR signature)
if tools.get('jarsigner'):
sig_checks += 1
r = run([tools['jarsigner'], '-verify', '-certs', apk])
txt = ((r.stdout or '') + (r.stderr or '')).strip()
log('[verify:jarsigner] rc=%d\n%s' % (r.returncode, txt[:2000]))
if 'jar verified' not in txt and 'verified' not in txt.lower():
log('[verify:jarsigner] note: not reported as verified (see output above)')
# 4) keytool cert dump
if tools.get('keytool'):
r = run([tools['keytool'], '-printcert', '-jarfile', apk])
log('[verify:keytool] rc=%d\n%s' % (r.returncode, (r.stdout or '').strip()[:1500]))
# 5) zipalign check
if tools.get('zipalign'):
r = run([tools['zipalign'], '-c', '-v', '4', apk])
log('[verify:zipalign] rc=%d' % r.returncode)
if r.returncode != 0:
log((r.stdout or '').strip()[:1500])
else:
log('[verify:zipalign] skipped: zipalign not on PATH')
# 6) v2/v3 APK Signing Block presence, read straight from the file
with open(apk, 'rb') as fh:
blob = fh.read()
log('[verify:v2block] %s' % ('present' if b'APK Sig Block 42' in blob else 'MISSING'))
# A build nobody could verify is not a passing build. Saying OK here would
# violate the first rule of this skill ("an APK is not done until it is
# verified"): report UNVERIFIED and fail instead.
if expect_signed and sig_checks == 0:
log('[result] UNVERIFIED: no signature verification tool was available '
'(apksigner / jarsigner / signer jar). This is NOT an OK result.')
return False
return ok
def fingerprint(apk):
r = run([KEYTOOL, '-printcert', '-jarfile', apk])
sha = re.findall(r'(SHA1|SHA256):\s*([0-9A-F:]+)', r.stdout)
return sha
def fingerprint(apk, tools):
if not tools.get('keytool'):
return []
r = run([tools['keytool'], '-printcert', '-jarfile', apk])
return re.findall(r'(SHA1|SHA256):\s*([0-9A-F:]+)', r.stdout or '')
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)
ap = argparse.ArgumentParser(
description='Repack an APK with replaced dex files, drop only the signature '
'entries, re-sign and verify.')
ap.add_argument('--apk', required=True, help='source apk to use as the template')
ap.add_argument('--out', required=True, help='output apk path')
ap.add_argument('--dexdir', help='directory of classes*.dex to swap in')
ap.add_argument('--dex', action='append', default=[],
help='name=path, repeatable, e.g. classes8.dex=work/patch/classes8.dex')
ap.add_argument('--workdir', default=None,
help='scratch directory (default: next to --out)')
ap.add_argument('--ks', default=None, help='keystore path (created on first use)')
ap.add_argument('--ks-alias', default='apkreverse', help='keystore alias')
ap.add_argument('--ks-pass', default=None,
help='keystore password; default: <ks>.pass.txt, else generated')
ap.add_argument('--signer-jar', default=None,
help='jar used for zipalign+sign (default: $APK_SIGNER_JAR or '
'uber-apk-signer.jar)')
ap.add_argument('--no-sign', action='store_true', help='stop after writing the apk')
ap.add_argument('--java', default=None, help='path to java (default: from PATH)')
ap.add_argument('--keytool', default=None, help='path to keytool (default: from PATH)')
ap.add_argument('--jarsigner', default=None, help='path to jarsigner (default: from PATH)')
ap.add_argument('--zipalign', default=None, help='path to zipalign (default: from PATH)')
ap.add_argument('--apksigner', default=None, help='path to apksigner (default: from PATH)')
args = ap.parse_args()
tools = resolve_tools(args)
signer_jar = args.signer_jar or os.environ.get('APK_SIGNER_JAR', 'uber-apk-signer.jar')
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)
# Default the scratch dir next to --out; never to a machine-specific path.
workdir = os.path.abspath(args.workdir) if args.workdir else os.path.dirname(out)
if not workdir:
workdir = os.getcwd()
os.makedirs(workdir, exist_ok=True)
out_folder = os.path.dirname(out)
if out_folder:
os.makedirs(out_folder, 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())))
log('[tools] java=%s keytool=%s apksigner=%s zipalign=%s'
% (tools['java'], tools['keytool'], tools['apksigner'], tools['zipalign']))
log('[repl] %s' % (', '.join('%s<-%s' % (k, v) for k, v in sorted(repl.items())) or '<none>'))
unsigned = os.path.join(args.workdir, 'unsigned.apk')
unsigned = os.path.join(workdir, 'unsigned.apk')
if os.path.exists(unsigned):
os.remove(unsigned)
build_unsigned(apk, repl, unsigned)
@@ -246,19 +433,31 @@ def main():
if args.no_sign:
shutil.copy2(unsigned, out)
else:
ks = ensure_keystore(args.ks)
signed, _ = sign_apk(unsigned, args.workdir, ks)
if not args.ks:
raise SystemExit(
"error: --ks is required when signing (there is no default keystore).\n"
" e.g. --ks work/out/release.keystore")
keytool = require(tools, 'keytool', 'create the keystore on first use')
password = ensure_keystore(os.path.abspath(args.ks), args.ks_alias,
args.ks_pass, keytool)
signed = sign_apk(unsigned, workdir, os.path.abspath(args.ks), args.ks_alias,
password, tools, signer_jar)
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'))
ok = verify_apk(out, tools, signer_jar, expect_signed=not args.no_sign)
if not args.no_sign:
for algo, val in fingerprint(out, tools):
log('[cert] %s %s' % (algo, val))
log('[result] %s' % ('OK' if ok else 'CHECK-FAILED'))
else:
log('[result] UNSIGNED (--no-sign): good for inspection, not installable as-is')
log('[reminder] repacking is not done until the app launches and the changed '
'behavior is exercised on a device: see scripts/install_test.py')
return 0 if ok else 2
scripts/run_probe.py
+360
View File
@@ -0,0 +1,360 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Inject a Frida probe script into a running Android app and stay resident.
WHY THIS EXISTS
---------------
`frida -U -f <pkg> -l script.js` has two failure modes that cost real debugging time:
1. With a real device AND an emulator both attached, the USB auto-selection picks
the emulator, so you attach to the wrong process and see nothing. This tool
prefers an explicit remote device over `adb forward`, which cannot be hijacked.
2. Probe output that only goes to a terminal is lost the moment the terminal
scrolls, the session detaches, or the app crashes. Here every message is
written to a log file *and* to stdout, with both the device timestamp and the
host arrival timestamp -- that pair is itself evidence when you suspect a
clock skew problem.
It also answers the three questions that otherwise produce a bare traceback:
* is the `frida` python module importable, and which version?
* is `adb` reachable, and is the target process actually running?
* is an on-device frida-server answering on the forwarded port?
USAGE
-----
python run_probe.py frida_probe.js 10 --pkg com.example.app
python run_probe.py frida_probe.js 5 --pkg com.example.app --device <serial>
python run_probe.py frida_probe.js 5 --pkg com.example.app --via usb
python run_probe.py frida_probe.js 2 --pkg com.example.app --spawn
python run_probe.py frida_probe.js 10 --pkg com.example.app --log my.log --port 27043
<script.js> the probe to inject (e.g. frida_probe.js, see its header to adapt)
<minutes> how long to stay attached (default 10)
NOTES
-----
* The host `frida` package and the on-device `frida-server` must be the SAME
version, and 16.x is the version to align on: 17.x can fail to locate the
Android dynamic linker and removes the built-in Java bridge.
* `--spawn` is the right choice when the problem happens at startup: attach-only
misses init and the first network calls. Hooks are installed before resume.
* Log file: a timestamped name is generated unless --log is given.
"""
import argparse
import os
import shutil
import subprocess
import sys
import time
from datetime import datetime
DEFAULT_PORT = 27042
DEFAULT_MINUTES = 10
DEFAULT_SERVER_PORT = 27042 # the port frida-server listens on, on the device
FRIDA_INSTALL_HINT = (
'install the matching host package and the matching device server, e.g.\n'
' pip install frida==16.7.19\n'
' # then push frida-server-16.7.19-android-<abi> to the device and run it as root'
)
def host_ts():
return datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
class Logger(object):
"""Write every line to stdout AND to a file, flushing both immediately."""
def __init__(self, path):
self.path = path
self.fh = open(path, 'a', encoding='utf-8', errors='replace')
def line(self, text):
print(text, flush=True)
self.fh.write(text + '\n')
self.fh.flush()
def close(self):
try:
self.fh.close()
except Exception:
pass
def adb_run(adb, serial, args, timeout=30):
cmd = [adb]
if serial:
cmd += ['-s', serial]
cmd += args
return subprocess.run(cmd, capture_output=True, text=True,
errors='replace', timeout=timeout)
def find_adb(explicit=None):
if explicit:
return explicit
return shutil.which('adb') or shutil.which('adb.exe')
def get_pid(adb, serial, pkg):
"""Return (pid, raw_output). pid is None when the process is not running."""
try:
r = adb_run(adb, serial, ['shell', 'pidof', pkg])
except subprocess.TimeoutExpired:
return None, 'adb timed out'
raw = ((r.stdout or '') + (r.stderr or '')).strip()
if r.returncode != 0 or not raw:
return None, raw
first = raw.split()[0]
if not first.isdigit():
return None, raw
return int(first), raw
def attach_remote(frida, port, logger):
mgr = frida.get_device_manager()
logger.line('[host %s] registering remote device 127.0.0.1:%d' % (host_ts(), port))
return mgr.add_remote_device('127.0.0.1:%d' % port)
def attach_usb(frida, device_id, logger):
if device_id:
logger.line('[host %s] selecting usb device %s' % (host_ts(), device_id))
return frida.get_device(device_id, timeout=5)
logger.line('[host %s] selecting usb device (auto; an emulator can win this race)'
% host_ts())
return frida.get_usb_device(timeout=5)
def format_message(message):
"""Turn a frida message into one log line. Never raises."""
kind = message.get('type')
if kind == 'send':
payload = message.get('payload')
if isinstance(payload, dict):
dev = payload.get('t') or '-'
tag = payload.get('tag') or 'SEND'
msg = payload.get('msg')
if msg is None:
msg = payload
return dev, '%s %s' % (tag, msg)
return '-', 'SEND %r' % (payload,)
if kind == 'error':
desc = message.get('description') or ''
where = '%s:%s' % (message.get('fileName') or '?', message.get('lineNumber') or '?')
stack = message.get('stack') or ''
out = 'FRIDA-ERROR %s (%s)' % (desc, where)
if stack:
out += '\n ' + stack.replace('\n', '\n ')
return '-', out
return '-', '%s %r' % (kind, message)
def main():
ap = argparse.ArgumentParser(
description='Inject a Frida probe into a running Android app, log everything '
'to stdout and a file, and stay attached for N minutes.')
ap.add_argument('script', help='path to the frida JS probe, e.g. frida_probe.js')
ap.add_argument('minutes', nargs='?', type=float, default=DEFAULT_MINUTES,
help='how long to stay attached (default %d)' % DEFAULT_MINUTES)
ap.add_argument('--pkg', default=None, help='target package name, e.g. com.example.app')
ap.add_argument('--device', default=None,
help='adb serial (remote mode) or frida device id (usb mode)')
ap.add_argument('--via', choices=['remote', 'usb'], default='remote',
help='remote = adb forward + explicit remote device (default, '
'cannot be hijacked by an emulator); usb = frida.get_usb_device()')
ap.add_argument('--port', type=int, default=DEFAULT_PORT,
help='host-side forwarded port (default %d)' % DEFAULT_PORT)
ap.add_argument('--server-port', type=int, default=DEFAULT_SERVER_PORT,
help='port frida-server listens on, on the device (default %d)'
% DEFAULT_SERVER_PORT)
ap.add_argument('--spawn', action='store_true',
help='spawn the app instead of attaching, so hooks are live before '
'the first network call')
ap.add_argument('--log', default=None, help='log file path (default: auto-named)')
ap.add_argument('--adb', default=None, help='path to adb (default: from PATH)')
ap.add_argument('--script-timeout', type=float, default=30.0,
help='seconds to wait for the script to load')
ap.add_argument('--keep-forward', action='store_true',
help='leave the adb forward in place after exit')
args = ap.parse_args()
if not os.path.isfile(args.script):
print('probe script not found: %s' % args.script)
return 2
if args.spawn and not args.pkg:
print('--spawn needs --pkg (there is no pid to attach to)')
return 2
log_name = args.log or ('probe-%s-%s.log'
% (args.pkg or 'target', datetime.now().strftime('%Y%m%d-%H%M%S')))
logger = Logger(log_name)
logger.line('# probe log %s' % os.path.abspath(log_name))
logger.line('# script=%s minutes=%s via=%s pkg=%s'
% (os.path.abspath(args.script), args.minutes, args.via, args.pkg))
# 1) frida python package
try:
import frida
except ImportError:
logger.line(host_ts() + ' FATAL cannot import the frida python module.\n' +
FRIDA_INSTALL_HINT)
logger.close()
return 3
logger.line('[host %s] frida python %s' % (host_ts(), getattr(frida, '__version__', '?')))
# 2) adb
adb = find_adb(args.adb)
if not adb:
logger.line(host_ts() + ' FATAL adb not found.\n'
' put platform-tools on PATH, or pass --adb <path>')
logger.close()
return 3
logger.line('[host %s] adb %s' % (host_ts(), adb))
pid = None
forward_added = False
device = None
session = None
script = None
counts = {}
deadline = time.monotonic() + max(args.minutes, 0) * 60.0
try:
# 3) remote device over adb forward (default), or plain USB selection
if args.via == 'remote':
try:
r = adb_run(adb, args.device,
['forward', 'tcp:%d' % args.port, 'tcp:%d' % args.server_port])
except subprocess.TimeoutExpired:
logger.line(host_ts() + ' FATAL adb forward timed out; is the device responsive?')
return 3
if r.returncode != 0:
msg = ((r.stdout or '') + (r.stderr or '')).strip()
logger.line(host_ts() + ' FATAL adb forward failed.\n'
' adb said: %s\n'
' if it mentions more than one device, pass --device <serial>;\n'
' check the device is visible with: adb devices' % msg)
return 3
forward_added = True
logger.line('[host %s] forwarded tcp:%d -> device tcp:%d'
% (host_ts(), args.port, args.server_port))
device = attach_remote(frida, args.port, logger)
else:
device = attach_usb(frida, args.device, logger)
# 4) pid: spawn when asked, otherwise attach to the running process
if args.spawn:
try:
pid = device.spawn([args.pkg])
except Exception as exc:
logger.line(host_ts() + ' FATAL spawn failed for %s: %s\n'
' check the package is installed: adb shell pm list packages | grep %s'
% (args.pkg, exc, args.pkg))
return 3
logger.line('[device] spawned %s pid=%s' % (args.pkg, pid))
else:
if not args.pkg:
logger.line(host_ts() + ' FATAL --pkg is required when attaching '
'(or use --spawn with --pkg)')
return 3
pid, raw = get_pid(adb, args.device, args.pkg)
if pid is None:
logger.line(host_ts() + ' FATAL process not found: %s\n'
' adb output: %s\n'
' start the app first, or use --spawn to launch it under the probe;\n'
' if pidof is unavailable on this ROM, try: adb shell ps -A'
% (args.pkg, raw or '<empty>'))
return 3
logger.line('[device] pid of %s = %d' % (args.pkg, pid))
# 5) attach + load
try:
session = device.attach(pid)
except Exception as exc:
logger.line(host_ts() + ' FATAL attach failed: %s\n'
' usual causes: frida-server not running on the device, a\n'
' host/server version mismatch, or an anti-instrumentation check.\n'
' %s' % (exc, FRIDA_INSTALL_HINT))
return 3
with open(args.script, encoding='utf-8') as fh:
source = fh.read()
def on_message(message, data):
dev, text = format_message(message)
tag = 'OTHER'
if isinstance(message.get('payload'), dict):
tag = message['payload'].get('tag') or 'SEND'
elif message.get('type') == 'error':
tag = 'FRIDA-ERROR'
counts[tag] = counts.get(tag, 0) + 1
logger.line('[host %s] [dev %s] %s' % (host_ts(), dev, text))
if message.get('type') == 'error':
desc = message.get('description') or ''
if 'Java API not available' in desc or 'Java is not defined' in desc:
logger.line('[hint] the script loaded but this runtime has no Java bridge. '
'That is a version problem, not a hook problem: align the host '
'frida package and the on-device frida-server to the same 16.x '
'version, then retry. Do not debug the hooks until READY appears.')
script = session.create_script(source)
script.on('message', on_message)
try:
script.load()
except Exception as exc:
logger.line(host_ts() + ' FATAL script load failed: %s\n'
' if this says "Java is not defined", the runtime has no Java\n'
' bridge: align host frida and on-device frida-server to 16.x.'
% exc)
return 3
if args.spawn:
device.resume(pid)
logger.line('[device] resumed pid=%d (hooks were installed before resume)' % pid)
logger.line('[host %s] attached; staying resident for %s minute(s). '
'Now drive the app UI.' % (host_ts(), args.minutes))
while time.monotonic() < deadline:
time.sleep(0.5)
except KeyboardInterrupt:
logger.line('[host %s] interrupted; detaching' % host_ts())
finally:
if script is not None:
try:
script.unload()
except Exception:
pass
if session is not None:
try:
session.detach()
except Exception:
pass
if forward_added and not args.keep_forward:
try:
adb_run(adb, args.device, ['forward', '--remove', 'tcp:%d' % args.port])
logger.line('[host %s] removed adb forward tcp:%d' % (host_ts(), args.port))
except Exception:
pass
logger.line('')
logger.line('== summary: %d message(s)' % sum(counts.values()))
for tag in sorted(counts):
logger.line(' %-16s %d' % (tag, counts[tag]))
if not counts:
logger.line(' no probe output at all. Ordered checklist:')
logger.line(' 1. did READY appear? if not, the script never loaded')
logger.line(' 2. is the class/overload name you configured really present?')
logger.line(' 3. ClassLoader: the hooks bind against the app classloader')
logger.line(' 4. did the UI action actually reach the business code?')
logger.line(' 5. check THROW lines: a local validation may have returned early')
logger.line('# log written to %s' % os.path.abspath(log_name))
logger.close()
return 0
if __name__ == '__main__':
sys.exit(main())
scripts/tls_check.py
+323
View File
@@ -0,0 +1,323 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""TLS certificate health check for one or more hosts.
WHY THIS EXISTS
---------------
When a repacked app's login/registration fails with
`Chain validation failed` / `SSLHandshakeException`, the cause is often the
*server* certificate, not your patch. This script answers that question
independently of the app, so you can rule the server side in or out **before**
touching the APK. It is step 2 of the three-step TLS triage:
1. take the real request URL / host out of the runtime probe (see frida_probe.js)
2. verify that host's certificate out-of-band <-- this script
3. compare with another host of the same app (usually one of them is fine),
which proves the failure is host-specific and not your clock/network
WHY IT STILL PRINTS THE CERTIFICATE WHEN VERIFICATION FAILS
-----------------------------------------------------------
`ssl.SSLSocket.getpeercert()` returns an empty dict whenever the handshake was
not validated, and `verification_mode=CERT_NONE` never validates. So on failure
we re-read the peer certificate in DER form and decode it locally. That is what
turns "it failed" into "it failed because notAfter is 39 days in the past".
WHAT IT REPORTS
---------------
Per host: strict verification result, protocol/cipher, subject, issuer,
notBefore, notAfter, days remaining, and a classification of the failure:
EXPIRED certificate has expired
NOT-YET-VALID certificate is not valid yet
HOSTNAME-MISMATCH chain is fine, the name does not match
UNTRUSTED-CA self-signed, or issuer not in the trust store
CHAIN-BROKEN bad signature / invalid CA / incomplete chain
REVOKED revoked by the issuer
OTHER anything else, with the raw OpenSSL message
Exit code: 0 if every host verified strictly, 2 if any host failed.
Pure standard library, Python 3.9+.
USAGE
-----
python tls_check.py api.example.com
python tls_check.py api.example.com cdn.example.com
python tls_check.py api.example.com --port 8443
python tls_check.py 203.0.113.10 --sni api.example.com # IP + real SNI name
python tls_check.py api.example.com --json # machine-readable
"""
import argparse
import hashlib
import json
import os
import socket
import ssl
import sys
import tempfile
from datetime import datetime, timezone
DEFAULT_PORT = 443
DEFAULT_TIMEOUT = 10
# OpenSSL X509_V_ERR_* -> (classification, human readable)
VERIFY_CODE_CLASS = {
7: ('CHAIN-BROKEN', 'certificate signature failure'),
9: ('NOT-YET-VALID', 'certificate is not yet valid'),
10: ('EXPIRED', 'certificate has expired'),
18: ('UNTRUSTED-CA', 'self-signed certificate'),
19: ('UNTRUSTED-CA', 'self-signed certificate in certificate chain'),
20: ('UNTRUSTED-CA', 'unable to get local issuer certificate'),
21: ('UNTRUSTED-CA', 'unable to verify the first certificate'),
24: ('CHAIN-BROKEN', 'invalid CA certificate'),
27: ('REVOKED', 'certificate revoked'),
62: ('HOSTNAME-MISMATCH', 'hostname mismatch'),
}
# What a failure means for the reverse-engineering workflow.
CLASS_ACTION = {
'EXPIRED': 'server-side problem: the certificate chain is not trustworthy '
'any more. Re-signing the APK cannot fix this, and a client patch '
'is the only way to keep the app usable.',
'NOT-YET-VALID': 'either the server is misconfigured or the device clock is '
'wrong. Check the device date before blaming the app.',
'HOSTNAME-MISMATCH': 'the chain is valid but the connection is using the '
'wrong name. Check for a proxy, a hosts entry, or a '
'hardcoded IP + wrong SNI.',
'UNTRUSTED-CA': 'the presented chain is not rooted in the system trust store. '
'Expected for a MITM/proxy setup; suspicious for a public API.',
'CHAIN-BROKEN': 'the server sent an incomplete or internally inconsistent '
'chain. Often a missing intermediate certificate.',
'REVOKED': 'the certificate was revoked by its issuer.',
'OTHER': 'read the raw OpenSSL message below; this is not one of the '
'well-known failure classes.',
}
def fmt_name(parts):
"""Flatten ((('commonName','x'),),) into 'commonName=x'."""
if not parts:
return None
flat = []
for rdn in parts:
for key, val in rdn:
flat.append('%s=%s' % (key, val))
return ', '.join(flat)
def parse_asn1_time(text):
"""'Nov 29 23:59:59 2026 GMT' -> aware UTC datetime (or None)."""
if not text:
return None
cleaned = text.strip()
if cleaned.endswith(' GMT') or cleaned.endswith(' UTC'):
cleaned = cleaned[:-4].rstrip()
try:
return datetime.strptime(cleaned, '%b %d %H:%M:%S %Y').replace(tzinfo=timezone.utc)
except ValueError:
return None
def decode_der(der):
"""Decode a DER certificate without ever verifying it.
Uses the CPython private helper that backs `getpeercert()`; if it is not
available we still return what we can (length + fingerprint), because a
partially decoded certificate beats no certificate at all.
"""
info = {'der_bytes': len(der),
'sha256': hashlib.sha256(der).hexdigest(),
'sha1': hashlib.sha1(der).hexdigest()}
helper = getattr(getattr(ssl, '_ssl', None), '_test_decode_cert', None)
if helper is None:
return info
path = None
try:
pem = ssl.DER_cert_to_PEM_cert(der)
with tempfile.NamedTemporaryFile('w', suffix='.pem', delete=False) as fh:
path = fh.name
fh.write(pem)
info.update(helper(path))
except Exception:
pass
finally:
if path and os.path.exists(path):
try:
os.remove(path)
except OSError:
pass
return info
def fetch_peer_cert(host, port, sni, timeout):
"""Read the peer certificate even when the chain cannot be validated."""
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with socket.create_connection((host, port), timeout=timeout) as raw:
with ctx.wrap_socket(raw, server_hostname=sni) as tls:
der = tls.getpeercert(binary_form=True)
return decode_der(der) if der else {}
def strict_connect(host, port, sni, timeout):
"""Full strict verification: system trust store + hostname match."""
ctx = ssl.create_default_context()
with socket.create_connection((host, port), timeout=timeout) as raw:
with ctx.wrap_socket(raw, server_hostname=sni) as tls:
return {
'cert': tls.getpeercert() or {},
'protocol': tls.version(),
'cipher': (tls.cipher() or ('', '', ''))[0],
}
def check_host(host, port, sni, timeout):
"""Return one result dict; never raises for network/TLS problems."""
res = {
'host': host,
'port': port,
'sni': sni,
'ok': False,
'classification': None,
'reason': None,
'verify_code': None,
'protocol': None,
'cipher': None,
'subject': None,
'issuer': None,
'not_before': None,
'not_after': None,
'days_remaining': None,
'fingerprint_sha256': None,
'action': None,
'error': None,
}
try:
hit = strict_connect(host, port, sni, timeout)
except ssl.SSLCertVerificationError as exc:
code = getattr(exc, 'verify_code', None)
cls, human = VERIFY_CODE_CLASS.get(code, ('OTHER', getattr(exc, 'verify_message', str(exc))))
res['classification'] = cls
res['reason'] = human
res['verify_code'] = code
res['action'] = CLASS_ACTION.get(cls)
res['error'] = '%s: %s' % (type(exc).__name__, exc)
except (ssl.SSLError, socket.error, OSError) as exc:
res['classification'] = 'OTHER'
res['reason'] = 'connection or handshake failed before certificate validation'
res['action'] = CLASS_ACTION['OTHER']
res['error'] = '%s: %s' % (type(exc).__name__, exc)
return res
else:
cert = hit['cert']
res.update({'ok': True, 'protocol': hit['protocol'], 'cipher': hit['cipher'],
'subject': fmt_name(cert.get('subject')),
'issuer': fmt_name(cert.get('issuer')),
'not_before': cert.get('notBefore'),
'not_after': cert.get('notAfter')})
res['days_remaining'] = days_left(res['not_after'])
return res
# Verification failed: still try to show the real certificate.
try:
cert = fetch_peer_cert(host, port, sni, timeout)
except Exception as exc:
res['error'] = (res['error'] or '') + ' | peek failed: %s' % exc
return res
res.update({'subject': fmt_name(cert.get('subject')),
'issuer': fmt_name(cert.get('issuer')),
'not_before': cert.get('notBefore'),
'not_after': cert.get('notAfter'),
'fingerprint_sha256': cert.get('sha256')})
res['days_remaining'] = days_left(res['not_after'])
return res
def days_left(not_after):
dt = parse_asn1_time(not_after)
if dt is None:
return None
return (dt - datetime.now(timezone.utc)).days
def report(res, out=sys.stdout):
w = out.write
w('== %s:%d%s\n' % (res['host'], res['port'],
'' if res['sni'] == res['host'] else ' (SNI: %s)' % res['sni']))
if res['ok']:
w(' status : OK (strict verification passed)\n')
else:
w(' status : FAIL - %s\n' % res['classification'])
w(' reason : %s\n' % res['reason'])
if res['verify_code'] is not None:
w(' verify_code : %d\n' % res['verify_code'])
if res['protocol']:
w(' protocol : %s cipher: %s\n' % (res['protocol'], res['cipher']))
w(' subject : %s\n' % (res['subject'] or '<undecoded>'))
w(' issuer : %s\n' % (res['issuer'] or '<undecoded>'))
w(' notBefore : %s\n' % (res['not_before'] or '<unknown>'))
w(' notAfter : %s\n' % (res['not_after'] or '<unknown>'))
days = res['days_remaining']
if days is not None:
if days < 0:
w(' validity : EXPIRED %d days ago\n' % (-days))
else:
w(' days remaining : %d\n' % days)
if res['fingerprint_sha256']:
w(' sha256 : %s\n' % res['fingerprint_sha256'])
if res['action'] and not res['ok']:
w(' action : %s\n' % res['action'])
if res['error']:
w(' raw error : %s\n' % res['error'])
w('\n')
def main():
ap = argparse.ArgumentParser(
description='Verify TLS certificates for one or more hosts, strictly, '
'and classify any failure (expired / hostname mismatch / untrusted CA).')
ap.add_argument('hosts', nargs='+', metavar='HOST',
help='hostname or IP to check, e.g. api.example.com')
ap.add_argument('--port', type=int, default=DEFAULT_PORT,
help='TCP port (default %d)' % DEFAULT_PORT)
ap.add_argument('--sni', default=None,
help='override the SNI/hostname to validate against; useful when '
'HOST is a bare IP, e.g. --sni api.example.com')
ap.add_argument('--timeout', type=float, default=DEFAULT_TIMEOUT,
help='connect timeout in seconds (default %d)' % DEFAULT_TIMEOUT)
ap.add_argument('--json', action='store_true', help='emit JSON instead of a report')
args = ap.parse_args()
results = []
for host in args.hosts:
sni = args.sni or host
results.append(check_host(host, args.port, sni, args.timeout))
if args.json:
json.dump(results, sys.stdout, indent=2, sort_keys=True)
sys.stdout.write('\n')
else:
for res in results:
report(res)
bad = [r for r in results if not r['ok']]
good = [r for r in results if r['ok']]
if len(results) > 1:
if bad and good:
print('SUMMARY: %d/%d hosts failed. At least one host on this app '
'verifies fine, so the failure is host-specific (server side), '
'not a clock/network problem on your side.'
% (len(bad), len(results)))
elif bad:
print('SUMMARY: every host failed. Suspect the network path, a proxy, '
'or the device/host clock before blaming the app.')
else:
print('SUMMARY: all %d hosts verified strictly.' % len(results))
for r in bad:
print('FAIL %s -> %s' % (r['host'], r['classification']))
return 0 if all(r['ok'] for r in results) else 2
if __name__ == '__main__':
sys.exit(main())