cratedb_mcp/settings.py (1)

7-8: Well-documented configuration parameter with sensible default.

The implementation correctly sets a secure default (only SELECT statements permitted) with the ability to override via environment variable.

However, consider adding a warning log message when PERMIT_ALL_STATEMENTS is enabled, as this could pose a security risk if accidentally enabled in production.

README.md (2)

43-49: Clear documentation of the read-only access policy.

The documentation effectively communicates the default behavior and override option to users.

Consider adjusting line 47 for better clarity:

-All other operations will raise a `ValueError` exception, unless the
+All other operations will raise a `ValueError` exception, unless
 `CRATEDB_MCP_PERMIT_ALL_STATEMENTS` environment variable will be set
 to a truthy value.

Additionally, it would be valuable to add a security warning about the risks of enabling all statements in production environments.

🧰 Tools

🪛 LanguageTool

[uncategorized] ~47-~47: This verb may not be in the correct tense. Consider changing the tense to fit the context better.
Context: ...IT_ALL_STATEMENTS` environment variable will be set to a truthy value. # Install ```sh...

(AI_EN_LECTOR_REPLACEMENT_VERB_TENSE)

---

47-47: Minor grammar issue in environment variable description.

The phrase "will be set" suggests a future action but should describe a current state.

-`CRATEDB_MCP_PERMIT_ALL_STATEMENTS` environment variable will be set
+`CRATEDB_MCP_PERMIT_ALL_STATEMENTS` environment variable is set

🧰 Tools

🪛 LanguageTool

[uncategorized] ~47-~47: This verb may not be in the correct tense. Consider changing the tense to fit the context better.
Context: ...IT_ALL_STATEMENTS` environment variable will be set to a truthy value. # Install ```sh...

(AI_EN_LECTOR_REPLACEMENT_VERB_TENSE)

cratedb_mcp/knowledge.py (1)

109-135: Well-implemented SQL validation with potential for enhancement

The implementation provides a solid foundation for enforcing read-only mode:

• Handles edge cases like empty or None inputs
• Properly parses SQL using a dedicated library
• Checks for both statement type and prohibited clauses
• Includes a configuration flag for flexibility

Consider these potential improvements:

1. The current implementation only examines the first statement if multiple statements are provided, potentially allowing SQL injection via statement chaining:

    # Parse the SQL statement.
    parsed = sqlparse.parse(expression.strip())
    if not parsed:
        return False

+   # Check for multiple statements (potential SQL injection)
+   if len(parsed) > 1:
+       return False
+
    # Check if the expression is valid and if it's a SELECT statement,
    # also trying to consider `SELECT ... INTO ...` statements.

2. Consider adding additional checks for potentially dangerous constructs within SELECT statements, like certain function calls or subqueries that might have side effects.

3. The FIXME comment correctly acknowledges limitations - exploring SQLAlchemy's read-only mode could be valuable for a future implementation.

tests/test_mcp.py (1)

31-40: Consider adding more complex SQL bypass tests

While the current tests cover basic scenarios well, consider adding more sophisticated test cases to verify that the validation can't be bypassed with more complex SQL constructs.

Add tests for these additional edge cases:

1. Statements with comments that might hide true intent
2. Multiple statements separated by semicolons
3. Complex subqueries within SELECT that might attempt data modification
4. Use of custom functions that could have side effects

def test_query_sql_forbidden_multiple_statements():
    with pytest.raises(ValueError) as ex:
        query_sql("SELECT 42; INSERT INTO foo VALUES (1)")
    assert ex.match("Only queries that have a SELECT statement are allowed")

def test_query_sql_forbidden_with_comments():
    with pytest.raises(ValueError) as ex:
        query_sql("/* Sneaky comment */ INSERT /* another comment */ INTO foo VALUES (1)")
    assert ex.match("Only queries that have a SELECT statement are allowed")

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

tests/test_knowledge.py (1)

22-75: Consider adding tests for other SQL operation types

The test suite is comprehensive, covering many edge cases and potential vulnerabilities. To ensure complete coverage, consider adding explicit tests for other SQL statement types like UPDATE, DELETE, TRUNCATE, DROP, and ALTER to verify they're also rejected.

