Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/workflows/tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -34,11 +34,15 @@ jobs:
- run: python -m pip install -r requirements.txt
- name: Install legacy baseline dependencies for performance comparison
run: python -m pip install -r .performance-base/requirements.txt
- name: Install virtual display for Tk GUI tests
run: sudo apt-get update && sudo apt-get install --no-install-recommends -y xvfb xauth
- run: python -m pip check
- name: Unit tests (no skips)
run: python tests/run_tests.py --require-decompiler --suite unit
- name: Integration tests (no skips)
run: python tests/run_tests.py --require-decompiler --suite integration
- name: GUI tests (no skips)
run: xvfb-run -a python tests/run_tests.py --require-decompiler --suite gui
- name: Reject material performance regressions against the target branch
run: python tests/benchmark_compare.py --baseline .performance-base --samples 31
- name: Check cold-process startup with --debug
Expand Down
1 change: 1 addition & 0 deletions droidasc/asc_client/apk/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Internal APK loading helpers."""
149 changes: 149 additions & 0 deletions droidasc/asc_client/apk/dex_reader.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
import struct


_U32_FROM = struct.Struct("<I").unpack_from


def _skip_uleb128(buf, off : int) -> int:
while buf[off] & 0x80:
off += 1
return off + 1


def _read_string_data_bytes(buf, str_off : int) -> bytes:
# The leading uleb128 stores UTF-16 code-unit count, not byte length.
# Reading to the terminating NUL is therefore required for MUTF-8.
ptr = _skip_uleb128(buf, str_off)
end = buf.find(b"\x00", ptr)
if end < 0:
raise ValueError("unterminated string_data_item")
return buf[ptr:end]


def _find_type_idx(buf, target_bytes : bytes) -> int:
if len(buf) < 0x70 or buf[:3] != b"dex":
return -1

string_ids_size = _U32_FROM(buf, 0x38)[0]
string_ids_off = _U32_FROM(buf, 0x3C)[0]
type_ids_size = _U32_FROM(buf, 0x40)[0]
type_ids_off = _U32_FROM(buf, 0x44)[0]

if string_ids_size == 0 or type_ids_size == 0:
return -1
if string_ids_off + string_ids_size * 4 > len(buf):
raise ValueError("bad string_ids range")
if type_ids_off + type_ids_size * 4 > len(buf):
raise ValueError("bad type_ids range")

left = 0
right = type_ids_size - 1
# type_ids follows string-id order. DEX MUTF-8 bytes preserve UTF-16
# code-unit ordering, so compare encoded descriptors without decoding.
while left <= right:
mid = (left + right) >> 1
str_idx = _U32_FROM(buf, type_ids_off + (mid << 2))[0]
if str_idx >= string_ids_size:
raise ValueError("bad type_id->string_idx")
str_off = _U32_FROM(buf, string_ids_off + (str_idx << 2))[0]
if str_off >= len(buf):
raise ValueError("bad string_data_off")

cls_bytes = _read_string_data_bytes(buf, str_off)
if cls_bytes == target_bytes:
return mid
if cls_bytes < target_bytes:
left = mid + 1
else:
right = mid - 1

return -1


def _class_defs_contains_type_idx(buf, type_idx : int) -> bool:
class_defs_size = _U32_FROM(buf, 0x60)[0]
class_defs_off = _U32_FROM(buf, 0x64)[0]

if class_defs_size == 0:
return False

class_defs_end = class_defs_off + (class_defs_size << 5)
if class_defs_end > len(buf):
raise ValueError("bad class_defs range")

needle = type_idx.to_bytes(4, "little")
pos = buf.find(needle, class_defs_off, class_defs_end)
while pos != -1:
# The same four bytes may occur in another class_def field. class_idx
# is valid only at the first word of each 32-byte class_def_item.
if ((pos - class_defs_off) & 0x1F) == 0:
return True
pos = buf.find(needle, pos + 1, class_defs_end)
return False


def _dex_defines_class(buf, target_bytes : bytes) -> bool:
type_idx = _find_type_idx(buf, target_bytes)
if type_idx < 0:
return False
return _class_defs_contains_type_idx(buf, type_idx)


def _decode_name(raw_name : bytes) -> str:
try:
return raw_name.decode("utf-8")
except UnicodeDecodeError:
# Surrogate-pair MUTF-8 is invalid strict UTF-8. Import the fallback
# only for such names so ordinary listclass startup stays cheap.
from droidasc.asc_core.utils.tinydex import _decode_mutf8_fallback

return _decode_mutf8_fallback(raw_name)


def list_dex_classes(buf, prefix : str = None, prefix_bytes : bytes = None):
if len(buf) < 0x70 or buf[:3] != b"dex":
raise ValueError("invalid DEX header")

string_ids_size = _U32_FROM(buf, 0x38)[0]
string_ids_off = _U32_FROM(buf, 0x3C)[0]
type_ids_size = _U32_FROM(buf, 0x40)[0]
type_ids_off = _U32_FROM(buf, 0x44)[0]
class_defs_size = _U32_FROM(buf, 0x60)[0]
class_defs_off = _U32_FROM(buf, 0x64)[0]

if class_defs_size == 0:
return []
if string_ids_off + string_ids_size * 4 > len(buf):
raise ValueError("bad string_ids range")
if type_ids_off + type_ids_size * 4 > len(buf):
raise ValueError("bad type_ids range")
if class_defs_off + class_defs_size * 32 > len(buf):
raise ValueError("bad class_defs range")

names = []
append = names.append
class_defs_end = class_defs_off + class_defs_size * 32
for class_def_off in range(class_defs_off, class_defs_end, 32):
type_idx = _U32_FROM(buf, class_def_off)[0]
if type_idx >= type_ids_size:
raise ValueError("bad class_def->type_idx")
string_idx = _U32_FROM(buf, type_ids_off + (type_idx << 2))[0]
if string_idx >= string_ids_size:
raise ValueError("bad type_id->string_idx")
string_off = _U32_FROM(buf, string_ids_off + (string_idx << 2))[0]
if string_off >= len(buf):
raise ValueError("bad string_data_off")
try:
raw_name = _read_string_data_bytes(buf, string_off)
except IndexError as error:
raise ValueError("bad string_data_off") from error

# ASCII prefixes can be rejected before decoding every class name.
if prefix_bytes is not None:
if raw_name.startswith(prefix_bytes):
append(_decode_name(raw_name))
else:
name = _decode_name(raw_name)
if prefix is None or name.startswith(prefix):
append(name)
return names
144 changes: 144 additions & 0 deletions droidasc/asc_client/apk/loader.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
import mmap
import struct
import zlib


_U16_FROM = struct.Struct("<H").unpack_from
_U32_FROM = struct.Struct("<I").unpack_from
_CD_SIG = b"PK\x01\x02"
_LH_SIG = b"PK\x03\x04"
_EOCD_SIG = b"PK\x05\x06"
_DEX_SUFFIX = b".dex"
_SLASH = ord("/")
_DEFLATE_CHUNK = 1 << 19


def _find_eocd(mm : mmap.mmap) -> int:
# EOCD can only be followed by a 65535-byte ZIP comment and its 22-byte
# fixed header, so searching the whole APK is unnecessary.
search_start = max(0, len(mm) - 65536 - 22)
return mm.rfind(_EOCD_SIG, search_start)


def _parse_cd_dex_entries(mm : mmap.mmap):
eocd_idx = _find_eocd(mm)
if eocd_idx < 0:
raise ValueError("EOCD not found")

cd_size = _U32_FROM(mm, eocd_idx + 12)[0]
cd_off = _U32_FROM(mm, eocd_idx + 16)[0]
cd_end = cd_off + cd_size
entries = []
seen_names = set()
pos = cd_off

# APKs normally name DEX files classes*.dex. Searching those bytes avoids
# walking every resource entry in a large central directory.
while True:
pos = mm.find(b"classes", pos, cd_end)
if pos < 0:
break
header_off = pos - 46
pos += 7
if header_off < cd_off or mm[header_off:header_off + 4] != _CD_SIG:
continue

name_len = _U16_FROM(mm, header_off + 28)[0]
name_start = pos - 7
name_end = name_start + name_len
if name_end > cd_end:
continue

name_bytes = mm[name_start:name_end]
if (
not name_bytes.endswith(_DEX_SUFFIX)
or _SLASH in name_bytes
or name_bytes in seen_names
):
continue

seen_names.add(name_bytes)
# name, uncompressed size, compressed size, local-header offset,
# compression method. Keep this tuple compact: it crosses processes.
entries.append((
name_bytes.decode("utf-8", errors="ignore"),
_U32_FROM(mm, header_off + 24)[0],
_U32_FROM(mm, header_off + 20)[0],
_U32_FROM(mm, header_off + 42)[0],
_U16_FROM(mm, header_off + 10)[0],
))

if entries:
return entries

# Fall back to a complete central-directory walk for valid but unusual ZIP
# layouts where the name search cannot recover an entry header.
ptr = cd_off
while ptr + 46 <= cd_end:
if mm[ptr:ptr + 4] != _CD_SIG:
break

name_len = _U16_FROM(mm, ptr + 28)[0]
extra_len = _U16_FROM(mm, ptr + 30)[0]
comment_len = _U16_FROM(mm, ptr + 32)[0]
name_start = ptr + 46
name_end = name_start + name_len
name_bytes = mm[name_start:name_end]
if name_bytes.endswith(_DEX_SUFFIX) and _SLASH not in name_bytes:
entries.append((
name_bytes.decode("utf-8", errors="ignore"),
_U32_FROM(mm, ptr + 24)[0],
_U32_FROM(mm, ptr + 20)[0],
_U32_FROM(mm, ptr + 42)[0],
_U16_FROM(mm, ptr + 10)[0],
))

ptr = name_end + extra_len + comment_len

return entries


def _inflate_deflate_chunks(comp_view, stop_event):
if stop_event is None:
# A single call is about 4% faster when cancellation is unnecessary.
return zlib.decompress(comp_view, -15)
decomp = zlib.decompressobj(-15)
out = bytearray()
pos = 0
total = len(comp_view)
while pos < total:
if stop_event.is_set():
return None
end = min(pos + _DEFLATE_CHUNK, total)
out.extend(decomp.decompress(comp_view[pos:end]))
pos = end
out.extend(decomp.flush())
if stop_event.is_set():
return None
return bytes(out)


def _inflate_dex(mm : mmap.mmap, entry, stop_event = None):
_name, uncomp_size, comp_size, local_header_off, comp_method = entry
if mm[local_header_off:local_header_off + 4] != _LH_SIG:
raise ValueError("bad local header signature")

name_len = _U16_FROM(mm, local_header_off + 26)[0]
extra_len = _U16_FROM(mm, local_header_off + 28)[0]
# File data starts after the variable local filename and extra fields, not
# at the corresponding central-directory record.
data_off = local_header_off + 30 + name_len + extra_len
comp_view = memoryview(mm)[data_off:data_off + comp_size]

if comp_method == 0:
data = bytes(comp_view)
elif comp_method == 8:
data = _inflate_deflate_chunks(comp_view, stop_event)
if data is None:
return None
else:
raise ValueError(f"unsupported compression method: {comp_method}")

if uncomp_size and len(data) != uncomp_size:
raise ValueError(f"size mismatch: expect {uncomp_size}, got {len(data)}")
return data
Loading
Loading