rivven-rdbc 0.0.22

//! Security utilities for SQL injection prevention in rivven-rdbc.
//!
//! Provides:
//! - Identifier validation for savepoints, table names, schema names
//! - String literal escaping for SQL string contexts
//!
//! These functions are used by all RDBC backends (PostgreSQL, MySQL, SQL Server)
//! and the dialect abstraction layer to prevent SQL injection attacks.

use crate::error::Error;

/// Validate a SQL identifier (savepoint, table, schema names).
///
/// Prevents SQL injection by enforcing strict character rules:
/// - Must not be empty
/// - Maximum 255 characters
/// - Must start with ASCII letter or underscore
/// - May only contain ASCII alphanumeric characters and underscores
///
/// This matches the `IDENTIFIER_REGEX` pattern from `rivven-core::validation`
/// (`^[a-zA-Z_][a-zA-Z0-9_]{0,254}$`) but uses char-iteration instead
/// of regex for zero-dependency, zero-allocation validation on the hot path.
///
/// # Examples
///
/// ```
/// use rivven_rdbc::security::validate_sql_identifier;
///
/// assert!(validate_sql_identifier("users").is_ok());
/// assert!(validate_sql_identifier("my_table_123").is_ok());
/// assert!(validate_sql_identifier("_private").is_ok());
///
/// // Rejects injection attempts
/// assert!(validate_sql_identifier("x; DROP TABLE users--").is_err());
/// assert!(validate_sql_identifier("").is_err());
/// assert!(validate_sql_identifier("123abc").is_err());
/// ```
pub fn validate_sql_identifier(name: &str) -> crate::Result<()> {
    if name.is_empty() {
        return Err(Error::config("SQL identifier cannot be empty"));
    }

    if name.len() > 255 {
        return Err(Error::config(format!(
            "SQL identifier too long: {} chars (max 255)",
            name.len()
        )));
    }

    let mut chars = name.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
        _ => {
            return Err(Error::config(format!(
                "Invalid SQL identifier '{}': must start with a letter or underscore",
                name
            )));
        }
    }

    for c in chars {
        if !c.is_ascii_alphanumeric() && c != '_' {
            return Err(Error::config(format!(
                "Invalid SQL identifier '{}': contains invalid character '{}'",
                name, c
            )));
        }
    }

    Ok(())
}

/// Escape a string value for safe interpolation into a SQL string literal context.
///
/// Replaces `'` with `''` (standard SQL escaping for single-quoted string literals).
/// This is used for `information_schema` queries where parameterized queries
/// are not practical (e.g., `table_exists_sql`, `list_columns_sql`) because
/// the SQL must be returned as a pre-built string.
///
/// **Prefer parameterized queries whenever possible.** This function is a fallback
/// for cases where the SQL generation API requires a complete SQL string.
///
/// # Examples
///
/// ```
/// use rivven_rdbc::security::escape_string_literal;
///
/// assert_eq!(escape_string_literal("users"), "users");
/// assert_eq!(escape_string_literal("don't"), "don''t");
/// assert_eq!(escape_string_literal("x'; DROP TABLE users--"), "x''; DROP TABLE users--");
/// ```
pub fn escape_string_literal(value: &str) -> String {
    // Fast path: no escaping needed (common case)
    if !value.contains('\'') {
        return value.to_string();
    }
    value.replace('\'', "''")
}

