Malte Krack

"""JADX decompiler subprocess wrapper.

Manages JADX subprocess execution for Java decompilation,
following the CAPA extractor subprocess pattern.
"""

import os
import re
import signal
import subprocess
from typing import Dict, Optional


class JADXDecompiler:
    """Subprocess wrapper for JADX Java decompilation."""

    def __init__(self, jadx_path: str = None, timeout: int = None, log=None):
        self.jadx_path = jadx_path or os.getenv("JADX_PATH", "jadx")
        self.timeout = timeout or int(os.getenv("JADX_TIMEOUT", "300"))
        self.log = log

    def decompile(self, apk_path: str, output_dir: str) -> bool:
        """Run JADX decompilation on an APK file.

        Returns True on success, False on failure.
        """
        cmd = [
            self.jadx_path,
            "--no-res",
            "--no-imports",
            "--threads-count", "2",
            "--output-dir", 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"JADX returned non-zero exit code {process.returncode}: "
                            f"{stderr[:500] if stderr else 'no stderr'}"
                        )
                    # JADX may still produce partial output on non-zero exit
                    return os.path.isdir(os.path.join(output_dir, "sources"))
                return True

            except subprocess.TimeoutExpired:
                if self.log:
                    self.log.error(
                        f"JADX timed out after {self.timeout}s"
                    )
                # Kill the process group
                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"JADX not found at '{self.jadx_path}'. "
                    "Install JADX or set JADX_PATH env var."
                )
            return False
        except Exception as e:
            if self.log:
                self.log.error(f"JADX execution error: {e}")
            return False

    def parse_java_methods(self, output_dir: str) -> Dict[str, str]:
        """Parse JADX output into per-method Java source.

        Returns dict keyed by 'package.ClassName.methodName(ParamType1, ParamType2)'.
        """
        sources_dir = os.path.join(output_dir, "sources")
        if not os.path.isdir(sources_dir):
            return {}

        methods = {}
        for root, _dirs, files in os.walk(sources_dir):
            for fname in files:
                if not fname.endswith(".java"):
                    continue
                fpath = os.path.join(root, fname)
                try:
                    with open(fpath, "r", encoding="utf-8", errors="replace") as f:
                        content = f.read()

                    # Derive the fully qualified class name from file path
                    rel_path = os.path.relpath(fpath, sources_dir)
                    class_fqn = rel_path.replace(os.sep, ".").replace("/", ".")
                    if class_fqn.endswith(".java"):
                        class_fqn = class_fqn[:-5]

                    file_methods = self._extract_methods_from_java(
                        content, class_fqn
                    )
                    methods.update(file_methods)

                except Exception:
                    continue

        return methods

    def _extract_methods_from_java(
        self, source: str, class_fqn: str
    ) -> Dict[str, str]:
        """Extract individual methods from a Java source file.

        Uses brace-depth tracking to find method boundaries.
        """
        methods = {}
        lines = source.split("\n")
        i = 0

        while i < len(lines):
            line = lines[i]
            match = self._is_method_declaration(line)
            if match:
                method_name, param_types = match
                # Track brace depth to find method end
                start_line = i
                brace_depth = 0
                found_open = False

                for j in range(i, len(lines)):
                    for ch in lines[j]:
                        if ch == "{":
                            brace_depth += 1
                            found_open = True
                        elif ch == "}":
                            brace_depth -= 1

                    if found_open and brace_depth == 0:
                        method_body = "\n".join(lines[start_line : j + 1])
                        key = f"{class_fqn}.{method_name}({param_types})"
                        methods[key] = method_body
                        i = j + 1
                        break
                else:
                    i += 1
            else:
                i += 1

        return methods

    # Regex for method declarations (not constructors, not class/interface decl)
    _METHOD_DECL_RE = re.compile(
        r"^\s*(?:(?:public|private|protected|static|final|synchronized|"
        r"native|abstract|strictfp)\s+)*"
        r"(?:[\w<>\[\]?.]+\s+)"  # return type
        r"(\w+)"  # method name
        r"\s*\(([^)]*)\)"  # parameters
        r"(?:\s*throws\s+[\w.,\s]+)?"
        r"\s*\{"
    )

    def _is_method_declaration(self, line: str) -> Optional[tuple]:
        """Check if a line is a method declaration.

        Returns (method_name, param_types_str) or None.
        """
        m = self._METHOD_DECL_RE.match(line)
        if not m:
            return None

        method_name = m.group(1)
        params_raw = m.group(2).strip()

        # Skip class-level blocks
        if method_name in ("if", "for", "while", "switch", "try", "catch"):
            return None

        # Extract just the type names from parameters
        if not params_raw:
            return (method_name, "")

        param_types = []
        for param in params_raw.split(","):
            param = param.strip()
            if not param:
                continue
            parts = param.split()
            # Last non-annotation token before the variable name is the type
            # Simple heuristic: type is second-to-last, name is last
            if len(parts) >= 2:
                param_types.append(parts[-2])
            elif len(parts) == 1:
                param_types.append(parts[0])

        return (method_name, ", ".join(param_types))