Filter internal __pgsqlite_* tables from client sqlite_master queries

src/catalog/query_interceptor.rs

@@ -18,6 +18,53 @@ use std::collections::HashMap;
 /// Type alias for the complex Future type returned by process_expression
 type ProcessExpressionFuture<'a> = Pin<Box<dyn Future<Output = Result<(), Box<dyn std::error::Error + Send + Sync>>> + Send + 'a>>;
 
+/// The pg_catalog tables that pgsqlite physically materializes as real SQLite
+/// tables. Source of truth: `src/migration/registry.rs`, which is the only place
+/// these are `CREATE TABLE`d. Every other `pg_*` / `information_schema_*` name is
+/// a function or an interceptor-synthesized virtual result, so only these four
+/// show up in a raw `sqlite_master` scan. They are pgsqlite-reserved names a user
+/// cannot own, and this is a closed set: filter them out of client-issued
+/// `sqlite_master` / `sqlite_schema` queries alongside the `__pgsqlite_*`
+/// bookkeeping tables. Keep this in step with the registry.
+const PG_CATALOG_INTERNAL_TABLES: [&str; 4] =
+    ["pg_attrdef", "pg_constraint", "pg_depend", "pg_index"];
+
+/// The pg_catalog / information_schema compatibility views pgsqlite materializes
+/// as real SQLite views. Source of truth: `src/migration/registry.rs`, the only
+/// place these are `CREATE VIEW`d. Like the tables above, every one is a
+/// pgsqlite-reserved name a user cannot own, so this is a closed, exact-name set
+/// (never a `pg_%` / `information_schema_%` wildcard, which would also hide a
+/// user's own `pg_my_report` view). Filter them out of client-issued
+/// `sqlite_master` / `sqlite_schema` queries so a raw `type = 'view'` scan (or an
+/// unqualified `SELECT *`) shows only the user's own schema. Keep this in step
+/// with the registry.
+const PG_CATALOG_INTERNAL_VIEWS: [&str; 24] = [
+    "information_schema_columns",
+    "information_schema_key_column_usage",
+    "information_schema_referential_constraints",
+    "information_schema_schemata",
+    "information_schema_table_constraints",
+    "information_schema_tables",
+    "pg_am",
+    "pg_attribute",
+    "pg_class",
+    "pg_database",
+    "pg_description",
+    "pg_enum",
+    "pg_foreign_data_wrapper",
+    "pg_namespace",
+    "pg_proc",
+    "pg_roles",
+    "pg_stat_activity",
+    "pg_stat_all_indexes",
+    "pg_stat_all_tables",
+    "pg_stat_database",
+    "pg_stat_user_indexes",
+    "pg_stat_user_tables",
+    "pg_type",
+    "pg_user",
+];
+
 /// Intercepts and handles queries to pg_catalog tables
 pub struct CatalogInterceptor;
 