/// Validate a SQL type name for safe interpolation into DDL statements.
///
/// Prevents SQL injection via `ColumnMetadata::type_name` by allowing only
/// characters that appear in legitimate SQL type specifications:
/// - ASCII letters, digits, underscores: `VARCHAR`, `INT`, `BIGINT`
/// - Parentheses and commas: `DECIMAL(10,2)`, `ENUM('a','b')`
/// - Spaces: `INT UNSIGNED`, `DOUBLE PRECISION`
/// - Single quotes: `ENUM('x','y')` (MySQL set/enum value lists)
/// - Periods: `NUMERIC(10.2)` (some dialects)
///
/// Rejects semicolons, double-dashes, newlines, backticks, and other
/// metacharacters that could escape the DDL context.
///
/// # Examples
///
/// ```
/// use rivven_rdbc::security::validate_sql_type_name;
///
/// assert!(validate_sql_type_name("INT").is_ok());
/// assert!(validate_sql_type_name("VARCHAR(255)").is_ok());
/// assert!(validate_sql_type_name("DECIMAL(10,2)").is_ok());
/// assert!(validate_sql_type_name("INT UNSIGNED").is_ok());
/// assert!(validate_sql_type_name("ENUM('a','b')").is_ok());
///
/// // Rejects injection attempts
/// assert!(validate_sql_type_name("INT; DROP TABLE users--").is_err());
/// assert!(validate_sql_type_name("").is_err());
/// ```
pub fn validate_sql_type_name(type_name: &str) -> crate::Result<()> {
    if type_name.is_empty() {
        return Err(Error::config("SQL type name cannot be empty"));
    }

    if type_name.len() > 255 {
        return Err(Error::config(format!(
            "SQL type name too long: {} chars (max 255)",
            type_name.len()
        )));
    }

    for c in type_name.chars() {
        if !(c.is_ascii_alphanumeric()
            || c == '_'
            || c == '('
            || c == ')'
            || c == ','
            || c == ' '
            || c == '\''
            || c == '.')
        {
            return Err(Error::config(format!(
                "Invalid SQL type name '{}': contains invalid character '{}'",
                type_name, c
            )));
        }
    }

    Ok(())
}

/// Validate a user-supplied WHERE clause for safe interpolation.
///
/// The clause is injected as a raw SQL fragment (via `Expr::cust()`), so it
/// **cannot** be parameterized. This function applies deny-list checks
/// to reject the most common SQL injection patterns:
///
/// - Semicolons (statement terminators / stacking)
/// - Double-dash `--` line comments
/// - C-style `/* */` block comments
/// - Backslash escapes (MySQL-specific injection vector)
/// - `UNION` keyword (second-order query injection)
/// - Subquery delimiters `SELECT` within parentheses
/// - `EXEC`/`EXECUTE` (stored procedure invocation)
/// - `xp_` prefixed calls (SQL Server extended procedures)
/// - `WAITFOR` / `BENCHMARK` / `SLEEP` (timing side-channels)
/// - `INTO OUTFILE` / `INTO DUMPFILE` (file-system writes)
/// - `LOAD_FILE` (file-system reads)
/// - Null bytes (string truncation)
/// - Newlines (multi-line injection)
///
/// # Security note
///
/// This is a **best-effort** safeguard, **not** a guarantee. A determined
/// attacker can craft payloads that bypass simple deny-list checks. Prefer
/// parameterized queries wherever possible. This clause should only come
/// from **trusted connector configuration**, never from end-user input.
///
/// # Examples
///
/// ```
/// use rivven_rdbc::security::validate_where_clause;
///
/// assert!(validate_where_clause("status = 'active'").is_ok());
/// assert!(validate_where_clause("id > 0 AND deleted = false").is_ok());
///
/// assert!(validate_where_clause("1=1; DROP TABLE users").is_err());
/// assert!(validate_where_clause("1=1 -- bypass").is_err());
/// assert!(validate_where_clause("1=1 /* comment */").is_err());
/// assert!(validate_where_clause("1=1 UNION SELECT * FROM passwords").is_err());
/// ```
pub fn validate_where_clause(clause: &str) -> crate::Result<()> {
    if clause.is_empty() {
        return Err(Error::config("WHERE clause cannot be empty"));
    }

    if clause.len() > 4096 {
        return Err(Error::config(format!(
            "WHERE clause too long: {} chars (max 4096)",
            clause.len()
        )));
    }

    // Null bytes — can truncate strings in C-backed drivers
    if clause.contains('\0') {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited null byte: {}",
            clause
        )));
    }

    // Newlines — multi-line injection
    if clause.contains('\n') || clause.contains('\r') {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited newline: {}",
            clause
        )));
    }

    if clause.contains(';') {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited character ';': {}",
            clause
        )));
    }

    if clause.contains("--") {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited pattern '--': {}",
            clause
        )));
    }

    if clause.contains("/*") || clause.contains("*/") {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited comment syntax: {}",
            clause
        )));
    }

    if clause.contains('\\') {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited backslash escape: {}",
            clause
        )));
    }

    // Case-insensitive keyword checks using word-boundary matching.
    // A "word boundary" means the character before/after the keyword is NOT
    // alphanumeric or underscore, preventing false positives like
    // `executor_status` matching `EXEC` or `SELECTIVITY` matching `SELECT`.
    let upper = clause.to_uppercase();

    // Single-word prohibited keywords (word-boundary matched to prevent false
    // positives like `executor_status` matching EXEC or `selectivity` matching SELECT)
    for keyword in &[
        // Query manipulation
        "UNION",
        "SELECT",
        "INSERT",
        "UPDATE",
        "DELETE",
        // DDL
        "DROP",
        "ALTER",
        "CREATE",
        "TRUNCATE",
        // Stored procedure / dynamic execution
        "EXEC",
        "EXECUTE",
        "DECLARE",
        "CALL",
        // Privilege escalation
        "GRANT",
        "REVOKE",
        // Timing side-channels
        "WAITFOR",
        "BENCHMARK",
        "SLEEP",
        "PG_SLEEP",
        // PostgreSQL file-access functions (exploitable without SELECT)
        "PG_READ_FILE",
        "PG_LS_DIR",
        "PG_READ_BINARY_FILE",
    ] {
        if contains_word(&upper, keyword) {
            return Err(Error::config(format!(
                "WHERE clause contains prohibited keyword '{}': {}",
                keyword, clause
            )));
        }
    }

    // SQL Server extended procedures (prefix match: xp_ followed by word chars)
    if contains_word_prefix(&upper, "XP_") {
        return Err(Error::config(format!(
            "WHERE clause contains prohibited pattern 'xp_': {}",
            clause
        )));
    }

    // Multi-word prohibited patterns (already act as their own boundary)
    for keyword in &["INTO OUTFILE", "INTO DUMPFILE", "LOAD_FILE"] {
        if upper.contains(keyword) {
            return Err(Error::config(format!(
                "WHERE clause contains prohibited keyword '{}': {}",
                keyword, clause
            )));
        }
    }

    // Note: MySQL `#` line comments are NOT blocked because `#` is a valid
    // PostgreSQL bitwise XOR operator and JSON path operator (#>, #>>).
    // The `--` and `/* */` checks cover the common cross-dialect comment vectors.

    Ok(())
}

