fix(cubejs): fifth review of the confinement (multi-hop row rules, query deadlines)

services/cubejs/Dockerfile

@@ -9,6 +9,9 @@ RUN DEBIAN_FRONTEND=noninteractive \
 
 ENV TERM=rxvt-unicode
 ENV NODE_ENV=production
+# DuckDB and SQLite datasources run queries on libuv workers (default 4): a
+# few slow tenant queries would starve DNS, crypto and file I/O for the pod.
+ENV UV_THREADPOOL_SIZE=64
 
 # Required for node-oracledb to buld on ARM64
 RUN apt-get update \

services/cubejs/index.js

@@ -15,7 +15,7 @@ import driverFactory, {
 } from "./src/utils/driverFactory.js";
 import { installProcessGuards } from "./src/utils/processGuards.js";
 import createQueryPreprocessor from "./src/utils/queryPreprocessor.js";
-import queryRewrite from "./src/utils/queryRewrite.js";
+import queryRewrite, { memberPathHops } from "./src/utils/queryRewrite.js";
 import repositoryFactory from "./src/utils/repositoryFactory.js";
 import scheduledRefreshContexts, {
   scheduledRefreshStalledFor,
@@ -77,7 +77,15 @@ const queryRewriteOnModel = async (query, context) => {
   });
   return queryRewrite(query, {
     ...context,
-    resolveMemberPath: (path) => cubeEvaluator.resolveMemberPath(path),
+    resolveMemberPath: (path) => {
+      const resolved = cubeEvaluator.resolveMemberPath(path);
+      return (
+        resolved && {
+          ...resolved,
+          hops: memberPathHops(path, cubeEvaluator.evaluatedCubes),
+        }
+      );
+    },
   });
 };
 

services/cubejs/scripts/patchCubeYamlCompiler.mjs

@@ -95,6 +95,14 @@ const LOADER_PATCHED = `                // pipe() drops source errors; forwarded
 //   what tenants need (httpfs for S3/HTTP, excel, ...) in
 //   CUBEJS_DB_DUCKDB_EXTENSIONS;
 // - queries: no logging/profiling calls (the lock doesn't cover them).
+// Tenant SQL a DuckDB datasource refuses: logging/profiling calls (the lock
+// doesn't cover them) and writes out of the database (COPY ... TO, EXPORT),
+// which no disk quota bounds.
+const DUCKDB_REFUSED_SQL = "/\\b(enable|disable)_(logging|profiling)\\b|\\bCOPY\\b[\\s\\S]*\\bTO\\b|\\bEXPORT\\s+DATABASE\\b/i";
+// In-process engines run each query on a libuv worker; one that never ends
+// holds it, and four freeze the process. Interrupted past this deadline.
+const LOCAL_QUERY_TIMEOUT = "Number(process.env.CUBEJS_DB_LOCAL_QUERY_TIMEOUT || 300) * 1000";
+
 const DUCKDB_SAFE_EXTENSIONS = "['httpfs', 'aws', 'azure', 'motherduck', 'excel', 'json', 'parquet', 'icu', 'fts', 'inet', 'tpch', 'tpcds', 'autocomplete', 'core_functions', 'jemalloc']";
 
 const DUCKDB_CREATE_ORIGINAL = `        let dbOptions;
