main.py

#!/usr/bin/env python3
"""
APK Forensics Tool — Modular static analysis pipeline.
Detects 30+ IOCs across permissions, C2 infrastructure, API abuse, and obfuscation.
Exports findings to structured JSON and HTML triage reports.
Integrates JADX-CLI decompilation for source-level review of obfuscated samples.

Usage: python main.py <path.apk> [--json] [--html] [--jadx /path/to/jadx] [--out DIR]
"""

import sys, os, re, zipfile, json, hashlib, subprocess, shutil, traceback, argparse
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Tuple, Optional
from datetime import datetime, timezone

# ── COLOUR HELPERS ───────────────────────────────────────────────────────────
RESET="\033[0m"; BOLD="\033[1m"; RED="\033[91m"; YELLOW="\033[93m"
GREEN="\033[92m"; CYAN="\033[96m"; GRAY="\033[90m"; MAGENTA="\033[95m"

def col(t, c): return f"{c}{t}{RESET}"
def bold(t): return col(t, BOLD)
def red(t): return col(t, RED)
def yellow(t): return col(t, YELLOW)
def green(t): return col(t, GREEN)
def cyan(t): return col(t, CYAN)
def gray(t): return col(t, GRAY)

# ══════════════════════════════════════════════════════════════════════════════
# IOC CATALOGUE (30+ indicators)
# ══════════════════════════════════════════════════════════════════════════════

# FR-3 Dangerous permissions (each = 1 IOC)
DANGEROUS_PERMISSIONS: Dict[str, int] = {
    "INTERNET": 1, "READ_SMS": 3, "RECEIVE_SMS": 3, "SEND_SMS": 3,
    "READ_CONTACTS": 2, "WRITE_CONTACTS": 2, "ACCESS_FINE_LOCATION": 3,
    "ACCESS_COARSE_LOCATION": 2, "RECORD_AUDIO": 4, "CAMERA": 3,
    "READ_CALL_LOG": 3, "WRITE_CALL_LOG": 3, "PROCESS_OUTGOING_CALLS": 3,
    "READ_PHONE_STATE": 2, "CALL_PHONE": 2, "READ_EXTERNAL_STORAGE": 2,
    "WRITE_EXTERNAL_STORAGE": 2, "RECEIVE_BOOT_COMPLETED": 2,
    "FOREGROUND_SERVICE": 1, "REQUEST_INSTALL_PACKAGES": 4,
    "BIND_DEVICE_ADMIN": 5, "SYSTEM_ALERT_WINDOW": 3,
    "PACKAGE_USAGE_STATS": 2, "ACCESS_WIFI_STATE": 1,
    "CHANGE_WIFI_STATE": 2, "DISABLE_KEYGUARD": 3,
    "QUERY_ALL_PACKAGES": 2, "HIDE_OVERLAY_WINDOWS": 2, "WAKE_LOCK": 1,
}

# FR-6 Suspicious APIs (each = 1 IOC)
SUSPICIOUS_APIS: Dict[str, Tuple[int, str]] = {
    "Ljava/lang/Runtime;->exec(": (5, "Process Execution"),
    "Ljava/lang/ProcessBuilder;": (5, "Process Execution"),
    "Ljava/lang/reflect/Method;->invoke(": (3, "Reflection / Code Injection"),
    "Ljava/lang/Class;->forName(": (2, "Dynamic Class Loading"),
    "Ljava/net/Socket;": (3, "Raw Socket"),
    "Ljava/net/ServerSocket;": (3, "Server Socket"),
    "Ljava/net/URL;->openConnection(": (2, "URL Connection"),
    "Ljava/net/HttpURLConnection;": (2, "HTTP Connection"),
    "Lokhttp3/": (2, "OkHttp"),
    "Lcom/squareup/okhttp/": (2, "OkHttp (legacy)"),
    "addJavascriptInterface": (4, "WebView JS Bridge"),
    "loadUrl": (2, "WebView loadUrl"),
    "evaluateJavascript": (3, "WebView JS Eval"),
    "DexClassLoader": (4, "Dynamic DEX Loading"),
    "PathClassLoader": (3, "Dynamic Class Loading"),
    "dalvik/system/DexFile": (4, "Direct DEX Access"),
    "Ljavax/crypto/Cipher;": (1, "Cryptographic Cipher"),
    "Landroid/app/admin/DevicePolicyManager;": (4, "Device Admin API"),
    "Landroid/content/pm/PackageInstaller;": (3, "Package Installer API"),
    "Landroid/accessibilityservice/": (3, "Accessibility Service"),
}

