以安全的方式从字典数据创建表

Posted 2023-02-19

【中文标题】以安全的方式从字典数据创建表

【英文标题】：Create table from dictionary data in a safe way

【发布时间】：2020-12-25 21:25:50

【问题描述】：

我有一个问题，我有一个字典列表，例如以下数据：

columns = [
    'name': 'column1',
    'type': 'varchar'
,

    'name': 'column2',
    'type': 'decimal'
,
.
.
.
]

从该列表中，我需要根据列表中包含列名称和类型的每个字典动态创建 CREATE TABLE 语句，并使用 psycopg2 适配器在 PostgreSQL 数据库上执行它。

我设法做到了：

columns = "(" + ",\n".join([" ".format(col['name'], col['type']) for col in columns]) + ")"
cursor.execute("CREATE TABLE some_table_name\n ".format(columns))

但是这种解决方案容易受到 SQL 注入的影响。我试图用 psycopg2 的 sql 模块做同样的事情，但没有运气。总是出现语法错误，因为它将类型括在引号中。

有什么方法可以安全地做到这一点吗？

【问题讨论】：

你能用psycopg2.sql发布你的尝试吗？

'columns' 有取消引号的字符串：是故意的吗？

没有时间看一个完整的例子，而是一个简单的例子：print(sql.SQL('decimal').as_string(con)) decimal

【参考方案1】：

扩展我的评论，一个完整的例子：

import psycopg2 
from psycopg2 import sql

columns = [
    'name': 'column1',
    'type': 'varchar'
,

    'name': 'column2',
    'type': 'decimal'
]

con = psycopg2.connect("dbname=test host=localhost user=aklaver") 
cur = con.cursor()

col_list = sql.SQL(',').join( [sql.Identifier(col["name"]) + sql.SQL(' ') + sql.SQL(col["type"]) for col in columns])
create_sql = sql.SQL("CREATE TABLE tablename ()").format(col_list)

print(create_sql.as_string(con))                                                                                                                                          
CREATE TABLE tablename ("column1" varchar,"column2" decimal)

cur.execute(create_sql)
con.commit()


test(5432)=> \d tablename 
                   Table "public.tablename"
 Column  |       Type        | Collation | Nullable | Default 
---------+-------------------+-----------+----------+---------
 column1 | character varying |           |          | 
 column2 | numeric           | 

遍历dicts的列列表并将列名分配为SQL标识符，将列类型直接分配为SQL到sql.SQL构造中。将此用作CREATE TABLE SQL 的参数。

警告：sql.SQL() 不会转义，因此在使用这些值之前必须对其进行验证。

【参考方案2】：

您可以使用AsIs 来获取添加的列类型不带引号：

import psycopg2
from psycopg2.extensions import AsIs
import psycopg2.sql as sql

conn = psycopg2.connect("dbname=mf port=5959 host=localhost user=mf_usr")

columns = [
    'name': "column1",
    'type': "varchar"
,

    'name': "column2",
    'type': "decimal"
]

# create a dict, so we can use dict placeholders in the CREATE TABLE query.
column_dict = c['name']: AsIs(c['type']) for c in columns

createSQL = sql.SQL("CREATE TABLE some_table_name\n (columns)").format(
    columns = sql.SQL(',').join(
        sql.SQL(' ').join([sql.Identifier(col), sql.Placeholder(col)]) for col in column_dict)
)

print(createSQL.as_string(conn))
cur = conn.cursor()
cur.execute(createSQL, column_dict)
cur.execute("insert into some_table_name (column1) VALUES ('foo')")
cur.execute("select * FROM some_table_name")
print('Result: ', cur.fetchall())

输出：

CREATE TABLE some_table_name
 ("column1" %(column1)s,"column2" %(column2)s)
Result:  [('foo', None)]

注意：psycopg2.sql 对 SQL 注入是安全的，AsIs 可能不是。 使用'type': "varchar; DROP TABLE foo" 进行测试导致 Postgres 语法错误：

b'CREATE TABLE some_table_name\n ("column1" varchar; DROP TABLE foo,"column2" decimal)'
Traceback (most recent call last):
  File "pct.py", line 28, in <module>
    cur.execute(createSQL, column_dict)
psycopg2.errors.SyntaxError: syntax error at or near ";"
LINE 2:  ("column1" varchar; DROP TABLE foo,"column2" decimal)

【讨论】：

这不安全。我只是尝试在字典的最后一列中添加足够的查询，然后它删除了数据库。示例：'type': "decimal); DROP table some_table_name; create table some_table_name_2(aaaaa varchar(255)" pastebin.pl/view/8bcb319b

是的，但很容易测试：';' in col_value。