+ def test_sql_expression_update_rejected():
+     """UPDATE statements are rejected"""
+     assert sql_expression_permitted("UPDATE foobar SET column = 'value'") is False
+
+
+ def test_sql_expression_delete_rejected():
+     """DELETE statements are rejected"""
+     assert sql_expression_permitted("DELETE FROM foobar") is False
+
+
+ def test_sql_expression_truncate_rejected():
+     """TRUNCATE statements are rejected"""
+     assert sql_expression_permitted("TRUNCATE TABLE foobar") is False
+
+
+ def test_sql_expression_drop_rejected():
+     """DROP statements are rejected"""
+     assert sql_expression_permitted("DROP TABLE foobar") is False
+
+
+ def test_sql_expression_alter_rejected():
+     """ALTER statements are rejected"""
+     assert sql_expression_permitted("ALTER TABLE foobar ADD COLUMN newcol INTEGER") is False

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

tests/test_knowledge.py (1)

22-99: Consider adding more edge case tests.

Your test suite is already quite thorough, but you might consider adding tests for:

1. Case-insensitive statements (e.g., "select" vs "SELECT")
2. Nested queries or complex SELECT statements with JOIN, GROUP BY, etc.
3. SQL injection patterns like string concatenation or comment termination
4. Statements with whitespace before the command (e.g., " SELECT")
5. Statements with Unicode characters or escapes

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

tests/test_knowledge.py (1)

75-82: Strong protection against obfuscation techniques.

These tests verify that common SQL obfuscation techniques (using multiple statements or embedding comments) are properly detected and rejected, which is critical for security.

Consider adding a few more tests for advanced SQL injection techniques such as:

def test_sql_expression_case_manipulation_rejected():
    """Statements with case manipulation to hide intent are rejected"""
    assert sql_expression_permitted("SeLeCt * FrOm users; DrOp TaBlE users") is False

def test_sql_expression_unicode_evasion_rejected():
    """Statements with unicode characters to evade filters are rejected"""
    assert sql_expression_permitted("SELECT * FROM users; \uFF1B DROP TABLE users") is False

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

tests/test_knowledge.py (2)

40-43: Consider consolidating redundant test cases

There appears to be redundancy between test_sql_expression_multiple_statements_rejected and test_sql_expression_select_multiple_rejected. Both verify that multiple statements are rejected, just with different SQL content.

You could consolidate these into a single test case with multiple assertions or make their purposes more distinct:

 def test_sql_expression_select_multiple_rejected():
-    """Multiple SQL statements are rejected"""
+    """Multiple SELECT statements are rejected"""
     assert sql_expression_permitted("SELECT 42; SELECT 42;") is False


 def test_sql_expression_multiple_statements_rejected():
+    """Mixed statement types (SELECT + DML) are rejected"""
     assert sql_expression_permitted("SELECT 42; INSERT INTO foo VALUES (1)") is False

Also applies to: 76-78

---

22-118: Consider additional test cases for complex SQL constructs

The test suite is already comprehensive, but to further strengthen it against potential filtering bypasses, consider adding tests for more complex SQL constructs like:

1. Common Table Expressions (CTEs)
2. Nested subqueries
3. UNION/INTERSECT/EXCEPT operations
4. Window functions
5. Vendor-specific syntax extensions

These could potentially be used in SQL injection attempts to bypass filtering.

def test_sql_expression_with_cte_permitted():
    """SELECT with CTE is permitted"""
    assert sql_expression_permitted(
        "WITH nums AS (SELECT 1) SELECT * FROM nums"
    ) is True

def test_sql_expression_with_union_permitted():
    """SELECT with UNION is permitted"""
    assert sql_expression_permitted(
        "SELECT 1 UNION SELECT 2"
    ) is True

def test_sql_expression_with_cte_dml_rejected():
    """CTE with DML is rejected"""
    assert sql_expression_permitted(
        "WITH nums AS (SELECT 1) INSERT INTO foo SELECT * FROM nums"
    ) is False

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

tests/test_knowledge.py (3)

29-31: Good validation of malformed SELECT statements.

This test correctly verifies that commented-out or malformed SELECT statements are rejected. Consider adding a few more examples of malformed statements for more comprehensive coverage.

def test_sql_expression_select_rejected():
    """Bogus SQL SELECT statements are rejected"""
    assert sql_expression_permitted(r"--\; select 42") is False