# FR-5 C2 keywords (each unique match = 1 IOC)
C2_KEYWORDS = [
    "command","shell","exec","payload","bot","socket","reverse","backdoor",
    "c2","cnc","c&c","admin","control","upload","exfil","steal","keylog",
    "intercept","inject","beacon","dropper","loader","implant","rat","trojan",
    "download","install","update","ping","heartbeat",
]

SUSPICIOUS_TLDS = {".pw",".tk",".ml",".ga",".cf",".gq",".xyz", ".top",".ru",".cn",".online",".site",".live"}
HARDCODED_IP_SCORE = 3
EXPORTED_COMP_SCORE = 2
SUSPICIOUS_DOM_SCORE = 2
SUSPICIOUS_KW_SCORE = 1

RISK_LEVELS = [(9,"HIGH",RED),(4,"MEDIUM",YELLOW),(0,"LOW",GREEN)]

_IP_RE = re.compile(r'\b(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b')
_URL_RE = re.compile(r'https?://[^\s\'"<>]{4,}', re.IGNORECASE)
_DOM_RE = re.compile(r'\b(?:[a-z0-9-]+\.)+(?:com|net|org|io|co|ru|cn|info|biz|xyz|top|pw|tk|ml|ga|cf|gq|online|site|club|app|dev|sh|live)\b', re.IGNORECASE)
_PRIV_IP = re.compile(r'^(?:10\.\d+.\d+.\d+|172\.(?:1[6-9]|2\d|3[01])\.\d+.\d+|192\.168\.\d+.\d+|127\.\d+.\d+.\d+|0\.0\.0\.0)$')

# ══════════════════════════════════════════════════════════════════════════════
# DATA MODEL
# ══════════════════════════════════════════════════════════════════════════════

@dataclass
class AnalysisResult:
    apk_path: str = ""
    package_name: str = "unknown"
    md5: str = ""
    sha1: str = ""
    sha256: str = ""
    file_size_bytes: int = 0
    timestamp: str = ""
    activities: List[str] = field(default_factory=list)
    services: List[str] = field(default_factory=list)
    receivers: List[str] = field(default_factory=list)
    providers: List[str] = field(default_factory=list)
    exported_components: List[str] = field(default_factory=list)
    permissions: List[str] = field(default_factory=list)
    urls: List[str] = field(default_factory=list)
    ips: List[str] = field(default_factory=list)
    domains: List[str] = field(default_factory=list)
    suspicious_apis: List[Dict] = field(default_factory=list)
    c2_keywords: List[str] = field(default_factory=list)
    ioc_count: int = 0
    risk_score: int = 0
    risk_level: str = "LOW"
    risk_breakdown: List[Dict] = field(default_factory=list)
    jadx_findings: List[str] = field(default_factory=list)
    jadx_used: bool = False
    analysis_method: str = "strings"
    errors: List[str] = field(default_factory=list)

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 1 — FILE VALIDATION & HASHING
# ══════════════════════════════════════════════════════════════════════════════

def validate_apk(path: str) -> bool:
    """Checks if the path exists and is a valid zip archive hosting an Android manifest."""
    if not os.path.isfile(path): return False
    if not zipfile.is_zipfile(path): return False
    try:
        with zipfile.ZipFile(path) as z:
            names = z.namelist()
            return "AndroidManifest.xml" in names or any(n.startswith("classes") for n in names)
    except Exception:
        return False

