Prevenindo ataques de injeção de SQL com Python

• O que injeção de SQL do Python é e como evitá-lo
• Como compor consultas com literais e identificadores como parâmetros
• Como executar consultas com segurança em um banco de dados

Compreendendo o Python SQL Injection

Configurando um banco de dados

Criando um banco de dados

$ createdb -O postgres psycopgtest

$ psql -U postgres -d psycopgtest
psql (11.2, server 10.5)
Type "help" for help.

Criando uma tabela com dados

psycopgtest=# CREATE TABLE users (
    username varchar(30),
    admin boolean
);
CREATE TABLE

psycopgtest=# INSERT INTO users
    (username, admin)
VALUES
    ('ran', true),
    ('haki', false);
INSERT 0 2

psycopgtest=# SELECT * FROM users;
 username | admin
----------+-------
 ran      | t
 haki     | f
(2 rows)

Configurando um ambiente virtual Python

(~/src) $ mkdir psycopgtest
(~/src) $ cd psycopgtest
(~/src/psycopgtest) $ python3 -m venv venv

Conectando ao banco de dados

(~/src/psycopgtest) $ source venv/bin/activate
(~/src/psycopgtest) $ python -m pip install psycopg2>=2.8.0
Collecting psycopg2
  Using cached https://....
  psycopg2-2.8.2.tar.gz
Installing collected packages: psycopg2
  Running setup.py install for psycopg2 ... done
Successfully installed psycopg2-2.8.2

import psycopg2

connection = psycopg2.connect(
    host="localhost",
    database="psycopgtest",
    user="postgres",
    password=None,
)
connection.set_session(autocommit=True)

• 
  host é o endereço IP ou o DNS do servidor onde seu banco de dados está localizado. Nesse caso, o host é sua máquina local ou localhost .
  
• 
  database é o nome do banco de dados ao qual se conectar. Você deseja se conectar ao banco de dados criado anteriormente, psycopgtest .
  
• 
  user é um usuário com permissões para o banco de dados. Neste caso, você deseja se conectar ao banco de dados como proprietário, então você passa o usuário postgres .
  
• 
  password é a senha para quem você especificou em user . Na maioria dos ambientes de desenvolvimento, os usuários podem se conectar ao banco de dados local sem uma senha.
  

from django.db import connection

Executando uma consulta

>>> with connection.cursor() as cursor:
...     cursor.execute('SELECT COUNT(*) FROM users')
...     result = cursor.fetchone()
... print(result)
(2,)

Usando parâmetros de consulta em SQL

# BAD EXAMPLE. DON'T DO THIS!
def is_admin(username: str) -> bool:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                admin
            FROM
                users
            WHERE
                username = '%s'
        """ % username)
        result = cursor.fetchone()
    admin, = result
    return admin

>>> is_admin('haki')
False
>>> is_admin('ran')
True

>>> is_admin('foo')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 12, in is_admin
TypeError: cannot unpack non-iterable NoneType object

# BAD EXAMPLE. DON'T DO THIS!
def is_admin(username: str) -> bool:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                admin
            FROM
                users
            WHERE
                username = '%s'
        """ % username)
        result = cursor.fetchone()

    if result is None:
        # User does not exist
        return False

    admin, = result
    return admin

>>> is_admin('haki')
False
>>> is_admin('ran')
True
>>> is_admin('foo')
False

Explorando parâmetros de consulta com Python SQL Injection

>>> is_admin("'; select true; --")
True

>>> print("select admin from users where username = '%s'" % "'; select true; --")
select admin from users where username = ''; select true; --'

select admin from users where username = '';

select true;

--'

>>> is_admin('haki')
False
>>> is_admin("'; update users set admin = 'true' where username = 'haki'; select true; --")
True
>>> is_admin('haki')
True

';

update users set admin = 'true' where username = 'haki';

select true; --

psycopgtest=# select * from users;
 username | admin
----------+-------
 ran      | t
 haki     | t
(2 rows)

psycopgtest=# update users set admin = false where username = 'haki';
UPDATE 1

