#!/usr/bin/env python3
"""
scan-integrations.py — static integration / SDK detector for harvested Appmiral APKs.

For every <pkg>/base.apk under the harvest root this:
  * parses the manifest (androguard) for package, label, version, sdk levels,
    permissions and declared components;
  * byte-scans the concatenated DEX (class names + string pool) plus resources.arsc
    and native-lib filenames for a data-driven SIGNATURE catalogue;
  * detects (a) every integration Appmiral advertises on
    https://appmiral.com/features/integrations and (b) extra tracking SDKs that are
    NOT advertised (Firebase, AppsFlyer, AWS Kinesis Firehose egress, ...).

Outputs (next to --out):
  <out>.csv   app x integration 1/0 matrix (+ id columns)
  <out>.json  per-app {label, version, sdk, location perms, matched evidence}
  <out>.identify.csv  identification table (label, version, abis, location perms)

Detection is presence-only and may have false negatives under heavy R8/obfuscation;
each match records the literal pattern hit so a reader can audit it. Reliability of
each signature is tagged high/med/low. Usage:

  python scan-integrations.py --apks <harvest_root> --out data/integration_matrix_<date>
"""
from __future__ import annotations
import argparse, csv, io, json, os, sys, zipfile

# Silence androguard's very verbose loguru DEBUG stream before importing it.
try:
    from loguru import logger as _loguru_logger
    _loguru_logger.remove()
except Exception:
    pass
import logging
logging.getLogger("androguard").setLevel(logging.ERROR)

try:
    from androguard.core.apk import APK            # androguard 4.x
except Exception:                                   # pragma: no cover
    from androguard.core.bytecodes.apk import APK   # androguard 3.x

