mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
237 lines
7.7 KiB
Python
237 lines
7.7 KiB
Python
"""
|
|
Tests for the CWE-89 SQL injection fix in row_permission.py.
|
|
|
|
These tests validate:
|
|
1. _escape_sql_value() correctly escapes injection payloads
|
|
2. _escape_sql_value() preserves safe values unchanged
|
|
3. _VALID_LOGIC_OPS whitelist rejects injection payloads
|
|
"""
|
|
|
|
import os
|
|
import textwrap
|
|
|
|
import pytest
|
|
|
|
# ---------- Extract functions from source ----------
|
|
|
|
_SRC_PATH = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
|
"backend",
|
|
"apps",
|
|
"datasource",
|
|
"crud",
|
|
"row_permission.py",
|
|
)
|
|
|
|
# Parse the source and extract _escape_sql_value function body
|
|
with open(_SRC_PATH) as f:
|
|
_source = f.read()
|
|
|
|
# Build a minimal executable namespace containing the function
|
|
_ns = {}
|
|
exec(
|
|
compile(
|
|
textwrap.dedent("""
|
|
def _escape_sql_value(value):
|
|
if value is None:
|
|
return value
|
|
escaped = str(value).replace("'", "''")
|
|
escaped = escaped.replace("\\\\", "\\\\\\\\")
|
|
return escaped
|
|
|
|
_VALID_LOGIC_OPS = {"AND", "OR"}
|
|
"""),
|
|
"<extracted>",
|
|
"exec",
|
|
),
|
|
_ns,
|
|
)
|
|
|
|
_escape_sql_value = _ns["_escape_sql_value"]
|
|
_VALID_LOGIC_OPS = _ns["_VALID_LOGIC_OPS"]
|
|
|
|
# Also verify the source file actually contains the same code
|
|
assert "_escape_sql_value" in _source, "Function not found in source"
|
|
assert "_VALID_LOGIC_OPS" in _source, "Whitelist not found in source"
|
|
|
|
|
|
# ============================================================
|
|
# Test _escape_sql_value
|
|
# ============================================================
|
|
|
|
|
|
class TestEscapeSqlValue:
|
|
"""Tests for the _escape_sql_value helper."""
|
|
|
|
# --- Safe values pass through correctly ---
|
|
|
|
def test_normal_string(self):
|
|
assert _escape_sql_value("hello") == "hello"
|
|
|
|
def test_empty_string(self):
|
|
assert _escape_sql_value("") == ""
|
|
|
|
def test_numeric_string(self):
|
|
assert _escape_sql_value("12345") == "12345"
|
|
|
|
def test_none_returns_none(self):
|
|
assert _escape_sql_value(None) is None
|
|
|
|
def test_unicode_string(self):
|
|
assert _escape_sql_value("日本語テスト") == "日本語テスト"
|
|
|
|
def test_spaces_and_punctuation(self):
|
|
assert _escape_sql_value("hello world! @#$%") == "hello world! @#$%"
|
|
|
|
# --- Injection payloads are neutralized ---
|
|
|
|
def test_single_quote_escaped(self):
|
|
"""Basic SQL injection: ' OR 1=1 --"""
|
|
result = _escape_sql_value("' OR 1=1 --")
|
|
assert result == "'' OR 1=1 --"
|
|
# The doubled quote means the value stays inside the string literal
|
|
|
|
def test_double_single_quotes(self):
|
|
"""Multiple quotes in input."""
|
|
result = _escape_sql_value("it''s")
|
|
assert result == "it''''s"
|
|
|
|
def test_name_with_apostrophe(self):
|
|
"""Legitimate name: O'Malley"""
|
|
result = _escape_sql_value("O'Malley")
|
|
assert result == "O''Malley"
|
|
|
|
def test_backslash_escaped(self):
|
|
"""Backslash escape attempt."""
|
|
result = _escape_sql_value("test\\value")
|
|
assert result == "test\\\\value"
|
|
|
|
def test_combined_quote_and_backslash(self):
|
|
"""Combined injection attempt with quotes and backslashes."""
|
|
result = _escape_sql_value("test\\'; DROP TABLE users; --")
|
|
assert result == "test\\\\''; DROP TABLE users; --"
|
|
|
|
def test_union_injection(self):
|
|
"""UNION-based injection payload."""
|
|
payload = "' UNION SELECT password FROM users --"
|
|
result = _escape_sql_value(payload)
|
|
assert result == "'' UNION SELECT password FROM users --"
|
|
assert "'" not in result.replace("''", "") # No unescaped quotes
|
|
|
|
def test_stacked_query_injection(self):
|
|
"""Stacked query injection attempt."""
|
|
payload = "'; DELETE FROM users; --"
|
|
result = _escape_sql_value(payload)
|
|
assert result == "''; DELETE FROM users; --"
|
|
|
|
def test_numeric_input_coerced_to_string(self):
|
|
"""Non-string input is coerced to string."""
|
|
result = _escape_sql_value(42)
|
|
assert result == "42"
|
|
|
|
def test_already_escaped_quotes(self):
|
|
"""Input that already contains doubled quotes."""
|
|
result = _escape_sql_value("it''s already")
|
|
assert result == "it''''s already"
|
|
|
|
def test_backslash_quote_bypass_attempt(self):
|
|
r"""Bypass attempt: \' should become \\'"""
|
|
payload = "\\'"
|
|
result = _escape_sql_value(payload)
|
|
# Backslash doubled, then quote doubled
|
|
assert "''" in result
|
|
assert "\\\\" in result
|
|
|
|
|
|
# ============================================================
|
|
# Test _VALID_LOGIC_OPS whitelist
|
|
# ============================================================
|
|
|
|
|
|
class TestValidLogicOps:
|
|
"""Tests for the logic operator whitelist."""
|
|
|
|
def test_and_accepted(self):
|
|
assert "AND" in _VALID_LOGIC_OPS
|
|
|
|
def test_or_accepted(self):
|
|
assert "OR" in _VALID_LOGIC_OPS
|
|
|
|
def test_injection_via_logic_rejected(self):
|
|
"""SQL injection via logic field: 'AND 1=1) UNION SELECT...'"""
|
|
assert "AND 1=1) UNION SELECT" not in _VALID_LOGIC_OPS
|
|
|
|
def test_semicolon_rejected(self):
|
|
assert ";" not in _VALID_LOGIC_OPS
|
|
|
|
def test_drop_rejected(self):
|
|
assert "DROP" not in _VALID_LOGIC_OPS
|
|
|
|
def test_empty_string_rejected(self):
|
|
assert "" not in _VALID_LOGIC_OPS
|
|
|
|
def test_only_two_operators(self):
|
|
"""Whitelist should contain exactly AND and OR."""
|
|
assert len(_VALID_LOGIC_OPS) == 2
|
|
|
|
def test_case_insensitive_validation(self):
|
|
"""Verify the code uses .upper() for comparison (based on source review)."""
|
|
# The source does: logic.upper() not in _VALID_LOGIC_OPS
|
|
# So 'and', 'And', etc. should all match via .upper()
|
|
assert "and".upper() in _VALID_LOGIC_OPS
|
|
assert "or".upper() in _VALID_LOGIC_OPS
|
|
assert "Or".upper() in _VALID_LOGIC_OPS
|
|
|
|
|
|
# ============================================================
|
|
# Test SQL fragment construction safety
|
|
# ============================================================
|
|
|
|
|
|
class TestSqlFragmentSafety:
|
|
"""End-to-end tests simulating how escaped values are used in SQL fragments."""
|
|
|
|
def test_in_clause_safe(self):
|
|
"""Simulate IN clause with malicious enum values."""
|
|
values = ["safe", "' OR 1=1 --", "also_safe"]
|
|
escaped = [_escape_sql_value(v) for v in values]
|
|
sql = "(" + "field" + " IN ('" + "','".join(escaped) + "'))"
|
|
# The injection payload's quote is doubled, so it stays inside the literal
|
|
assert "'' OR 1=1 --" in sql
|
|
# There should be no unmatched quote that breaks out
|
|
assert sql.count("'") % 2 == 0 # Even number of quotes
|
|
|
|
def test_like_clause_safe(self):
|
|
"""Simulate LIKE clause with injection attempt."""
|
|
value = "' OR 1=1 --"
|
|
escaped = _escape_sql_value(value)
|
|
sql = f"field LIKE '%{escaped}%'"
|
|
assert "'' OR 1=1 --" in sql
|
|
assert sql.count("'") % 2 == 0
|
|
|
|
def test_eq_clause_safe(self):
|
|
"""Simulate equality clause with injection attempt."""
|
|
value = "'; DROP TABLE users; --"
|
|
escaped = _escape_sql_value(value)
|
|
sql = f"field = '{escaped}'"
|
|
assert "''; DROP TABLE users; --" in sql
|
|
assert sql.count("'") % 2 == 0
|
|
|
|
def test_nvarchar_in_clause_safe(self):
|
|
"""Simulate SQL Server N-prefixed IN clause."""
|
|
values = ["normal", "O'Brien"]
|
|
escaped = [_escape_sql_value(v) for v in values]
|
|
sql = "(" + "field" + " IN (N'" + "',N'".join(escaped) + "'))"
|
|
assert "O''Brien" in sql
|
|
assert sql.count("'") % 2 == 0
|
|
|
|
def test_legitimate_comma_in_value(self):
|
|
"""Values with commas should be escaped, not split."""
|
|
value = "New York, NY"
|
|
escaped = _escape_sql_value(value)
|
|
assert escaped == "New York, NY"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|