def hash_file(path: str) -> Tuple[str,str,str]:
    """Generates MD5, SHA1, and SHA256 hashes for cryptographic identity verification."""
    md5 = hashlib.md5()
    sha1 = hashlib.sha1()
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            md5.update(chunk)
            sha1.update(chunk)
            sha256.update(chunk)
    return md5.hexdigest(), sha1.hexdigest(), sha256.hexdigest()

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 2 — ANDROGUARD PARSING
# ══════════════════════════════════════════════════════════════════════════════

def parse_with_androguard(apk_path: str, result: AnalysisResult):
    """Parses structural properties, structural architecture components, and extracts permissions."""
    try:
        from androguard.misc import AnalyzeAPK
        a, d_list, dx = AnalyzeAPK(apk_path)
        result.analysis_method = "androguard"
        result.package_name = a.get_package() or "unknown"
        result.activities = [str(x) for x in a.get_activities()]
        result.services = [str(x) for x in a.get_services()]
        result.receivers = [str(x) for x in a.get_receivers()]
        result.providers = [str(x) for x in a.get_providers()]
        result.permissions = [p.replace("android.permission.", "") for p in a.get_permissions()]
        
        # Parse exposed structure components dynamically
        for tag in ['activity', 'service', 'receiver', 'provider']:
            for item in a.get_android_manifest_xml().findall(f'.//{tag}'):
                name = item.get('{http://schemas.android.com/apk/res/android}name')
                exported = item.get('{http://schemas.android.com/apk/res/android}exported')
                if exported == "true" and name:
                    result.exported_components.append(f"{tag.upper()}: {name}")
                    
        scan_apis_androguard(dx, result)
    except ImportError:
        result.errors.append("Androguard not installed inside virtual environment — falling back to string mode.")
    except Exception as e:
        result.errors.append(f"Androguard structural extraction failure: {str(e)}")

def scan_apis_androguard(dx, result: AnalysisResult):
    """Utilizes Cross-References inside Dalvik analysis objects to map out explicit API targets."""
    seen = set()
    for pattern, (score, category) in SUSPICIOUS_APIS.items():
        try:
            for cls in dx.get_classes():
                if pattern in cls.name and (pattern, category) not in seen:
                    seen.add((pattern, category))
                    result.suspicious_apis.append({"api": pattern.strip(";"), "category": category, "score": score})
                    _add_score(result, f"API: {category}", score)
                    break
            for method in dx.find_methods(pattern):
                if (pattern, category) not in seen:
                    seen.add((pattern, category))
                    m = method.get_method()
                    label = f"{m.get_class_name()}->{m.get_name()}"
                    result.suspicious_apis.append({"api": label, "category": category, "score": score})
                    _add_score(result, f"API: {category}", score)
        except Exception:
            pass

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 3 — JADX-CLI DECOMPILATION (Source-Level Scanner)
# ══════════════════════════════════════════════════════════════════════════════

def run_jadx(apk_path: str, result: AnalysisResult, jadx_bin: str = "jadx"):
    """Invokes JADX command line suite to decompile and scan target application source blocks."""
    jadx_path = shutil.which(jadx_bin) or (jadx_bin if os.path.isfile(jadx_bin) else None)
    if not jadx_path:
        result.errors.append(f"JADX executable missing at context target: {jadx_bin}")
        return
    
    result.jadx_used = True
    temp_out = f"jadx_out_{result.md5}"
    try:
        cmd = [jadx_path, "-d", temp_out, "--no-res", apk_path]
        subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120, check=True)
        
        # Shallow evaluation on directory structure output to scan text anomalies
        for root, _, files in os.walk(temp_out):
            for file in files:
                if file.endswith(".java"):
                    with open(os.path.join(root, file), 'r', errors='ignore') as f:
                        content = f.read()
                        if "loadLibrary" in content:
                            result.jadx_findings.append(f"Native library invocation located inside java definitions: {file}")
    except subprocess.TimeoutExpired:
        result.errors.append("JADX processing exceeded maximum timeline window (Timeout).")
    except Exception as e:
        result.errors.append(f"JADX engine run error: {str(e)}")
    finally:
        if os.path.isdir(temp_out):
            shutil.rmtree(temp_out)

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 4 — STRING / NETWORK SCANNING
# ══════════════════════════════════════════════════════════════════════════════