@@ -157,8 +165,8 @@ const DUCKDB_PATCHED = `        const sqlLiteral = (v) => \`'\${String(v).replac
         // ponytail: a query-text filter can't see SQL built at run time
         // (query('...'||'...')), so logging can still be switched on to the
         // pod's stdout; a NetworkPolicy-free fix needs DuckDB to lock logging.
-        if (/\\b(enable|disable)_(logging|profiling)\\b/i.test(String(this.config.initSql || ''))) {
-            throw new Error('DuckDB: logging and profiling calls are not allowed');
+        if (${DUCKDB_REFUSED_SQL}.test(String(this.config.initSql || ''))) {
+            throw new Error('DuckDB: logging, profiling and COPY TO / EXPORT are not allowed');
         }
         // Nothing loads after the lock (the extension directory is out of
         // reach), so what is loaded now is all tenant SQL gets.
@@ -182,25 +190,81 @@ const DUCKDB_PATCHED = `        const sqlLiteral = (v) => \`'\${String(v).replac
         }
         if (this.config.initSql) {
             try {
-                await execAsync(this.config.initSql);`;
+                const initDeadline = setTimeout(() => connection.interrupt(), ${LOCAL_QUERY_TIMEOUT});
+                try {
+                    await execAsync(this.config.initSql);
+                }
+                finally {
+                    clearTimeout(initDeadline);
+                }`;
 
 const DUCKDB_QUERY_ORIGINAL = `    async query(query, values = [], _options) {
-        const { defaultConnection } = await this.getInitiatedState();`;
+        const { defaultConnection } = await this.getInitiatedState();
+        const result = await defaultConnection.run(query, (0, Transform_1.convertDuckDBParams)(values));
+        const transform = (0, Transform_1.buildTransform)(result.columnNames(), result.columnTypes());
+        const rows = [];
+        for (const chunk of await result.fetchAllChunks()) {
+            for (const row of (0, Transform_1.transformChunk)(chunk, transform)) {
+                rows.push(row);
+            }
+        }
+        return rows;
+    }`;
 
 const DUCKDB_QUERY_PATCHED = `    async query(query, values = [], _options) {
-        if (/\\b(enable|disable)_(logging|profiling)\\b/i.test(String(query))) {
-            throw new Error('DuckDB: logging and profiling calls are not allowed');
+        if (${DUCKDB_REFUSED_SQL}.test(String(query))) {
+            throw new Error('DuckDB: logging, profiling and COPY TO / EXPORT are not allowed');
+        }
+        const { defaultConnection } = await this.getInitiatedState();
+        const deadline = setTimeout(() => defaultConnection.interrupt(), ${LOCAL_QUERY_TIMEOUT});
+        try {
+            const result = await defaultConnection.run(query, (0, Transform_1.convertDuckDBParams)(values));
+            const transform = (0, Transform_1.buildTransform)(result.columnNames(), result.columnTypes());
+            const rows = [];
+            for (const chunk of await result.fetchAllChunks()) {
+                for (const row of (0, Transform_1.transformChunk)(chunk, transform)) {
+                    rows.push(row);
+                }
+            }
+            return rows;
         }
-        const { defaultConnection } = await this.getInitiatedState();`;
+        finally {
+            clearTimeout(deadline);
+        }
+    }`;
 
 const DUCKDB_STREAM_ORIGINAL = `    async stream(query, values, { highWaterMark }) {
-        const { instance } = await this.getInitiatedState();`;
+        const { instance } = await this.getInitiatedState();
+        // new connection, because stream can break with
+        // Attempting to execute an unsuccessful or closed pending query result
+        // PreAggregation queue has a concurrency limit, it's why pool is not needed here
+        const connection = await instance.connect();
+        let closed = false;
+        const close = () => {
+            if (!closed) {
+                closed = true;
+                connection.closeSync();
+            }
+        };`;
 
 const DUCKDB_STREAM_PATCHED = `    async stream(query, values, { highWaterMark }) {
-        if (/\\b(enable|disable)_(logging|profiling)\\b/i.test(String(query))) {
-            throw new Error('DuckDB: logging and profiling calls are not allowed');
+        if (${DUCKDB_REFUSED_SQL}.test(String(query))) {
+            throw new Error('DuckDB: logging, profiling and COPY TO / EXPORT are not allowed');
         }
-        const { instance } = await this.getInitiatedState();`;
+        const { instance } = await this.getInitiatedState();
+        // new connection, because stream can break with
+        // Attempting to execute an unsuccessful or closed pending query result
+        // PreAggregation queue has a concurrency limit, it's why pool is not needed here
+        const connection = await instance.connect();
+        const deadline = setTimeout(() => connection.interrupt(), ${LOCAL_QUERY_TIMEOUT});
+        let closed = false;
+        const close = () => {
+            if (!closed) {
+                closed = true;
+                clearTimeout(deadline);
+                connection.closeSync();
+            }
+        };`;
 
 // SQLite's ATTACH opens any path the process can reach, beside the
 // datasource's own file: no attached databases at all. An open failure
@@ -245,13 +309,13 @@ const SQLITE_QUERY_PATCHED = `  query(query, values) {
     }
     return Promise.resolve(this.opened).then(() => {
       if (this.openError) throw this.openError;
-      return new Promise(
-        (resolve, reject) => this.config.db.all(
-          query,
-          values || [],
-          (err, result) => (err ? reject(err) : resolve(result))
-        )
-      );
+      return new Promise((resolve, reject) => {
+        const deadline = setTimeout(() => this.config.db.interrupt(), ${LOCAL_QUERY_TIMEOUT});
+        this.config.db.all(query, values || [], (err, result) => {
+          clearTimeout(deadline);
+          return err ? reject(err) : resolve(result);
+        });
+      });
     });
   }`;
 

services/cubejs/src/__tests__/cube17Regression.test.js

@@ -282,8 +282,29 @@ after`;
                 await execAsync(this.config.initSql);
     async query(query, values = [], _options) {
         const { defaultConnection } = await this.getInitiatedState();
+        const result = await defaultConnection.run(query, (0, Transform_1.convertDuckDBParams)(values));
+        const transform = (0, Transform_1.buildTransform)(result.columnNames(), result.columnTypes());
+        const rows = [];
+        for (const chunk of await result.fetchAllChunks()) {
+            for (const row of (0, Transform_1.transformChunk)(chunk, transform)) {
+                rows.push(row);
+            }
+        }
+        return rows;
+    }
     async stream(query, values, { highWaterMark }) {
         const { instance } = await this.getInitiatedState();
+        // new connection, because stream can break with
+        // Attempting to execute an unsuccessful or closed pending query result
+        // PreAggregation queue has a concurrency limit, it's why pool is not needed here
+        const connection = await instance.connect();
+        let closed = false;
+        const close = () => {
+            if (!closed) {
+                closed = true;
+                connection.closeSync();
+            }
+        };
 y`);
     assert.equal(duck.changed, true);
     assert.match(duck.source, /autoinstall_known_extensions: 'false'/);
@@ -344,7 +365,6 @@ y`);
     try {
       assert.equal(Number((await driver.query("SELECT 41 + 1 AS answer", []))[0].answer), 42);
       await driver.query("CREATE OR REPLACE TABLE t AS SELECT range AS b FROM range(200000)", []);
-      await driver.query(`COPY (SELECT 1) TO '${dir}/out.csv'`, []);
       await assert.rejects(driver.query("SELECT * FROM leak", []), /leak/);
       for (const sql of [
         "SELECT content FROM read_text('/proc/self/environ')",
@@ -408,6 +428,28 @@ y`);
     }
   });
 
+  it("the installed DuckDB and SQLite drivers interrupt a query past the deadline, and DuckDB refuses COPY TO", async () => {
+    const { default: ServerCore } = await import("@cubejs-backend/server-core");
+    process.env.CUBEJS_DB_LOCAL_QUERY_TIMEOUT = "1";
+    const duck = ServerCore.createDriver("duckdb", {});
+    const lite = ServerCore.createDriver("sqlite", { database: ":memory:" });
+    try {
+      const started = Date.now();
+      await assert.rejects(duck.query("SELECT sum(hash(i)) FROM range(1000000000000) t(i)", []), /interrupt/i);
+      await assert.rejects(
+        lite.query("WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c) SELECT count(*) FROM c", []),
+        /interrupt/i,
+      );
+      assert.ok(Date.now() - started < 10000, "both stopped near the deadline");
+      await assert.rejects(duck.query("COPY (SELECT 1) TO 'x.csv'", []), /not allowed/);
+      await assert.rejects(duck.query("EXPORT DATABASE 'x'", []), /not allowed/);
+    } finally {
+      delete process.env.CUBEJS_DB_LOCAL_QUERY_TIMEOUT;
+      await duck.release();
+      await lite.release();
+    }
+  });
+
   it("the installed DuckDB driver keeps operator S3 credentials out of readable settings", async () => {
     const { default: ServerCore } = await import("@cubejs-backend/server-core");
     Object.assign(process.env, {

services/cubejs/src/utils/__tests__/joinAliasRowRules.test.js

@@ -27,8 +27,8 @@ describe("row rules follow join aliases (Cube 1.8)", () => {
     // Without the compiled model a cube the map lacks (a view) can't be read
     // for granularity paths: refused. Production resolves with Cube itself.
     assert.throws(() => memberInstance("sales_view.created_at.month", map), /cannot resolve/);
-    const resolved = memberInstance("sales_view.created_at.month", map, () => ({ instancePath: "sales_view", targetCube: "sales_view" }));
-    assert.deepEqual(resolved, { instance: "sales_view", cube: "sales_view" });
+    const { instance, cube } = memberInstance("sales_view.created_at.month", map, () => ({ instancePath: "sales_view", targetCube: "sales_view" }));
+    assert.deepEqual({ instance, cube }, { instance: "sales_view", cube: "sales_view" });
   });
 
   it("collects every instance a query reads, order included", () => {

services/cubejs/src/utils/__tests__/rowRulesEndToEnd.test.js

@@ -36,8 +36,38 @@ const MODEL = `cubes:
       - name: city
         sql: city
         type: string
+  - name: top
+    sql_table: shop.top
+    joins:
+      - name: events
+        alias: mid
+        relationship: many_to_one
+        sql: "{CUBE}.event_id = {mid}.id"
+    dimensions:
+      - name: id
+        sql: id
+        type: number
+        primary_key: true
+      - name: name
+        sql: name
+        type: string
+  - name: leaf
+    sql_table: shop.leaf
+    dimensions:
+      - name: id
+        sql: id
+        type: number
+        primary_key: true
+      - name: label
+        sql: label
+        type: string
   - name: events
     sql_table: cst.semantic_events
+    joins:
+      - name: leaf
+        alias: lf
+        relationship: many_to_one
+        sql: "{CUBE}.leaf_id = {lf}.id"
     dimensions:
       - name: id
         sql: id
@@ -59,14 +89,18 @@ mock.module("../dataSourceHelpers.js", {
     findDataSchemasByIds: async () => [{ name: "m.yml", code: MODEL }],
   },
 });
-const { default: queryRewrite } = await import("../queryRewrite.js");
+const { default: queryRewrite, memberPathHops } = await import("../queryRewrite.js");
 
 const compilers = prepareCompiler(
   { dataSchemaFiles: async () => [{ fileName: "m.yml", content: MODEL }] },
   { adapter: "postgres" },
 );
 await compilers.compiler.compile();
-const resolveMemberPath = (p) => compilers.cubeEvaluator.resolveMemberPath(p);
+// As index.js wires it: Cube's resolver plus every hop along the path.
+const resolveMemberPath = (p) => {
+  const resolved = compilers.cubeEvaluator.resolveMemberPath(p);
+  return resolved && { ...resolved, hops: memberPathHops(p, compilers.cubeEvaluator.evaluatedCubes) };
+};
 
 let version = 0;
 const context = () => ({
@@ -97,6 +131,16 @@ describe("row rules on Cube 1.8 member paths (end to end)", () => {
     ]);
   });
 
+  it("rules a ruled cube joined in the middle of a multi-hop path", async () => {
+    const query = await queryRewrite(
+      { dimensions: ["top.name", "top.mid.lf.label"] },
+      context(),
+    );
+    assert.deepEqual(query.filters, [
+      { member: "top.mid.partition", operator: "equals", values: ["tenant-a"] },
+    ]);
+  });
+
   it("rules the cubes of SQL API member expressions, and refuses ones it can't read", async () => {
     const query = await queryRewrite(
       { dimensions: [{ cubeName: "events", name: "e", expression: ["events", "LOWER(events.event)"] }], measures: ["events.count"] },

services/cubejs/src/utils/queryRewrite.js

@@ -344,6 +344,42 @@ const refuseMember = (member, why) => {
   throw error;
 };
 
+// Every cube instance a member path passes through, in order, following the
+// compiled model exactly as Cube's CubeEvaluator.resolveMemberPath does: a
+// ruled cube joined in the middle of orders.customer.events.city is read too.
+// Exported for index.js (which has the compiled cubes) and tests.
+export function memberPathHops(path, cubes) {
+  const parts = String(path).split(".");
+  if (parts.length < 2 || !cubes?.[parts[0]]) return null;
+  let instancePath = parts[0];
+  let targetCube = parts[0];
+  let aliased = false;
+  const hops = [{ instancePath, targetCube }];
+  for (let i = 1; i < parts.length; i += 1) {
+    const part = parts[i];
+    const rest = parts.length - i - 1;
+    const cube = cubes[targetCube];
+    const dimension = cube?.dimensions?.[part];
+    const isMember = dimension || cube?.measures?.[part] || cube?.segments?.[part];
+    if (isMember && (rest === 0 || (rest === 1 && dimension?.type === "time"))) return hops;
+    if (!aliased || part !== targetCube) {
+      const join = (cube?.joins || []).find((j) => (j.alias ?? j.name) === part);
+      if (join && (join.alias || aliased)) {
+        instancePath = `${instancePath}.${part}`;
+        targetCube = join.name;
+        aliased = true;
+      } else if (!aliased && cubes[part]) {
+        instancePath = part;
+        targetCube = part;
+      } else {
+        return null;
+      }
+      hops.push({ instancePath, targetCube });
+    }
+  }
+  return null;
+}
+
 export function memberInstance(member, cubeToTable, resolve) {
   const parts = member.split(".");
   if (parts.length <= 2) return { instance: parts[0], cube: parts[0] };
@@ -352,7 +388,11 @@ export function memberInstance(member, cubeToTable, resolve) {
   if (resolve) {
     const resolved = resolve(member);
     if (!resolved) refuseMember(member, "is not a path the model resolves");
-    return { instance: resolved.instancePath, cube: resolved.targetCube };
+    return {
+      instance: resolved.instancePath,
+      cube: resolved.targetCube,
+      hops: resolved.hops ?? [resolved],
+    };
   }
   // Without the compiled model (tests): the aliases the model map read.
   let cube = parts[0];
@@ -416,8 +456,10 @@ export function queryInstances(query, cubeToTable, { resolve, strict = false } =
   for (const member of [...queryMembers(query), ...orderMembers(query), ...subqueryOns]) {
     if (typeof member === "string") {
       if (member.indexOf(".") <= 0) continue;
-      const { instance, cube } = memberInstance(member, cubeToTable, resolve);
+      const { instance, cube, hops } = memberInstance(member, cubeToTable, resolve);
       instances.set(instance, cube);
+      // Every cube the path passes through is read, not only the last one.
+      for (const hop of hops || []) instances.set(hop.instancePath, hop.targetCube);
       continue;
     }
     const cubes = expressionCubes(member);