fix: Prevent OS command injection in Import/Export query export (CWE-78) (#9899)

User-supplied input was interpolated directly into a psql \copy
metacommand template without sanitization. An authenticated pgAdmin
user could inject ") TO PROGRAM 'cmd'" to achieve RCE on the pgAdmin
server, or ") TO '/path'" for arbitrary file write, by breaking out
of the \copy (...) context.

Add _is_query_parens_balanced() which tracks parenthesis depth in the
query, modeled on psql's strtokx tokenizer used by parse_slash_copy.
Only single-quoted strings ('...' with '' and \' escaping) and
double-quoted identifiers ("..." with "" escaping) are recognized;
line/block comments and dollar-quoting are deliberately NOT skipped
because psql's \copy parser does not recognize them either, so any
unbalanced ')' inside those constructs must remain visible to the
validator.

Whitelist-validate format, on_error, and log_verbosity which were
also raw-interpolated, and write the normalized lowercase value back
to data so downstream template equality checks (e.g. data.format ==
'csv') match regardless of input case.

Reject queries containing null bytes (which could cause a C-string
truncation mismatch between Python validation and psql execution)
and normalize \r\n / \r / \n to spaces (required for Windows; also
prevents psql metacommand termination).

Type-check 'query' before string operations so malformed payloads
return a clean 400 instead of a 500. Tighten is_query_export gating
to 'is True' in both update_data_for_import_export and the validation
block so a client cannot exploit a truthy-but-not-True value to
bypass one check while satisfying the other.

Add unit tests covering the parens parser (balanced cases, RCE
attempts, escape edge cases) and route-level tests asserting
BatchProcess is not invoked for malicious payloads.

Reported-by: chungkn (Chung Kim), OneMount Group <kimngocchung.k2a@gmail.com>

Author: asheshv

web/pgadmin/tools/import_export/__init__.py

@@ -258,14 +258,101 @@ def _save_import_export_settings(settings):
     store_setting('import_export_setting', settings)
 
 
+# Whitelisted values for fields raw-interpolated into the \copy
+# template. Defined at module level to avoid rebuilding per request.
+_VALID_FORMATS = frozenset({'csv', 'text', 'binary'})
+_VALID_ON_ERROR = frozenset({'stop', 'ignore'})
+_VALID_LOG_VERBOSITY = frozenset({'default', 'verbose'})
+
+
+def _is_query_parens_balanced(query):
+    """
+    Check that parentheses are balanced in the query, matching the
+    behavior of psql's \\copy (...) parser (strtokx in
+    parse_slash_copy). Returns False if depth goes negative, which
+    would allow breaking out of the \\copy (...) context.
+
+    IMPORTANT: This parser must be at least as restrictive as psql's
+    strtokx in parse_slash_copy. psql only recognizes:
+      - Single-quoted strings ('...' with '' and \\' escaping)
+      - Double-quoted identifiers ("..." with "" escaping)
+      - Parentheses as nesting delimiters
+    It does NOT recognize --, /* */, or $tag$...$tag$. We must not
+    skip content inside those constructs either, or an attacker can
+    hide an unbalanced ) inside them to bypass validation.
+
+    Backslash escapes (\\x) are always handled inside single-quoted
+    strings. psql enables this for E'...' strings unconditionally
+    and for all strings when standard_conforming_strings=off. By
+    always handling backslashes, we match psql in all modes. This
+    may reject some valid queries (e.g. '\\') with scs=on) but
+    will never accept a dangerous one.
+    """
+    depth = 0
+    i = 0
+    length = len(query)
+
+    while i < length:
+        ch = query[i]
+
+        # Single-quoted string literal: skip to closing quote.
+        # Handles both '' and \' escape sequences to match psql's
+        # strtokx which enables backslash escaping for E-strings
+        # (always) and all strings when scs=off.
+        if ch == "'":
+            i += 1
+            while i < length:
+                if query[i] == '\\':
+                    i += 2  # backslash escape: skip next char
+                elif query[i] == "'":
+                    if i + 1 < length and query[i + 1] == "'":
+                        i += 2  # '' escape
+                    else:
+                        break
+                else:
+                    i += 1
+            if i >= length:
+                return False  # Unterminated string literal
+            i += 1
+            continue
+
+        # Double-quoted identifier: skip to closing quote
+        # (handles "" escape sequences — matches strtokx behavior)
+        if ch == '"':
+            i += 1
+            while i < length:
+                if query[i] == '"':
+                    if i + 1 < length and query[i + 1] == '"':
+                        i += 2  # escaped quote
+                    else:
+                        break
+                else:
+                    i += 1
+            if i >= length:
+                return False  # Unterminated identifier
+            i += 1
+            continue
+
+        if ch == '(':
+            depth += 1
+        elif ch == ')':
+            depth -= 1
+            if depth < 0:
+                return False
+
+        i += 1
+
+    return depth >= 0
+
+
 def update_data_for_import_export(data):
     """
     This function will update the data. Remove unwanted keys based on
     import/export type
     """
     keys_not_required_with_query = ['on_error', 'log_verbosity', 'freeze',
                                     'default_string']
-    if 'is_query_export' in data and data['is_query_export']:
+    if data.get('is_query_export') is True:
         for k in keys_not_required_with_query:
             data.pop(k, None)
         data['is_import'] = False
@@ -339,6 +426,41 @@ def create_import_export_job(sid):
     else:
         return bad_request(errormsg=_('Please specify a valid file.'))
 
+    # Validate fields that are interpolated raw into the \copy template
+    # to prevent metacommand injection via these parameters. Normalize
+    # to lowercase so downstream template equality checks
+    # (e.g. data.format == 'csv') match regardless of input case.
+    fmt = data.get('format', '')
+    if not isinstance(fmt, str):
+        return bad_request(errormsg=_('Format must be a string.'))
+    if fmt.lower() not in _VALID_FORMATS:
+        return bad_request(errormsg=_('Invalid format specified.'))
+    data['format'] = fmt.lower()
+
+    if 'on_error' in data:
+        val = data['on_error']
+        if not isinstance(val, str):
+            return bad_request(
+                errormsg=_('on_error must be a string.')
+            )
+        if val.lower() not in _VALID_ON_ERROR:
+            return bad_request(
+                errormsg=_('Invalid on_error value.')
+            )
+        data['on_error'] = val.lower()
+
+    if 'log_verbosity' in data:
+        val = data['log_verbosity']
+        if not isinstance(val, str):
+            return bad_request(
+                errormsg=_('log_verbosity must be a string.')
+            )
+        if val.lower() not in _VALID_LOG_VERBOSITY:
+            return bad_request(
+                errormsg=_('Invalid log_verbosity value.')
+            )
+        data['log_verbosity'] = val.lower()
+
     # Get required and other columns list
     cols = _get_formatted_column_list(data, 'columns', driver, conn)
     not_null_cols = _get_formatted_column_list(data, NOT_NULL_COLUMNS,
@@ -353,9 +475,34 @@ def create_import_export_job(sid):
     # Remove unwanted keys from data
     update_data_for_import_export(data)
 
-    # Replace "\n" with " " as it is not working on Windows.
-    if 'query' in data:
-        data['query'] = data['query'].replace("\n", " ")
+    # Validate the query when this is a query-based export. The raw
+    # query is interpolated into psql's \copy (...) template, so we
+    # must ensure it cannot break out of that context (CWE-78).
+    if data.get('is_query_export') is True:
+        query = data.get('query')
+        if not isinstance(query, str) or not query.strip():
+            return bad_request(
+                errormsg=_('Query is required for query-based export.')
+            )
+        # Reject null bytes which could cause C-string truncation
+        # mismatch between Python validation and psql execution.
+        if '\x00' in query:
+            return bad_request(
+                errormsg=_('The query contains invalid characters.')
+            )
+        # Replace line breaks with space (required for Windows; also
+        # prevents psql metacommand termination on \r or \n).
+        query = query.replace('\r\n', ' ') \
+            .replace('\r', ' ').replace('\n', ' ')
+        # Validate that parentheses are balanced. An unbalanced ')'
+        # would close the \copy (...) context and allow injection of
+        # arbitrary metacommand syntax (e.g. ") TO PROGRAM 'cmd'"
+        # for RCE).
+        if not _is_query_parens_balanced(query):
+            return bad_request(
+                errormsg=_('The query contains unbalanced parentheses.')
+            )
+        data['query'] = query
 
     # Create the COPY FROM/TO  from template
     temp_path = 'import_export/sql/#{0}#/cmd.sql'.format(manager.version)