Scrub user mapping secrets in-place via ProcessUtility handler

When CREATE USER MAPPING or ALTER USER MAPPING targets an
iceberg_catalog server, overwrite client_id and client_secret values
with asterisks directly in the queryString buffer.  This prevents
credential leakage in pg_stat_statements and similar.

Why in-place mutation instead of a copy:

  pg_stat_statements captures the queryString pointer before calling
  prev_ProcessUtility, then passes the same local pointer to pgss_store
  after the call returns.  A pstrdup copy would be invisible to it.
  By overwriting the original buffer, every consumer sharing that pointer
  - like pg_stat_statements - sees the scrubbed version automatically.

  DDL execution is unaffected because CREATE/ALTER USER MAPPING reads
  option values from DefElem nodes in the parse tree, not queryString.

Why RegisterUtilityStatementHandler instead of ProcessUtility_hook:

  pg_lake_iceberg.so is preloaded by pg_extension_base during
  shared_preload_libraries (via pg_lake_table.control), which runs
  before pg_stat_statements installs its hook.  A hook installed in
  _PG_init would therefore sit below pg_stat_statements in the chain,
  defeating the purpose.  Using pg_lake_engine's handler mechanism
  (which also runs below pg_stat_statements) is fine because the
  in-place mutation affects the shared buffer regardless of call order.

Implementation:

- iceberg_catalog_secret_options: separates secret names (client_id,
  client_secret) from the full user mapping option list so scope
  remains visible.
- ScrubUserMappingSecrets: walks DefElem options, uses DefElem.location
  to find each secret quoted value in the query string, and overwrites
  value characters with * while preserving quotes and string length.
- ScrubIcebergUserMappingHandler: detects user mapping DDL targeting
  iceberg_catalog servers, calls ScrubUserMappingSecrets, returns false
  so normal processing continues.

Signed-off-by: sfc-gh-npuka <naisila.puka@snowflake.com>

Author: sfc-gh-npuka

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -21,6 +21,7 @@
 #include "pg_lake/http/http_client.h"
 #include "pg_lake/util/rel_utils.h"
 #include "pg_lake/parquet/field.h"
+#include "pg_lake/ddl/utility_hook.h"
 #include "pg_lake/iceberg/api/snapshot.h"
 
 #define REST_CATALOG_AUTH_TYPE_DEFAULT (0)
@@ -125,3 +126,6 @@ extern PGDLLEXPORT RestCatalogRequest * GetSetCurrentSchemaCatalogRequest(Oid re
 extern PGDLLEXPORT RestCatalogRequest * GetAddPartitionCatalogRequest(Oid relationId, List *partitionSpec);
 extern PGDLLEXPORT RestCatalogRequest * GetSetPartitionDefaultIdCatalogRequest(Oid relationId, int specId);
 extern PGDLLEXPORT RestCatalogRequest * GetRemoveSnapshotCatalogRequest(List *removedSnapshotIds, Oid relationId);
+
+/* ProcessUtility handler: scrubs secrets from user mapping DDL in-place */
+extern PGDLLEXPORT bool ScrubIcebergUserMappingHandler(ProcessUtilityParams *processUtilityParams, void *arg);

pg_lake_iceberg/src/init.c

@@ -317,6 +317,8 @@ _PG_init(void)
 							 NULL, NULL, NULL);
 
 	AvroInit();
+
+	RegisterUtilityStatementHandler(ScrubIcebergUserMappingHandler, NULL);
 }
 
 

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -39,7 +39,10 @@
 #include "utils/syscache.h"
 #include "utils/timestamp.h"
 
+#include "nodes/parsenodes.h"
+
 #include "pg_extension_base/base_workers.h"
+#include "pg_lake/ddl/utility_hook.h"
 #include "pg_lake/http/http_client.h"
 #include "pg_lake/iceberg/api/table_schema.h"
 #include "pg_lake/iceberg/catalog.h"
@@ -110,6 +113,7 @@ static const char *iceberg_catalog_user_mapping_options[] = {
 };
 
 
+
 static bool
 is_valid_option_in_list(const char *keyword, const char *const *options)
 {
@@ -202,6 +206,130 @@ iceberg_catalog_validator(PG_FUNCTION_ARGS)
 }
 
 