+   assert sql_expression_permitted(r"SEL ECT 42") is False
+   assert sql_expression_permitted(r"SELECT42") is False

---

22-117: Consider adding parameterized tests to reduce duplication.

While the test coverage is comprehensive, consider using pytest's parameterized testing to reduce code duplication and make it easier to add more test cases in the future.

import pytest

@pytest.mark.parametrize(
    "statement,expected",
    [
        ("SELECT 42", True),
        (" SELECT 42", True),
        ("select 42", True),
        (r"--\; select 42", False),
        ("SELECT 42; SELECT 42;", False),
        ("CREATE TABLE foobar AS SELECT 42", False),
        # Add more cases here
    ],
)
def test_sql_expression_permitted_cases(statement, expected):
    """Test various SQL statements against permission rules"""
    assert sql_expression_permitted(statement) is expected

This approach would maintain the same test coverage while making the code more maintainable.

---

22-117: Consider testing additional SQL injection techniques.

While the current tests cover many scenarios, consider adding tests for these additional SQL injection techniques:

• Inline comments within SQL keywords (SEL/**/ECT)
• Stacked queries with alternate delimiters
• UNION-based injection attempts

def test_sql_expression_inline_comments_rejected():
    """Statements with inline comments to hide intent are rejected"""
    assert sql_expression_permitted("SEL/**/ECT * FROM users") is False
    
def test_sql_expression_union_injection_rejected():
    """UNION-based injection attempts are rejected"""
    assert sql_expression_permitted("SELECT 1 UNION SELECT * FROM users") is False

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

cratedb_mcp/util/sql.py (1)

1-141: 💡 Verification agent

🧩 Analysis chain

Consider adding tests for SQL statements with embedded Unicode control characters.

While you check for Unicode semicolons, there are other Unicode characters that could be used to evade filters, such as zero-width spaces, bidirectional overrides, etc.

Run this script to check for additional evasion techniques:

---

🏁 Script executed:

#!/bin/bash
# Description: Test SQL filtering against additional Unicode evasion techniques

# Get current Unicode evasion test case
echo "Current test:"
grep -n "unicode_evasion" tests/test_util.py -A 3

# Test script with additional Unicode evasion test cases
cat << 'EOF' > /tmp/test_more_evasions.py
from cratedb_mcp.util.sql import sql_is_permitted

# Bidirectional override characters
print("Testing RIGHT-TO-LEFT OVERRIDE (U+202E):")
print(sql_is_permitted("SELECT * FROM users \u202E ELBAT PORD -- "))

# Zero-width spaces
print("Testing ZERO WIDTH SPACE (U+200B):")
print(sql_is_permitted("SELECT * FROM users;\u200BDROP TABLE users"))

# Line separator
print("Testing LINE SEPARATOR (U+2028):")
print(sql_is_permitted("SELECT * FROM users\u2028DROP TABLE users"))

# Paragraph separator
print("Testing PARAGRAPH SEPARATOR (U+2029):")
print(sql_is_permitted("SELECT * FROM users\u2029DROP TABLE users"))
EOF

# Run test script
python /tmp/test_more_evasions.py

Length of output: 547

---

🏁 Script executed:

#!/bin/bash
# Re-run Unicode evasion tests with the repository on PYTHONPATH
export PYTHONPATH="$(pwd)"

python - << 'EOF'
from cratedb_mcp.util.sql import sql_is_permitted

cases = {
    "RIGHT-TO-LEFT OVERRIDE (U+202E)":      "SELECT * FROM users \u202E ELBAT PORD -- ",
    "ZERO WIDTH SPACE (U+200B)":            "SELECT * FROM users;\u200BDROP TABLE users",
    "LINE SEPARATOR (U+2028)":              "SELECT * FROM users\u2028DROP TABLE users",
    "PARAGRAPH SEPARATOR (U+2029)":         "SELECT * FROM users\u2029DROP TABLE users",
}

for name, sql in cases.items():
    result = sql_is_permitted(sql)
    print(f"{name}: permitted? {result}")
EOF

Length of output: 318

---

Add tests for additional Unicode control‐character evasion cases

The current suite only covers the fullwidth semicolon (\uFF1B), but SQL filters can be bypassed with other invisible or bidirectional characters. Please extend tests/test_util.py with cases for:

