T09 · Insecure Skill Coding Practices
Warning
- Location
- patterns.md:123
- Finding
- SQL Injection in Incremental Loading Query## Vulnerability Details **File Location**: `patterns.md:123` **Vulnerability Type**: SQL injection through string interpolation **Risk Level**: Medium **Vulnerable Code**: ```python new_data = query(f"WHERE updated_at > '{last_loaded}'") ``` ### Technical Analysis The incremental-loading example directly embeds `last_loaded` into an SQL predicate using an f-string. It does not use a bound parameter, strict timestamp conversion, or escaping. This unsafe example also contradicts the parameterization requirement documented in `querying.md:41`. If an implementation follows this pattern and `last_loaded` can be influenced through user input, upstream data, or persisted workflow state, an attacker could insert quote characters and additional SQL syntax. Whether stacked statements are available depends on the database driver, but predicate manipulation or `UNION`-based extraction may remain possible even when multiple statements are disabled. ### Attack Path 1. The attacker influences the value stored or supplied as `last_loaded`. 2. The application inserts that value directly between SQL quotes. 3. A malicious value terminates the intended timestamp literal and adds SQL syntax. 4. The database parses the injected content as part of the query. 5. The attacker may alter filtering behavior or retrieve data outside the intended incremental window, subject to the database account's permissions. ### Impact Assessment Exploitation may expose records beyond the intended date range and, depending on the surrounding query, driver behavior, and database privileges, may enable broader unauthorized database reads. The resulting access is bounded by the permissions of the database connection used by the workflow. If the connection has write or administrative privileges, the potential scope may be greater.
- Remediation
- ## Remediation Suggestions - Replace f-string construction with the database driver's parameter-binding mechanism. - Parse and validate the watermark as a timestamp before executing the query. - Store watermarks in a typed state store rather than accepting arbitrary SQL fragments. - Run extraction with a read-only, least-privileged database account. - Add tests containing quotes, SQL metacharacters, malformed timestamps, and boundary values. - Update the example to follow a safe pattern such as: ```python last_loaded = parse_timestamp(get_last_watermark()) new_data = query( "SELECT required_columns FROM source_table WHERE updated_at > ?", parameters=(last_loaded,), ) ``` The exact placeholder syntax should be adjusted for the selected database driver.