def extract_strings_from_zip(apk_path: str) -> List[str]:
    """Fallback mechanism extracting asset string tables directly out of byte streams."""
    strings: List[str] = []
    try:
        with zipfile.ZipFile(apk_path) as z:
            for name in z.namelist():
                try:
                    data = z.read(name)
                    found = re.findall(rb'[\x20-\x7e]{6,}', data)
                    strings.extend(s.decode('ascii', errors='ignore') for s in found)
                except Exception:
                    pass
    except Exception:
        pass
    return strings

def scan_strings(strings: List[str], result: AnalysisResult):
    """RegEx processor targeting URLs, host targets, explicit foreign loops, and tracking keywords."""
    url_set, ip_set, domain_set, kw_set = set(), set(), set(), set()
    combined_context = []
    
    for s in strings:
        combined_context.append(s)
        for url in _URL_RE.findall(s):
            url_set.add(url[:120])
        for ip in _IP_RE.findall(s):
            if not _PRIV_IP.match(ip):
                ip_set.add(ip)
        for dom in _DOM_RE.findall(s):
            domain_set.add(dom.lower())
        
        sl = s.lower()
        for kw in C2_KEYWORDS:
            if re.search(r'\b' + kw + r'\b', sl):
                kw_set.add(kw)
                
    result.urls = list(url_set)
    result.ips = list(ip_set)
    result.domains = list(domain_set)
    result.c2_keywords = list(kw_set)
    
    # Calculate scores derived via flat string matches
    if result.ips:
        _add_score(result, f"Hardcoded IP Infrastructure detected ({len(result.ips)})", HARDCODED_IP_SCORE)
    if any(any(d.endswith(tld) for tld in SUSPICIOUS_TLDS) for d in result.domains):
        _add_score(result, "Suspicious Infrastructure TLD matched", SUSPICIOUS_DOM_SCORE)
    if len(result.c2_keywords) >= 3:
        _add_score(result, f"High volume C2 keywords discovered ({len(result.c2_keywords)})", SUSPICIOUS_KW_SCORE * 3)

def scan_apis_strings(strings: List[str], result: AnalysisResult):
    """Processes flat code tokens for structural API targets when a full framework analysis isn't viable."""
    seen = set()
    combined = "\n".join(strings)
    for pattern, (score, category) in SUSPICIOUS_APIS.items():
        short = pattern.strip(";->").split("/")[-1].strip(";->") or pattern
        if (pattern in combined or short in combined) and category not in seen:
            seen.add(category)
            result.suspicious_apis.append({"api": short, "category": category, "score": score})
            _add_score(result, f"API: {category}", score)

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 5 — SCORING & COGNITION ENGINE
# ══════════════════════════════════════════════════════════════════════════════

def _add_score(result: AnalysisResult, label: str, pts: int):
    """Enforces boundaries avoiding double counting the exact category metrics twice."""
    if not any(r["finding"] == label for r in result.risk_breakdown):
        result.risk_score += pts
        result.risk_breakdown.append({"finding": label, "score": pts})

def score_permissions(result: AnalysisResult):
    """Maps configured weight metrics to targeted permissions flags."""
    for perm in result.permissions:
        short = perm.split(".")[-1]
        score = DANGEROUS_PERMISSIONS.get(short, 0)
        if score > 0:
            _add_score(result, f"Permission: {short}", score)