/// Check whether `haystack` contains `word` at a word boundary.
///
/// A word boundary is any position where the adjacent character (or
/// start/end of string) is not ASCII alphanumeric or underscore.
///
/// # Safety invariant
/// This operates on bytes, not chars. The `u8 as char` cast is safe because:
/// - All keywords are ASCII, so byte-level comparison is correct (UTF-8
///   guarantees ASCII bytes never appear as continuation bytes).
/// - `is_word_boundary` only checks `is_ascii_alphanumeric()` and `!= '_'`,
///   both of which correctly classify non-ASCII bytes as boundaries.
/// - If `is_word_boundary` is ever extended to check Unicode letter categories
///   (e.g., `char::is_alphabetic()`), this must be rewritten to use char offsets.
#[inline]
fn is_word_boundary(c: Option<char>) -> bool {
    match c {
        None => true,
        Some(ch) => !ch.is_ascii_alphanumeric() && ch != '_',
    }
}

/// Returns `true` if `haystack` contains `word` surrounded by word boundaries.
fn contains_word(haystack: &str, word: &str) -> bool {
    let h = haystack.as_bytes();
    let w = word.as_bytes();
    if w.is_empty() || w.len() > h.len() {
        return false;
    }
    for start in 0..=(h.len() - w.len()) {
        if &h[start..start + w.len()] == w {
            let before = if start == 0 {
                None
            } else {
                Some(h[start - 1] as char)
            };
            let after = if start + w.len() >= h.len() {
                None
            } else {
                Some(h[start + w.len()] as char)
            };
            if is_word_boundary(before) && is_word_boundary(after) {
                return true;
            }
        }
    }
    false
}

