scalo 2.10.11

// Project:   scalo
// File:      src/expression/profile.rs
// Purpose:   Expression profile -- allowed/restricted CEL functions
// Language:  Rust
//
// License:   Apache-2.0
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! Expression profile -- allowed and restricted CEL functions.
//!
//! The profile restricts CEL to a high-performance subset suitable
//! for per-record evaluation at ingest/query time. Functions with
//! unbounded or unpredictable cost are blocked by default but can be
//! unlocked per-category via [`ProfileConfig`].

/// CEL functions allowed unconditionally in the profile.
///
/// Enforced as an ALLOWLIST: a function in neither this slice nor a restricted
/// category is rejected. Listing a function here is what permits it, so anything
/// the profile should accept has to appear -- the cost of an omission is a valid
/// expression refused. Kept in step with scalo-py's
/// `expression/profile.py::ALLOWED_FUNCTIONS`, except `matches`, which is
/// restricted here and unlocked via [`ProfileConfig::allow_regex`].
pub const ALLOWED_FUNCTIONS: &[&str] = &[
    // String operations (SIMD-friendly, bounded cost)
    "contains",
    "startsWith",
    "endsWith",
    // Collection
    "size",
    // Existence
    "has",
    // Type casts
    "int",
    "uint",
    "double",
    "string",
    "bool",
    "bytes",
    // Type introspection. Constant-time, and `dyn()` only relaxes static
    // type-checking; neither iterates.
    "type",
    "dyn",
    // Timestamp and duration accessors: cheap field reads on a value that is
    // already a timestamp. `timestamp()` and `duration()` are restricted, so
    // these are only reachable when the datastore supplies a typed value.
    "getDate",
    "getDayOfMonth",
    "getDayOfWeek",
    "getDayOfYear",
    "getFullYear",
    "getHours",
    "getMilliseconds",
    "getMinutes",
    "getMonth",
    "getSeconds",
];

/// Restricted function categories -- blocked by default, opt-in via config.
///
/// Each category has a reason for restriction and a config flag to unlock.
pub const RESTRICTED_REGEX: &[&str] = &["matches"];
pub const RESTRICTED_ITERATION: &[&str] = &["map", "filter", "exists", "all", "exists_one"];
pub const RESTRICTED_TIME: &[&str] = &["timestamp", "duration"];

/// All restricted functions (union of all categories).
pub const DISALLOWED_FUNCTIONS: &[&str] = &[
    "matches",
    "map",
    "filter",
    "exists",
    "all",
    "exists_one",
    "timestamp",
    "duration",
];

/// Configuration for the expression profile.
///
/// Each flag unlocks a category of restricted functions. All default
/// to `false` (blocked). Set explicitly in application config to opt in.
#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize)]
pub struct ProfileConfig {
    /// Allow `matches()` (regex). Unbounded cost per record -- use only
    /// when `contains()`/`startsWith()`/`endsWith()` are insufficient.
    pub allow_regex: bool,
    /// Allow `map()`, `filter()`, `exists()`, `all()`, `exists_one()`.
    /// O(n) per collection element -- cost proportional to data size.
    pub allow_iteration: bool,
    /// Allow `timestamp()`, `duration()`. Excluded because ClickHouse
    /// handles time natively -- rarely needed in CEL expressions.
    pub allow_time: bool,
}

impl ProfileConfig {
    /// Returns the set of functions that are currently blocked.
    #[must_use]
    pub fn blocked_functions(&self) -> Vec<&'static str> {
        let mut blocked = Vec::new();
        if !self.allow_regex {
            blocked.extend_from_slice(RESTRICTED_REGEX);
        }
        if !self.allow_iteration {
            blocked.extend_from_slice(RESTRICTED_ITERATION);
        }
        if !self.allow_time {
            blocked.extend_from_slice(RESTRICTED_TIME);
        }
        blocked
    }
}

/// CEL keywords and built-in names that look like function calls but
/// should be skipped during profile scanning.
const SKIP_NAMES: &[&str] = &[
    "true", "false", "null", "in", "has", "int", "uint", "double", "string", "bool",
];

