#!/usr/bin/env python3
"""Agent for testing Android intents for vulnerabilities.

Uses ADB and Drozer to enumerate exported components, test
intent injection, content provider SQL injection, broadcast
receiver abuse, and pending intent hijacking vulnerabilities.
"""

import json
import subprocess
import re
import sys
from pathlib import Path
from datetime import datetime


class AndroidIntentTestAgent:
    """Tests Android app IPC through intents for security flaws."""

    def __init__(self, package_name, output_dir="./android_intent_test"):
        self.package = package_name
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.findings = []

    def _adb(self, args, timeout=15):
        cmd = ["adb", "shell"] + args
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
            return result.stdout.strip(), result.returncode
        except (FileNotFoundError, subprocess.TimeoutExpired):
            return "", -1

    def _drozer(self, module, args=""):
        cmd_str = f"run {module} -a {self.package} {args}".strip()
        cmd = ["drozer", "console", "connect", "-c", cmd_str]
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            return result.stdout.strip()
        except (FileNotFoundError, subprocess.TimeoutExpired):
            return ""

    def enumerate_attack_surface(self):
        """Enumerate exported components via Drozer."""
        output = self._drozer("app.package.attacksurface")
        surface = {"activities": 0, "services": 0, "receivers": 0, "providers": 0}

        for m in re.finditer(r"(\d+)\s+(activities|broadcast receivers|content providers|services)\s+exported", output):
            count = int(m.group(1))
            comp_type = m.group(2)
            if "activities" in comp_type:
                surface["activities"] = count
            elif "services" in comp_type:
                surface["services"] = count
            elif "receivers" in comp_type:
                surface["receivers"] = count
            elif "providers" in comp_type:
                surface["providers"] = count

        total = sum(surface.values())
        if total > 0:
            self.findings.append({
                "severity": "info",
                "type": "Attack Surface",
                "detail": f"{total} exported components found",
                "breakdown": surface,
            })
        return surface

    def list_exported_activities(self):
        """List exported activities."""
        output = self._drozer("app.activity.info")
        activities = []
        for m in re.finditer(r"([\w.]+/[\w.$]+)", output):
            activities.append(m.group(1))
        return activities

    def test_activity_access(self, activity_name):
        """Test if an exported activity is accessible without auth."""
        output, rc = self._adb(["am", "start", "-n", f"{self.package}/{activity_name}"])
        accessible = "Error" not in output and rc == 0
        if accessible:
            self.findings.append({
                "severity": "high",
                "type": "Exported Activity Access",
                "detail": f"Activity {activity_name} accessible without authentication",
            })
        return {"activity": activity_name, "accessible": accessible, "output": output[:200]}

    def test_content_provider_query(self, uri):
        """Test content provider for data leakage."""
        output = self._drozer("app.provider.query", uri)
        has_data = bool(output) and "No results" not in output and "error" not in output.lower()
        if has_data:
            self.findings.append({
                "severity": "high",
                "type": "Content Provider Data Leakage",
                "detail": f"Data accessible via {uri}",
            })
        return {"uri": uri, "has_data": has_data, "preview": output[:300]}

    def test_sql_injection(self, uri):
        """Test content provider for SQL injection."""
        output = self._drozer("scanner.provider.injection")
        injectable = "Injectable" in output or "injection" in output.lower()
        if injectable:
            self.findings.append({
                "severity": "critical",
                "type": "Content Provider SQL Injection",
                "detail": f"SQL injection possible in {self.package}",
            })
        return {"package": self.package, "injectable": injectable}

    def test_path_traversal(self):
        """Test content providers for path traversal."""
        output = self._drozer("scanner.provider.traversal")
        vulnerable = "Vulnerable" in output or "traversal" in output.lower()
        if vulnerable:
            self.findings.append({
                "severity": "critical",
                "type": "Content Provider Path Traversal",
                "detail": f"Path traversal in {self.package}",
            })
        return {"vulnerable": vulnerable}

    def send_broadcast(self, action, extras=None):
        """Send broadcast to test exported receivers."""
        cmd = ["am", "broadcast", "-a", action, "-p", self.package]
        if extras:
            for key, val in extras.items():
                cmd.extend(["--es", key, val])
        output, rc = self._adb(cmd)
        return {"action": action, "result": output[:200], "returncode": rc}

    def check_debuggable(self):
        """Check if app is debuggable."""
        output, _ = self._adb(["run-as", self.package, "id"])
        debuggable = "uid=" in output
        if debuggable:
            self.findings.append({
                "severity": "high",
                "type": "Debuggable Application",
                "detail": f"{self.package} is debuggable",
            })
        return debuggable

    def generate_report(self):
        surface = self.enumerate_attack_surface()
        activities = self.list_exported_activities()
        self.check_debuggable()
        self.test_sql_injection("")
        self.test_path_traversal()

        report = {
            "report_date": datetime.utcnow().isoformat(),
            "package": self.package,
            "attack_surface": surface,
            "exported_activities": activities,
            "findings": self.findings,
            "total_findings": len(self.findings),
        }
        out = self.output_dir / "android_intent_report.json"
        with open(out, "w") as f:
            json.dump(report, f, indent=2)
        print(json.dumps(report, indent=2))
        return report