• RIGHT-TO-LEFT OVERRIDE (U+202E)
• ZERO WIDTH SPACE (U+200B)
• LINE SEPARATOR (U+2028)
• PARAGRAPH SEPARATOR (U+2029)

For example:

import pytest
from cratedb_mcp.util.sql import sql_is_permitted

@pytest.mark.parametrize("label,expr", [
    ("U+202E RTL OVERRIDE",      "SELECT * FROM users \u202E ELBAT PORD -- "),
    ("U+200B ZWSP",              "SELECT * FROM users;\u200BDROP TABLE users"),
    ("U+2028 LINE SEP",          "SELECT * FROM users\u2028DROP TABLE users"),
    ("U+2029 PARAGRAPH SEP",     "SELECT * FROM users\u2029DROP TABLE users"),
])
def test_sql_expression_unicode_evasion_rejected(label, expr):
    """Ensure statements with {} are rejected""".format(label)
    assert sql_is_permitted(expr) is False

This will help ensure the filter catches all known Unicode‐based evasion techniques.

cratedb_mcp/util/sql.py (5)

13-30: Add type validation for the expression parameter.

The function is well-documented, but it doesn't validate the type of the expression parameter before using it, which could lead to unexpected behavior with non-string inputs.

def sql_is_permitted(expression: str) -> bool:
    """
    Validate the SQL expression, only permit read queries by default.

    When the `CRATEDB_MCP_PERMIT_ALL_STATEMENTS` environment variable is set,
    allow all types of statements.

    FIXME: Revisit implementation, it might be too naive or weak.
           Issue:    https://github.com/crate/cratedb-mcp/issues/10
           Question: Does SQLAlchemy provide a solid read-only mode, or any other library?
    """
+   # Ensure expression is a string
+   if expression is None or not isinstance(expression, str):
+       logger.warning(f"Denied non-string SQL expression: {expression}")
+       return False
    is_dql = SqlFilter(expression=expression, permit_all=PERMIT_ALL_STATEMENTS).is_dql
    if is_dql is True:
        logger.info(f"Permitted SQL expression: {expression}")
    else:
        logger.warning(f"Denied SQL expression: {expression}")
    return is_dql

---

32-47: Add error handling in the post-initialization method.

The __post_init__ method assumes self.expression is a string if it's truthy, which could lead to an AttributeError if a non-string object is passed.

def __post_init__(self) -> None:
-   if self.expression:
+   if self.expression and isinstance(self.expression, str):
        self.expression = self.expression.strip()
+   elif self.expression and not isinstance(self.expression, str):
+       # Handle non-string inputs by converting to string
+       self.expression = str(self.expression).strip()

---

48-63: Add error handling for SQL parsing failures.

The parsing methods don't include error handling for failures, which could occur if the expression is not valid SQL or if the libraries raise exceptions.

def parse_cratedb(self):
    """
    Parse expression using `cratedb-sqlparse` library.
    """
    if self._parsed_cratedb is None:
-       self._parsed_cratedb = cratedb_sqlparse.sqlparse(self.expression)
+       try:
+           self._parsed_cratedb = cratedb_sqlparse.sqlparse(self.expression)
+       except Exception as e:
+           # If parsing fails, return an empty list to indicate invalid SQL
+           logger.warning(f"Failed to parse SQL with cratedb_sqlparse: {e}")
+           self._parsed_cratedb = []
    return self._parsed_cratedb

def parse_sqlparse(self):
    """
    Parse expression using traditional `sqlparse` library.
    """
    if self._parsed_sqlparse is None:
-       self._parsed_sqlparse = sqlparse.parse(self.expression)
+       try:
+           self._parsed_sqlparse = sqlparse.parse(self.expression)
+       except Exception as e:
+           # If parsing fails, return an empty list to indicate invalid SQL
+           logger.warning(f"Failed to parse SQL with sqlparse: {e}")
+           self._parsed_sqlparse = []
    return self._parsed_sqlparse

---

88-101: Add defensive check before accessing parsed SQL data.

The operation property assumes parsing succeeded and returned at least one statement, which could lead to an IndexError if parsing failed or returned an empty list.

@property
def operation(self) -> str:
    """
    The SQL operation: SELECT, INSERT, UPDATE, DELETE, CREATE, etc.
    """