/// Returns `true` if `haystack` contains `prefix` at a word boundary on the left.
/// Used for patterns like `XP_` where only the start boundary matters.
fn contains_word_prefix(haystack: &str, prefix: &str) -> bool {
    let h = haystack.as_bytes();
    let p = prefix.as_bytes();
    if p.len() > h.len() {
        return false;
    }
    for start in 0..=(h.len() - p.len()) {
        if &h[start..start + p.len()] == p {
            let before = if start == 0 {
                None
            } else {
                Some(h[start - 1] as char)
            };
            if is_word_boundary(before) {
                return true;
            }
        }
    }
    false
}

#[cfg(test)]
mod tests {
    use super::*;

    // -----------------------------------------------------------------------
    // validate_sql_identifier
    // -----------------------------------------------------------------------

    #[test]
    fn test_valid_identifiers() {
        assert!(validate_sql_identifier("users").is_ok());
        assert!(validate_sql_identifier("my_table").is_ok());
        assert!(validate_sql_identifier("_private").is_ok());
        assert!(validate_sql_identifier("a").is_ok());
        assert!(validate_sql_identifier("TABLE_123").is_ok());
        assert!(validate_sql_identifier("sp1").is_ok());
    }

    #[test]
    fn test_empty_identifier() {
        assert!(validate_sql_identifier("").is_err());
    }

    #[test]
    fn test_too_long_identifier() {
        let long = "a".repeat(256);
        assert!(validate_sql_identifier(&long).is_err());

        let max = "a".repeat(255);
        assert!(validate_sql_identifier(&max).is_ok());
    }

    #[test]
    fn test_starts_with_digit() {
        assert!(validate_sql_identifier("123abc").is_err());
        assert!(validate_sql_identifier("0").is_err());
    }

    #[test]
    fn test_injection_attempts() {
        // SQL injection via semicolon
        assert!(validate_sql_identifier("x; DROP TABLE users--").is_err());
        // SQL injection via quote
        assert!(validate_sql_identifier("x' OR '1'='1").is_err());
        // SQL injection via comment
        assert!(validate_sql_identifier("x--").is_err());
        // SQL injection via parentheses
        assert!(validate_sql_identifier("x()").is_err());
        // Unicode smuggling
        assert!(validate_sql_identifier("tabl\u{0435}").is_err()); // Cyrillic е
                                                                   // Whitespace
        assert!(validate_sql_identifier("user name").is_err());
        // Newlines
        assert!(validate_sql_identifier("x\nDROP TABLE").is_err());
        // Null bytes
        assert!(validate_sql_identifier("x\0").is_err());
        // Dots (schema.table injection)
        assert!(validate_sql_identifier("schema.table").is_err());
    }

    #[test]
    fn test_special_chars_rejected() {
        for ch in &[
            '.', '-', '@', '#', '$', '!', '%', '&', '*', '[', ']', '"', '`',
        ] {
            let name = format!("a{}", ch);
            assert!(
                validate_sql_identifier(&name).is_err(),
                "Should reject '{}'",
                name
            );
        }
    }

    // -----------------------------------------------------------------------
    // escape_string_literal
    // -----------------------------------------------------------------------

    #[test]
    fn test_escape_no_quotes() {
        assert_eq!(escape_string_literal("users"), "users");
        assert_eq!(escape_string_literal("my_table"), "my_table");
    }

    #[test]
    fn test_escape_single_quotes() {
        assert_eq!(escape_string_literal("don't"), "don''t");
        assert_eq!(escape_string_literal("'hello'"), "''hello''");
    }

    #[test]
    fn test_escape_injection_attempt() {
        assert_eq!(
            escape_string_literal("x'; DROP TABLE users--"),
            "x''; DROP TABLE users--"
        );
        assert_eq!(escape_string_literal("' OR '1'='1"), "'' OR ''1''=''1");
    }

    #[test]
    fn test_escape_empty_string() {
        assert_eq!(escape_string_literal(""), "");
    }

    // -----------------------------------------------------------------------
    // validate_sql_type_name
    // -----------------------------------------------------------------------