# ---------------------------------------------------------------------------
# SIGNATURE CATALOGUE
# (name, category, advertised, reliability, [byte patterns as ascii substrings])
# Patterns are matched against: concatenated classes*.dex + resources.arsc +
# native .so filenames + manifest component/permission names (one combined blob).
# DEX class paths use '/'; hosts use '.'.
# ---------------------------------------------------------------------------
SIGS = [
    # --- Content & Community (advertised) ---
    ("YouTube", "Content & Community", True, "med", ["com/google/android/youtube", "youtube.com", "youtubePlayer"]),
    ("Spotify", "Content & Community", True, "high", ["com/spotify", "api.spotify.com", "accounts.spotify.com"]),
    ("Instagram", "Content & Community", True, "med", ["com/instagram", "instagram.com", "graph.instagram.com"]),
    ("TikTok", "Content & Community", True, "high", ["com/tiktok", "com/bytedance", "tiktokapis.com", "tiktokv.com"]),
    ("Get Moments", "Content & Community", True, "low", ["getmoments", "get-moments", "getmoments.com"]),
    # --- SSO / Login (advertised) ---
    ("AWS Cognito", "SSO / Login", True, "high", ["cognito-idp", "cognito-identity", "amazoncognito", "/cognito"]),
    ("Azure / Microsoft", "SSO / Login", True, "high", ["com/microsoft/identity", "login.microsoftonline.com", "/msal"]),
    ("Sports Alliance", "SSO / Login", True, "low", ["sportsalliance", "sports-alliance"]),
    ("OpenID", "SSO / Login", True, "med", ["net/openid/appauth", "openid-configuration", "/.well-known/openid"]),
    # --- Ticketing (advertised, mostly server-side) ---
    ("CM.com", "Ticketing/Payments/CRM", True, "med", ["cm.com", "cmtelecom", "com/cmcom", "cm-telecom"]),
    ("ShowPass", "Ticketing", True, "med", ["showpass"]),
    ("Eventim", "Ticketing", True, "med", ["eventim"]),
    ("Ticketpay", "Ticketing", True, "low", ["ticketpay"]),
    ("Tickoweb", "Ticketing", True, "low", ["tickoweb"]),
    ("Ticketmaster", "Ticketing", True, "med", ["ticketmaster"]),
    ("Bleep", "Ticketing", True, "low", ["bleep.live", "/bleep"]),
    ("SeeTickets", "Ticketing", True, "med", ["seetickets", "see-tickets"]),
    ("Billetten", "Ticketing", True, "low", ["billetten"]),
    ("WeezEvents", "Ticketing/Cashless", True, "med", ["weezevent"]),
    ("GET (cashless)", "Cashless", True, "low", ["we-are-get", "getcashless"]),
    ("Mobile Locker", "Lockers", True, "low", ["mobilelocker", "mobile-locker"]),
    # --- CRM / CDP (advertised) ---
    ("Salesforce", "CRM / CDP", True, "high", ["com/salesforce/marketingcloud", "marketingcloudapis.com", "com/salesforce"]),
    ("Braze", "CRM / CDP", True, "high", ["com/braze", "com/appboy", "braze.com", "appboy"]),
    ("Insider", "CRM / CDP", True, "high", ["com/useinsider", "useinsider.com"]),
    ("Fanalist", "CRM / CDP", True, "low", ["fanalist"]),
    # --- Analytics & Location (advertised) — the core of this audit ---
    ("Crowd Connected", "Analytics & Location", True, "high", ["com/crowdconnected", "crowdconnected.com"]),
    ("Mapbox", "Analytics & Location", True, "high", ["com/mapbox", "api.mapbox.com", "events.mapbox.com", "libmapbox"]),
    ("Permutive", "Analytics & Location", True, "high", ["com/permutive", "permutive.com", "permutive.app"]),
    ("Tealium", "Analytics & Location", True, "high", ["com/tealium", "tiqcdn.com", "tealiumiq.com"]),
    ("PwC", "Analytics & Location", True, "low", ["/pwc/", "pwc.com"]),
    # --- AI / Chatbot (advertised) ---
    ("Halo by CM", "AI / Chatbot", True, "low", ["halo.cm.com", "cm.com/halo", "/halo"]),
    # --- Planning / Sponsorship (advertised) ---
    ("Beatswitch", "Planning", True, "low", ["beatswitch"]),
    ("Inventis", "Planning", True, "low", ["inventis"]),
    ("Jebbit", "Sponsorship", True, "low", ["jebbit"]),
    ("Komo", "Sponsorship", True, "low", ["komo.tech", "getkomo", "komoapp"]),
    # --- Extra tracking SDKs NOT on the marketing page ---
    ("AWS Kinesis Firehose (egress)", "Egress (not advertised)", False, "high", ["firehose", "PutRecordBatch", "/kinesis"]),
    ("Firebase Analytics", "Analytics (not advertised)", False, "high", ["com/google/firebase/analytics", "firebase-analytics"]),
    ("Firebase Messaging", "Push (not advertised)", False, "high", ["com/google/firebase/messaging", "firebase-messaging"]),
    ("Firebase Crashlytics", "Crash (not advertised)", False, "high", ["com/google/firebase/crashlytics", "crashlytics"]),
    ("Google Maps SDK", "Maps (not advertised)", False, "high", ["com/google/android/gms/maps", "maps.googleapis.com"]),
    ("Google Fused Location", "Location (not advertised)", False, "high", ["com/google/android/gms/location", "FusedLocation"]),
    ("Adjust", "Attribution (not advertised)", False, "high", ["com/adjust/sdk", "app.adjust.com"]),
    ("AppsFlyer", "Attribution (not advertised)", False, "high", ["com/appsflyer", "appsflyer.com"]),
    ("OneSignal", "Push (not advertised)", False, "high", ["com/onesignal", "onesignal.com"]),
    ("Airship", "Push (not advertised)", False, "high", ["com/urbanairship", "urbanairship.com"]),
    ("Segment", "CDP (not advertised)", False, "high", ["com/segment/analytics", "api.segment.io"]),
    ("Sentry", "Crash (not advertised)", False, "high", ["io/sentry", "sentry.io"]),
    ("OneTrust", "Consent (not advertised)", False, "high", ["com/onetrust", "cdn.cookielaw.org", "onetrust.com"]),
    ("Didomi", "Consent (not advertised)", False, "high", ["io/didomi", "didomi.io"]),
    ("Facebook SDK", "Social (not advertised)", False, "med", ["com/facebook/", "facebook.com/v"]),
    ("Branch", "Deeplink (not advertised)", False, "high", ["io/branch/referral", "branch.io", "app.link"]),
]

LOC_PERMS = {
    "android.permission.ACCESS_FINE_LOCATION": "fine",
    "android.permission.ACCESS_COARSE_LOCATION": "coarse",
    "android.permission.ACCESS_BACKGROUND_LOCATION": "background",
}


def read_blob(apk_path: str, extra_native: list[str]) -> tuple[bytes, list[str]]:
    """Return (scan_blob, abis). scan_blob = all dex + resources.arsc bytes."""
    parts: list[bytes] = []
    abis: set[str] = set()
    for p in [apk_path] + extra_native:
        if not os.path.exists(p):
            continue
        with zipfile.ZipFile(p) as z:
            for n in z.namelist():
                base = n.rsplit("/", 1)[-1]
                if n.startswith("classes") and n.endswith(".dex"):
                    parts.append(z.read(n))
                elif n == "resources.arsc":
                    parts.append(z.read(n))
                elif n.startswith("lib/") and n.endswith(".so"):
                    abi = n.split("/")[1]
                    abis.add(abi)
                    parts.append(b"\n" + base.encode("utf-8", "ignore") + b"\n")
    return b"".join(parts), sorted(abis)