-   return self._parsed_cratedb[0].type.upper()
+   parsed = self.parse_cratedb()
+   if not parsed:
+       return ""  # Return empty string for invalid SQL
+   return parsed[0].type.upper()

---

109-141: Add defensive checks in is_select_into and is_evasive properties.

Both properties make assumptions about the structure of the parsed SQL, which could lead to IndexErrors if parsing fails or returns unexpected results. Also, the approach for detecting SELECT INTO statements could be improved.

@property
def is_select_into(self) -> bool:
    """
    Use traditional `sqlparse` for catching `SELECT ... INTO ...` statements.

    Note: With `cratedb-sqlparse`, we can't find the `INTO` token at all?

    Examples:

        SELECT * INTO foobar FROM bazqux
        SELECT * FROM bazqux INTO foobar
    """
    parsed = self.parse_sqlparse()
+   if not parsed:
+       return False
    tokens = [str(item).upper() for item in parsed[0]]
    return "INTO" in tokens

@property
def is_evasive(self) -> bool:
    """
    Use traditional `sqlparse` for catching evasive SQL statements.

    A practice picked up from CodeRabbit was to reject multiple statements
    to prevent potential SQL injections. Is it a viable suggestion?

    Note: This currently does not work with `cratedb-sqlparse`?

    Examples:

        SELECT * FROM users; \uff1b DROP TABLE users
    """
    parsed = self.parse_sqlparse()
    return len(parsed) > 1

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

cratedb_mcp/util/sql.py (3)

56-78: Possible redundancy in is_dql property

The is_dql property has solid logic, but there's redundancy in checking for multiple statements.

Line 72-73 checks if len(parsed) > 1, but this same check is also implemented in the is_evasive property. Consider refactoring to use self.is_evasive instead of duplicating the logic:

-        # Reject multiple statements to prevent potential SQL injections.
-        if len(parsed) > 1:
-            return False
+        # Reject multiple statements to prevent potential SQL injections.
+        if self.is_evasive:
+            return False

Also, note that line 68 references cratedb-sqlparse in the comment but actually uses sqlparse. Consider updating the comment for clarity.

---

101-115: Consider more robust INTO detection

The is_select_into property could be susceptible to false positives if "INTO" appears in string literals.

Consider a more syntax-aware approach for detecting "INTO" clauses:

-        tokens = [str(token).upper() for token in parsed[0].tokens]
-        return "INTO" in tokens
+        # More robust approach to detect INTO not within string literals
+        for token in parsed[0].tokens:
+            if token.ttype in (sqlparse.tokens.Keyword, sqlparse.tokens.Keyword.DML) and str(token).upper() == "INTO":
+                return True
+        return False

This would check that the token is actually a keyword, not just a string containing "INTO".

---

117-132: Consider removing redundancy with is_dql

The is_evasive property duplicates logic that's already in the is_dql method.

To avoid maintenance issues if one implementation changes but not the other, consider refactoring to make one method call the other:

-        parsed = self.parse_sqlparse()
-        return len(parsed) > 1
+        return len(self.parse_sqlparse()) > 1

Also, since this check is already performed in is_dql, consider restructuring to avoid duplication.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

cratedb_mcp/util/sql.py (6)

12-29: Review logging of full SQL expressions for potential security issues

While the function is well-structured, logging the complete SQL expression might inadvertently expose sensitive data (e.g., data in WHERE clauses).

Consider redacting or truncating the SQL in the logs:

-        logger.info(f"Permitted SQL expression: {expression}")
+        logger.info(f"Permitted SQL expression: {expression[:50]}...")
-        logger.warning(f"Denied SQL expression: {expression}")
+        logger.warning(f"Denied SQL expression: {expression[:50]}...")

Also, the is_dql is True check is redundant:

-    if is_dql is True:
+    if is_dql:

---

31-46: Consider adding null check for expression in class constructor

The __post_init__ method strips whitespace if expression is truthy, but there's no validation that expression isn't None.

    def __post_init__(self) -> None:
+        if self.expression is None:
+            self.expression = ""
        if self.expression:
            self.expression = self.expression.strip()

Also, consider using a more specific type hint for _parsed_sqlparse instead of t.Any.

---

48-54: Add return type annotation and error handling to parse_sqlparse method

The method lacks a return type annotation and could fail if self.expression is None.