def score_exported(result: AnalysisResult):
    """Weights interface points accessible externally by alternate application suites."""
    if result.exported_components:
        total = min(len(result.exported_components) * EXPORTED_COMP_SCORE, 8)
        _add_score(result, f"Exported components ({len(result.exported_components)})", total)

def finalise_risk(result: AnalysisResult):
    """Applies metric classification windows to normalize numerical outputs into simple profiles."""
    for threshold, label, _ in RISK_LEVELS:
        if result.risk_score >= threshold:
            result.risk_level = label
            break
            
    result.ioc_count = (
        sum(1 for p in result.permissions if DANGEROUS_PERMISSIONS.get(p.split(".")[-1], 0) > 0) +
        len(result.ips) +
        len(result.suspicious_apis) +
        len(result.c2_keywords) +
        sum(1 for d in result.domains if any(d.endswith(t) for t in SUSPICIOUS_TLDS)) +
        len(result.exported_components) +
        len(result.jadx_findings)
    )

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 6 & 7 — REPORTS GENERATION
# ══════════════════════════════════════════════════════════════════════════════

def export_json(result: AnalysisResult, out_path: str):
    """Writes standardized execution tracking instances cleanly onto a filesystem structure."""
    with open(out_path, "w") as f:
        json.dump(asdict(result), f, indent=2)
    print(green(f"[+] JSON forensic report built completely → {out_path}"))

def export_html(result: AnalysisResult, out_path: str):
    """Builds an isolated HTML document capturing analytical output records."""
    html_template = """<!DOCTYPE html>
<html>
<head>
    <title>Triage Report: {pkg}</title>
    <style>
        body {{ font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Helvetica, Arial, sans-serif; margin: 30px; background: #f9f9f9; color: #333; }}
        .card {{ background: white; padding: 25px; border-radius: 8px; box-shadow: 0 4px 6px rgba(0,0,0,0.05); margin-bottom: 20px; }}
        h1, h2 {{ color: #111; margin-top: 0; }}
        .badge {{ padding: 6px 12px; border-radius: 20px; font-weight: bold; font-size: 14px; display: inline-block; }}
        .HIGH {{ background: #ffd1d1; color: #bc0000; }}
        .MEDIUM {{ background: #ffeaa7; color: #d63031; }}
        .LOW {{ background: #d4edda; color: #155724; }}
        table {{ width: 100%; border-collapse: collapse; margin-top: 15px; }}
        th, td {{ padding: 12px; text-align: left; border-bottom: 1px solid #eee; }}
        th {{ background: #f4f5f7; font-weight: 600; }}
    </style>
</head>
<body>
    <div class="card">
        <h1>APK Forensic Analysis Summary</h1>
        <p><strong>Package Target:</strong> {pkg}</p>
        <p><strong>Security Profile Flag:</strong> <span class="badge {level}">{level}</span></p>
        <p><strong>Calculated Score Assessment:</strong> {score} Points</p>
        <p><strong>Identified Indicators Count:</strong> {ioc}</p>
    </div>
    <div class="card">
        <h2>Breakdown Matrix Matrix</h2>
        <table>
            <tr><th>Vector / Context Rule Triggered</th><th>Score Contribution</th></tr>
            {matrix_rows}
        </table>
    </div>
</body>
</html>"""
    
    rows = ""
    for r in result.risk_breakdown:
        rows += f"<tr><td>{r['finding']}</td><td>+{r['score']}</td></tr>"
        
    compiled = html_template.format(
        pkg=result.package_name,
        level=result.risk_level,
        score=result.risk_score,
        ioc=result.ioc_count,
        matrix_rows=rows if rows else "<tr><td colspan='2'>No structural indicators scored.</td></tr>"
    )
    with open(out_path, "w") as f:
        f.write(compiled)
    print(green(f"[+] Static triage UI framework rendered successfully → {out_path}"))