def scan_app(pkg_dir: str):
    base = os.path.join(pkg_dir, "base.apk")
    if not os.path.exists(base):
        return None
    arm64 = os.path.join(pkg_dir, "split_config.arm64_v8a.apk")
    extra = [arm64] if os.path.exists(arm64) else []

    a = APK(base)
    pkg = a.get_package()
    try:
        label = a.get_app_name()
    except Exception:
        label = ""
    vname = a.get_androidversion_name() or ""
    vcode = a.get_androidversion_code() or ""
    minsdk = a.get_min_sdk_version() or ""
    tgtsdk = a.get_target_sdk_version() or ""
    perms = set(a.get_permissions())
    loc = sorted({tag for p, tag in LOC_PERMS.items() if p in perms})

    comp_names = "\n".join(
        list(a.get_activities()) + list(a.get_services())
        + list(a.get_receivers()) + list(a.get_providers())
    )
    blob, abis = read_blob(base, extra)
    blob = blob + b"\n" + comp_names.replace(".", "/").encode("utf-8", "ignore") \
                + b"\n" + "\n".join(sorted(perms)).encode("utf-8", "ignore")

    matched = {}
    for name, cat, advertised, rel, pats in SIGS:
        hits = [p for p in pats if p.encode("utf-8", "ignore") in blob]
        if hits:
            matched[name] = {"category": cat, "advertised": advertised,
                             "reliability": rel, "patterns": hits}
    return {
        "package": pkg, "label": label, "versionName": vname, "versionCode": str(vcode),
        "minSdk": str(minsdk), "targetSdk": str(tgtsdk), "abis": abis,
        "location_perms": loc, "matched": matched,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--apks", required=True, help="harvest root (<pkg>/base.apk dirs)")
    ap.add_argument("--out", required=True, help="output path prefix (no extension)")
    args = ap.parse_args()

    pkg_dirs = sorted(
        os.path.join(args.apks, d) for d in os.listdir(args.apks)
        if os.path.isdir(os.path.join(args.apks, d))
        and os.path.exists(os.path.join(args.apks, d, "base.apk"))
    )
    print(f"Scanning {len(pkg_dirs)} apps from {args.apks}")

    results = []
    for i, d in enumerate(pkg_dirs, 1):
        try:
            r = scan_app(d)
        except Exception as e:
            print(f"  [{i}/{len(pkg_dirs)}] ERROR {os.path.basename(d)}: {e}")
            continue
        if r:
            results.append(r)
            adv = sum(1 for v in r["matched"].values() if v["advertised"])
            print(f"  [{i}/{len(pkg_dirs)}] {r['package']:<45} loc={','.join(r['location_perms']) or '-':<22} integrations={len(r['matched'])} (advertised={adv})")

    cols = [s[0] for s in SIGS]
    os.makedirs(os.path.dirname(os.path.abspath(args.out)) or ".", exist_ok=True)
    with io.open(args.out + ".csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["package", "label", "versionName", "location_perms"] + cols)
        for r in sorted(results, key=lambda x: x["package"]):
            row = [r["package"], r["label"], r["versionName"], "|".join(r["location_perms"])]
            row += [("1" if c in r["matched"] else "0") for c in cols]
            w.writerow(row)

    with io.open(args.out + ".identify.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["package", "label", "versionName", "versionCode", "minSdk", "targetSdk", "abis",
                    "loc_fine", "loc_coarse", "loc_background"])
        for r in sorted(results, key=lambda x: x["package"]):
            lp = r["location_perms"]
            w.writerow([r["package"], r["label"], r["versionName"], r["versionCode"], r["minSdk"],
                        r["targetSdk"], "|".join(r["abis"]),
                        int("fine" in lp), int("coarse" in lp), int("background" in lp)])

    with io.open(args.out + ".json", "w", encoding="utf-8") as f:
        json.dump({"apps": results, "signatures": [
            {"name": s[0], "category": s[1], "advertised": s[2], "reliability": s[3]} for s in SIGS
        ]}, f, indent=2, ensure_ascii=False)

    print(f"\nWrote {args.out}.csv / .identify.csv / .json  ({len(results)} apps)")


if __name__ == "__main__":
    main()