Criação de parâmetros de consulta seguros

>>> # BAD EXAMPLE. DON'T DO THIS!
>>> username = username.replace("'", "''")

 1def is_admin(username: str) -> bool:
 2    with connection.cursor() as cursor:
 3        cursor.execute("""
 4            SELECT
 5                admin
 6            FROM
 7                users
 8            WHERE
 9                username = %(username)s
10        """, {
11            'username': username
12        })
13        result = cursor.fetchone()
14
15    if result is None:
16        # User does not exist
17        return False
18
19    admin, = result
20    return admin

• 
  Na linha 9, você usou um parâmetro nomeado username para indicar onde o nome de usuário deve ir. Observe como o parâmetro username não está mais entre aspas simples.
  
• 
  Na linha 11, você passou o valor de username como o segundo argumento para cursor.execute() . A conexão usará o tipo e o valor de username ao executar a consulta no banco de dados.
  

>>> is_admin('haki')
False
>>> is_admin('ran')
True
>>> is_admin('foo')
False
>>> is_admin("'; select true; --")
False

>>> with connection.cursor() as cursor:
...    cursor.execute("""
...        SELECT
...            admin
...        FROM
...            users
...        WHERE
...            username = %(username)s
...    """, {
...        'username': "'; select true; --"
...    })
...    print(cursor.query.decode('utf-8'))
SELECT
    admin
FROM
    users
WHERE
    username = '''; select true; --'

Passando parâmetros de consulta segura

# BAD EXAMPLES. DON'T DO THIS!
cursor.execute("SELECT admin FROM users WHERE username = '" + username + '");
cursor.execute("SELECT admin FROM users WHERE username = '%s' % username);
cursor.execute("SELECT admin FROM users WHERE username = '{}'".format(username));
cursor.execute(f"SELECT admin FROM users WHERE username = '{username}'");

# SAFE EXAMPLES. DO THIS!
cursor.execute("SELECT admin FROM users WHERE username = %s'", (username, ));
cursor.execute("SELECT admin FROM users WHERE username = %(username)s", {'username': username});

Usando a composição SQL

# BAD EXAMPLE. DON'T DO THIS!
def count_rows(table_name: str) -> int:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                count(*)
            FROM
                %(table_name)s
        """, {
            'table_name': table_name,
        })
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 9, in count_rows
psycopg2.errors.SyntaxError: syntax error at or near "'users'"
LINE 5:                 'users'
                        ^

from psycopg2 import sql

def count_rows(table_name: str) -> int:
    with connection.cursor() as cursor:
        stmt = sql.SQL("""
            SELECT
                count(*)
            FROM
                {table_name}
        """).format(
            table_name = sql.Identifier(table_name),
        )
        cursor.execute(stmt)
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

>>> count_rows('users')
2

>>> count_rows('foo')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in count_rows
psycopg2.errors.UndefinedTable: relation "foo" does not exist
LINE 5:                 "foo"
                        ^

from psycopg2 import sql

def count_rows(table_name: str, limit: int) -> int:
    with connection.cursor() as cursor:
        stmt = sql.SQL("""
            SELECT
                COUNT(*)
            FROM (
                SELECT
                    1
                FROM
                    {table_name}
                LIMIT
                    {limit}
            ) AS limit_query
        """).format(
            table_name = sql.Identifier(table_name),
            limit = sql.Literal(limit),
        )
        cursor.execute(stmt)
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

>>> count_rows('users', 1)
1
>>> count_rows('users', 10)
2

>>> count_rows("(select 1) as foo; update users set admin = true where name = 'haki'; --", 1)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 18, in count_rows
psycopg2.errors.UndefinedTable: relation "(select 1) as foo; update users set admin = true where name = '" does not exist
LINE 8:                     "(select 1) as foo; update users set adm...
                            ^

Conclusão

• O que injeção de SQL do Python é e como pode ser explorado
• Como evitar a injeção de SQL do Python usando parâmetros de consulta
• Como compor instruções SQL com segurança que usam literais e identificadores como parâmetros