    #[test]
    fn test_valid_type_names() {
        assert!(validate_sql_type_name("INT").is_ok());
        assert!(validate_sql_type_name("BIGINT").is_ok());
        assert!(validate_sql_type_name("VARCHAR(255)").is_ok());
        assert!(validate_sql_type_name("DECIMAL(10,2)").is_ok());
        assert!(validate_sql_type_name("INT UNSIGNED").is_ok());
        assert!(validate_sql_type_name("DOUBLE PRECISION").is_ok());
        assert!(validate_sql_type_name("ENUM('a','b','c')").is_ok());
        assert!(validate_sql_type_name("SET('x','y')").is_ok());
        assert!(validate_sql_type_name("NUMERIC(10.2)").is_ok());
        assert!(validate_sql_type_name("timestamp").is_ok());
        assert!(validate_sql_type_name("TINYINT(1)").is_ok());
    }

    #[test]
    fn test_empty_type_name() {
        assert!(validate_sql_type_name("").is_err());
    }

    #[test]
    fn test_type_name_injection_attempts() {
        assert!(validate_sql_type_name("INT; DROP TABLE users--").is_err());
        assert!(validate_sql_type_name("INT`; DROP TABLE").is_err());
        assert!(validate_sql_type_name("INT\nDROP TABLE").is_err());
        assert!(validate_sql_type_name("INT\0").is_err());
        assert!(validate_sql_type_name("INT--comment").is_err());
    }

    #[test]
    fn test_type_name_too_long() {
        let long = "A".repeat(256);
        assert!(validate_sql_type_name(&long).is_err());
    }

    // -----------------------------------------------------------------------
    // validate_where_clause
    // -----------------------------------------------------------------------

    #[test]
    fn test_valid_where_clauses() {
        assert!(validate_where_clause("status = 'active'").is_ok());
        assert!(validate_where_clause("id > 0 AND deleted = false").is_ok());
        assert!(validate_where_clause("age BETWEEN 18 AND 65").is_ok());
        assert!(validate_where_clause("name LIKE '%test%'").is_ok());
        assert!(validate_where_clause("id IN (1, 2, 3)").is_ok());
    }

    #[test]
    fn test_where_clause_injection_attacks() {
        // Statement stacking
        assert!(validate_where_clause("1=1; DROP TABLE users").is_err());
        // Line comments
        assert!(validate_where_clause("1=1 -- bypass").is_err());
        // Block comments
        assert!(validate_where_clause("1=1 /* comment */").is_err());
        // Backslash escape
        assert!(validate_where_clause("name = '\\' OR 1=1").is_err());
        // UNION injection
        assert!(validate_where_clause("1=1 UNION SELECT * FROM passwords").is_err());
        assert!(validate_where_clause("1=1 union select * from passwords").is_err());
        // Subquery
        assert!(validate_where_clause("id = (SELECT MAX(id) FROM users)").is_err());
        // EXEC
        assert!(validate_where_clause("1=1; EXEC sp_help").is_err());
        assert!(validate_where_clause("EXECUTE xp_cmdshell 'dir'").is_err());
        // xp_ extended procedures
        assert!(validate_where_clause("xp_cmdshell('dir')").is_err());
        // Timing attacks
        assert!(validate_where_clause("1=1 AND SLEEP(5)").is_err());
        assert!(validate_where_clause("1=1 AND BENCHMARK(1000000, SHA1('test'))").is_err());
        assert!(validate_where_clause("1=1; WAITFOR DELAY '0:0:5'").is_err());
        assert!(validate_where_clause("1=1 AND PG_SLEEP(5)").is_err());
        // File access
        assert!(validate_where_clause("1=1 INTO OUTFILE '/tmp/data'").is_err());
        assert!(validate_where_clause("1=1 INTO DUMPFILE '/tmp/data'").is_err());
        assert!(validate_where_clause("LOAD_FILE('/etc/passwd')").is_err());
        // Null bytes
        assert!(validate_where_clause("name = 'test\0").is_err());
        // Newlines
        assert!(validate_where_clause("1=1\nDROP TABLE users").is_err());
    }

    #[test]
    fn test_where_clause_empty() {
        assert!(validate_where_clause("").is_err());
    }

    #[test]
    fn test_where_clause_too_long() {
        let long = "a".repeat(4097);
        assert!(validate_where_clause(&long).is_err());
    }