/// Scan an expression for restricted function calls using default config.
///
/// Returns a list of error strings (empty if all function calls are
/// within the profile). Equivalent to `check_profile_with_config`
/// with [`ProfileConfig::default()`] (all restrictions active).
#[must_use]
pub fn check_profile(expr: &str) -> Vec<String> {
    check_profile_with_config(expr, &ProfileConfig::default())
}

/// Scan an expression for restricted function calls.
///
/// The scanner skips string literals to avoid false positives on
/// function names that appear inside quoted values.
///
/// Returns a list of error strings (empty if compliant).
#[must_use]
pub fn check_profile_with_config(expr: &str, config: &ProfileConfig) -> Vec<String> {
    // No early return on an empty blocked list. `blocked_functions()` is empty
    // when all three categories are unlocked, and returning here skipped the
    // allowlist check too -- so the most permissive config was also the one that
    // validated nothing at all.
    let blocked = config.blocked_functions();

    let mut errors = Vec::new();
    let bytes = expr.as_bytes();
    let len = bytes.len();
    let mut i = 0;

    while i < len {
        // Skip string literals (double-quoted and single-quoted)
        if bytes[i] == b'"' || bytes[i] == b'\'' {
            i = skip_string_literal(bytes, i);
            continue;
        }

        // Skip non-identifier characters
        if !is_ident_start(bytes[i]) {
            i += 1;
            continue;
        }

        // Read identifier
        let start = i;
        while i < len && is_ident_char(bytes[i]) {
            i += 1;
        }
        let name = &expr[start..i];

        // Skip whitespace between identifier and potential `(`
        let mut peek = i;
        while peek < len && bytes[peek] == b' ' {
            peek += 1;
        }

        // Check if followed by `(`
        if peek < len && bytes[peek] == b'(' {
            if SKIP_NAMES.contains(&name) {
                continue;
            }

            if blocked.contains(&name) {
                let reason = restriction_reason(name);
                errors.push(format!(
                    "Function '{name}()' is not allowed in the expression profile. {reason}"
                ));
            } else if !ALLOWED_FUNCTIONS.contains(&name) && !DISALLOWED_FUNCTIONS.contains(&name) {
                // In neither list. A restricted-but-unlocked function IS in
                // DISALLOWED_FUNCTIONS, so it falls through here and passes.
                errors.push(format!(
                    "Function '{name}()' is not in the expression profile. Check \
                     the spelling, or add it to ALLOWED_FUNCTIONS if it belongs \
                     in the profile."
                ));
            }
        }
    }

    errors
}

/// Skip past a string literal, handling escape sequences.
///
/// `start` must point to the opening quote character.
/// Returns the index after the closing quote (or end of input).
fn skip_string_literal(bytes: &[u8], start: usize) -> usize {
    let quote = bytes[start];
    let mut i = start + 1;
    while i < bytes.len() {
        if bytes[i] == b'\\' {
            // Skip escaped character
            i += 2;
            continue;
        }
        if bytes[i] == quote {
            return i + 1;
        }
        i += 1;
    }
    // Unterminated string -- return end of input
    bytes.len()
}

fn restriction_reason(name: &str) -> &'static str {
    match name {
        "matches" => {
            "Regex has unbounded cost per record. Use contains()/startsWith()/endsWith() instead, or set allow_regex: true in expression config."
        }
        "map" | "filter" | "exists" | "all" | "exists_one" => {
            "Per-element iteration has O(n) cost proportional to collection size. Set allow_iteration: true in expression config to permit."
        }
        "timestamp" | "duration" => {
            "Time functions excluded -- ClickHouse handles time natively. Set allow_time: true in expression config to permit."
        }
        _ => "Restricted by the expression profile.",
    }
}

fn is_ident_start(b: u8) -> bool {
    b.is_ascii_alphabetic() || b == b'_'
}

