Great question—this is such a common and valid worry, especially when Python's string formatting muscle memory kicks in. Let's break down your options step by step:

1. psycopg2's Built-in Multi-Statement Blocking #

First, a key safeguard you might not be aware of: psycopg2's default cursor.execute() does not allow executing multiple SQL statements in a single call. If someone accidentally writes unsafe code like:

cursor.execute("SELECT * FROM employees WHERE name = '%s'" % name)

And an attacker passes ''; DROP TABLE employees; as name, the resulting string would contain two separate statements—but psycopg2 will throw a ProgrammingError: can't execute multiple statements in a single call instead of running it. This is a native protection from the PostgreSQL protocol that psycopg2 leverages by default.

That said, this doesn't block all injection risks (like boolean-based injection within a single statement), but it does stop the catastrophic multi-statement attacks you're most worried about.

2. Custom Validation with psycopg2's SQL Parsing Tools #

If you want to add an extra layer of control (e.g., enforcing strict single-statement rules or blocking dangerous keywords), psycopg2 provides a parse() function in the psycopg2.sql module that can split SQL into individual statements. You can use this to build a custom cursor wrapper:

from psycopg2 import sql
from psycopg2.extensions import cursor

class SafeCursor(cursor):
    def execute(self, query, vars=None):
        # Parse the query into discrete statements
        parsed_statements = sql.parse(query)
        
        # Block multi-statement execution
        if len(parsed_statements) > 1:
            raise ValueError("Only single SQL statements are permitted")
        
        # Optional: Add custom checks for dangerous operations
        for stmt in parsed_statements:
            if stmt.segment.name.lower() in ('drop', 'truncate', 'delete'):
                # Adjust this list based on your app's allowed operations
                raise ValueError("Destructive SQL commands are not allowed here")
        
        # Pass through to the original execute method
        super().execute(query, vars)

# Use the safe cursor when connecting
conn = psycopg2.connect(...)
cur = conn.cursor(cursor_factory=SafeCursor)

This wrapper explicitly blocks multi-statement attempts and lets you add context-specific rules to prevent risky commands.

3. Additional Prevention Strategies #

Beyond custom wrappers, here are more robust ways to eliminate accidental injection risks:

• Lint for unsafe patterns: Use tools like flake8-psycopg2 (a flake8 plugin) that flags unsafe string formatting with cursor.execute() during development, catching mistakes early.
• Use psycopg2.sql for dynamic SQL: When you need dynamic elements (like table names), use sql.SQL() and sql.Identifier() instead of string concatenation to ensure proper escaping:

  table_name = "employees"
  query = sql.SQL("SELECT * FROM {} WHERE name = %s").format(sql.Identifier(table_name))
  cursor.execute(query, (name,))
  

• Adopt an ORM: Libraries like SQLAlchemy handle parameterization automatically, removing the need to write raw SQL in most cases. Even its core SQL expression language is designed to prevent injection.
• Enforce team standards: Document and require parameterized queries for all database interactions, and pair this with code reviews to catch slip-ups.

内容的提问来源于stack exchange，提问作者Mike