Add native tamper/suicide guidance, toolchain map, and three scripts
references/native-tamper-and-suicide.md - how a hardened library terminates its own process: the mechanisms and their signatures, how to tell which one actually fires, arranged faults that call nothing (defeating every PLT-level fix), resolving a stub to its symbol, forged section headers, function boundaries from PT_GNU_EH_FRAME, scanner traps, and neutralising by returning rather than spinning. references/toolchain.md - what to install, non-interactive invocation, GUI-only tools (ask for a human instead of silently substituting a weaker method), version-alignment traps, and working offline. scripts/elf_plt.py - PLT stub to imported symbol on x86_64 and aarch64 from the relocation table; caller listing; library byte-diff naming each changed stub's symbol. scripts/apk_diff.py - entry-level diff by content hash, so same-size replacements are caught. scripts/native_crash.py - locate a native death from a log or tombstone and flag an arranged fault. pitfalls P31-P36; SKILL.md gains a decision question, a hard constraint on neutralising terminate paths, and cross-references from packers/native-and-so.
This commit is contained in:
@@ -0,0 +1,164 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Entry-level APK diff: what actually changed between two builds.
|
||||
|
||||
Two jobs, both recurring:
|
||||
|
||||
1. **Audit a build you did not produce.** Given an original and a
|
||||
"cracked"/"modded" copy, this lists every entry that differs, every entry
|
||||
the author ADDED, and every entry they REMOVED. Added entries under
|
||||
`lib/`, `assets/`, or new dex files are the interesting ones -- that is
|
||||
where injected payloads live.
|
||||
|
||||
2. **Verify your own build was surgical.** After repacking, the diff against
|
||||
the original must contain exactly the entries you intended to touch and
|
||||
nothing else. A surprise entry is a bug in your pipeline, not in the app.
|
||||
|
||||
It compares *content*, not just sizes: two builds can have identical sizes and
|
||||
different bytes, and a same-size replacement is exactly the case a careless
|
||||
comparison misses.
|
||||
|
||||
Usage
|
||||
-----
|
||||
python apk_diff.py original.apk modified.apk
|
||||
python apk_diff.py original.apk modified.apk --hide-identical
|
||||
python apk_diff.py original.apk modified.apk --only 'classes.*\\.dex|lib/.*'
|
||||
python apk_diff.py original.apk modified.apk --summary
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import re
|
||||
import sys
|
||||
import zipfile
|
||||
|
||||
# Directories worth naming explicitly -- this is where injected code lives.
|
||||
INTERESTING = ("lib/", "assets/", "META-INF/services/")
|
||||
|
||||
|
||||
def snapshot(path):
|
||||
"""{entry_name: (size, sha256, compress_type, crc)}"""
|
||||
out = {}
|
||||
with zipfile.ZipFile(path) as z:
|
||||
for it in z.infolist():
|
||||
if it.is_dir():
|
||||
continue
|
||||
data = z.read(it.filename)
|
||||
out[it.filename] = (
|
||||
len(data),
|
||||
hashlib.sha256(data).hexdigest(),
|
||||
it.compress_type,
|
||||
it.CRC,
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def classify(name):
|
||||
if re.match(r"^classes\d*\.dex$", name):
|
||||
return "dex"
|
||||
if name.startswith("META-INF/") and name.endswith((".RSA", ".SF", ".MF", ".DSA", ".EC")):
|
||||
return "signature"
|
||||
if name.startswith(INTERESTING):
|
||||
return "code/assets"
|
||||
return "other"
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="Entry-level APK diff.")
|
||||
ap.add_argument("original")
|
||||
ap.add_argument("modified")
|
||||
ap.add_argument("--only", help="regex: restrict output to matching entry names")
|
||||
ap.add_argument("--hide-identical", action="store_true",
|
||||
help="omit the identical-count line")
|
||||
ap.add_argument("--summary", action="store_true",
|
||||
help="counts only, no per-entry listing")
|
||||
ap.add_argument("--max", type=int, default=200,
|
||||
help="max entries listed per section (default 200)")
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
a = snapshot(args.original)
|
||||
b = snapshot(args.modified)
|
||||
except Exception as e:
|
||||
sys.exit("error reading APK: %s" % e)
|
||||
|
||||
only_a = sorted(set(a) - set(b))
|
||||
only_b = sorted(set(b) - set(a))
|
||||
common = sorted(set(a) & set(b))
|
||||
changed = [n for n in common if a[n][1] != b[n][1]]
|
||||
|
||||
rx = re.compile(args.only) if args.only else None
|
||||
|
||||
def keep(n):
|
||||
return rx.search(n) if rx else True
|
||||
|
||||
def sha(path, n):
|
||||
return a[n][1][:16] if n in a else b[n][1][:16]
|
||||
|
||||
print("A (original) %s entries=%d" % (args.original, len(a)))
|
||||
print("B (modified) %s entries=%d" % (args.modified, len(b)))
|
||||
print("")
|
||||
print("only in A (removed) : %d" % len(only_a))
|
||||
print("only in B (added) : %d" % len(only_b))
|
||||
print("content changed : %d" % len(changed))
|
||||
if not args.hide_identical:
|
||||
print("identical : %d" % (len(common) - len(changed)))
|
||||
print("")
|
||||
|
||||
# Signature entries always differ; say so once instead of listing them.
|
||||
sig_only = [n for n in (only_a + only_b) if classify(n) == "signature"]
|
||||
if sig_only:
|
||||
print("(note: %d signature entries account for most of the add/remove noise)"
|
||||
% len(sig_only))
|
||||
|
||||
if args.summary:
|
||||
return
|
||||
|
||||
if only_b:
|
||||
print("\n=== ADDED in B (injection candidates) ===")
|
||||
shown = 0
|
||||
for n in only_b:
|
||||
if not keep(n) or classify(n) == "signature":
|
||||
continue
|
||||
print(" + %-52s %9d B [%s]" % (n, b[n][0], classify(n)))
|
||||
shown += 1
|
||||
if shown >= args.max:
|
||||
print(" ... (truncated)")
|
||||
break
|
||||
|
||||
if only_a:
|
||||
print("\n=== REMOVED from A ===")
|
||||
shown = 0
|
||||
for n in only_a:
|
||||
if not keep(n) or classify(n) == "signature":
|
||||
continue
|
||||
print(" - %-52s %9d B [%s]" % (n, a[n][0], classify(n)))
|
||||
shown += 1
|
||||
if shown >= args.max:
|
||||
print(" ... (truncated)")
|
||||
break
|
||||
|
||||
if changed:
|
||||
print("\n=== CONTENT CHANGED ===")
|
||||
shown = 0
|
||||
for n in changed:
|
||||
if not keep(n):
|
||||
continue
|
||||
sa, sb = a[n], b[n]
|
||||
cmark = "" if sa[2] == sb[2] else " ctype %d->%d" % (sa[2], sb[2])
|
||||
same_size = " (SAME SIZE)" if sa[0] == sb[0] else ""
|
||||
print(" ~ %-52s A %9d %s B %9d %s%s%s"
|
||||
% (n, sa[0], sa[1][:16], sb[0], sb[1][:16], same_size, cmark))
|
||||
shown += 1
|
||||
if shown >= args.max:
|
||||
print(" ... (truncated)")
|
||||
break
|
||||
if any(a[n][0] == b[n][0] for n in changed if keep(n)):
|
||||
print("\n (SAME SIZE means the change is not a length difference --")
|
||||
print(" a size-only comparison would have missed it entirely)")
|
||||
|
||||
if not (only_a or only_b or changed):
|
||||
print("no differences at all")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,457 @@
|
||||
#!/usr/bin/env python3
|
||||
"""PLT stub -> imported symbol mapping, and byte-level .so diffing.
|
||||
|
||||
Built for hardened/rebuilt native libraries where the usual paths lie:
|
||||
* section headers are often forged, so tools that walk sections (readelf -x,
|
||||
objdump -d on named sections, pyelftools' section API) return confidently
|
||||
wrong or empty results;
|
||||
* a linear disassembler can stop silently on a buffer that does not start at
|
||||
an instruction boundary, so "no matches" from it is not evidence.
|
||||
|
||||
Therefore this tool walks PT_LOAD / PT_DYNAMIC by hand and decodes the two
|
||||
stub shapes with its own decoders (no capstone dependency, no silent stops).
|
||||
|
||||
Why you need the stub -> symbol map: when a hardening library terminates the
|
||||
process, it does so through a PLT stub. To suppress it safely you must know
|
||||
*which* symbol a given stub resolves to -- inferring from position or from a
|
||||
comment is how you end up freezing `snprintf` or `pthread_exit` and reporting
|
||||
it as "the patch did not work".
|
||||
|
||||
Usage
|
||||
-----
|
||||
# full stub table
|
||||
python elf_plt.py libfoo.so
|
||||
|
||||
# what do these stubs call?
|
||||
python elf_plt.py libfoo.so --query 0x9920 0x9950
|
||||
|
||||
# find the stub for one symbol (and every caller of it)
|
||||
python elf_plt.py libfoo.so --symbol kill --callers
|
||||
|
||||
# what changed between the original and a rebuilt library?
|
||||
python elf_plt.py original.so --diff rebuilt.so
|
||||
|
||||
# same, but also name the symbol each changed stub belongs to
|
||||
python elf_plt.py original.so --diff rebuilt.so --name-regions
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import os
|
||||
import struct
|
||||
import sys
|
||||
|
||||
PT_LOAD = 1
|
||||
PT_DYNAMIC = 2
|
||||
|
||||
DT_PLTRELSZ = 2
|
||||
DT_PLTGOT = 3
|
||||
DT_STRTAB = 5
|
||||
DT_SYMTAB = 6
|
||||
DT_RELA = 7
|
||||
DT_RELASZ = 8
|
||||
DT_STRSZ = 10
|
||||
DT_SYMENT = 11
|
||||
DT_PLTREL = 20
|
||||
DT_JMPREL = 23
|
||||
|
||||
EM_X86_64 = 62
|
||||
EM_AARCH64 = 183
|
||||
|
||||
ARCH_NAME = {EM_X86_64: "x86_64", EM_AARCH64: "aarch64"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# ELF parsing (program headers only -- never trust the section table here)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
class Elf:
|
||||
def __init__(self, data, path=""):
|
||||
self.data = data
|
||||
self.path = path
|
||||
if data[:4] != b"\x7fELF":
|
||||
raise ValueError("not an ELF file")
|
||||
if data[4] != 2:
|
||||
raise ValueError("only 64-bit ELF is supported")
|
||||
self.machine = struct.unpack_from("<H", data, 18)[0]
|
||||
self.phoff = struct.unpack_from("<Q", data, 32)[0]
|
||||
self.phentsize = struct.unpack_from("<H", data, 54)[0]
|
||||
self.phnum = struct.unpack_from("<H", data, 56)[0]
|
||||
self.loads = [] # (vaddr, offset, filesz, memsz, flags)
|
||||
self.dynamic = None # (offset, size)
|
||||
self.gnu_eh_frame = None
|
||||
for i in range(self.phnum):
|
||||
off = self.phoff + i * self.phentsize
|
||||
p_type, p_flags = struct.unpack_from("<II", data, off)
|
||||
p_offset, p_vaddr, _pa, p_filesz, p_memsz, _al = struct.unpack_from(
|
||||
"<QQQQQQ", data, off + 8)
|
||||
if p_type == PT_LOAD:
|
||||
self.loads.append((p_vaddr, p_offset, p_filesz, p_memsz, p_flags))
|
||||
elif p_type == PT_DYNAMIC:
|
||||
self.dynamic = (p_offset, p_filesz)
|
||||
elif p_type == 0x6474e550:
|
||||
self.gnu_eh_frame = (p_vaddr, p_filesz)
|
||||
|
||||
@property
|
||||
def arch(self):
|
||||
return ARCH_NAME.get(self.machine, "machine=%d" % self.machine)
|
||||
|
||||
def exec_segment(self):
|
||||
for seg in self.loads:
|
||||
if seg[4] & 1:
|
||||
return seg
|
||||
raise ValueError("no executable PT_LOAD segment")
|
||||
|
||||
def vaddr_to_off(self, vaddr):
|
||||
for va, fo, _fs, ms, _fl in self.loads:
|
||||
if va <= vaddr < va + ms:
|
||||
return fo + (vaddr - va)
|
||||
return None
|
||||
|
||||
def dynamic_tags(self):
|
||||
if self.dynamic is None:
|
||||
raise ValueError("no PT_DYNAMIC segment")
|
||||
tags = {}
|
||||
off, size = self.dynamic
|
||||
for j in range(size // 16):
|
||||
t, v = struct.unpack_from("<qQ", self.data, off + j * 16)
|
||||
if t == 0:
|
||||
break
|
||||
tags[t] = v
|
||||
return tags
|
||||
|
||||
|
||||
def reloc_map(elf):
|
||||
"""{GOT vaddr: imported symbol name} from DT_JMPREL (+ DT_RELA when present)."""
|
||||
tags = elf.dynamic_tags()
|
||||
if DT_STRTAB not in tags or DT_SYMTAB not in tags:
|
||||
return {}
|
||||
strtab_o = elf.vaddr_to_off(tags[DT_STRTAB])
|
||||
symtab_o = elf.vaddr_to_off(tags[DT_SYMTAB])
|
||||
strsz = tags.get(DT_STRSZ, 0)
|
||||
if strtab_o is None or symtab_o is None:
|
||||
return {}
|
||||
|
||||
def symname(idx):
|
||||
so = symtab_o + idx * 24
|
||||
if so + 4 > len(elf.data):
|
||||
return "?"
|
||||
(st_name,) = struct.unpack_from("<I", elf.data, so)
|
||||
if st_name >= strsz:
|
||||
return "?"
|
||||
start = strtab_o + st_name
|
||||
end = elf.data.find(b"\x00", start)
|
||||
if end < 0:
|
||||
return "?"
|
||||
return elf.data[start:end].decode("utf-8", "replace")
|
||||
|
||||
out = {}
|
||||
for tag_off, tag_sz in ((DT_JMPREL, DT_PLTRELSZ), (DT_RELA, DT_RELASZ)):
|
||||
if tag_off not in tags:
|
||||
continue
|
||||
base = elf.vaddr_to_off(tags[tag_off])
|
||||
if base is None:
|
||||
continue
|
||||
n = tags.get(tag_sz, 0) // 24
|
||||
for k in range(n):
|
||||
if base + k * 24 + 16 > len(elf.data):
|
||||
break
|
||||
r_offset, r_info = struct.unpack_from("<QQ", elf.data, base + k * 24)
|
||||
out[r_offset] = symname(r_info >> 32)
|
||||
return out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Stub decoders -- written out rather than delegated, so nothing stops early
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def find_stubs(elf, relocs):
|
||||
"""{stub_vaddr: (symbol, got_vaddr)} for the library's stub table."""
|
||||
base, foff, fsz, _ms, _fl = elf.exec_segment()
|
||||
if elf.machine == EM_X86_64:
|
||||
return _stubs_x86(elf.data, base, foff, fsz, relocs)
|
||||
if elf.machine == EM_AARCH64:
|
||||
return _stubs_arm64(elf.data, base, foff, fsz, relocs)
|
||||
return {}
|
||||
|
||||
|
||||
def _stubs_x86(data, base, foff, fsz, relocs):
|
||||
"""x86_64 PLT stub: `ff 25 <disp32>` == jmp qword ptr [rip+disp32]."""
|
||||
out = {}
|
||||
i = 0
|
||||
while i < fsz - 6:
|
||||
if data[foff + i] == 0xFF and data[foff + i + 1] == 0x25:
|
||||
disp = struct.unpack_from("<i", data, foff + i + 2)[0]
|
||||
va = base + i
|
||||
target = va + 6 + disp
|
||||
if target in relocs:
|
||||
out[va] = (relocs[target], target)
|
||||
i += 1
|
||||
return out
|
||||
|
||||
|
||||
def _stubs_arm64(data, base, foff, fsz, relocs):
|
||||
"""aarch64 PLT stub: adrp x16 / ldr x17,[x16,#off] / add x16,x16,#off / br x17.
|
||||
|
||||
Note the stub is FOUR instructions (16 bytes), not four bytes. Assuming the
|
||||
shorter form leads to "borrowing" the next slot, which belongs to a
|
||||
different symbol.
|
||||
"""
|
||||
out = {}
|
||||
off = 0
|
||||
while off < fsz - 16:
|
||||
w0, w1, w2, w3 = struct.unpack_from("<IIII", data, foff + off)
|
||||
dec = _decode_adrp(w0, base + off)
|
||||
if dec is not None:
|
||||
page, rd0 = dec
|
||||
ldr = _decode_ldr_unsigned(w1)
|
||||
add = _decode_add_imm(w2)
|
||||
br = _decode_br(w3)
|
||||
if (ldr is not None and add is not None and br is not None
|
||||
and rd0 == 16 and ldr[0] == 16 and ldr[1] == 17
|
||||
and add[0] == 16 and add[1] == 16
|
||||
and ldr[2] == add[2] and br == 17):
|
||||
got = page + add[2]
|
||||
if got in relocs:
|
||||
out[base + off] = (relocs[got], got)
|
||||
off += 4
|
||||
return out
|
||||
|
||||
|
||||
def _sign_extend(value, bits):
|
||||
if value & (1 << (bits - 1)):
|
||||
value -= (1 << bits)
|
||||
return value
|
||||
|
||||
|
||||
def _decode_adrp(word, pc):
|
||||
"""adrp Rd, #imm -> (target_page, Rd). None if not an adrp."""
|
||||
if (word >> 31) & 1 != 1:
|
||||
return None
|
||||
if ((word >> 24) & 0x1F) != 0x10:
|
||||
return None
|
||||
immlo = (word >> 29) & 0x3
|
||||
immhi = (word >> 5) & 0x7FFFF
|
||||
imm = (immhi << 2) | immlo
|
||||
imm = _sign_extend(imm, 21)
|
||||
return (pc & ~0xFFF) + (imm << 12), (word & 0x1F)
|
||||
|
||||
|
||||
def _decode_ldr_unsigned(word):
|
||||
"""ldr Rt, [Rn, #imm] (64-bit unsigned offset) -> (Rn, Rt, byte_offset)."""
|
||||
if (word >> 30) != 0b11: # size == 64-bit
|
||||
return None
|
||||
if ((word >> 27) & 0x7) != 0b111:
|
||||
return None
|
||||
if ((word >> 24) & 0x3) != 0b01:
|
||||
return None
|
||||
imm12 = (word >> 10) & 0xFFF
|
||||
rn = (word >> 5) & 0x1F
|
||||
rt = word & 0x1F
|
||||
return rn, rt, imm12 * 8
|
||||
|
||||
|
||||
def _decode_add_imm(word):
|
||||
"""add Rd, Rn, #imm -> (Rn, Rd, imm).
|
||||
|
||||
Encoding: sf op S 100010 sh imm12 Rn Rd
|
||||
bits[31]=sf bits[30]=op bits[29]=S bits[28:23]=100010 (6 bits, not 9)
|
||||
"""
|
||||
if (word >> 31) & 1 != 1: # sf must be 1 (64-bit form)
|
||||
return None
|
||||
if ((word >> 23) & 0x3F) != 0b100010:
|
||||
return None
|
||||
if (word >> 22) & 1: # shifted form -- not what stubs use
|
||||
return None
|
||||
imm12 = (word >> 10) & 0xFFF
|
||||
rn = (word >> 5) & 0x1F
|
||||
rd = word & 0x1F
|
||||
return rn, rd, imm12
|
||||
|
||||
|
||||
def _decode_br(word):
|
||||
"""br Rn -> Rn, or None."""
|
||||
if (word & 0xFFFFFC1F) == 0xD61F0000:
|
||||
return (word >> 5) & 0x1F
|
||||
return None
|
||||
|
||||
|
||||
def find_branch_callers(elf, target_vaddr, limit=500):
|
||||
"""Addresses of BL/B instructions that branch to target_vaddr."""
|
||||
base, foff, fsz, _ms, _fl = elf.exec_segment()
|
||||
hits = []
|
||||
if elf.machine == EM_AARCH64:
|
||||
off = 0
|
||||
while off < fsz - 4:
|
||||
(w,) = struct.unpack_from("<I", data_slice(elf.data, foff + off, 4))
|
||||
if (w & 0xFC000000) in (0x94000000, 0x14000000): # BL / B
|
||||
imm = _sign_extend(w & 0x03FFFFFF, 26) << 2
|
||||
if base + off + imm == target_vaddr:
|
||||
hits.append(base + off)
|
||||
if len(hits) >= limit:
|
||||
break
|
||||
off += 4
|
||||
elif elf.machine == EM_X86_64:
|
||||
i = 0
|
||||
while i < fsz - 5:
|
||||
if elf.data[foff + i] == 0xE8:
|
||||
disp = struct.unpack_from("<i", elf.data, foff + i + 1)[0]
|
||||
if base + i + 5 + disp == target_vaddr:
|
||||
hits.append(base + i)
|
||||
if len(hits) >= limit:
|
||||
break
|
||||
i += 1
|
||||
return hits
|
||||
|
||||
|
||||
def data_slice(data, off, n):
|
||||
return data[off:off + n]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Commands
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def load(path):
|
||||
with open(path, "rb") as fh:
|
||||
return Elf(fh.read(), path)
|
||||
|
||||
|
||||
def cmd_list(elf, args):
|
||||
relocs = reloc_map(elf)
|
||||
stubs = find_stubs(elf, relocs)
|
||||
base, foff, fsz, _ms, _fl = elf.exec_segment()
|
||||
print("== %s" % elf.path)
|
||||
print(" arch=%s exec seg: vaddr=0x%x file=0x%x size=0x%x" % (elf.arch, base, foff, fsz))
|
||||
print(" imported relocations=%d stubs resolved=%d" % (len(relocs), len(stubs)))
|
||||
if not stubs:
|
||||
print(" NOTE: zero stubs resolved. Either the library has no PLT, or the")
|
||||
print(" hardening removed/forged the relocation tables. Do not read")
|
||||
print(" this as 'the library imports nothing'.")
|
||||
for va in sorted(stubs):
|
||||
sym, got = stubs[va]
|
||||
print(" 0x%08x -> %-34s (GOT 0x%x)" % (va, sym, got))
|
||||
return stubs
|
||||
|
||||
|
||||
def cmd_query(elf, args):
|
||||
relocs = reloc_map(elf)
|
||||
stubs = find_stubs(elf, relocs)
|
||||
for a in args.query:
|
||||
va = int(a, 0)
|
||||
hit = stubs.get(va)
|
||||
if hit:
|
||||
print("0x%x -> %s (GOT 0x%x)" % (va, hit[0], hit[1]))
|
||||
else:
|
||||
print("0x%x -> not a resolved stub" % va)
|
||||
|
||||
|
||||
def cmd_symbol(elf, args):
|
||||
relocs = reloc_map(elf)
|
||||
stubs = find_stubs(elf, relocs)
|
||||
wanted = args.symbol
|
||||
found = [(va, sym, got) for va, (sym, got) in stubs.items() if sym == wanted]
|
||||
if not found:
|
||||
near = sorted({s for _v, (s, _g) in stubs.items()
|
||||
if wanted in s or s in wanted})[:10]
|
||||
print("no stub resolves to %r" % wanted)
|
||||
if near:
|
||||
print(" similar: %s" % ", ".join(near))
|
||||
return
|
||||
for va, sym, got in sorted(found):
|
||||
print("stub 0x%x -> %s (GOT 0x%x)" % (va, sym, got))
|
||||
if args.callers:
|
||||
callers = find_branch_callers(elf, va)
|
||||
print(" callers of the stub: %d" % len(callers))
|
||||
for c in callers:
|
||||
print(" 0x%x" % c)
|
||||
print(" NOTE: a stub-level patch covers all of these. A call-site-level")
|
||||
print(" patch must cover every one of them, plus any caller reached")
|
||||
print(" through a cached function pointer (not listed here).")
|
||||
|
||||
|
||||
def cmd_diff(elf_a, args):
|
||||
path_b = args.diff
|
||||
elf_b = load(path_b)
|
||||
a, b = elf_a.data, elf_b.data
|
||||
print("== diff")
|
||||
print(" A %s %d bytes sha256=%s" % (elf_a.path, len(a), hashlib.sha256(a).hexdigest()[:16]))
|
||||
print(" B %s %d bytes sha256=%s" % (path_b, len(b), hashlib.sha256(b).hexdigest()[:16]))
|
||||
if len(a) != len(b):
|
||||
print(" SIZE DIFFERS: %d vs %d -- a length change is itself a finding"
|
||||
% (len(a), len(b)))
|
||||
n = min(len(a), len(b))
|
||||
runs = []
|
||||
i = 0
|
||||
while i < n:
|
||||
if a[i] != b[i]:
|
||||
j = i
|
||||
while j < n and a[j] != b[j]:
|
||||
j += 1
|
||||
runs.append((i, j - i))
|
||||
i = j
|
||||
else:
|
||||
i += 1
|
||||
total = sum(r[1] for r in runs)
|
||||
print(" differing bytes=%d in %d run(s)" % (total, len(runs)))
|
||||
|
||||
regions = {}
|
||||
if args.name_regions:
|
||||
relocs_a = reloc_map(elf_a)
|
||||
stubs_a = find_stubs(elf_a, relocs_a)
|
||||
|
||||
def describe(off):
|
||||
for va, (sym, _got) in stubs_a.items():
|
||||
size = 16 if elf_a.machine == EM_AARCH64 else 16
|
||||
if va <= off < va + size:
|
||||
return "PLT stub for %s" % sym
|
||||
return ""
|
||||
|
||||
regions = {"describe": describe}
|
||||
|
||||
for off, ln in runs:
|
||||
note = ""
|
||||
if regions:
|
||||
note = regions["describe"](off)
|
||||
if note:
|
||||
note = " <- " + note
|
||||
print(" off=0x%06x len=%-4d A: %s" % (off, ln, a[off:off + ln].hex(" ")))
|
||||
print(" %s B: %s%s" % (" " * 14, b[off:off + ln].hex(" "), note))
|
||||
if args.name_regions and not runs:
|
||||
print(" (identical)")
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(
|
||||
description="PLT stub -> symbol mapping and .so diffing for hardened libraries.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||||
ap.add_argument("so", help="path to the ELF (.so)")
|
||||
ap.add_argument("--query", nargs="+", metavar="ADDR",
|
||||
help="resolve specific stub addresses (hex ok)")
|
||||
ap.add_argument("--symbol", metavar="NAME",
|
||||
help="find the stub that resolves to this symbol")
|
||||
ap.add_argument("--callers", action="store_true",
|
||||
help="with --symbol: also list branch call sites")
|
||||
ap.add_argument("--diff", metavar="OTHER_SO",
|
||||
help="byte-compare against another library")
|
||||
ap.add_argument("--name-regions", action="store_true",
|
||||
help="with --diff: name the symbol each changed stub belongs to")
|
||||
args = ap.parse_args()
|
||||
|
||||
try:
|
||||
elf = load(args.so)
|
||||
except Exception as e:
|
||||
sys.exit("error: %s" % e)
|
||||
|
||||
if args.diff:
|
||||
cmd_diff(elf, args)
|
||||
elif args.query:
|
||||
cmd_query(elf, args)
|
||||
elif args.symbol:
|
||||
cmd_symbol(elf, args)
|
||||
else:
|
||||
cmd_list(elf, args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,279 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract native crash blocks from a log or tombstone and locate the fault.
|
||||
|
||||
Native deaths are the expensive kind: no Java stack, often no obvious cause, and
|
||||
the interesting question ("which library, which offset, is this a real fault or
|
||||
an arranged one?") is buried in a debug dump that is tedious to read by hand.
|
||||
|
||||
This reads a saved `logcat` capture or a tombstone file and reports, per crash:
|
||||
|
||||
* the signal, fault address, and cause
|
||||
* the register state that matters (the faulting pointer)
|
||||
* the backtrace, split into your own libraries vs system libraries
|
||||
* a verdict hint when the fault address looks *arranged* rather than accidental
|
||||
|
||||
With `--lib NAME=PATH` it also resolves a frame to a local copy of the library
|
||||
and prints the bytes at that offset, plus a disassembly window when a decoder is
|
||||
available.
|
||||
|
||||
Why the verdict hint matters: a hardening layer that decides the environment is
|
||||
hostile rarely calls `kill`. It more often arranges a fault -- load a small
|
||||
constant, use it as a pointer -- so the death looks like an ordinary bug. The
|
||||
tell is a fault address that is a small integer combined with a register holding
|
||||
that same integer. This script flags that shape so you look at the right place
|
||||
instead of hunting a bug that is not there.
|
||||
|
||||
Usage
|
||||
-----
|
||||
python native_crash.py capture.txt
|
||||
python native_crash.py tombstone_07
|
||||
python native_crash.py capture.txt --lib libfoo.so=/path/to/local/libfoo.so
|
||||
python native_crash.py capture.txt --lib libfoo.so=./libfoo.so --disasm 16
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import struct
|
||||
import sys
|
||||
|
||||
SIGNAL_RE = re.compile(
|
||||
r"signal\s+(\d+)\s+\((\w+)\)\s*,?\s*code\s+(\d+)\s*\(([^)]+)\)\s*,?\s*fault addr\s+(0x[0-9a-fA-F]+)")
|
||||
PID_RE = re.compile(r"pid:\s*(\d+),\s*tid:\s*(\d+),\s*name:\s*(\S+)\s*>>>\s*(\S+)")
|
||||
PROC_RE = re.compile(r"Cmdline:\s*(\S+)")
|
||||
UPTIME_RE = re.compile(r"Process uptime:\s*(\d+)")
|
||||
FRAME_RE = re.compile(r"#(\d+)\s+pc\s+([0-9a-fA-F]+)\s+(\S+)")
|
||||
REG_RE = re.compile(r"^\s*([xX]\d{1,2}|sp|lr|pc|pst)\s+([0-9a-fA-F]{8,16})")
|
||||
SYSTEM_LIB_RE = re.compile(r"^/(apex|system|vendor|data/dalvik-cache|memfd)")
|
||||
|
||||
|
||||
def is_system_lib(path):
|
||||
if not path.startswith("/"):
|
||||
return True
|
||||
return bool(SYSTEM_LIB_RE.match(path))
|
||||
|
||||
|
||||
def parse(text):
|
||||
"""Yield crash dicts."""
|
||||
lines = text.splitlines()
|
||||
crashes = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
m = SIGNAL_RE.search(lines[i])
|
||||
# some ROMs print the signal line without "F DEBUG"; accept both
|
||||
if not m:
|
||||
i += 1
|
||||
continue
|
||||
block = {"signal": int(m.group(1)), "signame": m.group(2),
|
||||
"si_code": m.group(4), "fault": m.group(5),
|
||||
"frames": [], "regs": {}, "raw": []}
|
||||
# walk up a little to pick up pid/cmdline/uptime printed just before
|
||||
for j in range(max(0, i - 25), i):
|
||||
l = lines[j]
|
||||
mm = PID_RE.search(l)
|
||||
if mm and "pid" not in block:
|
||||
block["pid"], block["tid"], block["tname"], block["proc"] = mm.groups()
|
||||
mm = PROC_RE.search(l)
|
||||
if mm and "proc" not in block:
|
||||
block["proc"] = mm.group(1)
|
||||
mm = UPTIME_RE.search(l)
|
||||
if mm:
|
||||
block["uptime"] = mm.group(1)
|
||||
if "Cause:" in l and "cause" not in block:
|
||||
block["cause"] = l.split("Cause:", 1)[1].strip()
|
||||
# walk down through the dump
|
||||
j = i
|
||||
while j < len(lines) and j < i + 120:
|
||||
l = lines[j]
|
||||
if j > i + 3 and SIGNAL_RE.search(l) and j > i + 3:
|
||||
break
|
||||
block["raw"].append(l.strip())
|
||||
mm = PID_RE.search(l)
|
||||
if mm and "pid" not in block:
|
||||
block["pid"], block["tid"], block["tname"], block["proc"] = mm.groups()
|
||||
mm = PROC_RE.search(l)
|
||||
if mm and "proc" not in block:
|
||||
block["proc"] = mm.group(1)
|
||||
mm = UPTIME_RE.search(l)
|
||||
if mm:
|
||||
block["uptime"] = mm.group(1)
|
||||
if "Cause:" in l and "cause" not in block:
|
||||
block["cause"] = l.split("Cause:", 1)[1].strip()
|
||||
mm = FRAME_RE.search(l)
|
||||
if mm:
|
||||
# Backtrace lines look like either of:
|
||||
# #00 pc 00000000000128cc /data/app/.../lib/arm64/libfoo.so
|
||||
# #02 pc 00000000000cccd0 /apex/.../libc.so (__pthread_start+256)
|
||||
# The path is group(3); anything after it is a symbol in parens.
|
||||
path = mm.group(3)
|
||||
rest = l.split(path, 1)[1].strip() if path in l else ""
|
||||
sym = ""
|
||||
m2 = re.search(r"\(([^)]+)\)", rest)
|
||||
if m2:
|
||||
sym = m2.group(1)
|
||||
block["frames"].append({
|
||||
"n": mm.group(1), "pc": int(mm.group(2), 16),
|
||||
"path": path, "symbol": sym,
|
||||
})
|
||||
# Register dumps put several registers on one line, so collect all of
|
||||
# them rather than only the leading one.
|
||||
for k, v in re.findall(r"\b([xX]\d{1,2}|sp|lr|pc|pst)\s+([0-9a-fA-F]{8,16})\b",
|
||||
l.replace("F DEBUG :", " ")):
|
||||
block["regs"][k] = int(v, 16)
|
||||
if "backtrace:" in l:
|
||||
pass
|
||||
j += 1
|
||||
i = j if j > i else i + 1
|
||||
crashes.append(block)
|
||||
return crashes
|
||||
|
||||
|
||||
def verdict_hint(crash):
|
||||
hints = []
|
||||
fault = int(crash["fault"], 16)
|
||||
regs = crash["regs"]
|
||||
if fault <= 0x100:
|
||||
holder = [k for k, v in regs.items() if v == fault]
|
||||
hints.append("fault address is a small integer (%s)." % crash["fault"])
|
||||
if holder:
|
||||
hints.append("register(s) %s hold exactly that value -- this is the shape of an "
|
||||
"ARRANGED fault, not an accident." % ", ".join(sorted(holder)))
|
||||
hints.append("check whether the instruction at the faulting pc loaded that constant "
|
||||
"a few instructions earlier and used it as a pointer.")
|
||||
if crash["signame"] == "SIGKILL":
|
||||
hints.append("SIGKILL with no tombstone: this is a deliberate termination, not a bug. "
|
||||
"No crash dump exists because nothing faulted.")
|
||||
if crash.get("cause") and "null pointer" in crash["cause"] and fault > 0x100:
|
||||
hints.append("a genuine null-ish dereference; treat as an ordinary bug unless a small "
|
||||
"constant is involved.")
|
||||
return hints
|
||||
|
||||
|
||||
def load_libs(specs):
|
||||
out = {}
|
||||
for s in specs or []:
|
||||
if "=" not in s:
|
||||
sys.exit("--lib expects NAME=PATH, got %r" % s)
|
||||
name, path = s.split("=", 1)
|
||||
with open(path, "rb") as fh:
|
||||
out[name] = fh.read()
|
||||
return out
|
||||
|
||||
|
||||
def maybe_disasm(data, offset, count):
|
||||
try:
|
||||
from capstone import Cs, CS_ARCH_ARM64, CS_MODE_ARM, CS_ARCH_X86, CS_MODE_64
|
||||
except Exception:
|
||||
return None
|
||||
# guess arch from ELF e_machine
|
||||
if data[:4] != b"\x7fELF":
|
||||
return None
|
||||
machine = struct.unpack_from("<H", data, 18)[0]
|
||||
if machine == 183:
|
||||
md = Cs(CS_ARCH_ARM64, CS_MODE_ARM)
|
||||
start = offset & ~3
|
||||
elif machine == 62:
|
||||
md = Cs(CS_ARCH_X86, CS_MODE_64)
|
||||
start = max(0, offset - 16)
|
||||
else:
|
||||
return None
|
||||
out = []
|
||||
for ins in md.disasm(data[start:start + count * 8], start):
|
||||
mark = " <<<" if ins.address <= offset < ins.address + ins.size else ""
|
||||
out.append(" 0x%06x %-10s %s%s" % (ins.address, ins.mnemonic, ins.op_str, mark))
|
||||
if len(out) >= count:
|
||||
break
|
||||
if not out:
|
||||
# capstone stopped immediately -- say so rather than implying emptiness
|
||||
return [" (decoder produced nothing from this offset; a silent stop is not "
|
||||
"evidence there is no code here)"]
|
||||
return out
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description="Locate native crashes from a log or tombstone.")
|
||||
ap.add_argument("logfile", help="logcat capture or tombstone file")
|
||||
ap.add_argument("--lib", action="append", metavar="NAME=PATH",
|
||||
help="resolve frames for this library name to a local file")
|
||||
ap.add_argument("--disasm", type=int, default=0, metavar="N",
|
||||
help="disassemble N instructions around the faulting pc")
|
||||
ap.add_argument("--frames", type=int, default=12, help="how many frames to print")
|
||||
args = ap.parse_args()
|
||||
|
||||
if not os.path.exists(args.logfile):
|
||||
sys.exit("no such file: %s" % args.logfile)
|
||||
with open(args.logfile, "r", encoding="utf-8", errors="replace") as fh:
|
||||
text = fh.read()
|
||||
|
||||
crashes = parse(text)
|
||||
if not crashes:
|
||||
print("no native crash block found in %s" % args.logfile)
|
||||
print("")
|
||||
print("This is a real negative only if the capture covers the death. Check:")
|
||||
print(" - was the crash buffer cleared *after* the build under test was installed?")
|
||||
print(" - does the capture include the tombstone? (a SIGKILL leaves none)")
|
||||
print(" - did the process die at all, or did it merely restart?")
|
||||
return
|
||||
|
||||
libs = load_libs(args.lib)
|
||||
|
||||
for idx, c in enumerate(crashes, 1):
|
||||
print("=" * 72)
|
||||
print("CRASH %d" % idx)
|
||||
print(" process : %s" % c.get("proc", "?"))
|
||||
print(" pid/tid : %s / %s" % (c.get("pid", "?"), c.get("tid", "?")))
|
||||
if c.get("uptime"):
|
||||
print(" uptime : %ss <-- time from launch to death; your test window must "
|
||||
"exceed this" % c["uptime"])
|
||||
print(" signal : %s (%d), si_code=%s" % (c["signame"], c["signal"], c["si_code"]))
|
||||
print(" fault addr: %s" % c["fault"])
|
||||
if c.get("cause"):
|
||||
print(" cause : %s" % c["cause"])
|
||||
|
||||
hints = verdict_hint(c)
|
||||
if hints:
|
||||
print(" HINTS")
|
||||
for h in hints:
|
||||
print(" - %s" % h)
|
||||
|
||||
app_frames = [f for f in c["frames"] if not is_system_lib(f["path"])]
|
||||
if app_frames:
|
||||
print(" backtrace (your libraries):")
|
||||
for f in app_frames[:args.frames]:
|
||||
print(" #%s pc 0x%x %s%s"
|
||||
% (f["n"], f["pc"], os.path.basename(f["path"]),
|
||||
(" (%s)" % f["symbol"]) if f["symbol"] else ""))
|
||||
else:
|
||||
print(" backtrace: no frame inside an app library")
|
||||
print(" -> the fault is in the runtime or a system library; an app-side")
|
||||
print(" patch is unlikely to be the cause. Look at what called in.")
|
||||
|
||||
if c["frames"]:
|
||||
print(" full backtrace:")
|
||||
for f in c["frames"][:args.frames]:
|
||||
tag = "system" if is_system_lib(f["path"]) else "APP"
|
||||
print(" #%s [%s] pc 0x%x %s%s"
|
||||
% (f["n"], tag, f["pc"], (f["path"] or "?"),
|
||||
(" (%s)" % f["symbol"]) if f["symbol"] else ""))
|
||||
|
||||
# resolve against local libraries
|
||||
for f in c["frames"]:
|
||||
base = os.path.basename(f["path"])
|
||||
for name, data in libs.items():
|
||||
if name in base or base in name:
|
||||
print(" local %s @ 0x%x:" % (name, f["pc"]))
|
||||
if args.disasm and f is c["frames"][0]:
|
||||
for line in maybe_disasm(data, f["pc"], args.disasm) or []:
|
||||
print(line)
|
||||
else:
|
||||
print(" raw: %s" % data[f["pc"]:f["pc"] + 16].hex(" "))
|
||||
print("")
|
||||
|
||||
print("=" * 72)
|
||||
print("RECORD THIS: signal, fault addr, and the time-to-death above. They are the")
|
||||
print("baseline an attempted fix must be measured against -- without them a later")
|
||||
print("run cannot distinguish 'fixed' from 'the timing changed'.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user