Overview
CVE-2026-4231 describes a server-side request forgery (SSRF) vulnerability in vanna-ai's Flask-based Endpoint component, specifically in the update_sql/run_sql path. The weakness allows an attacker to coerce the server into issuing HTTP requests to arbitrary URLs, including internal resources, by manipulating user-provided input. Because the exploit was publicly disclosed and can be triggered remotely, affected deployments risk data leakage, internal resource access, or other collateral damage associated with SSRF. The vulnerability aligns with CWE-918, since it enables unauthorized requests initiated by the server as part of processing user data. In real Flask apps this class of flaw often arises when input is used directly to fetch or run content from external URLs without proper validation. The CVE notes the issue affects vanna up to version 2.0.2, underscoring the risk in older Flask integration code paths when handling dynamic SQL or content retrieved from user-supplied URLs.
Affected Versions
vanna up to 2.0.2 (≤ 2.0.2)
Code Fix Example
Flask API Security Remediation
Vulnerable pattern (Flask):
from flask import Flask, request
import requests
import psycopg2
app = Flask(__name__)
@app.route('/Endpoint/update_sql', methods=['POST'])
def update_sql():
sql_url = request.form['sql_url'] # attacker-controlled URL
sql = requests.get(sql_url, timeout=5).text # SSRF + dynamic SQL
conn = psycopg2.connect(dsn='dbname=app user=app')
cur = conn.cursor()
cur.execute(sql) # executing remote SQL directly
conn.commit()
cur.close()
conn.close()
return 'OK'
FIXED pattern (Flask):
from flask import Flask, request, abort
import requests
import urllib.parse
import psycopg2
ALLOWED_DOMAINS = {'trusted.example.com', 'internal.example.local'}
STATEMENTS = {
'get_active_users': 'SELECT id, username FROM users WHERE active = TRUE',
'count_logs': 'SELECT COUNT(*) FROM logs'
}
def is_allowed_url(url: str) -> bool:
try:
p = urllib.parse.urlparse(url)
except Exception:
return False
if p.scheme not in ('http', 'https') or not p.hostname:
return False
host = p.hostname
if host in ALLOWED_DOMAINS or any(host.endswith('.' + d) for d in ALLOWED_DOMAINS):
return True
return False
app = Flask(__name__)
@app.route('/Endpoint/update_sql', methods=['POST'])
def update_sql():
sql_url = request.form.get('sql_url')
# Enforce SSRF controls: only allow known domains
if not sql_url or not is_allowed_url(sql_url):
abort(403)
resp = requests.get(sql_url, timeout=5, allow_redirects=False)
if resp.status_code != 200:
abort(502)
# Do not execute remote SQL directly; require a predefined statement key
stmt_key = request.form.get('stmt')
sql = STATEMENTS.get(stmt_key)
if not sql:
abort(400)
conn = psycopg2.connect(dsn='dbname=app user=app')
cur = conn.cursor()
cur.execute(sql)
conn.commit()
cur.close()
conn.close()
return 'OK'