Processing many APKs¶

Malware collections contain truncated downloads, files that aren't APKs, and archives broken on purpose. Your batch job should keep going past all of them.

CLI¶

show --json walks directories recursively and prints one JSON object per APK. It reports files that fail to parse on stderr and skips them, so stdout stays valid JSONL:

apk-info show --json ./samples/ > results.jsonl 2> errors.log

# package name per file
jq -r '[.path, .package_name] | @tsv' results.jsonl

# APKs that target an old SDK
jq -r 'select((.target_sdk_version | tonumber) < 29) | .path' results.jsonl

Python¶

Catch APKError per file, and use a thread pool to parse in parallel:

def triage(path: Path) -> dict:
    try:
        apk = APK(path)
    except APKError as e:
        return {"file": str(path), "error": str(e)}

    return {
        "file": str(path),
        "package": apk.get_package_name(),
        "version": apk.get_version_name(),
        "min_sdk": apk.get_min_sdk_version(),
        "main_activity": apk.get_main_activity(),
        "permissions": sorted(apk.get_permissions()),
    }


if __name__ == "__main__":
    files = [p for p in Path(sys.argv[1]).rglob("*") if p.suffix.lower() in EXTENSIONS]

    # parsing releases the GIL, so threads run in parallel
    with ThreadPoolExecutor(max_workers=8) as pool:
        for result in pool.map(triage, files):
            print(json.dumps(result))

{"file": "samples/junk.apk", "error": "got error while parsing apk archive: provided file is not a zip archive"}
{"file": "samples/org.adaway.apk", "package": "org.adaway", "version": "6.1.4", "min_sdk": "26", ...}

APK(), get_signatures() and read() release the GIL while they work, so threads parse files in parallel, about 3× faster with 8 threads than with one. An APK object can be passed between threads.

Rust¶

Apk reads files independently, so threads parallelize well:

fn triage(path: &Path) -> serde_json::Value {
    match Apk::new(path) {
        Ok(apk) => serde_json::json!({
            "file": path,
            "package": apk.get_package_name(),
            "version": apk.get_version_name(),
            "min_sdk": apk.get_min_sdk_version(),
            "permissions": apk.get_permissions().collect::<Vec<_>>(),
        }),
        Err(e) => serde_json::json!({ "file": path, "error": e.to_string() }),
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let dir = std::env::args().nth(1).ok_or("usage: batch <folder>")?;
    let mut files = Vec::new();
    collect(Path::new(&dir), &mut files)?;

    // `Apk` only reads, so each thread can parse its own files independently
    let workers = std::thread::available_parallelism()?.get();
    let chunk = files.len().div_ceil(workers).max(1);
    std::thread::scope(|scope| {
        for part in files.chunks(chunk) {
            scope.spawn(move || {
                for path in part {
                    println!("{}", triage(path));
                }
            });
        }
    });

    Ok(())
}

Tips¶

Filter by extension (.apk, .xapk, .apkm, .apks), but don't rely on it: many malware samples have no extension. Files that aren't zip archives fail fast with APKError.
APK() reads only the central directory, the manifest and resources.arsc. It reads DEX files and other entries on demand, so triage stays cheap for large APKs.
get_signatures() parses the signing block when you call it.