Script to get strace logs from APKs.

dataflow.py

import json
import re
import argparse
from glob import glob
from os.path import join, basename, splitext


PATTERN = r"\{EPOLLIN, \{u32=\d+, u64=\d+}}"
SENDTO_FLOW = {}
ALL_FLOWS = {}


def process_epoll(line):
    all_fd = re.findall(PATTERN, line)
    for fd in all_fd:
        uid = re.search(r"u32=(\d+)", fd).group(1)
        if uid not in SENDTO_FLOW:
            SENDTO_FLOW[uid] = {}


def process_recvfrom(line):
    recvfrom = {}
    line = line.split()
    id = re.search(r"\d+", line[0]).group(0)
    # recvfrom["total_data"] = line[-1]xx
    # recvfrom["max_allowed"] = line[-6][:-1]

    if id in SENDTO_FLOW:
        SENDTO_FLOW[id]["recvfrom_total_data"] = line[-1]
        SENDTO_FLOW[id]["recvfrom_max_allowed"] = line[-6][:-1]


def process_sendto(line):
    sendto = {}
    line = line.split()
    id = re.search(r"\d+", line[0]).group(0)

    if id in SENDTO_FLOW:
        SENDTO_FLOW[id]["sendto_total_data"] = line[-1]


def process_directory(directory_path):
    all_files = glob(join(directory_path, "*.log"))
    all_data_flow = {}
    for file in all_files:
        file_name = basename(file)
        package_name = splitext(file_name)[0]
        ALL_FLOWS[package_name] = main(file)
        SENDTO_FLOW = {}
    print(ALL_FLOWS)
    # with open("dataflow_sendto.json", "w") as f:
    #     json.dumps(f)


def main(file):
    with open(file, "r") as f:
        data = f.read().splitlines()
    for line in data:
        if line.startswith("epoll_pwait"):
            process_epoll(line)
        if line.startswith("recvfrom"):
            if "EAGAIN" not in line:
                process_recvfrom(line)
        if line.startswith("sendto"):
            process_sendto(line)

    clean_output = {key: value for key, value in SENDTO_FLOW.items() if value}

    return clean_output


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Process APK files with ADB commands")
    parser.add_argument("-d", "--directory", help="Directory containing APK files")
    parser.add_argument("-f", "--file", help="Single APK file to process")
    args = parser.parse_args()

    if args.directory:
        process_directory(args.directory)
    elif args.file:
        main(args.file)

dupSearcher.sh

#!/bin/bash

if [ "$#" -eq 0 ]; then
    echo "Usage: $0 <log_directory_or_file>"
    exit 1
fi

input="$1"

if [ -d "$input" ]; then
    # If the input is a directory
    logfiles="$input"/*.log
elif [ -f "$input" ]; then
    # If the input is a single log file
    logfiles="$input"
else
    echo "Error: Invalid input. Please provide a directory or a single log file."
    exit 1
fi

for logfile in $logfiles; do
    output=$(grep dup -n "$logfile" | awk '{print $1}')
    startline=1
    echo "Working on $logfile\n\n"
    for line in $output; do
        IFS=':'
        read -r lno fd <<< "$line"
        fd=$(echo "$fd" | sed 's/.*(\([0-9]*\)).*/\1/')
        IFS=$' \t\n'
        syscall=$(sed -n "${startline},${lno}p" "$logfile" | grep "($fd)" | grep -v "close($fd)")
        echo "For $fd, in range $startline - $lno: $syscall"
        startline=$lno
    done
done

strace_apk_logs.py

import os
import subprocess
import glob
import time
import argparse


def install_apk(file_path):
    subprocess.run(["adb", "install", file_path])

    
def start_apk(package_name):
    subprocess.run(["adb", "shell", "monkey", "-p", package_name, "1"])


def find_pid(apk_name):
    pid_output = subprocess.check_output(
        ["adb", "shell", "ps", "|", "grep", apk_name]
    ).decode()
    pid_lines = pid_output.split("\n")
    for line in pid_lines:
        if apk_name in line:
            pid = line.split()[1]
            return pid


def connect_strace(pid, log_file):
    subprocess.Popen(
        [
            "adb",
            "shell",
            "strace",
            "-p",
            pid,
            "-o",
            os.path.join("/sdcard/Documents", log_file),
        ]
    )


def start_monkey(package_name):
    subprocess.Popen(
        [
            "adb",
            "shell",
            "monkey",
            "-p",
            package_name,
            "--pct-trackball",
            "0",
            "-v",
            "10000",
        ]
    )


def process_single_apk(file_path):
    install_apk(file_path)
    apk_name = os.path.basename(file_path)
    package_name = os.path.splitext(apk_name)[0]
    start_apk(package_name)
    time.sleep(3)
    apk_pid = find_pid(package_name)
    log_file = package_name + ".log"
    if apk_pid:
        connect_strace(apk_pid, log_file)
        start_monkey(package_name)
    else:
        print(f"PID for {apk_name} not found.")


def process_apk_directory(directory_path):
    apk_files = glob.glob(os.path.join(directory_path, "*.apk"))
    for apk_file in apk_files:
        process_single_apk(apk_file)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Process APK files with ADB commands")
    parser.add_argument("-d", "--directory", help="Directory containing APK files")
    parser.add_argument("-f", "--file", help="Single APK file to process")
    args = parser.parse_args()

    if args.directory:
        process_apk_directory(args.directory)
    elif args.file:
        process_single_apk(args.file)
    else:
        print("Please specify either a directory or a single APK file.")