Postgres的jsonb查询用于动态值

Question

在用户表中，我有一个名为experience的json列，其结构如下：

[
    {
        "field": "devops",
        "years": 9
    },
    {
        "field": "backend dev",
        "years": 7
    } 
... // could be N number of objects with different values
]

业务需求

客户可以请求具有任何领域经验且在每个领域拥有相应年限经验的人员。

以下是一个查询示例

SELECT * FROM users
WHERE
jsonb_path_exists(experience, '$[*] ? (@.field == "devops" && @.years > 5)') and
jsonb_path_exists(experience, '$[*] ? (@.field == "backend dev" && @.years > 5)')
LIMIT 3;

问题

假设我收到一个请求：

[
  { field: "devops", years: 5 }, 
  { field: "java", years: 6 }, 
  { field: "ui/ux", years: 2 }] // and so on

如何动态创建查询而不必担心 SQL 注入？

技术栈

• Nodejs
• Typescript
• TypeORM
• Postgres

- Karan Kumar

Answer 1

索引

首先，你想要 索引支持。我建议使用一个像这样的 jsonb_path_ops 索引：

CREATE INDEX users_experience_gin_idx ON users USING gin (experience jsonb_path_ops);

请参见:

• 在JSON数组中查找元素的索引

查询

还有一个查询可以利用该索引（与原始查询完全等效）：

SELECT *
FROM   users
WHERE  experience @? '$[*] ? (@.field == "devops" && @.years > 5 )'
AND    experience @? '$[*] ? (@.field == "backend dev" && @.years > 5)'
LIMIT  3;

需要 Postgres 12 或更高版本，该版本添加了SQL/JSON路径语言。

索引支持与 Postgres 中的运算符相关联。操作符 @? 是 jsonb_path_exists() 的等效操作符。请参见：

• 查找包含 JSONB 记录数组中键的行

动态生成查询

SELECT 'SELECT * FROM users
WHERE  experience @? '
       || string_agg(quote_nullable(format('$[*] ? (@.field == %s && @.years > %s)'
                                         , f->'field'
                                         , f->'years')) || '::jsonpath'
                   , E'\nAND    experience @? ')
       || E'\nLIMIT  3'
FROM   jsonb_array_elements('[{"field": "devops", "years": 5 }, 
                              {"field": "java", "years": 6 }, 
                              {"field": "ui/ux", "years": 2 }]') f;

生成上述形式的查询：

SELECT * FROM users
WHERE  experience @? '$[*] ? (@.field == "devops" && @.years > 5)'::jsonpath
AND    experience @? '$[*] ? (@.field == "java" && @.years > 6)'::jsonpath
AND    experience @? '$[*] ? (@.field == "ui/ux" && @.years > 2)'::jsonpath
LIMIT  3;

全自动化

如何动态创建查询而不必担心SQL注入问题？

将上述查询生成过程放入一个PL/pgSQL函数中以实现动态执行：

CREATE OR REPLACE FUNCTION f_users_with_experience(_filter_arr jsonb, _limit int = 3)
  RETURNS SETOF users
  LANGUAGE plpgsql PARALLEL SAFE STABLE STRICT AS
$func$
DECLARE
   _sql text;
BEGIN
   -- assert (you may want to be stricter?)
   IF jsonb_path_exists (_filter_arr, '$[*] ? (!exists(@.field) || !exists(@.years))') THEN
      RAISE EXCEPTION 'Parameter $2 (_filter_arr) must be a JSON array with keys "field" and "years" in every object. Invalid input was: >>%<<', _filter_arr;
   END IF;

   -- generate query string
   SELECT INTO _sql
'SELECT * FROM users
WHERE  experience @? '
       || string_agg(quote_nullable(format('$[*] ? (@.field == %s && @.years > %s)'
                                         , f->'field'
                                         , f->'years'))
                   , E'\nAND    experience @? ')
       || E'\nLIMIT   ' || _limit
   FROM   jsonb_array_elements(_filter_arr) f;

   -- execute
   IF _sql IS NULL THEN
      RAISE EXCEPTION 'SQL statement is NULL. Should not occur!';
   ELSE
   -- RAISE NOTICE '%', _sql;     -- debug first if in doubt
      RETURN QUERY EXECUTE _sql;
   END IF;
END
$func$;

调用：

SELECT * FROM f_users_with_experience('[{"field": "devops", "years": 5 }, 
                                      , {"field": "backend dev", "years": 6}]');

或者使用不同的LIMIT：

SELECT * FROM f_users_with_experience('[{"field": "devops", "years": 5 }]', 123);

db<>fiddle 这里

如果您要操作此内容并理解它，则应熟悉PL/pgSQL。

SQL注入攻击是不可能的，因为...

1. 有效的JSON输入被强制执行
2. JSON值与原始JSON双引号拼接。
3. 最重要的是，每个生成的jsonpath值都用quote_nullable()单引号括起来。

在谈论SQL / JSON路径表达式时，我使用路径表达式来验证有效输入：

jsonb_path_exists (_filter_arr, '$[*] ? (!exists(@.field) || !exists(@.years))')

检查JSON数组中的每个对象，看是否缺少两个必需键之一（field，years）。

- Erwin Brandstetter

Answer 2

这是一个参数化查询，更或少避免了注入攻击。 qualifies 标量子查询计算是否满足所有请求项的 experience。 参数为 $1（请求参数的 jsonb 数组）和 $2（限制值）。根据您的环境风格，您可能需要更改它们的语法。

select t.* from 
(
  select u.*,
    (
      select count(*) = jsonb_array_length($1)
      from jsonb_array_elements(u.experience) ej -- jsonb list of experiences 
      inner join jsonb_array_elements($1) rj     -- jsonb list of request items
         on ej ->> 'field' =  rj ->> 'field'
        and (ej ->> 'years')::numeric >= (rj ->> 'years')::numeric
    ) as qualifies
 from users as u
) as t
where t.qualifies
limit $2;

一些解释

qualifies子查询的逻辑是：首先将experience和请求的jsonb数组'规范化'成'tables'，然后基于目标条件(在这种情况下是 field_a = field_b and years_a >= years_b) 进行内连接并计算匹配数量。如果数量等于请求项的数量(即count(*) = jsonb_array_length($1)), 那么所有请求项都得到满足，因此experience符合条件。
因此，不需要动态SQL。我认为这种方法可能也是可重复使用的。

- Stefanov.sm