+/*
+ * IsIcebergCatalogServer returns true if the named server exists and uses
+ * the iceberg_catalog FDW.  Returns false (without error) when the server
+ * does not exist yet (e.g. during CREATE SERVER ... CREATE USER MAPPING
+ * inside the same transaction before the server is committed).
+ */
+static bool
+IsIcebergCatalogServer(const char *serverName)
+{
+	ForeignServer *server = GetForeignServerByName(serverName, true);
+
+	if (server == NULL)
+		return false;
+
+	ForeignDataWrapper *fdw = GetForeignDataWrapper(server->fdwid);
+
+	return strcmp(fdw->fdwname, ICEBERG_CATALOG_FDW_NAME) == 0;
+}
+
+
+/*
+ * ScrubUserMappingSecrets overwrites secret option values in the query
+ * string in-place with asterisks.
+ *
+ * In-place mutation is essential: pg_stat_statements captures the queryString
+ * pointer before calling prev_ProcessUtility, then stores it after the call
+ * returns. A copy with pstrdup would be invisible to pg_stat_statements
+ * because its local pointer still references the original memory. By
+ * overwriting the original buffer, every holder of that pointer — including
+ * pg_stat_statements — sees the scrubbed version.
+ *
+ * The actual DDL execution is unaffected because CREATE/ALTER USER MAPPING
+ * reads option values from the parse tree (DefElem nodes), not from
+ * queryString.
+ */
+static void
+ScrubUserMappingSecrets(const char *queryString, List *options)
+{
+	const char *secret_options[] = {"client_id", "client_secret", NULL};
+	ListCell   *lc;
+
+	foreach(lc, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(lc);
+
+		if (!is_valid_option_in_list(def->defname, secret_options))
+			continue;
+
+		if (def->location < 0)
+			continue;
+
+		char	   *p = (char *) queryString + def->location;
+
+		/* skip past the key name */
+		while (*p && !isspace((unsigned char) *p) && *p != '\'')
+			p++;
+
+		/* skip whitespace between key and opening quote */
+		while (*p && *p != '\'')
+			p++;
+
+		if (*p != '\'')
+			continue;
+
+		p++;					/* skip opening quote */
+
+		/* overwrite value characters with '*', handling '' escapes */
+		while (*p && *p != '\'')
+			*p++ = '*';
+		while (*(p + 1) == '\'')
+		{
+			*p++ = '*';			/* first quote of '' pair */
+			*p++ = '*';			/* second quote of '' pair */
+			while (*p && *p != '\'')
+				*p++ = '*';
+		}
+	}
+}
+
+
+/*
+ * ScrubIcebergUserMappingHandler is a ProcessUtility handler registered via
+ * pg_lake_engine's RegisterUtilityStatementHandler.  When it detects a
+ * CREATE/ALTER USER MAPPING targeting an iceberg_catalog server, it scrubs
+ * secret values in the queryString in-place and returns false so normal
+ * processing continues.
+ */
+bool
+ScrubIcebergUserMappingHandler(ProcessUtilityParams *processUtilityParams,
+							   void *arg)
+{
+	Node	   *parsetree = processUtilityParams->plannedStmt->utilityStmt;
+	const char *serverName = NULL;
+	List	   *options = NIL;
+
+	if (IsA(parsetree, CreateUserMappingStmt))
+	{
+		CreateUserMappingStmt *stmt = (CreateUserMappingStmt *) parsetree;
+
+		serverName = stmt->servername;
+		options = stmt->options;
+	}
+	else if (IsA(parsetree, AlterUserMappingStmt))
+	{
+		AlterUserMappingStmt *stmt = (AlterUserMappingStmt *) parsetree;
+
+		serverName = stmt->servername;
+		options = stmt->options;
+	}
+	else
+		return false;
+
+	if (serverName == NULL || options == NIL)
+		return false;
+
+	if (!IsIcebergCatalogServer(serverName))
+		return false;
+
+	ScrubUserMappingSecrets(processUtilityParams->queryString, options);
+
+	return false;
+}
+
+
 /*
  * GetRestCatalogConnectionFromGUCs returns a RestCatalogConnectionInfo
  * populated from the current GUC variables. Used for backward-compatible

pg_lake_table/tests/pytests/test_iceberg_catalog_server.py

@@ -809,3 +809,204 @@ def test_catalog_object_store_literal_still_works(
         pg_conn,
     )
     pg_conn.rollback()
+
+
+# ── Query string scrubbing for user mapping DDL ───────────────────────────
+
+
+def test_scrub_create_user_mapping_in_pg_stat_statements(
+    installcheck, superuser_conn, extension
+):
+    """CREATE USER MAPPING secrets should be scrubbed in pg_stat_statements."""
+    if installcheck:
+        return
+
+    run_command("CREATE EXTENSION IF NOT EXISTS pg_stat_statements", superuser_conn)
+    run_command("SELECT pg_stat_statements_reset()", superuser_conn)
+    superuser_conn.commit()
+
+    run_command(
+        """
+        CREATE SERVER test_scrub_srv TYPE 'rest'
+            FOREIGN DATA WRAPPER iceberg_catalog
+            OPTIONS (rest_endpoint 'http://localhost:8181')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    run_command(
+        """
+        CREATE USER MAPPING FOR CURRENT_USER SERVER test_scrub_srv
+            OPTIONS (client_id 'secret_id_value', client_secret 'secret_key_value')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    result = run_query(
+        """
+        SELECT query FROM pg_stat_statements
+        WHERE query ILIKE '%CREATE USER MAPPING%test_scrub_srv%'
+        """,
+        superuser_conn,
+    )
+
+    assert len(result) >= 1
+    query_text = result[0]["query"]
+    assert "secret_id_value" not in query_text
+    assert "secret_key_value" not in query_text
+    assert "'***" in query_text
+
+    run_command(
+        "DROP USER MAPPING FOR CURRENT_USER SERVER test_scrub_srv", superuser_conn
+    )
+    run_command("DROP SERVER test_scrub_srv", superuser_conn)
+    superuser_conn.commit()
+
+
+def test_scrub_alter_user_mapping_in_pg_stat_statements(
+    installcheck, superuser_conn, extension
+):
+    """ALTER USER MAPPING secrets should also be scrubbed in pg_stat_statements."""
+    if installcheck:
+        return
+
+    run_command("CREATE EXTENSION IF NOT EXISTS pg_stat_statements", superuser_conn)
+    run_command("SELECT pg_stat_statements_reset()", superuser_conn)
+    superuser_conn.commit()
+
+    run_command(
+        """
+        CREATE SERVER test_scrub_alter_srv TYPE 'rest'
+            FOREIGN DATA WRAPPER iceberg_catalog
+            OPTIONS (rest_endpoint 'http://localhost:8181')
+        """,
+        superuser_conn,
+    )
+    run_command(
+        """
+        CREATE USER MAPPING FOR CURRENT_USER SERVER test_scrub_alter_srv
+            OPTIONS (client_id 'old_id', client_secret 'old_secret')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    run_command("SELECT pg_stat_statements_reset()", superuser_conn)
+    superuser_conn.commit()
+
+    run_command(
+        """
+        ALTER USER MAPPING FOR CURRENT_USER SERVER test_scrub_alter_srv
+            OPTIONS (SET client_id 'new_secret_id', SET client_secret 'new_secret_key')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    result = run_query(
+        """
+        SELECT query FROM pg_stat_statements
+        WHERE query ILIKE '%ALTER USER MAPPING%test_scrub_alter_srv%'
+        """,
+        superuser_conn,
+    )
+
+    assert len(result) >= 1
+    query_text = result[0]["query"]
+    assert "new_secret_id" not in query_text
+    assert "new_secret_key" not in query_text
+    assert "'***" in query_text
+
+    run_command(
+        "DROP USER MAPPING FOR CURRENT_USER SERVER test_scrub_alter_srv",
+        superuser_conn,
+    )
+    run_command("DROP SERVER test_scrub_alter_srv", superuser_conn)
+    superuser_conn.commit()
+
+
+def test_scrub_preserves_actual_credentials(superuser_conn, extension):
+    """Scrubbing the query string should not affect the stored credentials."""
+    run_command(
+        """
+        CREATE SERVER test_scrub_creds_srv TYPE 'rest'
+            FOREIGN DATA WRAPPER iceberg_catalog
+            OPTIONS (rest_endpoint 'http://localhost:8181')
+        """,
+        superuser_conn,
+    )
+    run_command(
+        """
+        CREATE USER MAPPING FOR CURRENT_USER SERVER test_scrub_creds_srv
+            OPTIONS (client_id 'real_id', client_secret 'real_secret')
+        """,
+        superuser_conn,
+    )
+
+    result = run_query(
+        """
+        SELECT umoptions FROM pg_user_mapping um
+            JOIN pg_foreign_server fs ON um.umserver = fs.oid
+        WHERE fs.srvname = 'test_scrub_creds_srv'
+        """,
+        superuser_conn,
+    )
+
+    assert len(result) == 1
+    opts = result[0]["umoptions"]
+    assert "client_id=real_id" in opts
+    assert "client_secret=real_secret" in opts
+
+    superuser_conn.rollback()
+
+
+def test_scrub_leaves_scope_visible(installcheck, superuser_conn, extension):
+    """scope is not a secret — it should remain visible after scrubbing."""
+    if installcheck:
+        return
+
+    run_command("CREATE EXTENSION IF NOT EXISTS pg_stat_statements", superuser_conn)
+    run_command("SELECT pg_stat_statements_reset()", superuser_conn)
+    superuser_conn.commit()
+
+    run_command(
+        """
+        CREATE SERVER test_scrub_scope_srv TYPE 'rest'
+            FOREIGN DATA WRAPPER iceberg_catalog
+            OPTIONS (rest_endpoint 'http://localhost:8181')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    run_command(
+        """
+        CREATE USER MAPPING FOR CURRENT_USER SERVER test_scrub_scope_srv
+            OPTIONS (client_id 'id123', client_secret 'secret456', scope 'PRINCIPAL_ROLE:ALL')
+        """,
+        superuser_conn,
+    )
+    superuser_conn.commit()
+
+    result = run_query(
+        """
+        SELECT query FROM pg_stat_statements
+        WHERE query ILIKE '%CREATE USER MAPPING%test_scrub_scope_srv%'
+        """,
+        superuser_conn,
+    )
+
+    assert len(result) >= 1
+    query_text = result[0]["query"]
+    assert "id123" not in query_text
+    assert "secret456" not in query_text
+    assert "PRINCIPAL_ROLE:ALL" in query_text
+
+    run_command(
+        "DROP USER MAPPING FOR CURRENT_USER SERVER test_scrub_scope_srv",
+        superuser_conn,
+    )
+    run_command("DROP SERVER test_scrub_scope_srv", superuser_conn)
+    superuser_conn.commit()