def main():
    if len(sys.argv) < 2:
        print("Usage: agent.py <package_name>")
        sys.exit(1)
    agent = AndroidIntentTestAgent(sys.argv[1])
    agent.generate_report()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Android Intent Vulnerability Scanner

Parses AndroidManifest.xml to identify exported components and generate
Drozer/ADB test commands for IPC security assessment.

Usage:
    python process.py --manifest AndroidManifest.xml [--package com.target.app] [--output report.json]
"""

import argparse
import json
import sys
import xml.etree.ElementTree as ET
from datetime import datetime
from pathlib import Path


def parse_manifest(manifest_path: str) -> dict:
    """Parse AndroidManifest.xml for exported components."""
    tree = ET.parse(manifest_path)
    root = tree.getroot()
    ns = {"android": "http://schemas.android.com/apk/res/android"}

    package = root.get("package", "unknown")
    target_sdk = ""
    for sdk in root.findall(".//uses-sdk"):
        target_sdk = sdk.get(f"{{{ns['android']}}}targetSdkVersion", "unknown")

    components = {"activities": [], "services": [], "receivers": [], "providers": []}

    for comp_type, tag in [("activities", "activity"), ("services", "service"),
                            ("receivers", "receiver"), ("providers", "provider")]:
        for elem in root.findall(f".//{tag}"):
            name = elem.get(f"{{{ns['android']}}}name", "")
            exported = elem.get(f"{{{ns['android']}}}exported", "")
            permission = elem.get(f"{{{ns['android']}}}permission", "")
            has_intent_filter = len(elem.findall("intent-filter")) > 0

            # Determine effective export status
            if exported == "true":
                is_exported = True
            elif exported == "false":
                is_exported = False
            else:
                is_exported = has_intent_filter  # Default: exported if has intent-filter (pre API 31)

            if is_exported:
                component = {
                    "name": name,
                    "exported": True,
                    "permission": permission,
                    "has_intent_filter": has_intent_filter,
                    "protected": bool(permission),
                }

                # Get intent filter actions
                actions = []
                for intent_filter in elem.findall("intent-filter"):
                    for action in intent_filter.findall("action"):
                        actions.append(action.get(f"{{{ns['android']}}}name", ""))
                component["actions"] = actions

                # Provider-specific attributes
                if tag == "provider":
                    component["authorities"] = elem.get(f"{{{ns['android']}}}authorities", "")
                    component["read_permission"] = elem.get(f"{{{ns['android']}}}readPermission", "")
                    component["write_permission"] = elem.get(f"{{{ns['android']}}}writePermission", "")

                components[comp_type].append(component)

    return {"package": package, "target_sdk": target_sdk, "components": components}


def generate_test_commands(parsed: dict) -> list:
    """Generate Drozer and ADB test commands."""
    commands = []
    pkg = parsed["package"]

    for activity in parsed["components"]["activities"]:
        commands.append({
            "component": activity["name"],
            "type": "activity",
            "tool": "drozer",
            "command": f'run app.activity.start --component {pkg} {activity["name"]}',
            "risk": "HIGH" if not activity["protected"] else "LOW",
        })

    for receiver in parsed["components"]["receivers"]:
        for action in receiver.get("actions", []):
            commands.append({
                "component": receiver["name"],
                "type": "receiver",
                "tool": "adb",
                "command": f'adb shell am broadcast -a {action} -n {pkg}/{receiver["name"]}',
                "risk": "HIGH" if not receiver["protected"] else "LOW",
            })

    for provider in parsed["components"]["providers"]:
        auth = provider.get("authorities", "")
        if auth:
            commands.append({
                "component": provider["name"],
                "type": "provider_query",
                "tool": "drozer",
                "command": f'run app.provider.query content://{auth}/',
                "risk": "CRITICAL" if not provider.get("read_permission") else "MEDIUM",
            })
            commands.append({
                "component": provider["name"],
                "type": "provider_injection",
                "tool": "drozer",
                "command": f'run scanner.provider.injection -a {pkg}',
                "risk": "CRITICAL",
            })

    return commands


def assess_findings(parsed: dict) -> list:
    """Assess security of exported components."""
    findings = []
    components = parsed["components"]

    for comp_type, items in components.items():
        for item in items:
            if not item.get("protected"):
                findings.append({
                    "component": item["name"],
                    "type": comp_type,
                    "issue": f"Exported {comp_type[:-1]} without permission protection",
                    "severity": "HIGH" if comp_type in ("providers", "receivers") else "MEDIUM",
                    "owasp_mobile": "M8",
                    "cwe": "CWE-926",
                })

    # Check for sensitive-looking unprotected components
    sensitive_keywords = ["admin", "debug", "internal", "settings", "config", "payment", "auth"]
    for comp_type, items in components.items():
        for item in items:
            name_lower = item["name"].lower()
            if any(kw in name_lower for kw in sensitive_keywords) and not item.get("protected"):
                findings.append({
                    "component": item["name"],
                    "type": comp_type,
                    "issue": f"Sensitive component '{item['name']}' exported without protection",
                    "severity": "CRITICAL",
                    "owasp_mobile": "M8",
                    "cwe": "CWE-926",
                })

    return findings


def main():
    parser = argparse.ArgumentParser(description="Android Intent Vulnerability Scanner")
    parser.add_argument("--manifest", required=True, help="AndroidManifest.xml path")
    parser.add_argument("--output", default="intent_scan.json", help="Output report")
    args = parser.parse_args()

    if not Path(args.manifest).exists():
        print(f"[-] Not found: {args.manifest}")
        sys.exit(1)

    parsed = parse_manifest(args.manifest)
    commands = generate_test_commands(parsed)
    findings = assess_findings(parsed)

    total_exported = sum(len(v) for v in parsed["components"].values())

    report = {
        "scan": {"manifest": args.manifest, "package": parsed["package"],
                 "target_sdk": parsed["target_sdk"], "date": datetime.now().isoformat()},
        "attack_surface": {
            "total_exported": total_exported,
            "activities": len(parsed["components"]["activities"]),
            "services": len(parsed["components"]["services"]),
            "receivers": len(parsed["components"]["receivers"]),
            "providers": len(parsed["components"]["providers"]),
        },
        "findings": findings,
        "test_commands": commands,
    }

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)

    print(f"[+] Package: {parsed['package']}")
    print(f"[+] Exported components: {total_exported}")
    print(f"[+] Findings: {len(findings)}")
    print(f"[+] Test commands generated: {len(commands)}")
    print(f"[+] Report saved: {args.output}")


if __name__ == "__main__":
    main()