[fix] validate sql reaching the alert query executor

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/expr/AlertExpressionEvalVisitor.java

@@ -39,6 +39,11 @@ public class AlertExpressionEvalVisitor extends AlertExpressionBaseVisitor<List<
     private static final String VALUE = "__value__";
     private static final String TIMESTAMP = "__timestamp__";
 
+    /**
+     * Every statement this visitor evaluates goes to this executor, so whether a statement is
+     * allowed to run is decided there rather than at each visit method, see
+     * {@code DataSourceServiceImpl}.
+     */
     private final QueryExecutor executor;
     private final CommonTokenStream tokens;
 
@@ -259,17 +264,16 @@ public List<Map<String, Object>> visitSqlExpr(AlertExpressionParser.SqlExprConte
 
     @Override
     public List<Map<String, Object>> visitSqlCallExpr(AlertExpressionParser.SqlCallExprContext ctx) {
-        return callSqlOrPromql(tokens.getText(ctx.string()));
+        return executor.execute(unquote(tokens.getText(ctx.string())));
     }
 
     @Override
     public List<Map<String, Object>> visitPromqlCallExpr(AlertExpressionParser.PromqlCallExprContext ctx) {
-        return callSqlOrPromql(tokens.getText(ctx.string()));
+        return executor.execute(unquote(tokens.getText(ctx.string())));
     }
 
-    private List<Map<String, Object>> callSqlOrPromql(String text) {
-        String script = text.substring(1, text.length() - 1);
-        return executor.execute(script);
+    private String unquote(String text) {
+        return text.substring(1, text.length() - 1);
     }
 
     /**

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/DataSourceServiceImpl.java

@@ -31,7 +31,6 @@
 import org.apache.hertzbeat.alert.expr.AlertExpressionParser;
 import org.apache.hertzbeat.alert.service.DataSourceService;
 import org.apache.hertzbeat.common.support.exception.AlertExpressionException;
-import org.apache.hertzbeat.common.support.valid.SqlSecurityException;
 import org.apache.hertzbeat.common.support.valid.SqlSecurityValidator;
 import org.apache.hertzbeat.common.util.ResourceBundleUtil;
 import org.apache.hertzbeat.warehouse.constants.WarehouseConstants;
@@ -58,6 +57,13 @@ public class DataSourceServiceImpl implements DataSourceService {
      */
     private static final List<String> DEFAULT_ALLOWED_TABLES = List.of(WarehouseConstants.LOG_TABLE_NAME);
 
+    /**
+     * The policy for an alert expression is read only and nothing narrower: which tables it
+     * may read is not constrained, because metric tables are created per metric on demand and
+     * a whitelist would reject every legitimate metric query.
+     */
+    private static final SqlSecurityValidator EXPRESSION_SQL_VALIDATOR = SqlSecurityValidator.selectOnly();
+
     protected ResourceBundle bundle = ResourceBundleUtil.getBundle("alerter");
 
     @Setter
@@ -93,7 +99,7 @@ public List<Map<String, Object>> calculate(String datasource, String expr) {
         // replace all white space
         expr = expr.replaceAll("\\s+", " ");
         try {
-            return evaluate(expr, executor);
+            return evaluate(expr, guardSql(executor, EXPRESSION_SQL_VALIDATOR));
         } catch (AlertExpressionException ae) {
             log.error("Calculate query parse error, datasource: {}, expr: {}, msg: {}", datasource, expr, ae.getMessage(), ae);
             throw ae;
@@ -119,36 +125,34 @@ public List<Map<String, Object>> query(String datasource, String expr) {
         // replace all white space
         expr = expr.replaceAll("\\s+", " ");
 
-        // SQL security validation for SQL-based datasources
-        if (isSqlDatasource(datasource)) {
-            validateSqlSecurity(expr);
-        }
-
         try {
-            return executor.execute(expr);
+            return guardSql(executor, sqlSecurityValidator).execute(expr);
+        } catch (AlertExpressionException ae) {
+            // A statement the policy rejected, whose message names the part it broke.
+            throw ae;
         } catch (Exception e) {
             log.error("Error executing query on datasource {}: {}", datasource, e.getMessage());
             throw new AlertExpressionException(e.getMessage());
         }
     }
 
     /**
-     * Check if the datasource is SQL-based
-     */
-    private boolean isSqlDatasource(String datasource) {
-        return datasource != null && datasource.equalsIgnoreCase(WarehouseConstants.SQL);
-    }
-
-    /**
-     * Validate SQL statement for security
+     * Wraps an executor that speaks sql so that nothing runs on it unvalidated.
+     *
+     * <p>The decision is made from the executor rather than from the datasource string the
+     * caller passed, because the executor is what actually holds the database credentials.
+     * A datasource that does not speak sql is handed back untouched: a promql endpoint takes
+     * a query string, not a statement, and running it through a sql parser would only reject
+     * valid promql.
+     * @param executor Executor chosen for this datasource
+     * @param validator Policy to enforce, read only for expressions and whitelisting for raw log queries
+     * @return The executor, guarded when it speaks sql
      */
-    private void validateSqlSecurity(String sql) {
-        try {
-            sqlSecurityValidator.validate(sql);
-        } catch (SqlSecurityException e) {
-            log.warn("SQL security validation failed: {}", e.getMessage());
-            throw new AlertExpressionException("SQL security validation failed: " + e.getMessage());
+    private QueryExecutor guardSql(QueryExecutor executor, SqlSecurityValidator validator) {
+        if (!executor.support(WarehouseConstants.SQL)) {
+            return executor;
         }
+        return new SqlValidatingQueryExecutor(executor, validator);
     }
 
     private List<Map<String, Object>> evaluate(String expr, QueryExecutor executor) {

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/service/impl/SqlValidatingQueryExecutor.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.alert.service.impl;
+
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
+import org.apache.hertzbeat.common.support.exception.AlertExpressionException;
+import org.apache.hertzbeat.common.support.valid.SqlSecurityException;
+import org.apache.hertzbeat.common.support.valid.SqlSecurityValidator;
+import org.apache.hertzbeat.warehouse.db.QueryExecutor;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A sql executor that validates before it runs anything.
+ *
+ * <p>An alert expression reaches a query executor by several routes: the {@code sql("...")}
+ * and {@code promql("...")} spellings both carry an arbitrary string, a bare select is
+ * parsed by the expression grammar itself, and each of them is evaluated for the preview
+ * endpoint and for the periodic evaluation loop alike. All of them end at
+ * {@link QueryExecutor#execute(String)}, which runs the string with the server side database
+ * credentials.
+ *
+ * <p>Guarding that one method rather than each route is what makes the check complete: the
+ * spelling an expression happens to use does not decide whether the statement is checked,
+ * the database it lands on does. It also means a route added later is covered without anyone
+ * remembering to add a call.
+ */
+public class SqlValidatingQueryExecutor implements QueryExecutor {
+
+    private final QueryExecutor delegate;
+
+    private final SqlSecurityValidator validator;
+
+    public SqlValidatingQueryExecutor(QueryExecutor delegate, SqlSecurityValidator validator) {
+        this.delegate = delegate;
+        this.validator = validator;
+    }
+
+    @Override
+    public List<Map<String, Object>> execute(String query) {
+        try {
+            validator.validate(query);
+        } catch (SqlSecurityException e) {
+            // AlertExpressionException rather than a generic failure: it is the type
+            // DataSourceServiceImpl rethrows untouched and the preview endpoint turns into a
+            // 400, so the author of the rule sees which part of the policy the statement broke
+            throw new AlertExpressionException("SQL security validation failed: " + e.getMessage());
+        }
+        return delegate.execute(query);
+    }
+
+    @Override
+    public DatasourceQueryData query(DatasourceQuery datasourceQuery) {
+        return delegate.query(datasourceQuery);
+    }
+
+    @Override
+    public String getDatasource() {
+        return delegate.getDatasource();
+    }
+
+    @Override
+    public boolean support(String queryLanguage) {
+        return delegate.support(queryLanguage);
+    }
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/expr/AlertExpressionEvalVisitorTest.java

@@ -864,10 +864,55 @@ void testComparisonExpr() {
         assertEquals(0, result.get(0).get("__value__"));
     }
 
+    @Test
+    void testSqlCallRunsPlainRead() {
+        when(mockExecutor.execute("select value from cpu where host = 'server1'"))
+                .thenReturn(List.of(new HashMap<>(Map.of("__value__", 80.0))));
+
+        final List<Map<String, Object>> result =
+                evaluate("sql(\"select value from cpu where host = 'server1'\") > 70");
+
+        assertEquals(1, result.size());
+        assertEquals(80.0, result.get(0).get("__value__"));
+    }
+
+    /**
+     * Subqueries and nested aggregation are supported in alert expressions, and with every
+     * metric table already readable, rejecting them would cost features without denying an
+     * attacker anything. The policy stops at read only on purpose.
+     */
+    @Test
+    void testSqlCallKeepsSubqueriesWorking() {
+        final String sql = "select value from cpu where host = (select host from hosts limit 1)";
+        when(mockExecutor.execute(sql)).thenReturn(List.of(new HashMap<>(Map.of("__value__", 80.0))));
+
+        final List<Map<String, Object>> result = evaluate("sql(\"" + sql + "\") > 70");
+
+        assertEquals(1, result.size());
+        assertEquals(80.0, result.get(0).get("__value__"));
+    }
+
+    /**
+     * The visitor hands both spellings to the executor as written. Whether a statement is
+     * allowed to run is decided by the executor it lands on, see
+     * {@code DataSourceServiceTest}, so promql keeps working for text that is not valid sql
+     * at all.
+     */
+    @Test
+    void testBothCallSpellingsReachTheExecutorAsWritten() {
+        when(mockExecutor.execute("rate(http_requests_total[5m])"))
+                .thenReturn(List.of(new HashMap<>(Map.of("__value__", 80.0))));
+
+        final List<Map<String, Object>> result = evaluate("promql(\"rate(http_requests_total[5m])\") > 70");
+
+        assertEquals(1, result.size());
+        assertEquals(80.0, result.get(0).get("__value__"));
+    }
+
     private List<Map<String, Object>> evaluate(String expression) {
         AlertExpressionLexer lexer = new AlertExpressionLexer(CharStreams.fromString(expression));
         CommonTokenStream tokens = new CommonTokenStream(lexer);
         AlertExpressionParser parser = new AlertExpressionParser(tokens);
         return new AlertExpressionEvalVisitor(mockExecutor, tokens).visit(parser.expression());
     }
-}
+}

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/DataSourceServiceTest.java

@@ -787,4 +787,64 @@ void query12() {
                 () -> dataSourceService.query("sql", "SELEC * FORM hertzbeat_logs"));
         verify(mockExecutor, never()).execute(anyString());
     }
+
+    /**
+     * The expression grammar offers three ways to reach the executor, and the datasource the
+     * caller names is what picks that executor. So a statement written with the
+     * {@code promql("...")} spelling still lands on the sql executor, with the server side
+     * database credentials behind it, whenever the caller names the sql datasource: whether a
+     * statement may run cannot be decided from the spelling.
+     */
+    @Test
+    void calculateRejectsWritesWhateverSpellingTheyArrivedIn() {
+        final QueryExecutor mockExecutor = Mockito.mock(QueryExecutor.class);
+        when(mockExecutor.support("sql")).thenReturn(true);
+        dataSourceService.setExecutors(List.of(mockExecutor));
+
+        assertThrows(AlertExpressionException.class,
+                () -> dataSourceService.calculate("sql", "sql(\"drop table cpu\") > 0"));
+        assertThrows(AlertExpressionException.class,
+                () -> dataSourceService.calculate("sql", "promql(\"drop table cpu\") > 0"));
+        assertThrows(AlertExpressionException.class,
+                () -> dataSourceService.calculate("sql", "sql(\"select 1; drop table cpu\") > 0"));
+        verify(mockExecutor, never()).execute(anyString());
+    }
+
+    /**
+     * Only a datasource that speaks sql is guarded. A promql endpoint takes a query string
+     * rather than a statement, so running it through a sql parser would reject valid promql
+     * without denying an attacker anything.
+     */
+    @Test
+    void calculateLeavesPromqlDatasourcesAlone() {
+        final QueryExecutor mockExecutor = Mockito.mock(QueryExecutor.class);
+        when(mockExecutor.support("promql")).thenReturn(true);
+        when(mockExecutor.execute(anyString())).thenReturn(List.of(new HashMap<>(Map.of("__value__", 100.0))));
+        dataSourceService.setExecutors(List.of(mockExecutor));
+
+        final List<Map<String, Object>> result = dataSourceService.calculate(
+                "promql", "promql(\"rate(http_requests_total[5m])\") > 70");
+
+        assertEquals(1, result.size());
+        verify(mockExecutor).execute("rate(http_requests_total[5m])");
+    }
+
+    /**
+     * A read still has to run, including the GreptimeDB range query syntax that the sql
+     * parser cannot read.
+     */
+    @Test
+    void calculateStillRunsReadsOnSqlDatasources() {
+        final QueryExecutor mockExecutor = Mockito.mock(QueryExecutor.class);
+        when(mockExecutor.support("sql")).thenReturn(true);
+        when(mockExecutor.execute(anyString())).thenReturn(List.of(new HashMap<>(Map.of("__value__", 100.0))));
+        dataSourceService.setExecutors(List.of(mockExecutor));
+
+        final String rangeQuery = "select avg(value) RANGE '10s' from cpu ALIGN '5s'";
+        final List<Map<String, Object>> result =
+                dataSourceService.calculate("sql", "sql(\"" + rangeQuery + "\") > 70");
+
+        assertEquals(1, result.size());
+        verify(mockExecutor).execute(rangeQuery);
+    }
 }