@@ -46,7 +93,21 @@ impl CatalogInterceptor {
            lower_query.trim() == "select version()" {
             return None;
         }
-
+
+        // Filter pgsqlite's internal bookkeeping tables (__pgsqlite_*) out of
+        // client-issued sqlite_master / sqlite_schema queries. pgsqlite's own
+        // synthesized catalog queries already exclude these tables, but a query
+        // written by the client is otherwise passed through verbatim and leaks
+        // the internal metadata tables. This runs before the pg_catalog /
+        // information_schema gate below because a sqlite_master query contains
+        // none of those substrings. Fails open (returns None) on any parse
+        // failure or when sqlite_master is not an actual FROM relation.
+        if lower_query.contains("sqlite_master") || lower_query.contains("sqlite_schema") {
+            if let Some(result) = Self::intercept_sqlite_master_query(query, &db).await {
+                return Some(result);
+            }
+        }
+
         // Check for catalog tables
         let has_catalog_tables = lower_query.contains("pg_catalog") || lower_query.contains("pg_type") ||
            lower_query.contains("pg_namespace") || lower_query.contains("pg_range") ||
@@ -208,6 +269,128 @@ impl CatalogInterceptor {
         None
     }
 
+    /// Rewrite a client-issued `sqlite_master` / `sqlite_schema` query so that
+    /// pgsqlite's internal `__pgsqlite_*` bookkeeping tables are excluded, then
+    /// execute it. Returns `None` (letting the original query run unmodified) if
+    /// the query cannot be parsed as a single SELECT or does not actually select
+    /// from a `sqlite_master` relation, so this fails open in every ambiguous
+    /// case.
+    async fn intercept_sqlite_master_query(query: &str, db: &Arc<DbHandler>) -> Option<Result<DbResponse, PgSqliteError>> {
+        let rewritten = Self::rewrite_sqlite_master_query(query)?;
+        match db.query(&rewritten).await {
+            Ok(response) => Some(Ok(response)),
+            Err(e) => Some(Err(PgSqliteError::Sqlite(e))),
+        }
+    }
+
+    /// Rewrite a `sqlite_master` / `sqlite_schema` SELECT so that pgsqlite's
+    /// internal `__pgsqlite_*` bookkeeping tables are excluded, returning the
+    /// rewritten SQL. Returns `None` (fail open, run the original query) when the
+    /// input is not a single SELECT or does not select from a `sqlite_master`
+    /// relation.
+    fn rewrite_sqlite_master_query(query: &str) -> Option<String> {
+        let dialect = PostgreSqlDialect {};
+        let mut statements = Parser::parse_sql(&dialect, query).ok()?;
+        if statements.len() != 1 {
+            return None;
+        }
+        let Statement::Query(query_stmt) = &mut statements[0] else {
+            return None;
+        };
+        let SetExpr::Select(select) = &mut *query_stmt.body else {
+            return None;
+        };
+
+        // Collect a column qualifier for every sqlite_master / sqlite_schema
+        // relation referenced in the FROM clause (alias-aware and JOIN-safe).
+        let mut qualifiers: Vec<String> = Vec::new();
+        for table in &select.from {
+            Self::collect_sqlite_master_qualifiers(&table.relation, &mut qualifiers);
+            for join in &table.joins {
+                Self::collect_sqlite_master_qualifiers(&join.relation, &mut qualifiers);
+            }
+        }
+        if qualifiers.is_empty() {
+            // "sqlite_master" only appeared in a literal or column name, not as
+            // a FROM relation. Leave the query untouched.
+            return None;
+        }
+
+        // AND the pgsqlite-reserved-name filter into the WHERE clause for each
+        // referenced sqlite_master relation (hides the __pgsqlite_* bookkeeping
+        // tables, the materialized pg_catalog tables, and any index/trigger that
+        // belongs to one of them).
+        for qualifier in qualifiers {
+            if let Some(filter) = Self::pgsqlite_name_filter_expr(&qualifier) {
+                select.selection = Some(match select.selection.take() {
+                    Some(existing) => Expr::BinaryOp {
+                        left: Box::new(existing),
+                        op: sqlparser::ast::BinaryOperator::And,
+                        right: Box::new(filter),
+                    },
+                    None => filter,
+                });
+            }
+        }
+
+        Some(statements[0].to_string())
+    }
+
+    /// If `factor` is a `sqlite_master` / `sqlite_schema` table, push the column
+    /// qualifier to use for it (its alias when present, otherwise the bare table
+    /// name) onto `qualifiers`.
+    fn collect_sqlite_master_qualifiers(factor: &TableFactor, qualifiers: &mut Vec<String>) {
+        if let TableFactor::Table { name, alias, .. } = factor {
+            let full_name = name.to_string().to_lowercase();
+            let bare_name = full_name.rsplit('.').next().unwrap_or(&full_name);
+            if bare_name == "sqlite_master" || bare_name == "sqlite_schema" {
+                let qualifier = match alias {
+                    Some(alias) => alias.name.to_string(),
+                    None => bare_name.to_string(),
+                };
+                qualifiers.push(qualifier);
+            }
+        }
+    }
+
+    /// Build the pgsqlite-reserved-name filter predicate by parsing it, so the
+    /// exact AST shape stays in step with the sqlparser version in use. The
+    /// predicate keeps only rows whose `name` and `tbl_name` are neither
+    /// `__pgsqlite_*` bookkeeping tables, one of the materialized pg_catalog
+    /// tables, nor one of the materialized pg_catalog / information_schema views.
+    /// Filtering on `tbl_name` as well as `name` hides the indexes and triggers
+    /// that belong to an internal object, so an unqualified
+    /// `SELECT * FROM sqlite_master` (not just a `type = 'table'` / `'view'`
+    /// scan) comes back clean. Returns `None` if it cannot be parsed.
+    fn pgsqlite_name_filter_expr(qualifier: &str) -> Option<Expr> {
+        let quote = |names: &[&str]| {
+            names
+                .iter()
+                .map(|name| format!("'{name}'"))
+                .collect::<Vec<_>>()
+                .join(", ")
+        };
+        let table_list = quote(&PG_CATALOG_INTERNAL_TABLES);
+        let view_list = quote(&PG_CATALOG_INTERNAL_VIEWS);
+        let sql = format!(
+            "SELECT 1 WHERE {qualifier}.name NOT LIKE '__pgsqlite_%' \
+             AND {qualifier}.name NOT IN ({table_list}) \
+             AND {qualifier}.name NOT IN ({view_list}) \
+             AND {qualifier}.tbl_name NOT LIKE '__pgsqlite_%' \
+             AND {qualifier}.tbl_name NOT IN ({table_list}) \
+             AND {qualifier}.tbl_name NOT IN ({view_list})"
+        );
+        let dialect = PostgreSqlDialect {};
+        let statements = Parser::parse_sql(&dialect, &sql).ok()?;
+        let Statement::Query(query) = statements.into_iter().next()? else {
+            return None;
+        };
+        let SetExpr::Select(select) = *query.body else {
+            return None;
+        };
+        select.selection
+    }
+
     async fn handle_catalog_query(query: &sqlparser::ast::Query, db: Arc<DbHandler>, session: Option<Arc<SessionState>>) -> Option<DbResponse> {
         debug!("handle_catalog_query called");
         println!("HANDLE_CATALOG_QUERY: called with query");
@@ -3894,3 +4077,134 @@ impl CatalogInterceptor {
         Ok(filtered)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const FILTER: &str = "NOT LIKE '__pgsqlite_%'";
+
+    #[test]
+    fn plain_sqlite_master_select_is_filtered() {
+        let rewritten =
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_master")
+                .expect("plain sqlite_master query should be rewritten");
+        assert!(rewritten.contains(FILTER), "rewritten = {rewritten}");
+        assert!(rewritten.contains("sqlite_master.name"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn materialized_pg_catalog_tables_are_filtered_by_name_and_tbl_name() {
+        let rewritten =
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_master")
+                .expect("plain sqlite_master query should be rewritten");
+        // Each of the four materialized pg_catalog tables is excluded by name.
+        for name in ["pg_attrdef", "pg_constraint", "pg_depend", "pg_index"] {
+            assert!(rewritten.contains(&format!("'{name}'")), "missing {name}: {rewritten}");
+        }
+        assert!(rewritten.contains("sqlite_master.name NOT IN"), "rewritten = {rewritten}");
+        // tbl_name is filtered too, so indexes/triggers owned by an internal
+        // table are hidden even in a SELECT * (non type='table') scan.
+        assert!(
+            rewritten.contains("sqlite_master.tbl_name NOT LIKE '__pgsqlite_%'"),
+            "rewritten = {rewritten}"
+        );
+        assert!(rewritten.contains("sqlite_master.tbl_name NOT IN"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn materialized_pg_catalog_views_are_filtered_by_name_and_tbl_name() {
+        let rewritten =
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_master")
+                .expect("plain sqlite_master query should be rewritten");
+        // Every reserved pg_catalog / information_schema view is excluded.
+        // Generated from the const so the test cannot drift from the source list.
+        for name in PG_CATALOG_INTERNAL_VIEWS {
+            assert!(rewritten.contains(&format!("'{name}'")), "missing {name}: {rewritten}");
+        }
+        // Both name and tbl_name carry the view exclusion, so an unqualified
+        // SELECT * (not just a type='view' scan) stays clean.
+        assert!(rewritten.contains("sqlite_master.name NOT IN"), "rewritten = {rewritten}");
+        assert!(rewritten.contains("sqlite_master.tbl_name NOT IN"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn user_named_pg_view_not_in_reserved_set_is_not_filtered() {
+        // A user object whose name is `pg_*` but NOT in the reserved set must
+        // NOT appear in the exclusion list: the filter is exact-name, not a
+        // `pg_%` wildcard.
+        let rewritten =
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_master")
+                .expect("plain sqlite_master query should be rewritten");
+        assert!(!PG_CATALOG_INTERNAL_VIEWS.contains(&"pg_my_report"));
+        assert!(!PG_CATALOG_INTERNAL_TABLES.contains(&"pg_my_report"));
+        assert!(!rewritten.contains("'pg_my_report'"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn existing_where_clause_is_preserved_with_and() {
+        let rewritten = CatalogInterceptor::rewrite_sqlite_master_query(
+            "SELECT name FROM sqlite_master WHERE type = 'table'",
+        )
+        .expect("query with WHERE should be rewritten");
+        assert!(rewritten.contains("type = 'table'"), "rewritten = {rewritten}");
+        assert!(rewritten.contains(" AND "), "rewritten = {rewritten}");
+        assert!(rewritten.contains(FILTER), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn aliased_sqlite_master_uses_the_alias_qualifier() {
+        let rewritten = CatalogInterceptor::rewrite_sqlite_master_query(
+            "SELECT m.name FROM sqlite_master m WHERE m.type = 'table'",
+        )
+        .expect("aliased sqlite_master query should be rewritten");
+        assert!(rewritten.contains("m.name NOT LIKE '__pgsqlite_%'"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn joined_sqlite_master_is_filtered_and_join_safe() {
+        let rewritten = CatalogInterceptor::rewrite_sqlite_master_query(
+            "SELECT sm.name FROM sqlite_master sm JOIN pragma_table_info('t') pti",
+        )
+        .expect("joined sqlite_master query should be rewritten");
+        // The qualifier is the alias, keeping the filter unambiguous across the join.
+        assert!(rewritten.contains("sm.name NOT LIKE '__pgsqlite_%'"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn sqlite_schema_alias_is_filtered() {
+        let rewritten =
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_schema")
+                .expect("sqlite_schema query should be rewritten");
+        assert!(rewritten.contains(FILTER), "rewritten = {rewritten}");
+        assert!(rewritten.contains("sqlite_schema.name"), "rewritten = {rewritten}");
+    }
+
+    #[test]
+    fn unparseable_sql_fails_open() {
+        assert!(
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT FROM WHERE sqlite_master ???")
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn sqlite_master_only_as_a_literal_is_left_untouched() {
+        // The substring appears, but not as a FROM relation, so nothing is rewritten.
+        assert!(
+            CatalogInterceptor::rewrite_sqlite_master_query(
+                "SELECT 'sqlite_master' AS label FROM users"
+            )
+            .is_none()
+        );
+    }
+
+    #[test]
+    fn similarly_named_table_is_not_matched() {
+        // A user table whose name merely contains the substring must not be filtered.
+        assert!(
+            CatalogInterceptor::rewrite_sqlite_master_query("SELECT name FROM sqlite_master_backup")
+                .is_none()
+        );
+    }
+}