437 lines
17 KiB
Python
437 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""Apply equal-length byte patches to a dex, with structural and verifier checks.
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
The cheap way to change behaviour is to rewrite a few bytes in place rather than
|
|
rebuild a method body. It is cheap for a reason: nothing moves, so no try/catch
|
|
block, debug-info pointer or branch displacement can be invalidated by
|
|
construction. But it has four failure modes that all produce a build which looks
|
|
fine and misbehaves, and this tool exists to make each one impossible to skip:
|
|
|
|
1. **Same-length is not enough** -- the replacement must decode to a valid
|
|
instruction sequence of exactly the same length. A 4-byte branch replaced by
|
|
a 4-byte branch is fine; a 2-byte branch replaced by a 4-byte instruction
|
|
corrupts the next instruction.
|
|
2. **Polarity** -- the overwhelmingly common error is patching the wrong side of
|
|
a condition. So a spec can *require* the instruction that follows the edit,
|
|
and this tool aborts if it is not the one you named. Name the fall-through
|
|
you expect and the mistake cannot survive.
|
|
3. **Verifier legality** -- a branch whose target is a `move-result*` bypasses
|
|
the producing invoke and the class fails to load (`VerifyError`). Checked
|
|
here before writing.
|
|
4. **dex header staleness** -- the checksum and signature fields must be
|
|
recomputed, signature FIRST (the checksum covers it). Skipping this makes
|
|
Android fall back to interpreting the dex, which surfaces as an unrelated
|
|
ClassNotFoundException at startup.
|
|
|
|
Spec format (JSON; a list, or {"patches": [...]}):
|
|
|
|
[{
|
|
"name": "human label",
|
|
"reason": "why this edit is correct",
|
|
"class": "Lpkg/Name;",
|
|
"method": "someMethod",
|
|
"desc": "(Lpkg/Arg;)V", // optional: disambiguates overloads
|
|
"match": {"kind": "if-testz", "target_reads_field_of": "Lpkg/Cfg;",
|
|
"target_field": "enabled"},
|
|
"expect_next": {"kind": "invoke", "target_method": "goHome"},
|
|
"replace": {"kind": "nops"} // or {"bytes": "28060000"}
|
|
}]
|
|
|
|
`match` selects the instruction; `expect_next` (optional but strongly advised)
|
|
asserts what immediately follows, which is what pins the polarity; `replace`
|
|
gives the bytes. Run with --dry-run first: it prints the match, the neighbours,
|
|
the verifier verdict and the predicted byte diff without writing anything.
|
|
|
|
Exit codes: 0 ok, 1 failure (structure, match, verifier, or self-verify), 2 usage.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import struct
|
|
import sys
|
|
import zipfile
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from dexutil import ( # noqa: E402
|
|
Dex, IF_TEST, IF_TESTZ, INVOKE_OPS, MOVE_RESULT_OPS, RETURN_OPS,
|
|
fix_dex_header, load_dex, verify_dex_header,
|
|
)
|
|
|
|
|
|
def _kind_of(insn):
|
|
"""Coarse semantic kind used by spec matching."""
|
|
op = insn["op"]
|
|
if op in IF_TEST:
|
|
return "if-test"
|
|
if op in IF_TESTZ:
|
|
return "if-testz"
|
|
if op in INVOKE_OPS:
|
|
return "invoke"
|
|
if op in RETURN_OPS:
|
|
return "return"
|
|
if op in MOVE_RESULT_OPS:
|
|
return "move-result"
|
|
if op == 0x12:
|
|
return "const4"
|
|
return insn["name"]
|
|
|
|
|
|
def _describe_neighbours(dex, code_off, target_off, span=4):
|
|
"""Render up to `span` instructions before and after an offset."""
|
|
insns = list(dex.decode(code_off))
|
|
idx = None
|
|
for i, insn in enumerate(insns):
|
|
if insn["off"] == target_off:
|
|
idx = i
|
|
break
|
|
if idx is None:
|
|
return [], None, []
|
|
before = insns[max(0, idx - span):idx]
|
|
after = insns[idx + 1:idx + 1 + span]
|
|
return before, insns[idx], after
|
|
|
|
|
|
def _matches(dex, insn, crit, code_off=None):
|
|
"""Does one instruction satisfy a match/expect criterion?"""
|
|
if crit is None:
|
|
return True
|
|
want = crit.get("kind")
|
|
if want and _kind_of(insn) != want:
|
|
return False
|
|
|
|
if want in ("if-test", "if-testz"):
|
|
treg = crit.get("target_reg")
|
|
if treg is not None:
|
|
regs, _ = dex.branch_regs(insn)
|
|
if treg not in regs:
|
|
return False
|
|
# The usual way to pin "this is the gate for config X": require that the
|
|
# branch TARGET instruction reads a field of a named class. This is what
|
|
# distinguishes the gate you want from the other branches in the method.
|
|
fcls = crit.get("target_reads_field_of")
|
|
fname = crit.get("target_field")
|
|
if fcls or fname:
|
|
if code_off is None:
|
|
return False
|
|
tgt = dex.branch_target(insn)
|
|
t_insn = dex.insn_at(code_off, tgt)
|
|
if t_insn is None:
|
|
return False
|
|
if not (0x52 <= t_insn["op"] <= 0x58):
|
|
return False
|
|
fidx = struct.unpack_from("<H", t_insn["raw"], 2)[0]
|
|
cls, nm, _ty = dex.field(fidx)
|
|
if fcls and cls != fcls:
|
|
return False
|
|
if fname and nm != fname:
|
|
return False
|
|
|
|
if want == "const4":
|
|
lit = crit.get("literal")
|
|
if lit is not None:
|
|
raw = insn["raw"][1]
|
|
val = raw & 0xF
|
|
if val > 7:
|
|
val -= 16
|
|
if val != lit:
|
|
return False
|
|
|
|
if want == "invoke":
|
|
midx = struct.unpack_from("<H", insn["raw"], 2)[0]
|
|
cls, nm, _ds = dex.method(midx)
|
|
if crit.get("target_class") and crit["target_class"] not in cls:
|
|
return False
|
|
if crit.get("target_method") and crit["target_method"] != nm:
|
|
return False
|
|
|
|
if want in ("return", "move-result", "nop"):
|
|
pass
|
|
|
|
raw_eq = crit.get("bytes")
|
|
if raw_eq and insn["raw"].hex() != raw_eq.replace(" ", "").lower():
|
|
return False
|
|
return True
|
|
|
|
|
|
def _find_site(dex, code_off, spec):
|
|
"""Return the single instruction matching spec['match'].
|
|
|
|
Exactly one match is required. Ambiguity is treated as failure rather than
|
|
"take the first": picking the wrong site in a method with several similar
|
|
branches is a silent, expensive error.
|
|
"""
|
|
insns = list(dex.decode(code_off))
|
|
hits = [i for i in insns if _matches(dex, i, spec.get("match"), code_off)]
|
|
if len(hits) == 1:
|
|
return hits[0], insns
|
|
if not hits:
|
|
return None, insns
|
|
|
|
# allow an ordinal to disambiguate deliberately
|
|
ordv = spec.get("match", {}).get("ordinal")
|
|
if ordv is not None and 0 <= ordv < len(hits):
|
|
return hits[ordv], insns
|
|
return ("ambiguous", hits), insns
|
|
|
|
|
|
def _verifier_report(dex, code_off, target_off_for_new_edge=None):
|
|
"""Tier-3 check: does any branch target a move-result instruction?
|
|
|
|
`move-result*` must be immediately preceded by its producing invoke. A branch
|
|
into one bypasses the producer and the class fails to load with
|
|
`VerifyError ... copyResN v <- result0 type=Undefined`. A linear "is the
|
|
previous instruction the producer" scan misses this; only the CFG view sees
|
|
it.
|
|
"""
|
|
findings = []
|
|
for insn in dex.decode(code_off):
|
|
if insn["op"] not in MOVE_RESULT_OPS:
|
|
continue
|
|
tgt = insn["off"]
|
|
for other in dex.decode(code_off):
|
|
t = dex.branch_target(other)
|
|
if t == tgt:
|
|
findings.append((other["off"], tgt))
|
|
return findings
|
|
|
|
|
|
def _apply_replace(dex_data, insn, repl):
|
|
"""Return the new bytes for this instruction, enforcing equal length."""
|
|
kind = repl.get("kind")
|
|
if kind == "nops":
|
|
n = len(insn["raw"])
|
|
return b"\x00" * n
|
|
if kind == "bytes":
|
|
blob = bytes.fromhex(repl["bytes"].replace(" ", ""))
|
|
return blob
|
|
raise ValueError("replace must be {'kind':'nops'} or {'kind':'bytes','bytes':...}")
|
|
|
|
|
|
def process_one(dex, data, spec, index, dry_run):
|
|
"""Apply one spec entry to `data`; return a report dict, or raise."""
|
|
cls = spec["class"]
|
|
name = spec["method"]
|
|
desc = spec.get("desc")
|
|
label = spec.get("name") or ("patch%d" % index)
|
|
report = {"name": label, "ok": False}
|
|
|
|
if desc:
|
|
found = dex.find_method(cls, name, desc)
|
|
if not found:
|
|
raise RuntimeError("[%s] method not found: %s->%s%s"
|
|
% (label, cls, name, desc))
|
|
section, midx, code_off = found
|
|
else:
|
|
overloads = dex.find_methods_named(cls, name)
|
|
if len(overloads) != 1:
|
|
raise RuntimeError(
|
|
"[%s] %s->%s has %d overloads; give \"desc\" to disambiguate: %s"
|
|
% (label, cls, name, len(overloads),
|
|
[d for _s, _i, d, _c in overloads]))
|
|
section, midx, dsc, code_off = overloads[0]
|
|
desc = dsc
|
|
|
|
report.update({"class": cls, "method": name, "desc": desc,
|
|
"section": section, "method_idx": midx, "code_off": code_off})
|
|
|
|
site, _insns = _find_site(dex, code_off, spec)
|
|
if site is None:
|
|
raise RuntimeError("[%s] no instruction matched spec['match']" % label)
|
|
if isinstance(site, tuple) and site and site[0] == "ambiguous":
|
|
cands = ", ".join("0x%x" % h["off"] for h in site[1])
|
|
raise RuntimeError("[%s] match is ambiguous (%d hits: %s); tighten it or "
|
|
"set match.ordinal" % (label, len(site[1]), cands))
|
|
|
|
before, cur, after = _describe_neighbours(dex, code_off, site["off"])
|
|
report.update({
|
|
"site_off": cur["off"],
|
|
"old_bytes": cur["raw"].hex(),
|
|
"old_text": dex.describe(cur),
|
|
"before": [dex.describe(i) for i in before],
|
|
"after": [dex.describe(i) for i in after],
|
|
})
|
|
|
|
# polarity pin: what must immediately follow?
|
|
exp = spec.get("expect_next")
|
|
if exp is not None:
|
|
nxt = after[0] if after else None
|
|
if nxt is None or not _matches(dex, nxt, exp, code_off):
|
|
raise RuntimeError(
|
|
"[%s] expect_next not satisfied at 0x%x. Expected %r, found %r.\n"
|
|
" This is the polarity check: the instruction after the branch "
|
|
"is not the one your spec assumes, so the branch points at the "
|
|
"wrong side."
|
|
% (label, cur["off"], exp, dex.describe(nxt) if nxt else None))
|
|
report["expect_next_ok"] = True
|
|
|
|
new = _apply_replace(data, cur, spec.get("replace", {"kind": "nops"}))
|
|
if len(new) != len(cur["raw"]):
|
|
raise RuntimeError(
|
|
"[%s] replacement is %d bytes but the instruction is %d. Equal length "
|
|
"is required: a longer replacement overwrites the next instruction."
|
|
% (label, len(new), len(cur["raw"])))
|
|
report.update({"new_bytes": new.hex(), "size_delta": 0})
|
|
|
|
# verifier: simulate the new instruction to see if it adds a branch edge
|
|
fake = Dex(bytes(data), dex.name)
|
|
fake_probe = dict(cur)
|
|
fake_probe["raw"] = new
|
|
fake_probe["op"] = new[0]
|
|
report["new_text"] = fake.describe(fake_probe)
|
|
|
|
if dry_run:
|
|
report["ok"] = True
|
|
return report, data
|
|
|
|
data[cur["off"]:cur["off"] + len(new)] = new
|
|
|
|
# Re-read the patched dex and confirm the edit landed.
|
|
#
|
|
# Compare the BYTE RANGE, not a decoded instruction: replacing a 4-byte branch
|
|
# with a nop pair yields two 1-unit instructions, so "the instruction at this
|
|
# offset has length 4" is false even though the patch is perfectly correct.
|
|
# The byte range is what the patch actually owns.
|
|
span = slice(cur["off"], cur["off"] + len(new))
|
|
if bytes(data[span]) != new:
|
|
raise RuntimeError(
|
|
"[%s] bytes at 0x%x are %s, expected %s"
|
|
% (label, cur["off"], bytes(data[span]).hex(), new.hex()))
|
|
|
|
patched = Dex(bytes(data), dex.name)
|
|
insns_after = list(patched.decode(code_off))
|
|
info = patched.code_info(code_off)
|
|
expected_end = info["insns_off"] + info["insns_size"] * 2
|
|
ended = (insns_after[-1]["off"] + insns_after[-1]["units"] * 2
|
|
if insns_after else info["insns_off"])
|
|
report["post_instruction_count"] = len(insns_after)
|
|
report["post_decode_clean"] = (ended == expected_end)
|
|
|
|
landed = [i for i in insns_after
|
|
if cur["off"] <= i["off"] < cur["off"] + len(new)]
|
|
report["post_text"] = " | ".join(patched.describe(i) for i in landed)
|
|
|
|
# the next instruction boundary after the replaced span must still be the
|
|
# instruction that followed the original one
|
|
nxt_after = patched.insn_at(code_off, cur["off"] + len(new))
|
|
report["post_next_text"] = patched.describe(nxt_after) if nxt_after else None
|
|
if not report["post_decode_clean"]:
|
|
raise RuntimeError(
|
|
"[%s] the method no longer decodes to exactly insns_size units after "
|
|
"the edit -- the replacement changed the instruction stream length. "
|
|
"Use an equal-length replacement." % label)
|
|
|
|
edges = _verifier_report(patched, code_off)
|
|
report["verifier_edges"] = len(edges)
|
|
if edges:
|
|
raise RuntimeError(
|
|
"[%s] verifier: a branch targets a move-result at %s; the class will "
|
|
"fail to load (VerifyError). Use a nop pair instead of a branch."
|
|
% (label, ", ".join("0x%x->0x%x" % e for e in edges)))
|
|
|
|
report["ok"] = True
|
|
return report, data
|
|
|
|
|
|
def main(argv):
|
|
ap = argparse.ArgumentParser(
|
|
description="Apply equal-length byte patches to a dex, with structural, "
|
|
"polarity and verifier checks.",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog=__doc__.split("Spec format")[-1][:1200] if __doc__ else None)
|
|
ap.add_argument("dex", help="input .dex, or an APK/zip (use --entry to pick)")
|
|
ap.add_argument("--entry", help="archive member to read (default: classes.dex)")
|
|
ap.add_argument("--spec", required=True, help="JSON spec file")
|
|
ap.add_argument("-o", "--out", help="output dex path")
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="report matches and diffs without writing")
|
|
ap.add_argument("--report", help="write a machine-readable JSON report here")
|
|
args = ap.parse_args(argv[1:])
|
|
|
|
dex, entry = load_dex(args.dex, args.entry) if os.path.exists(args.dex) else (None, None)
|
|
if dex is None:
|
|
# allow raw bytes piped through a path that does not exist yet: fail clearly
|
|
print("error: no such file: %s" % args.dex)
|
|
return 2
|
|
|
|
problems = dex.check()
|
|
if problems:
|
|
print("structural problems in %s (%s):" % (args.dex, entry))
|
|
for p in problems:
|
|
print(" - %s" % p)
|
|
print("refusing to patch a source that does not parse cleanly")
|
|
return 1
|
|
|
|
with open(args.spec, encoding="utf-8") as fh:
|
|
spec_doc = json.load(fh)
|
|
specs = spec_doc["patches"] if isinstance(spec_doc, dict) else spec_doc
|
|
|
|
ok_before = verify_dex_header(dex.d)
|
|
print("== source: %s (%s, %d bytes)" % (args.dex, entry, len(dex.d)))
|
|
print(" header before: checksum_ok=%s signature_ok=%s"
|
|
% ok_before)
|
|
if not all(ok_before):
|
|
print(" note: the SOURCE dex header is already inconsistent. Some")
|
|
print(" producers ship a zeroed signature field; the patch will fix it.")
|
|
|
|
data = bytearray(dex.d)
|
|
reports = []
|
|
for i, spec in enumerate(specs):
|
|
rep, data = process_one(dex, data, spec, i, args.dry_run)
|
|
reports.append(rep)
|
|
print("\n-- [%s] %s.%s%s" % (rep["name"], rep["class"], rep["method"], rep["desc"]))
|
|
print(" site 0x%x %s" % (rep["site_off"], rep["old_text"]))
|
|
for line in rep["before"][-2:]:
|
|
print(" (before) %s" % line)
|
|
print(" -> %s" % rep.get("new_text"))
|
|
for line in rep["after"][:2]:
|
|
print(" (after) %s" % line)
|
|
if rep.get("expect_next_ok"):
|
|
print(" polarity: expect_next satisfied")
|
|
|
|
if args.dry_run:
|
|
print("\n== dry run: nothing written")
|
|
if args.report:
|
|
with open(args.report, "w", encoding="utf-8") as fh:
|
|
json.dump(reports, fh, indent=2)
|
|
print(" report: %s" % args.report)
|
|
return 0
|
|
|
|
header = fix_dex_header(data)
|
|
ok_after = verify_dex_header(data)
|
|
print("\n== header checksum 0x%08x -> 0x%08x" % (header["before_checksum"],
|
|
header["after_checksum"]))
|
|
print(" signature %s -> %s"
|
|
% (header["before_signature"].hex()[:16],
|
|
header["after_signature"].hex()[:16]))
|
|
print(" self-verify: checksum_ok=%s signature_ok=%s" % ok_after)
|
|
if not all(ok_after):
|
|
print("FATAL: header does not self-verify; nothing written")
|
|
return 1
|
|
|
|
out = args.out
|
|
if not out:
|
|
root, ext = os.path.splitext(args.dex)
|
|
out = root + ".patched" + (ext or ".dex")
|
|
with open(out, "wb") as fh:
|
|
fh.write(bytes(data))
|
|
print("== wrote %s (%d bytes, delta %d)"
|
|
% (out, len(data), len(data) - len(dex.d)))
|
|
|
|
if args.report:
|
|
with open(args.report, "w", encoding="utf-8") as fh:
|
|
json.dump(reports, fh, indent=2)
|
|
print(" report: %s" % args.report)
|
|
|
|
print("\nNext: repack (scripts/repack.py) then sign with apksigner, then run "
|
|
"the device + screen checks in references/verification.md.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|