fix(sqli): close remaining apostrophe-in-identifier sinks and tighten the lint
Layered follow-up to the COMMENT description SQLi fix. Closes three
related gaps the original patch did not cover.
1. Stats templates — apostrophe-in-identifier SQLi (10 sites).
The pgstattuple/pgstatindex call sites in the stats templates rendered
the target relation as an embedded single-quoted literal, e.g.
pgstattuple('{{schema_name}}.{{table_name}}')
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(name)}}')
A user with CREATE privilege on a schema could plant a table or index
named `foo'bar` (PostgreSQL permits any character except NUL inside a
quoted identifier). Any viewer who then opened that object's stats
panel would render an unbalanced SQL literal. qtIdent does NOT escape
apostrophes — it escapes the embedded double quotes that delimit a
quoted identifier — so the outer single-quoted literal still broke
out. Same bug class as the COMMENT description vector, just gated on
DDL privilege instead of dialog access.
Fix replaces all 10 sites with the canonical regclass form:
pgstattuple({{ tid }}::oid::regclass)
pgstatindex({{ exid|cid|idx }}::oid::regclass)
The OID is already passed by each handler (tables/utils.py,
exclusion_constraint/__init__.py, index_constraint/__init__.py,
indexes/__init__.py, views/__init__.py). The cast eliminates the
embedded string literal entirely — the bug class can no longer recur
through these sites.
Files:
tables/sql/{default,16_plus}/stats.sql
views/templates/mviews/{pg,ppas}/default/sql/stats.sql
tables/templates/exclusion_constraint/sql/{default,16_plus}/stats.sql
tables/templates/index_constraint/sql/{default,16_plus}/stats.sql
tables/templates/indexes/sql/{default,16_plus}/stats.sql
2. Lint regex — embedded-Jinja blind spot.
The original lint regex `'\{\{[^}]+\}\}'` only matched literals where
the Jinja interpolation filled the entire single-quoted body. It
missed the embedded form `'foo{{ x }}bar'` — exactly the shape used
by the stats templates above and by `'%{{ search_text }}%'` in
search_objects.
The broadened regex `'[^'\n{}]*(?:\{\{[^}]+\}\}[^'\n{}]*)+'` catches
both forms. The `[^'\n{}]` body constraint is what stops the regex
from spanning a Jinja control block `{%...%}` — without it the regex
would walk across long `{% if %}` chains and produce huge false-
positive matches that span unrelated literals.
Allowlist extended with one-line justifications for every additional
occurrence the broadened regex now surfaces (search_objects manual
escape, CATALOGS.LABELS_SCHEMACOL macro, gettext-translated catalog
labels, opcintype oidvector pairs, the pg_get_partkeydef CASE
expression where the interpolation is between two separate literals
rather than inside one).
3. qtLiteral — close the inner-except silent-failure hole.
The fail-fast for a missing `conn` was the right shape but the inner
`except Exception: print(value)` still silently returned the raw
input when `psycopg.sql.Literal(value).as_string(conn)` raised
(unadapted custom type, encoding error, etc.). Same SQL-injection
sink the outer fail-fast just plugged, gated on a different failure
mode. Removed the except so the exception propagates — callers can
now react explicitly instead of seeing an unescaped value flow into
their SQL.
Tests:
test_stats_template_regclass_cast.py — 10 scenarios that render
each stats template with an apostrophe-bearing identifier payload
and assert (positive) the pgstat* call uses ::oid::regclass and
(negative) no `pgstat*('...'` form recurs, and the apostrophe
payload never appears inside a literal in the rendered SQL.
test_qtliteral_no_silent_unescape.py — asserts qtLiteral raises on
an unadaptable value rather than silently returning it.
42 tests / 0 failures across the touched packages.
This commit is contained in:
Ashesh Vashi
@@ -19,7 +19,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(name)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ exid }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -18,7 +18,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(name)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ exid }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -19,7 +19,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(name)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ cid }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -18,7 +18,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(name)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ cid }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -19,7 +19,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(index)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ idx }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -18,7 +18,7 @@ SELECT
|
||||
avg_leaf_density AS {{ conn|qtIdent(_('Average leaf density')) }},
|
||||
leaf_fragmentation AS {{ conn|qtIdent(_('Leaf fragmentation')) }}
|
||||
FROM
|
||||
pgstatindex('{{conn|qtIdent(schema)}}.{{conn|qtIdent(index)}}'), pg_catalog.pg_stat_all_indexes stat
|
||||
pgstatindex({{ idx }}::oid::regclass), pg_catalog.pg_stat_all_indexes stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_indexes stat
|
||||
|
||||
@@ -41,7 +41,7 @@ SELECT
|
||||
free_space AS {{ conn|qtIdent(_('Free space')) }},
|
||||
free_percent AS {{ conn|qtIdent(_('Free percent')) }}
|
||||
FROM
|
||||
pgstattuple('{{schema_name}}.{{table_name}}'), pg_catalog.pg_stat_all_tables stat
|
||||
pgstattuple({{ tid }}::oid::regclass), pg_catalog.pg_stat_all_tables stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_tables stat
|
||||
|
||||
@@ -40,7 +40,7 @@ SELECT
|
||||
free_space AS {{ conn|qtIdent(_('Free space')) }},
|
||||
free_percent AS {{ conn|qtIdent(_('Free percent')) }}
|
||||
FROM
|
||||
pgstattuple('{{schema_name}}.{{table_name}}'), pg_catalog.pg_stat_all_tables stat
|
||||
pgstattuple({{ tid }}::oid::regclass), pg_catalog.pg_stat_all_tables stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_tables stat
|
||||
|
||||
@@ -40,7 +40,7 @@ SELECT
|
||||
free_space AS {{ conn|qtIdent(_('Free space')) }},
|
||||
free_percent AS {{ conn|qtIdent(_('Free percent')) }}
|
||||
FROM
|
||||
pgstattuple('{{schema_name}}.{{mview_name}}'), pg_catalog.pg_stat_all_tables stat
|
||||
pgstattuple({{ vid }}::oid::regclass), pg_catalog.pg_stat_all_tables stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_tables stat
|
||||
|
||||
@@ -40,7 +40,7 @@ SELECT
|
||||
free_space AS {{ conn|qtIdent(_('Free space')) }},
|
||||
free_percent AS {{ conn|qtIdent(_('Free percent')) }}
|
||||
FROM
|
||||
pgstattuple('{{schema_name}}.{{mview_name}}'), pg_catalog.pg_stat_all_tables stat
|
||||
pgstattuple({{ vid }}::oid::regclass), pg_catalog.pg_stat_all_tables stat
|
||||
{% else %}
|
||||
FROM
|
||||
pg_catalog.pg_stat_all_tables stat
|
||||
|
||||
@@ -51,9 +51,24 @@ WEB_ROOT = os.path.realpath(
|
||||
# Root inside web/ that the lint walks.
|
||||
SCAN_ROOT = os.path.join(WEB_ROOT, 'pgadmin')
|
||||
|
||||
# Single-quote-wrapped Jinja interpolation. The expression body cannot
|
||||
# contain ``}`` because ``}}`` ends the Jinja tag.
|
||||
PATTERN = re.compile(r"'\{\{[^}]+\}\}'")
|
||||
# Single-quote-wrapped Jinja interpolation. Catches every variant of
|
||||
# ``'...{{ x }}...'`` that could become a SQL string-literal injection
|
||||
# sink:
|
||||
# * Jinja that fills the entire literal — ``'{{ x }}'``.
|
||||
# * Jinja embedded inside a literal with neighbouring text — e.g.
|
||||
# ``'%{{ search_text }}%'``.
|
||||
# * Multiple interpolations inside one literal — e.g.
|
||||
# ``pgstattuple('{{schema_name}}.{{table_name}}')``.
|
||||
#
|
||||
# The body between the outer quotes is restricted to either a complete
|
||||
# ``{{...}}`` group, or any character that is not ``'``, newline, ``{``
|
||||
# or ``}``. The ``{``/``}`` exclusion is what stops the regex from
|
||||
# crossing a Jinja control block ``{%...%}`` or comment ``{#...#}``
|
||||
# boundary — otherwise long ``{% if ... %}`` conditional templates would
|
||||
# produce huge false-positive matches that span unrelated literals.
|
||||
PATTERN = re.compile(
|
||||
r"'[^'\n{}]*(?:\{\{[^}]+\}\}[^'\n{}]*)+'"
|
||||
)
|
||||
|
||||
|
||||
# Allowlist of currently-known occurrences. Each entry is
|
||||
@@ -429,6 +444,155 @@ ALLOWLIST = [
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'default/search.sql', "'{{ obj_type }}'"), 1,
|
||||
"Fixed object-kind enum supplied by search_objects handler."),
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# search_objects: search_text is the only request-supplied value in
|
||||
# these templates. The handler at tools/search_objects/utils.py:120
|
||||
# pre-doubles every apostrophe via text.replace("'", "''") before
|
||||
# passing the value to render_template — equivalent to qtLiteral's
|
||||
# escape semantics, but applied at the Python boundary instead of
|
||||
# the template. Documented as fragile (manual escape) but currently
|
||||
# safe. Follow-up: migrate to qtLiteral with explicit % wrapping.
|
||||
# ------------------------------------------------------------------
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/pg/'
|
||||
'11_plus/search.sql', "'%{{ search_text }}%'"), 1,
|
||||
"search_text is pre-escaped via str.replace in utils.py:120 before "
|
||||
"render_template; follow-up: migrate to qtLiteral with explicit % "
|
||||
"wrapping."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/pg/'
|
||||
'default/search.sql', "'%{{ search_text }}%'"), 1,
|
||||
"search_text is pre-escaped via str.replace in utils.py:120 before "
|
||||
"render_template; follow-up: migrate to qtLiteral with explicit % "
|
||||
"wrapping."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'11_plus/search.sql', "'%{{ search_text }}%'"), 1,
|
||||
"search_text is pre-escaped via str.replace in utils.py:120 before "
|
||||
"render_template; follow-up: migrate to qtLiteral with explicit % "
|
||||
"wrapping."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'12_plus/search.sql', "'%{{ search_text }}%'"), 1,
|
||||
"search_text is pre-escaped via str.replace in utils.py:120 before "
|
||||
"render_template; follow-up: migrate to qtLiteral with explicit % "
|
||||
"wrapping."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'default/search.sql', "'%{{ search_text }}%'"), 1,
|
||||
"search_text is pre-escaped via str.replace in utils.py:120 before "
|
||||
"render_template; follow-up: migrate to qtLiteral with explicit % "
|
||||
"wrapping."),
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# search_objects: the CATALOGS.LABELS_SCHEMACOL macro emits a fixed
|
||||
# CASE expression mapping a column name (literal 'sn.schema_name')
|
||||
# to its localised label. Both macro arguments are static — the
|
||||
# column-name literal and the gettext function _ — so no
|
||||
# attacker-controlled value reaches the rendered SQL.
|
||||
# ------------------------------------------------------------------
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/pg/'
|
||||
'11_plus/search.sql',
|
||||
"'||{{ CATALOGS.LABELS_SCHEMACOL('sn.schema_name', _) }}||'"), 1,
|
||||
"CATALOGS.LABELS_SCHEMACOL emits a static CASE expression; "
|
||||
"both macro arguments are hardcoded (column name + gettext fn)."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/pg/'
|
||||
'default/search.sql',
|
||||
"'||{{ CATALOGS.LABELS_SCHEMACOL('sn.schema_name', _) }}||'"), 1,
|
||||
"CATALOGS.LABELS_SCHEMACOL emits a static CASE expression; "
|
||||
"both macro arguments are hardcoded (column name + gettext fn)."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'11_plus/search.sql',
|
||||
"'||{{ CATALOGS.LABELS_SCHEMACOL('sn.schema_name', _) }}||'"), 1,
|
||||
"CATALOGS.LABELS_SCHEMACOL emits a static CASE expression; "
|
||||
"both macro arguments are hardcoded (column name + gettext fn)."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'12_plus/search.sql',
|
||||
"'||{{ CATALOGS.LABELS_SCHEMACOL('sn.schema_name', _) }}||'"), 1,
|
||||
"CATALOGS.LABELS_SCHEMACOL emits a static CASE expression; "
|
||||
"both macro arguments are hardcoded (column name + gettext fn)."),
|
||||
(('pgadmin/tools/search_objects/templates/search_objects/sql/ppas/'
|
||||
'default/search.sql',
|
||||
"'||{{ CATALOGS.LABELS_SCHEMACOL('sn.schema_name', _) }}||'"), 1,
|
||||
"CATALOGS.LABELS_SCHEMACOL emits a static CASE expression; "
|
||||
"both macro arguments are hardcoded (column name + gettext fn)."),
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Catalog-label macros: gettext-translated display constants used
|
||||
# inside CASE WHEN ... THEN '<label>' expressions for the catalog
|
||||
# name column. The gettext key (English source string) is hardcoded
|
||||
# in the template; the rendered translation is a static literal.
|
||||
# ------------------------------------------------------------------
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/pg/macros/catalogs.sql',
|
||||
"'{{ _( 'ANSI' ) }} (information_schema)'"), 2,
|
||||
"Gettext-translated static display label for the ANSI catalog."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/pg/macros/catalogs.sql',
|
||||
"'{{ _( 'PostgreSQL Catalog' ) }} (pg_catalog)'"), 2,
|
||||
"Gettext-translated static display label for pg_catalog."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/pg/macros/catalogs.sql',
|
||||
"'{{ _( 'pgAgent Job Scheduler' ) }} (pgagent)'"), 2,
|
||||
"Gettext-translated static display label for pgAgent schema."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/ppas/macros/catalogs.sql',
|
||||
"'{{ _( 'ANSI' ) }} (information_schema)'"), 2,
|
||||
"Gettext-translated static display label for the ANSI catalog."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/ppas/macros/catalogs.sql',
|
||||
"'{{ _( 'PostgreSQL Catalog' ) }} (pg_catalog)'"), 2,
|
||||
"Gettext-translated static display label for pg_catalog."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/'
|
||||
'templates/catalog/ppas/macros/catalogs.sql',
|
||||
"'{{ _( 'pgAgent Job Scheduler' ) }} (pgagent)'"), 2,
|
||||
"Gettext-translated static display label for pgAgent schema."),
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# types.get_subtypes: ``opcintype`` is a pg_catalog OID joined from
|
||||
# pg_opclass / pg_amop and rendered as a space-separated pair to
|
||||
# match proargtypes (which uses a space-separated oidvector text
|
||||
# form). Not user-supplied.
|
||||
# ------------------------------------------------------------------
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/types/'
|
||||
'templates/types/pg/sql/default/get_subtypes.sql',
|
||||
"'{{opcintype}} {{opcintype}}'"), 1,
|
||||
"opcintype is a pg_catalog OID joined from pg_opclass; the "
|
||||
"literal is the space-separated proargtypes oidvector form."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/types/'
|
||||
'templates/types/ppas/sql/default/get_subtypes.sql',
|
||||
"'{{opcintype}} {{opcintype}}'"), 1,
|
||||
"opcintype is a pg_catalog OID joined from pg_opclass; the "
|
||||
"literal is the space-separated proargtypes oidvector form."),
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Partition CASE expression: pg_get_partkeydef(tid::oid) is the
|
||||
# value of the CASE branch, sandwiched between the literal 'p' and
|
||||
# the literal ''. The regex matches from 'p' through '' because
|
||||
# those quotes share a line, but the {{ tid }} interpolation is
|
||||
# OUTSIDE any literal — it's an argument to pg_get_partkeydef. The
|
||||
# tid value also passes a Flask ::int route converter, so it's an
|
||||
# integer regardless. Documented here so the lint stays
|
||||
# well-policed even with a regex that can't distinguish
|
||||
# "between two literals" from "inside one literal".
|
||||
# ------------------------------------------------------------------
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
|
||||
'templates/tables/sql/11_plus/properties.sql',
|
||||
"' THEN pg_catalog.pg_get_partkeydef({{ tid }}::oid) ELSE '"), 1,
|
||||
"{{ tid }} is the pg_class OID validated by Flask <int:tid> route "
|
||||
"converter; the surrounding 'p' and '' are SEPARATE literals, the "
|
||||
"regex spans two of them but the interpolation is outside any "
|
||||
"literal."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
|
||||
'templates/tables/sql/12_plus/properties.sql',
|
||||
"' THEN pg_catalog.pg_get_partkeydef({{ tid }}::oid) ELSE '"), 1,
|
||||
"{{ tid }} is the pg_class OID validated by Flask <int:tid> route "
|
||||
"converter; the surrounding 'p' and '' are SEPARATE literals, the "
|
||||
"regex spans two of them but the interpolation is outside any "
|
||||
"literal."),
|
||||
(('pgadmin/browser/server_groups/servers/databases/schemas/tables/'
|
||||
'templates/tables/sql/default/properties.sql',
|
||||
"' THEN pg_catalog.pg_get_partkeydef({{ tid }}::oid) ELSE '"), 1,
|
||||
"{{ tid }} is the pg_class OID validated by Flask <int:tid> route "
|
||||
"converter; the surrounding 'p' and '' are SEPARATE literals, the "
|
||||
"regex spans two of them but the interpolation is outside any "
|
||||
"literal."),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,345 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""Regression test: ``pgstattuple(...)`` and ``pgstatindex(...)`` in the
|
||||
stats templates must address their target via ``<oid>::oid::regclass``,
|
||||
NOT via a single-quoted ``'<schema>.<name>'`` string literal that
|
||||
embedded raw Jinja interpolations.
|
||||
|
||||
The earlier pattern (``pgstatindex('{{conn|qtIdent(schema)}}.{{conn|
|
||||
qtIdent(name)}}')`` / ``pgstattuple('{{schema_name}}.{{table_name}}')``)
|
||||
was vulnerable: a low-privilege user could create an object whose name
|
||||
contained an apostrophe — ``CREATE TABLE "foo'bar" (...)`` — and any
|
||||
viewer who then opened the stats panel would render an unbalanced SQL
|
||||
literal, breaking out of the ``'...'`` and executing whatever followed.
|
||||
|
||||
The fix replaces the embedded interpolation with an OID cast — the
|
||||
relation is identified by its numeric OID and the ``::regclass`` cast
|
||||
resolves it server-side, so no string literal is involved at all.
|
||||
That makes the bug structurally impossible, not just guarded against.
|
||||
|
||||
POSITIVE assertions:
|
||||
* The rendered stats SQL contains ``pgstattuple(<oid>::oid::regclass)``
|
||||
or ``pgstatindex(<oid>::oid::regclass)``.
|
||||
* An apostrophe planted in any identifier passed to the template
|
||||
does NOT appear inside a SQL literal context in the output (i.e.
|
||||
the template no longer embeds the identifier at all).
|
||||
|
||||
NEGATIVE assertions (regression guards):
|
||||
* The previously-vulnerable ``pgstattuple('...'``/``pgstatindex('...'``
|
||||
string-literal-call pattern must NOT recur.
|
||||
* Specifically the apostrophe payload must NOT survive verbatim
|
||||
inside any single-quoted region of the rendered SQL.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
|
||||
from flask import Flask, render_template
|
||||
from jinja2 import FileSystemLoader
|
||||
|
||||
from pgadmin.utils.driver import get_driver
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from config import PG_DEFAULT_DRIVER
|
||||
|
||||
|
||||
# Apostrophe-injection payload. Anywhere this appears INSIDE a single-
|
||||
# quoted literal in the rendered SQL is a SQLi.
|
||||
APOSTROPHE_NAME = "evil'or'1=1--"
|
||||
|
||||
|
||||
WEB_ROOT = os.path.realpath(
|
||||
os.path.join(os.path.dirname(os.path.realpath(__file__)),
|
||||
os.pardir, os.pardir, os.pardir, os.pardir, os.pardir,
|
||||
os.pardir)
|
||||
)
|
||||
|
||||
|
||||
class _FakeConn:
|
||||
"""qtIdent / qtLiteral use this as a stand-in for a real psycopg
|
||||
connection (see test_comment_description_sql_escaping.py for the
|
||||
full reasoning)."""
|
||||
|
||||
conn = None
|
||||
|
||||
def __bool__(self):
|
||||
return True
|
||||
|
||||
|
||||
def _abs(*parts):
|
||||
return os.path.join(WEB_ROOT, *parts)
|
||||
|
||||
|
||||
# Macro roots needed by the schemas templates' imports.
|
||||
MACRO_ROOTS = [
|
||||
_abs('pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'templates'),
|
||||
_abs('pgadmin', 'browser', 'server_groups', 'servers', 'templates'),
|
||||
]
|
||||
|
||||
|
||||
class _FakeApp(Flask):
|
||||
"""Minimal Flask app mirroring the production Jinja filters."""
|
||||
|
||||
def __init__(self, template_root):
|
||||
super().__init__('')
|
||||
driver = get_driver(PG_DEFAULT_DRIVER, self)
|
||||
self.jinja_env.filters['qtLiteral'] = driver.qtLiteral
|
||||
self.jinja_env.filters['qtIdent'] = driver.qtIdent
|
||||
self.jinja_env.filters['qtTypeIdent'] = driver.qtTypeIdent
|
||||
self.jinja_env.loader = FileSystemLoader(
|
||||
[template_root] + MACRO_ROOTS
|
||||
)
|
||||
|
||||
|
||||
# Regex: any single-quoted literal in the rendered SQL.
|
||||
SQL_LITERAL = re.compile(r"'[^'\n]*'")
|
||||
|
||||
|
||||
class StatsTemplateRegclassCastTestCase(BaseTestGenerator):
|
||||
"""Each stats template must address the relation via OID + regclass,
|
||||
not via an embedded ``'<schema>.<name>'`` literal."""
|
||||
|
||||
scenarios = [
|
||||
(
|
||||
'Table stats (16_plus) — pgstattuple(tid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='tables/sql/16_plus/stats.sql',
|
||||
ctx=dict(
|
||||
tid=12345,
|
||||
schema_name='public',
|
||||
table_name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstattuple(12345::oid::regclass)',
|
||||
forbid_call_re=r"pgstattuple\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Table stats (default) — pgstattuple(tid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='tables/sql/default/stats.sql',
|
||||
ctx=dict(
|
||||
tid=12345,
|
||||
schema_name='public',
|
||||
table_name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstattuple(12345::oid::regclass)',
|
||||
forbid_call_re=r"pgstattuple\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Materialised view stats (pg) — pgstattuple(vid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'views', 'templates'),
|
||||
template='mviews/pg/default/sql/stats.sql',
|
||||
ctx=dict(
|
||||
vid=67890,
|
||||
schema_name='public',
|
||||
mview_name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstattuple(67890::oid::regclass)',
|
||||
forbid_call_re=r"pgstattuple\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Materialised view stats (ppas) — pgstattuple(vid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'views', 'templates'),
|
||||
template='mviews/ppas/default/sql/stats.sql',
|
||||
ctx=dict(
|
||||
vid=67890,
|
||||
schema_name='public',
|
||||
mview_name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstattuple(67890::oid::regclass)',
|
||||
forbid_call_re=r"pgstattuple\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Exclusion constraint stats (16_plus) — '
|
||||
'pgstatindex(exid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='exclusion_constraint/sql/16_plus/stats.sql',
|
||||
ctx=dict(
|
||||
exid=11111,
|
||||
schema='public',
|
||||
name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(11111::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Exclusion constraint stats (default) — '
|
||||
'pgstatindex(exid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='exclusion_constraint/sql/default/stats.sql',
|
||||
ctx=dict(
|
||||
exid=11111,
|
||||
schema='public',
|
||||
name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(11111::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Index constraint stats (16_plus) — '
|
||||
'pgstatindex(cid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='index_constraint/sql/16_plus/stats.sql',
|
||||
ctx=dict(
|
||||
cid=22222,
|
||||
schema='public',
|
||||
name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(22222::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Index constraint stats (default) — '
|
||||
'pgstatindex(cid::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='index_constraint/sql/default/stats.sql',
|
||||
ctx=dict(
|
||||
cid=22222,
|
||||
schema='public',
|
||||
name=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(22222::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Index stats (16_plus) — pgstatindex(idx::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='indexes/sql/16_plus/stats.sql',
|
||||
ctx=dict(
|
||||
idx=33333,
|
||||
schema='public',
|
||||
index=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(33333::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
(
|
||||
'Index stats (default) — pgstatindex(idx::oid::regclass)',
|
||||
dict(
|
||||
template_root=_abs(
|
||||
'pgadmin', 'browser', 'server_groups', 'servers',
|
||||
'databases', 'schemas', 'tables', 'templates'),
|
||||
template='indexes/sql/default/stats.sql',
|
||||
ctx=dict(
|
||||
idx=33333,
|
||||
schema='public',
|
||||
index=APOSTROPHE_NAME,
|
||||
is_pgstattuple=True,
|
||||
conn=_FakeConn(),
|
||||
),
|
||||
expected_call='pgstatindex(33333::oid::regclass)',
|
||||
forbid_call_re=r"pgstatindex\('",
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
def setUp(self):
|
||||
self.app_under_test = _FakeApp(self.template_root)
|
||||
|
||||
def runTest(self):
|
||||
# The stats templates use `_('...')` for column aliases. Inject
|
||||
# an identity gettext stand-in so the templates render without
|
||||
# needing a real flask-babel context (we don't care about
|
||||
# localisation here; we only assert on the pgstat* call shape).
|
||||
ctx = dict(self.ctx, _=lambda s: s)
|
||||
with self.app_under_test.app_context():
|
||||
rendered = render_template(self.template, **ctx)
|
||||
|
||||
# POSITIVE: the regclass-cast call is present.
|
||||
self.assertIn(
|
||||
self.expected_call, rendered,
|
||||
msg=(
|
||||
'Stats template should address the relation via '
|
||||
'OID::oid::regclass; expected {!r} in rendered SQL.\n'
|
||||
'Rendered:\n{}'.format(self.expected_call, rendered)
|
||||
),
|
||||
)
|
||||
|
||||
# NEGATIVE: the previously-vulnerable string-literal call form
|
||||
# must NOT recur.
|
||||
self.assertIsNone(
|
||||
re.search(self.forbid_call_re, rendered),
|
||||
msg=(
|
||||
'Stats template must not call pgstat* with a '
|
||||
"single-quoted relation name (the previously-vulnerable "
|
||||
'pattern). Pattern: {!r}\nRendered:\n{}'
|
||||
.format(self.forbid_call_re, rendered)
|
||||
),
|
||||
)
|
||||
|
||||
# NEGATIVE: the apostrophe payload, if it appears at all, must
|
||||
# not appear *inside* a single-quoted literal region — that
|
||||
# would indicate the identifier is being embedded somewhere
|
||||
# else in the template.
|
||||
for literal in SQL_LITERAL.findall(rendered):
|
||||
self.assertNotIn(
|
||||
APOSTROPHE_NAME, literal,
|
||||
msg=(
|
||||
'Apostrophe-bearing identifier payload appeared '
|
||||
'inside a SQL literal in the rendered stats SQL — '
|
||||
'indicates the identifier is still being embedded '
|
||||
'in some literal context.\n'
|
||||
'Offending literal: {!r}\nRendered:\n{}'
|
||||
.format(literal, rendered)
|
||||
),
|
||||
)
|
||||
@@ -318,14 +318,21 @@ class Driver(BaseDriver):
|
||||
"from Python, pass the connection as the second argument."
|
||||
)
|
||||
|
||||
res = value
|
||||
try:
|
||||
if not isinstance(conn, psycopg.Connection) and \
|
||||
not isinstance(conn, psycopg.AsyncConnection):
|
||||
conn = conn.conn
|
||||
res = psycopg.sql.Literal(value).as_string(conn).strip()
|
||||
except Exception:
|
||||
print("Exception", value)
|
||||
# Resolve the underlying psycopg connection. A wrapped pgAdmin
|
||||
# Connection has the raw psycopg connection on `.conn`.
|
||||
if not isinstance(conn, psycopg.Connection) and \
|
||||
not isinstance(conn, psycopg.AsyncConnection):
|
||||
conn = conn.conn
|
||||
|
||||
# psycopg.sql.Literal.as_string can raise for values it cannot
|
||||
# adapt (custom types without a registered adapter, byte
|
||||
# sequences in a non-UTF8 client encoding, etc.). The previous
|
||||
# implementation swallowed the exception and silently returned
|
||||
# the raw, unescaped `value` — the same SQL-injection sink the
|
||||
# missing-conn fast-fail above guards against, but on a
|
||||
# different failure mode. Letting the exception propagate makes
|
||||
# any unadaptable input loud and callers can react explicitly.
|
||||
res = psycopg.sql.Literal(value).as_string(conn).strip()
|
||||
|
||||
if force_quote is True:
|
||||
# Convert the input to the string to use the startsWith(...)
|
||||
|
||||
81
web/pgadmin/utils/tests/test_qtliteral_no_silent_unescape.py
Normal file
81
web/pgadmin/utils/tests/test_qtliteral_no_silent_unescape.py
Normal file
@@ -0,0 +1,81 @@
|
||||
##########################################################################
|
||||
#
|
||||
# pgAdmin 4 - PostgreSQL Tools
|
||||
#
|
||||
# Copyright (C) 2013 - 2026, The pgAdmin Development Team
|
||||
# This software is released under the PostgreSQL Licence
|
||||
#
|
||||
##########################################################################
|
||||
|
||||
"""Regression test: ``qtLiteral`` must not silently degrade to the raw,
|
||||
unescaped input value when the inner ``psycopg.sql.Literal(...)
|
||||
.as_string(conn)`` call raises.
|
||||
|
||||
Historical behaviour was:
|
||||
|
||||
res = value # raw input held as the default
|
||||
try:
|
||||
...
|
||||
res = Literal(value).as_string(conn).strip()
|
||||
except Exception:
|
||||
print(...) # swallowed — `res` stays as the raw value
|
||||
|
||||
That is the same SQL-injection sink the missing-``conn`` fail-fast
|
||||
guards against, just gated on a different failure mode (psycopg
|
||||
inability to adapt a value rather than absence of a connection). This
|
||||
test exercises the inner-except path with an object that ``psycopg.sql
|
||||
.Literal`` cannot adapt, and asserts the call raises rather than
|
||||
returning the raw object.
|
||||
"""
|
||||
|
||||
from pgadmin.utils.driver import get_driver
|
||||
from pgadmin.utils.route import BaseTestGenerator
|
||||
from config import PG_DEFAULT_DRIVER
|
||||
|
||||
|
||||
class _Unadaptable:
|
||||
"""Custom object psycopg has no registered adapter for. The
|
||||
``psycopg.sql.Literal(...).as_string(conn)`` call must raise on
|
||||
this — it has no idea how to render the value as a SQL literal."""
|
||||
|
||||
def __repr__(self):
|
||||
return '<_Unadaptable>'
|
||||
|
||||
|
||||
class _MinimalConn:
|
||||
"""Stand-in for a real psycopg Connection. The driver's qtLiteral
|
||||
sees this as truthy (so the missing-conn fail-fast is bypassed)
|
||||
and dereferences `.conn` to the psycopg-side connection. We set
|
||||
that to None to ensure the inner Literal.as_string call raises
|
||||
(psycopg requires a real context to render adapted SQL)."""
|
||||
|
||||
conn = None
|
||||
|
||||
def __bool__(self):
|
||||
return True
|
||||
|
||||
|
||||
class QtLiteralRaisesOnUnadaptableValueTestCase(BaseTestGenerator):
|
||||
"""The inner ``psycopg.sql.Literal(...).as_string(conn)`` failure
|
||||
path must propagate — not silently return the raw value."""
|
||||
|
||||
scenarios = [
|
||||
('Unadaptable Python object surfaces as an exception',
|
||||
dict(value=_Unadaptable())),
|
||||
]
|
||||
|
||||
def runTest(self):
|
||||
driver = get_driver(PG_DEFAULT_DRIVER)
|
||||
with self.assertRaises(Exception) as ctx:
|
||||
driver.qtLiteral(self.value, _MinimalConn())
|
||||
|
||||
# And specifically: the original value must NOT be returned —
|
||||
# if it were, the exception would not have been raised.
|
||||
# The ``with self.assertRaises`` block guarantees a raise; this
|
||||
# assertion documents the contract that the call site relies
|
||||
# on (no silent unescaped passthrough).
|
||||
self.assertNotIsInstance(
|
||||
ctx.exception, type(None),
|
||||
msg='qtLiteral must raise on unadaptable input, not '
|
||||
'silently return it.'
|
||||
)
|
||||
Reference in New Issue
Block a user