-def parse_sqlparse(self):
+def parse_sqlparse(self) -> t.List[sqlparse.sql.Statement]:
    """
    Parse expression using traditional `sqlparse` library.
    """
    if self._parsed_sqlparse is None:
+        if not self.expression:
+            return []
        self._parsed_sqlparse = sqlparse.parse(self.expression)
    return self._parsed_sqlparse

---

56-79: Incorrect comment in is_dql about cratedb-sqlparse

The comment "Parse the SQL statement using cratedb-sqlparse" is inaccurate since the code is using the standard sqlparse library.

-    # Parse the SQL statement using `cratedb-sqlparse`.
+    # Parse the SQL statement using the sqlparse library.

Also, there's a duplicate check for multiple statements here and in is_evasive, which could be confusing for maintainers.

---

117-133: Duplicate check for multiple statements

The is_evasive property duplicates the check already performed in is_dql.

To avoid redundancy and improve maintainability, consider refactoring to use is_evasive in is_dql without repeating the logic:

@property
def is_dql(self) -> bool:
    """
    Whether the statement is a DQL statement, which effectively invokes read-only operations only.
    """

    if not self.expression:
        return False

    if self.permit_all:
        return True

    # Parse the SQL statement using the sqlparse library.
    parsed = self.parse_sqlparse()

-    # Reject multiple statements to prevent potential SQL injections.
-    if len(parsed) > 1:
-        return False

    # Check if the expression is valid and if it's a DQL/SELECT statement,
    # also trying to consider `SELECT ... INTO ...` and evasive
    # `SELECT * FROM users; \uff1b DROP TABLE users` statements.
    return self.is_select and not self.is_camouflage

Additionally, the docstring mentions cratedb-sqlparse but the code uses standard sqlparse.

---

1-133: Enhance security-critical module with additional safeguards

This is an excellent start for a security-critical filter, but consider these additional protections:

1. Handle malicious SQL encodings (hex, comments, etc.)
2. Add specific checks for common SQL injection patterns
3. Consider implementing a deny list of dangerous keywords
4. Add unit tests with comprehensive edge cases

The current implementation mainly focuses on DQL vs. non-DQL distinction. For enhanced security, consider a more comprehensive approach with a proper SQL parser or query executor that runs in a sandbox with appropriate permissions.

For maximum security, you might want to explore ORM-based solutions or database view-based approaches instead of SQL filtering.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

cratedb_mcp/util/sql.py (1)

83-89: ⚠️ Potential issue

Add error handling to the operation property.

The operation property accesses parsed[0] without checking if the parsed result contains any elements, which could cause an IndexError.

@property
def operation(self) -> str:
    """
    The SQL operation: SELECT, INSERT, UPDATE, DELETE, CREATE, etc.
    """
    parsed = self.parse_sqlparse()
+    if not parsed:
+        return ""
    return parsed[0].get_type().upper()

cratedb_mcp/util/sql.py (3)

1-11: Import structure looks good but consider organizing them alphabetically.

The imports are well-organized by category (standard library, third-party, and local imports), but within each category, they could be sorted alphabetically for better maintainability.

---

32-50: Consider validating permit_all flag in post_init

The class handles None for expression but doesn't validate the permit_all flag. While it's unlikely to be None since it has a default value, being thorough with input validation is good practice.

def __post_init__(self) -> None:
    if self.expression is None:
        self.expression = ""
    if self.expression:
        self.expression = self.expression.strip()
+    # Ensure permit_all is a boolean
+    self.permit_all = bool(self.permit_all)

---

59-74: Consider adding explicit validation of SQL statement structure

The is_dql property checks if the statement is SELECT and not camouflage, but it doesn't validate that the SQL structure is correct. Consider adding a check that the parsed statement is valid SQL.

@property
def is_dql(self) -> bool:
    """
    Whether the statement is a DQL statement, which effectively invokes read-only operations only.
    """

    if not self.expression:
        return False

    if self.permit_all:
        return True

+    # Ensure we have a valid SQL statement
+    parsed = self.parse_sqlparse()
+    if not parsed:
+        return False
+
    # Check if the expression is valid and if it's a DQL/SELECT statement,
    # also trying to consider `SELECT ... INTO ...` and evasive
    # `SELECT * FROM users; \uff1b DROP TABLE users` statements.
    return self.is_select and not self.is_camouflage

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro