fix: security hardening for SQL query API (#81)

* fix: replace string-based SQL injection in inject_block_filter with AST manipulation

The previous implementation used string position matching (finding WHERE,
ORDER BY, LIMIT keywords) and format! interpolation to splice block_num
filters into user-provided SQL. This was vulnerable to structural SQL
injection where crafted queries could exploit the naive keyword matching
(e.g. WHERE inside string literals, UNION bypasses).

Replace with sqlparser AST parsing and manipulation:
- Parse user SQL into AST, requiring a single simple SELECT statement
- Determine filter column from the FROM table (num for blocks, block_num
  for others)
- Safely AND the block filter into the existing WHERE clause (or add one)
- Serialize modified AST back to SQL

Also:
- Reject UNION/INTERSECT/set operations in live mode (ambiguous filtering)
- Return Result<String, ApiError> instead of String for proper error handling
- Add Display impl for ApiError
- Add tests for UNION rejection, non-SELECT rejection, and WHERE keyword
  in string literals

Amp-Thread-ID: https://ampcode.com/threads/T-019c3272-f632-763c-8078-504a90852a67
Co-authored-by: Amp <amp@ampcode.com>

* fix: add table allowlist to query validator and read-only API role

Replace the blocklist-only approach with a table allowlist so API users
can only query: blocks, txs, logs, receipts, token_holders,
token_balances, and CTE-defined tables.

Previously, users could query sync_state (internal), pg_tables (schema
enumeration), or any other table accessible to the tidx DB user.

Changes:
- Allowlist in validator: only permitted tables + CTE-defined names pass
- Block dblink function family (cross-database access)
- Add db/api_role.sql migration creating a tidx_api read-only role with
  SELECT-only grants on indexed tables (defense-in-depth)
- Thread CTE names through all validate_* functions
- 6 new tests: sync_state rejected, pg_tables rejected, unknown table
  rejected, CTE tables allowed, dblink blocked, analytics tables allowed

Amp-Thread-ID: https://ampcode.com/threads/T-019c3272-f632-763c-8078-504a90852a67
Co-authored-by: Amp <amp@ampcode.com>

* fix: close DoS, privilege escalation, and file read bypass vectors

Block three categories of attacks:

1. DoS via resource exhaustion:
   - Reject WITH RECURSIVE (endless loop CTEs)
   - Block generate_series() (billion-row generation)
   - Block SELECT INTO (object creation)

2. Privilege escalation / validator bypass:
   - Validate expressions inside VALUES rows (previously VALUES(pg_sleep(10))
     bypassed the entire function blocklist)
   - Reject TABLE statement (TABLE pg_shadow bypassed table allowlist)
   - Validate GROUP BY, HAVING, JOIN ON expressions (could hide function calls)
   - Walk IsNull/IsNotNull/IsTrue/IsFalse/Like expressions recursively

3. File read hardening:
   - Block lo_get/lo_open/lo_close/loread/lo_creat/lo_create/lo_unlink/lo_put
   - Block pg_file_read/pg_file_write/pg_file_rename/pg_file_unlink/pg_logdir_ls
   - VALUES bypass closure prevents pg_read_file via VALUES(...)

11 new tests covering all vectors.

Amp-Thread-ID: https://ampcode.com/threads/T-019c3272-f632-763c-8078-504a90852a67
Co-authored-by: Amp <amp@ampcode.com>

* fix: replace function blocklist with allowlist and harden API role

Switch from a function blocklist (reject known-bad) to an allowlist
(permit known-good only). This eliminates the risk of missing dangerous
functions as PostgreSQL adds new ones.

Validator changes:
- ALLOWED_FUNCTIONS allowlist: ABI helpers, aggregates, scalars, string,
  numeric, time, window functions, and type casting
- Reject ALL table functions (FROM func(...)) unconditionally
- Reject unsupported TableFactor variants (catch-all _ => Err)
- Remove is_dangerous_function() and is_dangerous_table_function()

API role hardening (db/api_role.sql):
- Deny-by-default: REVOKE ALL on tables, sequences, and functions
  before granting specific access
- Add token_holders and token_balances to SELECT grants
- CONNECTION LIMIT 64
- statement_timeout = 30s, work_mem = 64MB, temp_file_limit = 256MB

5 new tests, all 147 lib tests passing.

Amp-Thread-ID: https://ampcode.com/threads/T-019c499a-f07e-73a9-9526-6c18fd511372
Co-authored-by: Amp <amp@ampcode.com>

* fix: reject-by-default expression validation, LIMIT/depth/size caps

Switch validate_expr to reject-by-default: only explicitly allowed
expression types are permitted (identifiers, literals, binary/unary ops,
CASE, CAST, BETWEEN, IN, LIKE, subqueries, SQL builtins like EXTRACT,
SUBSTRING, TRIM, etc). Unknown expression variants are rejected.

Query structure hardening:
- Reject FOR UPDATE/SHARE locking clauses
- Validate ORDER BY expressions through validate_expr
- Validate LIMIT/OFFSET: must be numeric literals, capped at 10,000
- Reject LIMIT BY (ClickHouse-specific)
- Subquery depth limit: max 4 levels of nesting
- Query size limit: max 64KB
- Validate window function OVER clause (PARTITION BY, ORDER BY)

Service layer:
- Replace string-based LIMIT detection (contains("LIMIT")) with
  AST-based detection via append_limit_if_missing()
- Use HARD_LIMIT_MAX constant (10,000) across validator, service, and API
- API param clamping uses HARD_LIMIT_MAX instead of hardcoded 100,000

15 new tests, all 162 lib tests passing.

Amp-Thread-ID: https://ampcode.com/threads/T-019c499a-f07e-73a9-9526-6c18fd511372
Co-authored-by: Amp <amp@ampcode.com>

* fix: close FILTER clause bypass, LIMIT NULL, FETCH, negative limit

Fixes found during oracle review:

- Validate function FILTER (WHERE ...) clause: previously
  COUNT(*) FILTER (WHERE pg_sleep(1) IS NOT NULL) bypassed
  the function allowlist entirely
- Validate WITHIN GROUP (ORDER BY ...) expressions
- Reject LIMIT NULL (effectively means no limit, bypasses cap)
- Reject negative LIMIT/OFFSET values
- Reject FETCH clause (FETCH FIRST N ROWS ONLY bypasses LIMIT cap)

4 new tests, all 166 lib tests passing.

Amp-Thread-ID: https://ampcode.com/threads/T-019c499a-f07e-73a9-9526-6c18fd511372
Co-authored-by: Amp <amp@ampcode.com>

---------

Co-authored-by: Amp <amp@ampcode.com>

Author: gakonst

db/api_role.sql

@@ -0,0 +1,35 @@
+-- Read-only API role for query connections.
+-- Defense-in-depth: even if the query validator is bypassed,
+-- this role cannot modify data, execute arbitrary functions,
+-- or exhaust server resources.
+DO $$
+BEGIN
+    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'tidx_api') THEN
+        CREATE ROLE tidx_api WITH LOGIN PASSWORD 'tidx_api' NOSUPERUSER NOCREATEDB NOCREATEROLE;
+    END IF;
+END $$;
+
+-- Revoke all privileges first (deny-by-default)
+REVOKE ALL ON ALL TABLES IN SCHEMA public FROM tidx_api;
+REVOKE ALL ON ALL SEQUENCES IN SCHEMA public FROM tidx_api;
+REVOKE EXECUTE ON ALL FUNCTIONS IN SCHEMA public FROM tidx_api;
+
+-- Grant read-only access to indexed tables only
+GRANT USAGE ON SCHEMA public TO tidx_api;
+GRANT SELECT ON blocks, txs, logs, receipts, token_holders, token_balances TO tidx_api;
+
+-- Grant execute only on ABI decode helper functions
+GRANT EXECUTE ON FUNCTION abi_uint(bytea) TO tidx_api;
+GRANT EXECUTE ON FUNCTION abi_int(bytea) TO tidx_api;
+GRANT EXECUTE ON FUNCTION abi_address(bytea) TO tidx_api;
+GRANT EXECUTE ON FUNCTION abi_bool(bytea) TO tidx_api;
+GRANT EXECUTE ON FUNCTION abi_bytes(bytea, int) TO tidx_api;
+GRANT EXECUTE ON FUNCTION abi_string(bytea, int) TO tidx_api;
+GRANT EXECUTE ON FUNCTION format_address(bytea) TO tidx_api;
+GRANT EXECUTE ON FUNCTION format_uint(bytea) TO tidx_api;
+
+-- Resource limits (prevent DoS)
+ALTER ROLE tidx_api CONNECTION LIMIT 64;
+ALTER ROLE tidx_api SET statement_timeout = '30s';
+ALTER ROLE tidx_api SET work_mem = '64MB';
+ALTER ROLE tidx_api SET temp_file_limit = '256MB';