# ══════════════════════════════════════════════════════════════════════════════
# MODULE 8 & 9 — ENGINE TERMINAL LAYOUT AND ENGINE PIPELINE ORCHESTRATION
# ══════════════════════════════════════════════════════════════════════════════

def render_terminal(result: AnalysisResult):
    """Prints a styled summary of findings directly to the terminal."""
    W = 66
    hr = lambda c="─": gray(c * W)
    sec = lambda t: f"\n{bold(cyan('▸ ' + t))}\n{hr()}"
    
    lv = result.risk_level
    lvc = RED if lv == "HIGH" else (YELLOW if lv == "MEDIUM" else GREEN)
    
    print(sec("Analysis Summary Output"))
    print(f"Target App Identity  : {bold(result.package_name)}")
    print(f"Cryptographic MD5    : {gray(result.md5)}")
    print(f"Execution Model Mode : {bold(result.analysis_method.upper())}")
    print(f"Risk Class Context   : {col(lv, lvc)} ({result.risk_score} points)")
    print(f"Total Confirmed IOCs : {bold(str(result.ioc_count))}")
    print(hr())

def analyse(apk_path: str, jadx_bin: Optional[str] = None) -> AnalysisResult:
    """Core analysis orchestrator tying all modules together."""
    result = AnalysisResult(
        apk_path=apk_path,
        timestamp=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        file_size_bytes=os.path.getsize(apk_path),
    )
    
    # 1. Capture Identity Signature
    result.md5, result.sha1, result.sha256 = hash_file(apk_path)
    
    # 2. Extract Structure components via Androguard 
    parse_with_androguard(apk_path, result)
    
    # 3. Pull file strings data blocks
    strings = extract_strings_from_zip(apk_path)
    scan_strings(strings, result)
    
    # Fallback to string scanning for APIs if Androguard wasn't available
    if result.analysis_method == "strings":
        scan_apis_strings(strings, result)
        
    # 4. If requested, parse decompilation 
    if jadx_bin:
        run_jadx(apk_path, result, jadx_bin)
        
    # 5. Core logic calculations evaluation closure
    score_permissions(result)
    score_exported(result)
    finalise_risk(result)
    
    return result

def main():
    parser = argparse.ArgumentParser(description="APK Static Forensics Tool Engine")
    parser.add_argument("apk", help="Path to target package archive asset")
    parser.add_argument("--json", action="store_true", help="Build out analytical raw JSON dataset object")
    parser.add_argument("--html", action="store_true", help="Compile human readable HTML profile summary documentation sheet")
    parser.add_argument("--jadx", metavar="PATH", help="Explicit configuration link location referencing compilation assets directory structure")
    parser.add_argument("--out", metavar="DIR", default=".", help="Target tracking location directory tree")
    
    args = parser.parse_args()
    
    if not validate_apk(args.apk):
        print(red(f"[-] Evaluation context target does not present structural properties expected of clear APK contents: {args.apk}"))
        sys.exit(1)
        
    print(cyan(f"[*] Beginning active binary inspection processing context: {args.apk}"))
    
    try:
        analysis_data_model = analyse(args.apk, args.jadx)
        render_terminal(analysis_data_model)
        
        if args.json:
            export_json(analysis_data_model, os.path.join(args.out, f"forensic_report_{analysis_data_model.md5}.json"))
        if args.html:
            export_html(analysis_data_model, os.path.join(args.out, f"forensic_report_{analysis_data_model.md5}.html"))
            
        if analysis_data_model.errors:
            print(yellow(f"\n[!] Complete Execution tracking reported context processing warnings ({len(analysis_data_model.errors)}):"))
            for err in analysis_data_model.errors:
                print(gray(f"  - {err}"))
                
    except Exception as fatal_err:
        print(red(f"[-] Forensic scanning pipeline broken by critical failure scenario: {str(fatal_err)}"))
        traceback.print_exc()
        sys.exit(2)

if __name__ == "__main__":
    main()