fn is_ident_char(b: u8) -> bool {
    b.is_ascii_alphanumeric() || b == b'_'
}

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

    // ── Default config (all restricted) ─────────────────────────

    #[test]
    fn allowed_function_passes() {
        assert!(check_profile(r#"msg.contains("error")"#).is_empty());
    }

    #[test]
    fn starts_with_passes() {
        assert!(check_profile(r#"path.startsWith("/api/")"#).is_empty());
    }

    #[test]
    fn ends_with_passes() {
        assert!(check_profile(r#"file.endsWith(".log")"#).is_empty());
    }

    #[test]
    fn matches_blocked_by_default() {
        let errors = check_profile(r#"name.matches("^web-[0-9]+$")"#);
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("matches()"));
        assert!(errors[0].contains("allow_regex"));
    }

    /// A function in neither the allowlist nor a restricted category must be
    /// rejected. The scan only consulted the blocked list, so a typo passed the
    /// profile, and the CEL crate resolves functions at evaluation -- the caller
    /// was told a misspelt expression was valid.
    #[test]
    fn unknown_function_rejected() {
        for expr in [
            "nosuchfunc(x)",
            r#"severity == "critical" && bogus(z)"#,
            "startWith(host)",
            "sizeof(tags) > 3",
        ] {
            let errors = check_profile(expr);
            assert!(
                !errors.is_empty(),
                "{expr} passed the profile despite an unknown function"
            );
        }
    }

    /// Unlocking every restricted category must permit those categories, not
    /// switch the scan off. `blocked_functions()` returns empty when all three
    /// flags are set, and the early return then skipped the allowlist check too,
    /// so the most permissive config was also the one that validated nothing.
    #[test]
    fn all_categories_unlocked_still_rejects_an_unknown_function() {
        let config = ProfileConfig {
            allow_regex: true,
            allow_iteration: true,
            allow_time: true,
        };
        assert!(
            check_profile_with_config(r#"msg.contains("x")"#, &config).is_empty(),
            "an allowed function must still pass with every category unlocked"
        );
        assert!(
            !check_profile_with_config("nosuchfunc(x)", &config).is_empty(),
            "unlocking every category disabled the profile scan entirely"
        );
    }

    /// The allowlist must not refuse legitimate profile expressions. An omission
    /// here rejects a valid operator filter, which is the cost of enforcing it.
    #[test]
    fn profile_functions_still_pass() {
        for expr in [
            r#"has(user.id) && host.startsWith("web-")"#,
            "size(tags) > 3",
            r#"int(count) > 100 && message.contains("error")"#,
            r#"double(score) >= 0.5 && string(id).endsWith("-x")"#,
            "bytes(payload).size() > 0",
            "type(value) == string",
            r#"dyn(meta).contains("k")"#,
            "created.getFullYear() == 2026",
        ] {
            let errors = check_profile(expr);
            assert!(errors.is_empty(), "{expr} was rejected: {errors:?}");
        }
    }

    #[test]
    fn disallowed_map_rejected() {
        let errors = check_profile("[1,2,3].map(x, x * 2)");
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("map()"));
    }

    #[test]
    fn disallowed_filter_rejected() {
        let errors = check_profile("[1,2,3].filter(x, x > 1)");
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("filter()"));
    }

    #[test]
    fn disallowed_timestamp_rejected() {
        let errors = check_profile(r#"timestamp("2024-01-01T00:00:00Z")"#);
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("timestamp()"));
    }

    #[test]
    fn disallowed_duration_rejected() {
        let errors = check_profile(r#"duration("1h")"#);
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("duration()"));
    }

    #[test]
    fn keywords_skipped() {
        assert!(check_profile("has(user.name)").is_empty());
        assert!(check_profile("int(x) > 10").is_empty());
        assert!(check_profile("bool(y)").is_empty());
    }

    #[test]
    fn plain_comparison_passes() {
        assert!(check_profile(r#"severity == "critical""#).is_empty());
    }

    #[test]
    fn compound_expression_passes() {
        assert!(check_profile(r#"severity == "critical" && amount > 10000"#).is_empty());
    }

    // ── String literal false-positive prevention ────────────────

    #[test]
    fn function_name_inside_string_not_flagged() {
        // "filter" appears inside a string literal, not as a function call
        assert!(check_profile(r#"msg.contains("filter")"#).is_empty());
    }

    #[test]
    fn function_name_inside_string_with_parens_not_flagged() {
        // "map(" appears inside a string -- should not be flagged
        assert!(check_profile(r#"msg.contains("map(x)")"#).is_empty());
    }

    #[test]
    fn matches_inside_string_not_flagged() {
        assert!(check_profile(r#"msg.contains("matches")"#).is_empty());
    }

    #[test]
    fn timestamp_inside_string_not_flagged() {
        assert!(check_profile(r#"label == "timestamp""#).is_empty());
    }

    #[test]
    fn escaped_quote_inside_string_handled() {
        // String with escaped quote: "filter\"(" -- scanner must not exit early
        assert!(check_profile(r#"msg.contains("filter\"(")"#).is_empty());
    }

    #[test]
    fn single_quoted_string_handled() {
        assert!(check_profile("msg.contains('filter')").is_empty());
    }

    #[test]
    fn real_call_after_string_still_caught() {
        // String contains "ok" but then a real map() call follows
        let errors = check_profile(r#""ok" + items.map(x, x)"#);
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("map()"));
    }

    // ── Config overrides ────────────────────────────────────────

    #[test]
    fn matches_allowed_with_regex_config() {
        let config = ProfileConfig {
            allow_regex: true,
            ..Default::default()
        };
        assert!(check_profile_with_config(r#"name.matches("^web-[0-9]+$")"#, &config).is_empty());
    }

    #[test]
    fn map_still_blocked_with_regex_config() {
        let config = ProfileConfig {
            allow_regex: true,
            ..Default::default()
        };
        let errors = check_profile_with_config("[1,2].map(x, x)", &config);
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("map()"));
    }

    #[test]
    fn iteration_allowed_with_config() {
        let config = ProfileConfig {
            allow_iteration: true,
            ..Default::default()
        };
        assert!(check_profile_with_config("[1,2].map(x, x * 2)", &config).is_empty());
        assert!(check_profile_with_config("[1,2].filter(x, x > 1)", &config).is_empty());
        assert!(check_profile_with_config("[1,2].exists(x, x > 1)", &config).is_empty());
    }

    #[test]
    fn time_allowed_with_config() {
        let config = ProfileConfig {
            allow_time: true,
            ..Default::default()
        };
        assert!(
            check_profile_with_config(r#"timestamp("2024-01-01T00:00:00Z")"#, &config).is_empty()
        );
        assert!(check_profile_with_config(r#"duration("1h")"#, &config).is_empty());
    }

    #[test]
    fn all_restrictions_lifted() {
        let config = ProfileConfig {
            allow_regex: true,
            allow_iteration: true,
            allow_time: true,
        };
        assert!(config.blocked_functions().is_empty());
        assert!(
            check_profile_with_config(r#"name.matches("x") && [1].map(x, x)"#, &config).is_empty()
        );
    }

    // ── Edge cases ──────────────────────────────────────────────

    #[test]
    fn identifier_not_followed_by_paren_is_fine() {
        // "filter" as a variable name, not a function call
        assert!(check_profile("filter > 10").is_empty());
    }

    #[test]
    fn identifier_with_space_before_paren() {
        let errors = check_profile("[1,2].map (x, x)");
        assert_eq!(errors.len(), 1);
        assert!(errors[0].contains("map()"));
    }

    #[test]
    fn empty_expression() {
        assert!(check_profile("").is_empty());
    }

    #[test]
    fn whitespace_only() {
        assert!(check_profile("   ").is_empty());
    }

    #[test]
    fn multiple_violations_reported() {
        let errors = check_profile("[1].map(x, x).filter(y, y > 0)");
        assert_eq!(errors.len(), 2);
    }
}