Haipeng Ren

"""Apktool disassembler subprocess wrapper.

Manages apktool subprocess execution for smali disassembly,
following the CAPA extractor subprocess pattern.
"""

import os
import signal
import subprocess
from typing import List, Optional


class ApktoolDisassembler:
    """Subprocess wrapper for apktool smali disassembly."""

    def __init__(
        self, apktool_path: str = None, timeout: int = None, log=None
    ):
        self.apktool_path = apktool_path or os.getenv("APKTOOL_PATH", "apktool")
        self.timeout = timeout or int(os.getenv("APKTOOL_TIMEOUT", "120"))
        self.log = log

    def disassemble(self, apk_path: str, output_dir: str) -> bool:
        """Run apktool disassembly on an APK file.

        Returns True on success, False on failure.
        """
        cmd = [
            self.apktool_path,
            "d",
            "--no-res",
            "--force",
            "--output", output_dir,
            apk_path,
        ]

        try:
            process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                preexec_fn=os.setsid,
            )

            try:
                stdout, stderr = process.communicate(timeout=self.timeout)
                if process.returncode != 0:
                    if self.log:
                        self.log.warning(
                            f"apktool returned non-zero exit code {process.returncode}: "
                            f"{stderr[:500] if stderr else 'no stderr'}"
                        )
                    return False
                return True

            except subprocess.TimeoutExpired:
                if self.log:
                    self.log.error(
                        f"apktool timed out after {self.timeout}s"
                    )
                try:
                    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
                    process.wait(timeout=3)
                except (ProcessLookupError, subprocess.TimeoutExpired):
                    try:
                        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
                    except ProcessLookupError:
                        pass
                return False

        except FileNotFoundError:
            if self.log:
                self.log.error(
                    f"apktool not found at '{self.apktool_path}'. "
                    "Install apktool or set APKTOOL_PATH env var."
                )
            return False
        except Exception as e:
            if self.log:
                self.log.error(f"apktool execution error: {e}")
            return False

    def get_smali_directories(self, output_dir: str) -> List[str]:
        """Return paths to all smali output directories.

        Handles multi-DEX: smali/, smali_classes2/, smali_classes3/, etc.
        """
        smali_dirs = []
        if not os.path.isdir(output_dir):
            return smali_dirs

        for entry in sorted(os.listdir(output_dir)):
            if entry == "smali" or entry.startswith("smali_classes"):
                full_path = os.path.join(output_dir, entry)
                if os.path.isdir(full_path):
                    smali_dirs.append(full_path)

        return smali_dirs