    #[test]
    fn test_where_clause_word_boundary_no_false_positives() {
        // Should NOT reject: keyword is part of a larger word
        assert!(validate_where_clause("executor_status = 'running'").is_ok());
        assert!(validate_where_clause("execution_count > 0").is_ok());
        assert!(validate_where_clause("selectivity > 0.5").is_ok());
        assert!(validate_where_clause("selected = true").is_ok());
        assert!(validate_where_clause("preselected = true").is_ok());
        assert!(validate_where_clause("reunionist = 'alice'").is_ok());
        assert!(validate_where_clause("asleep = false").is_ok());
    }

    #[test]
    fn test_where_clause_word_boundary_still_catches_keywords() {
        // Should STILL reject: keyword at word boundary
        assert!(validate_where_clause("1=1 UNION ALL").is_err());
        assert!(validate_where_clause("(SELECT 1)").is_err());
        assert!(validate_where_clause("EXEC sp_help").is_err());
        assert!(validate_where_clause("id=1 AND SLEEP(5)").is_err());
        assert!(validate_where_clause("EXECUTE('cmd')").is_err());
        // DDL keywords
        assert!(validate_where_clause("1=1; DROP TABLE users").is_err()); // also blocked by ;
        assert!(validate_where_clause("ALTER TABLE users").is_err());
        assert!(validate_where_clause("CREATE TABLE evil").is_err());
        assert!(validate_where_clause("TRUNCATE TABLE users").is_err());
        // DML keywords
        assert!(validate_where_clause("INSERT INTO users").is_err());
        assert!(validate_where_clause("UPDATE users SET x=1").is_err());
        assert!(validate_where_clause("DELETE FROM users").is_err());
        // Privilege escalation
        assert!(validate_where_clause("GRANT ALL ON users").is_err());
        assert!(validate_where_clause("REVOKE ALL ON users").is_err());
    }

    #[test]
    fn test_where_clause_hash_comment_allowed() {
        // `#` is NOT blocked because it is a valid PostgreSQL operator
        // (bitwise XOR, JSON path operators #>, #>>)
        assert!(validate_where_clause("flags # 4 > 0").is_ok());
        assert!(validate_where_clause("data #>> '{key}'").is_ok());
    }

    #[test]
    fn test_where_clause_ddl_word_boundary_no_false_positives() {
        // DDL keywords as substrings should NOT be rejected
        assert!(validate_where_clause("droplet_count > 0").is_ok());
        assert!(validate_where_clause("created_at > '2024-01-01'").is_ok());
        assert!(validate_where_clause("alteration = 'none'").is_ok());
        assert!(validate_where_clause("undeleted = true").is_ok());
        assert!(validate_where_clause("inserted = false").is_ok());
        assert!(validate_where_clause("updated_at IS NOT NULL").is_ok());
        assert!(validate_where_clause("revoked = false").is_ok());
        assert!(validate_where_clause("granted = true").is_ok());
    }

    // -----------------------------------------------------------------------
    // contains_word / contains_word_prefix
    // -----------------------------------------------------------------------

    #[test]
    fn test_contains_word_boundaries() {
        assert!(contains_word("HELLO WORLD", "HELLO"));
        assert!(contains_word("HELLO WORLD", "WORLD"));
        assert!(contains_word("(EXEC)", "EXEC"));
        assert!(!contains_word("EXECUTOR", "EXEC"));
        assert!(!contains_word("PRESELECT", "SELECT"));
        assert!(!contains_word("SELECTIVITY", "SELECT"));
        assert!(contains_word("SELECT", "SELECT"));
        assert!(contains_word(" SELECT ", "SELECT"));
        // Underscore is NOT a word boundary
        assert!(!contains_word("A_EXEC_B", "EXEC"));
        assert!(!contains_word("DROP_TABLE", "DROP"));
        // Empty inputs
        assert!(!contains_word("", "EXEC"));
        assert!(!contains_word("EX", "EXEC")); // keyword longer than haystack
                                               // Tab/special chars ARE word boundaries
        assert!(contains_word("EXEC\tSOMETHING", "EXEC"));
        assert!(contains_word("X=EXEC", "EXEC")); // = is boundary
    }
}