src/api/mod.rs

@@ -257,7 +257,7 @@ fn default_timeout() -> u64 {
     5000
 }
 fn default_limit() -> i64 {
-    10000
+    crate::query::HARD_LIMIT_MAX
 }
 
 #[derive(Serialize)]
@@ -299,7 +299,7 @@ async fn handle_query_once(
 
     let options = QueryOptions {
         timeout_ms: params.timeout_ms.clamp(100, 30000),
-        limit: params.limit.clamp(1, 100000),
+        limit: params.limit.clamp(1, crate::query::HARD_LIMIT_MAX),
     };
 
     // Route to appropriate engine
@@ -397,7 +397,7 @@ async fn handle_query_live(
     let signature = params.signature;
     let options = QueryOptions {
         timeout_ms: params.timeout_ms.clamp(100, 30000),
-        limit: params.limit.clamp(1, 100000),
+        limit: params.limit.clamp(1, crate::query::HARD_LIMIT_MAX),
     };
 
     // Detect if this is an OLAP query (aggregations, etc.)
@@ -477,7 +477,16 @@ async fn handle_query_live(
                     } else {
                         let catch_up_start = last_block_num + 1;
                         for block_num in catch_up_start..=end {
-                            let filtered_sql = inject_block_filter(&sql, block_num);
+                            let filtered_sql = match inject_block_filter(&sql, block_num) {
+                                Ok(s) => s,
+                                Err(e) => {
+                                    yield Ok(SseEvent::default()
+                                        .event("error")
+                                        .json_data(serde_json::json!({ "ok": false, "error": e.to_string() }))
+                                        .unwrap());
+                                    return;
+                                }
+                            };
                             match crate::service::execute_query_postgres(&pool, &filtered_sql, signature.as_deref(), &options).await {
                                 Ok(result) => {
                                     yield Ok(SseEvent::default()
@@ -516,50 +525,85 @@ async fn handle_query_live(
 /// Inject a block number filter into SQL query for live streaming.
 /// Transforms queries to only return data for the specific block.
 /// Uses 'num' for blocks table, 'block_num' for txs/logs tables.
+///
+/// Uses sqlparser AST manipulation to safely add the filter condition,
+/// avoiding SQL injection risks from string-based splicing.
 #[doc(hidden)]
-pub fn inject_block_filter(sql: &str, block_num: u64) -> String {
-    let sql_upper = sql.to_uppercase();
-    
-    // Determine column name based on table being queried
-    let col = if sql_upper.contains("FROM BLOCKS") || sql_upper.contains("FROM \"BLOCKS\"") {
-        "num"
-    } else {
-        "block_num"
+pub fn inject_block_filter(sql: &str, block_num: u64) -> Result<String, ApiError> {
+    use sqlparser::ast::{
+        BinaryOperator, Expr, Ident, SetExpr, Statement, Value,
     };
-    
-    // Find WHERE clause position
-    if let Some(where_pos) = sql_upper.find("WHERE") {
-        // Insert after WHERE
-        let insert_pos = where_pos + 5;
-        format!(
-            "{} {} = {} AND {}",
-            &sql[..insert_pos],
-            col,
-            block_num,
-            &sql[insert_pos..]
-        )
-    } else if let Some(order_pos) = sql_upper.find("ORDER BY") {
-        // Insert WHERE before ORDER BY
-        format!(
-            "{} WHERE {} = {} {}",
-            &sql[..order_pos],
-            col,
-            block_num,
-            &sql[order_pos..]
-        )
-    } else if let Some(limit_pos) = sql_upper.find("LIMIT") {
-        // Insert WHERE before LIMIT
-        format!(
-            "{} WHERE {} = {} {}",
-            &sql[..limit_pos],
-            col,
-            block_num,
-            &sql[limit_pos..]
-        )
-    } else {
-        // Append WHERE at end
-        format!("{sql} WHERE {col} = {block_num}")
+    use sqlparser::dialect::GenericDialect;
+    use sqlparser::parser::Parser;
+
+    let dialect = GenericDialect {};
+    let mut statements = Parser::parse_sql(&dialect, sql)
+        .map_err(|e| ApiError::BadRequest(format!("SQL parse error: {e}")))?;
+
+    if statements.len() != 1 {
+        return Err(ApiError::BadRequest(
+            "Live mode requires exactly one SQL statement".to_string(),
+        ));
     }
+
+    let stmt = &mut statements[0];
+    let query = match stmt {
+        Statement::Query(q) => q,
+        _ => {
+            return Err(ApiError::BadRequest(
+                "Live mode requires a SELECT query".to_string(),
+            ))
+        }
+    };
+
+    let select = match query.body.as_mut() {
+        SetExpr::Select(s) => s,
+        _ => {
+            return Err(ApiError::BadRequest(
+                "Live mode requires a simple SELECT query (UNION/INTERSECT not supported)"
+                    .to_string(),
+            ))
+        }
+    };
+
+    let table_name: String = select
+        .from
+        .first()
+        .and_then(|twj| match &twj.relation {
+            sqlparser::ast::TableFactor::Table { name, .. } => {
+                name.0.last().and_then(|part| part.as_ident()).map(|ident| ident.value.to_lowercase())
+            }
+            _ => None,
+        })
+        .ok_or_else(|| {
+            ApiError::BadRequest(
+                "Live mode requires a query with a FROM table clause".to_string(),
+            )
+        })?;
+
+    let col_name = if table_name == "blocks" { "num" } else { "block_num" };
+
+    let col_expr = Expr::CompoundIdentifier(vec![
+        Ident::new(&table_name),
+        Ident::new(col_name),
+    ]);
+
+    let block_filter = Expr::BinaryOp {
+        left: Box::new(col_expr),
+        op: BinaryOperator::Eq,
+        right: Box::new(Expr::Value(Value::Number(block_num.to_string(), false).into())),
+    };
+
+    select.selection = Some(match select.selection.take() {
+        Some(existing) => Expr::BinaryOp {
+            left: Box::new(Expr::Nested(Box::new(existing))),
+            op: BinaryOperator::And,
+            right: Box::new(block_filter),
+        },
+        None => block_filter,
+    });
+
+    Ok(stmt.to_string())
 }
 
 /// Rewrite analytics table references to include chain-specific database prefix.
@@ -599,6 +643,19 @@ pub enum ApiError {
     NotFound(String),
 }
 
+impl std::fmt::Display for ApiError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ApiError::BadRequest(msg) => write!(f, "{msg}"),
+            ApiError::Timeout => write!(f, "Query timeout"),
+            ApiError::QueryError(msg) => write!(f, "{msg}"),
+            ApiError::Internal(msg) => write!(f, "{msg}"),
+            ApiError::Forbidden(msg) => write!(f, "{msg}"),
+            ApiError::NotFound(msg) => write!(f, "{msg}"),
+        }
+    }
+}
+
 impl IntoResponse for ApiError {
     fn into_response(self) -> axum::response::Response {
         let (status, message) = match self {

src/db/schema.rs

@@ -39,6 +39,9 @@ pub async fn run_migrations(pool: &Pool) -> Result<()> {
     // Load any optional extensions
     conn.batch_execute(include_str!("../../db/extensions.sql")).await?;
 
+    // Create read-only API role with SELECT-only access to indexed tables
+    conn.batch_execute(include_str!("../../db/api_role.sql")).await?;
+
     Ok(())
 }
 

src/query/mod.rs

@@ -7,7 +7,7 @@ pub use parser::{
     extract_order_by_columns, AbiParam, AbiType, EventSignature,
 };
 pub use router::{route_query, QueryEngine};
-pub use validator::validate_query;
+pub use validator::{validate_query, HARD_LIMIT_MAX};
 
 use regex_lite::Regex;
 use std::sync::LazyLock;