Apk forge
""" APK Forge: Advanced APK Mutation Framework Educational use only. User assumes all liability. """
import os import sys import shutil import hashlib import subprocess import multiprocessing from pathlib import Path from typing import Dict, List, Optional, Union from functools import lru_cache from dataclasses import dataclass import yaml import struct import mmap import re from concurrent.futures import ThreadPoolExecutor
@dataclass class PatchConfig: """Configuration for a single patch.""" patch_type: str target: Optional[str] = None replacement: Optional[str] = None resource_id: Optional[int] = None new_value: Optional[str] = None class_name: Optional[str] = None smali_code: Optional[str] = None remove_permission: Optional[str] = None add_meta_data: Optional[str] = None
@dataclass class APKInfo: """Metadata for an APK.""" path: Path sha256: str is_split: bool = False split_paths: List[Path] = None
class DependencyResolver: """Automatically downloads and caches required tools."""
CACHE_DIR = Path.home() / ".apk_forge" / "cache"
TOOLS = {
"apktool": {
"url": "https://github.com/iBotPeaches/Apktool/releases/download/v2.9.3/apktool_2.9.3.jar",
"filename": "apktool.jar"
},
"bundletool": {
"url": "https://github.com/google/bundletool/releases/download/1.15.6/bundletool-all-1.15.6.jar",
"filename": "bundletool.jar"
},
"zipalign": {
"url": "https://dl.google.com/android/repository/commandlinetools-linux-9477386_latest.zip",
"filename": "zipalign",
"binary_path": "cmdline-tools/bin/zipalign"
}
}
def __init__(self):
self.CACHE_DIR.mkdir(parents=True, exist_ok=True)
def _download_file(self, url: str, destination: Path) -> None:
"""Download a file from a URL."""
import urllib.request
if not destination.exists():
print(f"Downloading {url}...")
urllib.request.urlretrieve(url, destination)
def resolve(self, tool_name: str) -> Path:
"""Resolve and return the path to a tool."""
tool_info = self.TOOLS[tool_name]
cached_path = self.CACHE_DIR / tool_info["filename"]
if not cached_path.exists():
self._download_file(tool_info["url"], cached_path)
if tool_name == "zipalign":
# Extract zipalign from the ZIP file
import zipfile
with zipfile.ZipFile(cached_path, 'r') as zip_ref:
zip_ref.extractall(self.CACHE_DIR)
binary_path = self.CACHE_DIR / tool_info["binary_path"]
if not binary_path.exists():
raise FileNotFoundError(f"zipalign binary not found at {binary_path}")
return binary_path
return cached_path
class APKProcessor: """Handles APK disassembly, patching, and reassembly."""
def __init__(self, dependency_resolver: DependencyResolver):
self.dependency_resolver = dependency_resolver
self.apktool_path = dependency_resolver.resolve("apktool")
self.zipalign_path = dependency_resolver.resolve("zipalign")
def disassemble_apk(self, apk_path: Path, output_dir: Path) -> None:
"""Disassemble an APK using apktool."""
cmd = ["java", "-jar", str(self.apktool_path), "d", str(apk_path), "-o", str(output_dir)]
subprocess.run(cmd, check=True)
def assemble_apk(self, input_dir: Path, output_path: Path) -> None:
"""Reassemble an APK using apktool."""
cmd = ["java", "-jar", str(self.apktool_path), "b", str(input_dir), "-o", str(output_path)]
subprocess.run(cmd, check=True)
def zipalign_apk(self, input_path: Path, output_path: Path) -> None:
"""Zipalign the APK."""
cmd = [str(self.zipalign_path), "-p", "4", str(input_path), str(output_path)]
subprocess.run(cmd, check=True)
@lru_cache(maxsize=10)
def get_decompiled_dir(self, apk_path: Path) -> Path:
"""Return cached decompiled directory or decompile the APK."""
sha256 = self._calculate_sha256(apk_path)
cached_dir = Path(f"~/.apk_forge/cache/{sha256}").expanduser()
if cached_dir.exists():
return cached_dir
cached_dir.mkdir(parents=True, exist_ok=True)
self.disassemble_apk(apk_path, cached_dir)
return cached_dir
def _calculate_sha256(self, file_path: Path) -> str:
"""Calculate the SHA-256 hash of a file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
class SmaliASTParser: """Rudimentary Smali parser for AST-based patching."""
def __init__(self, smali_content: str):
self.smali_content = smali_content
self.lines = smali_content.splitlines()
def find_method(self, method_signature: str) -> Optional[Dict]:
"""Find a method by its signature and return its start/end lines."""
method_pattern = re.compile(rf"\.method.*{re.escape(method_signature)}")
for i, line in enumerate(self.lines):
if method_pattern.search(line):
# Find the end of the method
end_line = i + 1
while end_line < len(self.lines) and not self.lines[end_line].startswith(".end method"):
end_line += 1
return {"start": i, "end": end_line}
return None
def replace_method_body(self, method_signature: str, new_body: str) -> str:
"""Replace the body of a method with new Smali code."""
method_info = self.find_method(method_signature)
if not method_info:
raise ValueError(f"Method {method_signature} not found.")
# Preserve the method declaration and .end method
new_lines = self.lines[:method_info["start"] + 1]
new_lines.append(new_body)
new_lines.extend(self.lines[method_info["end"]:])
return "\n".join(new_lines)
class ResourceBinaryPatcher: """Patches binary resources in resources.arsc."""
RES_TABLE_HEADER_FORMAT = "<I" # Simplified header format
def __init__(self, arsc_path: Path):
self.arsc_path = arsc_path
def patch_string_resource(self, resource_id: int, new_value: str) -> None:
"""Patch a string resource by its ID."""
with open(self.arsc_path, "r+b") as f:
mm = mmap.mmap(f.fileno(), 0)
# Simplified: Find and replace the string (actual implementation requires parsing ResTable)
# This is a placeholder for the actual binary patching logic
print(f"Patching resource {hex(resource_id)} to '{new_value}' (simplified)")
mm.close()
class PatchApplier: """Applies patches to decompiled APK directories."""
def __init__(self, decompiled_dir: Path, patch_configs: List[PatchConfig]):
self.decompiled_dir = decompiled_dir
self.patch_configs = patch_configs
def apply_patches(self) -> None:
"""Apply all patches in the config."""
for patch in self.patch_configs:
if patch.patch_type == "smali_ast_patch":
self._apply_smali_ast_patch(patch)
elif patch.patch_type == "binary_resource_patch":
self._apply_binary_resource_patch(patch)
elif patch.patch_type == "inject_smali_class":
self._apply_inject_smali_class(patch)
elif patch.patch_type == "manifest_patch":
self._apply_manifest_patch(patch)
def _apply_smali_ast_patch(self, patch: PatchConfig) -> None:
"""Apply a Smali AST patch."""
smali_dir = self.decompiled_dir / "smali"
for smali_file in smali_dir.rglob("*.smali"):
with open(smali_file, "r", encoding="utf-8") as f:
content = f.read()
parser = SmaliASTParser(content)
try:
new_content = parser.replace_method_body(patch.target, patch.replacement)
with open(smali_file, "w", encoding="utf-8") as f:
f.write(new_content)
print(f"Patched {smali_file}")
except ValueError:
continue
def _apply_binary_resource_patch(self, patch: PatchConfig) -> None:
"""Apply a binary resource patch."""
arsc_path = self.decompiled_dir / "resources.arsc"
if arsc_path.exists():
patcher = ResourceBinaryPatcher(arsc_path)
patcher.patch_string_resource(patch.resource_id, patch.new_value)
def _apply_inject_smali_class(self, patch: PatchConfig) -> None:
"""Inject a new Smali class."""
smali_dir = self.decompiled_dir / "smali"
class_path = smali_dir / f"{patch.class_name.replace('.', '/')}.smali"
class_path.parent.mkdir(parents=True, exist_ok=True)
with open(class_path, "w", encoding="utf-8") as f:
f.write(patch.smali_code)
print(f"Injected {class_path}")
def _apply_manifest_patch(self, patch: PatchConfig) -> None:
"""Apply a manifest patch."""
manifest_path = self.decompiled_dir / "AndroidManifest.xml"
if manifest_path.exists():
with open(manifest_path, "r", encoding="utf-8") as f:
content = f.read()
if patch.remove_permission:
content = re.sub(
rf'<uses-permission android:name="{re.escape(patch.remove_permission)}" />',
"",
content
)
if patch.add_meta_data:
content = content.replace(
"</application>",
f"{patch.add_meta_data}</application>"
)
with open(manifest_path, "w", encoding="utf-8") as f:
f.write(content)
print(f"Patched {manifest_path}")
class ReportGenerator: """Generates an HTML report of applied patches."""
def __init__(self, original_apk: Path, patched_apk: Path, decompiled_dir: Path):
self.original_apk = original_apk
self.patched_apk = patched_apk
self.decompiled_dir = decompiled_dir
def generate_report(self, output_path: Path = Path("patch_report.html")) -> None:
"""Generate an HTML report."""
original_sha256 = self._calculate_sha256(self.original_apk)
patched_sha256 = self._calculate_sha256(self.patched_apk)
report_content = f"""
<html>
<head><title>APK Patch Report</title></head>
<body>
<h1>APK Patch Report</h1>
<h2>Original APK: {self.original_apk.name}</h2>
<h2>Patched APK: {self.patched_apk.name}</h2>
<h3>SHA-256 Hashes:</h3>
<ul>
<li>Original: {original_sha256}</li>
<li>Patched: {patched_sha256}</li>
</ul>
<h3>Modified Files:</h3>
<ul>
{"".join(f"<li>{f.relative_to(self.decompiled_dir)}</li>" for f in self.decompiled_dir.rglob("*") if f.is_file())}
</ul>
</body>
</html>
"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(report_content)
print(f"Report generated: {output_path}")
def _calculate_sha256(self, file_path: Path) -> str:
"""Calculate the SHA-256 hash of a file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
class APKForge: """Main framework class."""
def __init__(self):
self.dependency_resolver = DependencyResolver()
self.processor = APKProcessor(self.dependency_resolver)
def patch_apk(
self,
apk_path: Path,
config_path: Path,
output_path: Path,
mapping_file: Optional[Path] = None
) -> None:
"""Patch an APK based on a YAML config."""
# Load config
with open(config_path, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
patch_configs = [
PatchConfig(**patch) for patch in config.get("patches", [])
]
# Handle obfuscation mapping
if mapping_file:
self._translate_patches_with_mapping(patch_configs, mapping_file)
# Decompile APK
decompiled_dir = self.processor.get_decompiled_dir(apk_path)
# Apply patches
patch_applier = PatchApplier(decompiled_dir, patch_configs)
patch_applier.apply_patches()
# Reassemble APK
temp_apk_path = output_path.parent / f"temp_{output_path.name}"
self.processor.assemble_apk(decompiled_dir, temp_apk_path)
self.processor.zipalign_apk(temp_apk_path, output_path)
temp_apk_path.unlink()
# Generate report
report_generator = ReportGenerator(apk_path, output_path, decompiled_dir)
report_generator.generate_report()
print(f"Patched APK saved to: {output_path}")
def _translate_patches_with_mapping(
self,
patch_configs: List[PatchConfig],
mapping_file: Path
) -> None:
"""Translate patch targets using a ProGuard mapping file."""
with open(mapping_file, "r", encoding="utf-8") as f:
mapping_lines = f.readlines()
for patch in patch_configs:
if patch.target:
for line in mapping_lines:
if line.startswith(patch.target.split("->")[0]):
obfuscated_name = line.split(" -> ")[1].strip()
patch.target = patch.target.replace(
patch.target.split("->")[0],
obfuscated_name
)
break
def main(): import argparse parser = argparse.ArgumentParser(description="APK Forge: Advanced APK Mutation Framework") parser.add_argument("--apk", type=Path, required=True, help="Path to the input APK") parser.add_argument("--config", type=Path, required=True, help="Path to the YAML config file") parser.add_argument("--output", type=Path, required=True, help="Path to the output APK") parser.add_argument("--mapping-file", type=Path, help="Path to the ProGuard mapping file") args = parser.parse_args()
forge = APKForge()
forge.patch_apk(args.apk, args.config, args.output, args.mapping_file)
if name == "main": main()