diff --git a/data/timesketch.conf b/data/timesketch.conf index 20aba88730..7c36df8202 100644 --- a/data/timesketch.conf +++ b/data/timesketch.conf @@ -39,6 +39,10 @@ OPENSEARCH_SSL = False OPENSEARCH_VERIFY_CERTS = True OPENSEARCH_CA_CERTS = None OPENSEARCH_TIMEOUT = 10 +# Connections kept alive per node. Needs to cover the requests a single +# process can have in flight at once: the web server's threads, or a +# worker's Celery concurrency. +OPENSEARCH_POOL_MAXSIZE = 20 OPENSEARCH_FLUSH_INTERVAL = 5000 OPENSEARCH_FLUSH_BYTE_SIZE = 52428800 OPENSEARCH_INDEX_WAIT_TIMEOUT = 10 diff --git a/docs/guides/user/search-query-guide.md b/docs/guides/user/search-query-guide.md index 1101e9fbdf..fb0cfdc1c6 100644 --- a/docs/guides/user/search-query-guide.md +++ b/docs/guides/user/search-query-guide.md @@ -265,3 +265,218 @@ Here are some common searches: ## Common questions There is a frequent question around Windows Event logs and how they are represented in Timesketch when imported from Plaso. For that we recommend reading up on [Common misconception about Windows EventLogs](https://osdfir.blogspot.com/2021/10/common-misconceptions-about-windows.html) + +## PPL and SQL + +Query String and Wildcard search return events. +[PPL](https://docs.opensearch.org/latest/sql-and-ppl/ppl/index/) and +[SQL](https://docs.opensearch.org/latest/sql-and-ppl/sql/index/) are sent to +OpenSearch directly, so they can also aggregate: count events per host, group +events by `data_type`, or rank field values by how often they occur. Use them +for questions about totals and distributions, and use Query String when you +want to read the events themselves. + +Both languages need OpenSearch 3.7.0 or later with the SQL plugin installed. If +`PPL` and `SQL` are missing from the search mode menu, the cluster does not +provide them. Timesketch re-checks the cluster periodically, so the modes appear +on their own after an upgrade. + +### Selecting a search mode + +The button on the left of the query bar selects the language: + +| Mode | Language | +| ----- | ------------------------------------- | +| `QS` | Query String (Lucene), the default | +| `WC` | [Wildcard](#wildcard-search-mode) | +| `PPL` | OpenSearch Piped Processing Language | +| `SQL` | OpenSearch SQL | + +### Indexes and query scope + +PPL and SQL run their searches directly against the OpenSearch indexes that hold +a sketch's events. In both languages an index takes the place of a table, and +Timesketch's own structure does not exist at that level: one index can hold +several timelines, and some of them may belong to other sketches. + +Timesketch closes that gap for you. It names the sketch's indexes in the query +and adds a filter for the timelines the sketch contains, so results never reach +events from outside it. Leave both out of what you type: + +* In PPL, begin with the first command, for example `stats count() by data_type`. + Do not write `source=` and do not begin with a pipe. +* In SQL, begin with `SELECT` and write no `FROM` clause. + +A query naming any other index is rejected. SQL is limited to `SELECT`, `SHOW` +and `DESCRIBE`, so nothing can be modified through this interface. + +### PPL + +| Description | Example query | +| ------------------------------- | --------------------------------------------------------- | +| First 100 events | `head 100` | +| Filter on message text | `where message like 'Failed%' \| head 100` | +| Count events per data type | `stats count() as cnt by data_type \| sort - cnt` | +| Ten most common hostnames | `stats count() as cnt by hostname \| sort - cnt \| head 10` | +| One event per value of a field | `dedup source_short \| head 50` | +| Newest events first | `sort - datetime \| head 100` | + +`sort` takes field names and aliases, never function calls. Name the +aggregation with `as` and sort on that name: + +``` +stats count() by hostname | sort - count() <- rejected +stats count() as cnt by hostname | sort - cnt <- works +``` + +Inside `stats`, the aggregation comes before `by`, as in +`stats count() as cnt by hostname`. Row limits use `head N`; PPL has no `LIMIT`. + +OpenSearch documents every command and its options in +[PPL commands](https://docs.opensearch.org/latest/sql-and-ppl/ppl/commands/index/). + +### SQL + +| Description | Example query | +| -------------------------- | -------------------------------------------------------------------------------------- | +| First 100 events | `SELECT datetime, message, timestamp_desc LIMIT 100` | +| Filter on message text | `SELECT datetime, message WHERE message LIKE 'Failed%' LIMIT 100` | +| Count events per data type | `SELECT data_type, COUNT(*) AS cnt GROUP BY data_type ORDER BY cnt DESC LIMIT 20` | + +Always give `ORDER BY` a `LIMIT`. Without one, OpenSearch sorts the whole result +set in memory and may return nothing. + +The supported clauses and their order of execution are documented in +[Basic SQL queries](https://docs.opensearch.org/latest/sql-and-ppl/sql/basic/). + +### Field types + +`datetime` is the only date field, and date detection is switched off, so a +field holding a formatted timestamp is text rather than a date. + +Every string value is indexed as `text`, with a `keyword` sub-field for exact +matches and a `wildcard` sub-field for substring matches. The `keyword` +sub-field is skipped for values longer than 256 characters, so very long values +cannot be matched exactly or aggregated on. + +Numeric-looking fields are often not numeric. `plaso.mappings` maps several of +them as text on purpose, among them `file_size`, `offset`, `sequence_number`, +`source_port`, `exit_status`, `severity`, `version` and `http_response_bytes`. +Comparing one of these against a bare number matches nothing, because a text +field holds the digits as characters: + +``` +file_size = 4096 <- no rows, file_size is text +file_size = '4096' <- works +``` + +Fields absent from `plaso.mappings` and `generic.mappings` are typed from the +first value indexed, so their type follows the data rather than a fixed schema. +If a comparison against a number returns no rows, quote the value and run it +again. [Index Mappings](../admin/index-mappings.md) describes the mapping files +in full. + +### Filtering by time + +Use the time-range picker above the query bar rather than comparing `datetime` +inside the query. The picker builds the range filter itself, and it applies to +PPL and SQL exactly as it does to Query String. + +A range written in the query is easy to get wrong: an unquoted timestamp is +parsed as arithmetic, and a quoted one is compared against whatever type the +field turned out to be. When a range does belong in the query text, Query String +mode handles it predictably with `datetime:[2024-01-01 TO 2024-01-31]`. + +### Matching text with LIKE + +`LIKE 'Failed%'` matches values starting with `Failed`. OpenSearch can use the +index to find them, because the pattern begins with a literal. + +A pattern beginning with `%` gives OpenSearch nothing to start from, so it reads +every value in the field instead. On a large timeline that is slow. + +| Pattern | Matches | Uses the index | +| ----------------- | ---------------------------------- | -------------- | +| `'Failed%'` | values starting with `Failed` | yes | +| `'%@example.com'` | values ending with `@example.com` | no | +| `'%failed%'` | values containing `failed` | no | + +Anchor the beginning of the pattern whenever the data allows it. If you cannot, +anchor the end instead: `'%@example.com'` still scans the field, but far fewer +values match it than `'%@example%'`. + +### Grouping on a field that not every event has + +A sketch usually holds timelines from several sources, and a field one source +populates is missing from the others. Grouping on that field collects every +event lacking it into a single bucket. That bucket is frequently the largest, so +it sorts to the top and pushes the values you wanted out of view. + +No event actually holds that bucket's label as a value, so searching for it in +Query String mode returns nothing. Its count says only how many events had no +value for the field. + +Narrow the query to the source that populates the field, then group: + +``` +where data_type = 'windows:evtx:record' | stats count() as cnt by event_identifier | sort - cnt +``` + +If a count looks wrong, search for the same value in Query String mode. That +confirms the number and shows whether the value is one you can search for at +all. + +### Warnings shown next to the query + +The query bar flags the mistakes above before the query runs, and explains what +to change. A warning does not block anything, so a query with an unaliased sort +or an unlimited `ORDER BY` can still be submitted. + +A query OpenSearch rejects is reported in the results panel as a query error. +Failing to reach the cluster is reported as a separate kind of error, so a typo +in the query is distinguishable from OpenSearch being unavailable. + +### Reading the results + +Results appear as a table of the columns the query selected, rather than the +event list used by Query String search. + +Select a cell to search for that value in Query String mode. This is how you get +from a count back to the events behind it. Values too long for the `keyword` +sub-field cannot be searched this way. + +The panel can also show the execution plan, which is the query as OpenSearch +resolved it and therefore includes the index and timeline filters Timesketch +added. Compare it with what you typed when a result is not what you expected. +[Explain API](https://docs.opensearch.org/latest/sql-and-ppl/sql-and-ppl-api/index/#explain-api) +describes the plan formats. + +### Exporting results + +Export writes rows to a file as they arrive instead of holding them in memory, +so an export can be far larger than the table on screen. + +Use SQL for large exports. The SQL plugin pages with a +[cursor](https://docs.opensearch.org/latest/sql-and-ppl/sql-and-ppl-api/index/#paginating-results), +and every page costs roughly the same. Cursors are a SQL-only feature, so a PPL +export pages by re-running the query with a larger offset each time; the pages +grow slower as the export continues, and on a large timeline the later ones can +time out and end the export early. + +An export that stopped early ends with a JSON object containing +`"incomplete": true`, the number of rows written and the offset reached. Check +the last line before treating a file as complete. + +### Limits + +| Limit | Value | +| ---------------------------- | ------------------------------------------ | +| Rows returned by a SQL query | 1000 by default, 10000 at most | +| Rows returned by a PPL query | the SQL plugin's own row limit | +| Query and explain timeout | 30 seconds | +| Export page size | 10000 rows | +| Export page timeout | 60 seconds for SQL, 120 seconds for PPL | + +An aggregation reaching the row limit is truncated rather than wrong, but the +tail of the distribution is missing. Add a `where` stage or a `WHERE` clause to +bring the number of groups down. diff --git a/timesketch/api/v1/resources/direct_query/__init__.py b/timesketch/api/v1/resources/direct_query/__init__.py new file mode 100644 index 0000000000..1b852914e4 --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/__init__.py @@ -0,0 +1,43 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Direct PPL / SQL query resources for the Timesketch API. + +Proxies the OpenSearch `_plugins/_ppl` and `_plugins/_sql` endpoints, enforces +sketch-level ACLs, and injects a `source=`/`FROM` clause that restricts every +query to the sketch's own indices. + +PPL and SQL are separated into dedicated dialect modules (`ppl.py`, `sql.py`) +behind a common interface, following the same shape as the query-string / +wildcard split in `explore.py`: a thin per-language entry point over one shared +execution shell. The REST resources are re-exported here because they are the +package's public surface; everything else is imported from its own module. +""" + +from timesketch.api.v1.resources.direct_query.endpoints import ( + PplQueryExplainResource, + PplQueryExportResource, + PplQueryResource, + SqlQueryExplainResource, + SqlQueryExportResource, + SqlQueryResource, +) + +__all__ = [ + "PplQueryExplainResource", + "PplQueryExportResource", + "PplQueryResource", + "SqlQueryExplainResource", + "SqlQueryExportResource", + "SqlQueryResource", +] diff --git a/timesketch/api/v1/resources/direct_query/base.py b/timesketch/api/v1/resources/direct_query/base.py new file mode 100644 index 0000000000..bee7106753 --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/base.py @@ -0,0 +1,441 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Shared infrastructure for the direct PPL / SQL query resources. + +Everything in this module is dialect agnostic: OpenSearch connection details, +sketch index/timeline scoping, and response envelope construction. The +dialect-specific pieces live in `ppl.py` and `sql.py`. +""" + +import datetime +import logging +import re +import threading +import time + +from opensearchpy import exceptions as opensearch_exceptions + +from timesketch.lib.datastores.opensearch import build_opensearch_client + +logger = logging.getLogger("timesketch.direct_query_api") + +# Rows requested per round trip when streaming an export. Both dialects +# paginate at the same size; only the per-request timeout differs, and that +# lives with the dialect that uses it. +DIRECT_QUERY_EXPORT_PAGE_SIZE = 10000 + +# Applies to the execute and explain endpoints, which are dialect agnostic. +EXECUTE_TIMEOUT_SECONDS = 30 + +# The mapping probe runs before the query it guards, so it is kept short. +MAPPING_TIMEOUT_SECONDS = 5 + +# Planning only, no execution, so this stays well under the execute timeout. +EXPLAIN_VERIFY_TIMEOUT_SECONDS = 10 + +# Field carrying the timeline a document belongs to. Indices written by older +# Timesketch versions predate it. +TIMELINE_FIELD = "__ts_timeline_id" + +# Timesketch stores event time twice: `datetime` as a date and `timestamp` as +# microseconds since the epoch. Range filtering goes through the numeric field +# because date comparison in the SQL and PPL plugins depends on how the index +# declared its date format, and silently matches nothing when the two disagree. +# An integer comparison on a long has no such ambiguity. +TIME_RANGE_FIELD = "timestamp" + +# An index only gains or loses the field when a timeline is (re)indexed, so a +# few minutes of staleness is harmless and keeps the probe off the hot path. +TIMELINE_FIELD_CACHE_TTL_SECONDS = 300 + +# Bounds cache growth across many sketches. Entries are cheap and the whole map +# is discarded on overflow rather than evicted one by one; refilling costs one +# mapping call per pattern still in use. +TIMELINE_FIELD_CACHE_MAX_ENTRIES = 1024 + +_timeline_field_cache = {} +_timeline_field_cache_lock = threading.Lock() + +# One OpenSearch client for every call this package makes. The client is built +# once rather than per request: its transport holds the connection pool, and +# rebuilding it each time would pay TCP and TLS setup on every page of an +# export. The transport is thread safe, so it is shared across worker threads. +_client_holder = {"client": None} +_client_lock = threading.Lock() + + +def configure_client(app): + """Build the shared client once, at startup. + + The datastore's builder is used rather than a second copy of the host, + credential and TLS handling, so direct queries reach the cluster the way + the rest of Timesketch does -- across every node in ``OPENSEARCH_HOSTS`` + rather than whichever one happens to be listed first. + + Args: + app (Flask): application whose config carries the OpenSearch settings. + """ + with app.app_context(): + with _client_lock: + _client_holder["client"] = build_opensearch_client() + + +def get_client(): + """Return the shared OpenSearch client, building it on first use.""" + client = _client_holder["client"] + if client is not None: + return client + + with _client_lock: + if _client_holder["client"] is None: + _client_holder["client"] = build_opensearch_client() + return _client_holder["client"] + + +def reset_client(): + """Drop the shared client. Used by tests and after a config change.""" + with _client_lock: + _client_holder["client"] = None + + +def _parse_boundary(value, label, end_of_day): + """Parse one ISO 8601 boundary into epoch microseconds. + + A date with no time part covers the whole day: a start snaps to 00:00:00 + and an end to the last microsecond before midnight, so ``2026-04-07`` to + ``2026-04-07`` is that entire day rather than an empty instant. + + Raises: + ValueError: if the value is not a string or not ISO 8601. + """ + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{label} must be an ISO 8601 date or datetime string.") + + text = value.strip() + # `fromisoformat` reads "+00:00" but not the "Z" that browsers emit. + normalised = text[:-1] + "+00:00" if text.endswith("Z") else text + + try: + parsed = datetime.datetime.fromisoformat(normalised) + except ValueError as e: + raise ValueError( + f"{label} is not a valid ISO 8601 date or datetime: {value}" + ) from e + + date_only = len(text) == 10 + if date_only and end_of_day: + parsed = parsed + datetime.timedelta(days=1, microseconds=-1) + + if parsed.tzinfo is None: + # Timesketch timestamps are UTC, and an unqualified boundary that + # silently took the server's zone would shift every result. + parsed = parsed.replace(tzinfo=datetime.timezone.utc) + + return int(parsed.timestamp() * 1_000_000) + + +def parse_time_range(req_json): + """Return ``(start_micros, end_micros)`` for the request, or ``None``. + + Either boundary may be omitted for an open-ended range. Returns None when + neither is given. + + Raises: + ValueError: if a boundary is unparseable or the range is inverted. + """ + raw_start = req_json.get("start_time") + raw_end = req_json.get("end_time") + if raw_start is None and raw_end is None: + return None + + start = ( + None if raw_start is None else _parse_boundary(raw_start, "start_time", False) + ) + end = None if raw_end is None else _parse_boundary(raw_end, "end_time", True) + + if start is not None and end is not None and start > end: + raise ValueError("start_time must not be later than end_time.") + + return start, end + + +def time_range_predicate(time_range, conjunction="and"): + """Return a numeric range predicate for ``timestamp``, or an empty string. + + The expression itself is valid in both dialects: an integer comparison needs + no quoting, casting or date-format agreement. Only the conjunction is + spelled to match the surrounding dialect's style. + """ + if not time_range: + return "" + start, end = time_range + clauses = [] + if start is not None: + clauses.append(f"{TIME_RANGE_FIELD} >= {start}") + if end is not None: + clauses.append(f"{TIME_RANGE_FIELD} <= {end}") + return f" {conjunction} ".join(clauses) + + +def get_sketch_scope(sketch, timeline_ids=None): + """Return ``(index_pattern, timeline_ids)`` for a sketch in one pass. + + ``sketch.timelines`` is a lazily loaded relationship, so the index pattern + and the active timeline IDs are collected together rather than walking it + twice. + + The timeline IDs drive the ``__ts_timeline_id`` filter that scopes results + to the timelines actually in the sketch. Multiple timelines can share one + index, and deleting a timeline does not delete its documents, so index-name + scoping alone would surface events from co-located or removed timelines + ("orphaned" records). + + Args: + sketch (Sketch): the sketch whose timelines are being scoped. + timeline_ids (list): optional timeline IDs to restrict to. If empty or + None, all sketch timelines are used. + """ + selected = set(timeline_ids) if timeline_ids else None + seen = set() + indices = [] + active_timeline_ids = [] + + for timeline in sketch.timelines: + if not timeline.searchindex: + continue + if selected is not None and timeline.id not in selected: + continue + active_timeline_ids.append(timeline.id) + index_name = timeline.searchindex.index_name + if index_name and index_name not in seen: + seen.add(index_name) + indices.append(index_name) + + return ",".join(indices), active_timeline_ids + + +def _probe_timeline_field(index_pattern): + """Ask OpenSearch whether any index in the pattern maps the timeline field. + + Returns True when the mapping cannot be read. Injecting the predicate and + risking a loud query error is safer than silently widening a query's scope + because a mapping call happened to fail. + """ + try: + body = get_client().indices.get_field_mapping( + fields=TIMELINE_FIELD, + index=index_pattern, + ignore_unavailable=True, + request_timeout=MAPPING_TIMEOUT_SECONDS, + ) + except opensearch_exceptions.OpenSearchException as e: + logger.warning( + "Mapping probe for %s failed (%s); assuming %s is present.", + index_pattern, + e, + TIMELINE_FIELD, + ) + return True + + if not isinstance(body, dict): + return True + # OpenSearch answers per index, and reports the field only where it exists. + # One index carrying it is enough: a multi-index query resolves the field + # from the union of the mappings. + return any( + isinstance(entry, dict) and entry.get("mappings") for entry in body.values() + ) + + +def index_pattern_has_timeline_field(index_pattern): + """Return whether ``index_pattern`` maps the timeline field, with caching. + + The answer decides whether the ``__ts_timeline_id`` predicate can be used. + Under the Calcite query engine (the default from OpenSearch 3.0, with V2 + fallback disabled) naming a field that no index maps is a hard error rather + than a null comparison, so injecting it into a legacy index fails the whole + query. + """ + now = time.monotonic() + cached = _timeline_field_cache.get(index_pattern) + if cached and cached[0] > now: + return cached[1] + + has_field = _probe_timeline_field(index_pattern) + with _timeline_field_cache_lock: + if len(_timeline_field_cache) >= TIMELINE_FIELD_CACHE_MAX_ENTRIES: + _timeline_field_cache.clear() + _timeline_field_cache[index_pattern] = ( + now + TIMELINE_FIELD_CACHE_TTL_SECONDS, + has_field, + ) + return has_field + + +# Index references as they appear in an OpenSearch execution plan. The Calcite +# engine writes a text plan; the V2 engine writes a JSON tree that names the +# index either in a request string or, for joins, in a `tableName` field. +_PLAN_CALCITE_SCAN = re.compile( + r"CalciteLogicalIndexScan\(table=\[\[OpenSearch,\s*([^\]]+?)\s*\]\]" +) +_PLAN_INDEX_NAME = re.compile(r"indexName=([^\s,]+(?:,[^\s,]+)*)") + + +def plan_indices(plan): + """Return every index named in an OpenSearch execution plan. + + The plan is walked rather than pattern matched as a whole, because the + shape differs by engine and by query: Calcite emits a text plan, the V2 SQL + engine emits a JSON tree, and a V2 join names its tables in `tableName` + fields instead of a request string. A comma-joined multi-index pattern is + split, so the result is always individual index names. + """ + found = set() + + def walk(node): + if isinstance(node, dict): + for key, value in node.items(): + if key == "tableName" and isinstance(value, str): + found.add(value) + walk(value) + elif isinstance(node, list): + for item in node: + walk(item) + elif isinstance(node, str): + for pattern in (_PLAN_CALCITE_SCAN, _PLAN_INDEX_NAME): + for match in pattern.finditer(node): + found.add(match.group(1)) + + walk(plan) + + indices = set() + for name in found: + indices.update(part.strip() for part in name.split(",") if part.strip()) + return indices + + +def verify_scope_with_explain(dialect, scoped_query, allowed): + """Ask OpenSearch which indices the scoped query really reads. + + This is a second, independent check on top of the dialect's own scoping. + The dialect works from the query text with regexes, so it can only look for + syntax it knows about; the plan comes from the engine's own parser and + names every index the query will actually touch, including those reached + through PPL's `lookup`, `join` and subsearch commands. + + Returns an error message when the plan names an index outside the sketch, + otherwise None. + + A plan that cannot be read raises no objection, and the dialect's own + result stands. That keeps an unfamiliar plan format from taking the feature + down on an OpenSearch upgrade, at the cost of quietly falling back to + text-based scoping -- which is why it is logged. Note the dialect layer is + itself fail-closed, so this is a narrowing of defence in depth rather than + a hole. + """ + try: + plan = dialect.api(get_client()).explain( + body=dialect.explain_payload(scoped_query), + request_timeout=EXPLAIN_VERIFY_TIMEOUT_SECONDS, + ) + except ( + opensearch_exceptions.ConnectionError, + opensearch_exceptions.SerializationError, + ) as e: + logger.warning("Could not verify query scope against the plan: %s", e) + return None + except opensearch_exceptions.TransportError: + # Usually a query the plugin rejects outright; executing it will + # surface the real error to the user. + return None + + named = plan_indices(plan) + if not named: + logger.warning( + "No index found in the %s execution plan; falling back to " + "text-based scoping. The plan format may have changed.", + dialect.name.upper(), + ) + return None + + outside = sorted(named - set(allowed)) + if outside: + logger.warning( + "%s plan reads indices outside the sketch: %s", + dialect.name.upper(), + ", ".join(outside), + ) + return f"{dialect.name.upper()} query targets indices outside this sketch." + return None + + +def validate_query(query, dialect): + """Validate a query for non-emptiness and dialect read-only safety. + + Returns None if valid, else an error message. + """ + if not query or not query.strip(): + return "Query cannot be empty." + return dialect.validate(query) + + +def format_opensearch_error(data): + """Flatten an OpenSearch error body into a single message string.""" + error = data.get("error", {}) + if not isinstance(error, dict): + return str(error) + reason = error.get("reason", str(error)) + error_type = error.get("type", "") + details = error.get("details", "") + message = f"{error_type}: {reason}" + if details: + message += f"\n{details}" + return message + + +def error_message(exc): + """Return a readable message for an exception raised by the client. + + A query the plugin rejects carries its error document on the exception, + which names the syntax problem. A connection failure carries the + underlying exception there instead, and reads better as its own message. + """ + try: + info = exc.info + except (AttributeError, LookupError): + # `info` is a property over the third argument, which not every + # exception class carries. This runs on the error path, so it must not + # raise an error of its own. + info = None + + if isinstance(info, dict) and info.get("error"): + return format_opensearch_error(info) + return str(exc) + + +def empty_result(dialect, error, result_type="direct"): + """Build a result envelope carrying an error and no rows.""" + envelope = { + "result_type": result_type, + "language": dialect.name, + "error": str(error), + } + if result_type == "direct": + envelope.update({"columns": [], "datarows": [], "total": 0, "size": 0}) + return envelope + + +def columns_from_schema(data): + """Extract column names from an OpenSearch PPL/SQL response schema.""" + return [col.get("name", f"col_{i}") for i, col in enumerate(data.get("schema", []))] diff --git a/timesketch/api/v1/resources/direct_query/capability.py b/timesketch/api/v1/resources/direct_query/capability.py new file mode 100644 index 0000000000..09e9997c71 --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/capability.py @@ -0,0 +1,187 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Cluster capability checks for the direct-query languages. + +PPL and SQL are served by the OpenSearch SQL plugin rather than by the search +API the rest of Timesketch uses, so whether they work is a property of the +cluster and not of the sketch. Both the sketch metadata, which drops the +languages from the search-mode menu, and the query endpoints, which refuse the +request outright, read their answer from here. + +The probe goes over the same client the dialects use, so a cluster that answers +here is one that can serve the queries. +""" + +import logging +import threading +import time + +from flask import current_app +from opensearchpy import exceptions as opensearch_exceptions +from packaging import version + +from timesketch.api.v1.resources.direct_query.base import MAPPING_TIMEOUT_SECONDS +from timesketch.api.v1.resources.direct_query.base import get_client + +logger = logging.getLogger("timesketch.api.direct_query") + +# The dialects rely on behaviour that settled in 3.7.0, the Calcite engine +# being the default for PPL among it. Older clusters accept the same queries +# but differ in how a filter injected ahead of a stats stage is pushed down, +# and the scoping this feature depends on is exactly such a filter. +MINIMUM_OPENSEARCH_VERSION = "3.7.0" + +# Cluster properties change on a restart or an upgrade, not per request. They +# are cached, but not for the life of the process, so that a cluster upgraded +# underneath a running worker starts offering the languages without one. +PROBE_TTL_SECONDS = 300 + +_probe_lock = threading.Lock() +_probe = {"checked_at": 0.0, "result": None} + + +class DirectQuerySupport: + """Whether a cluster can serve the direct-query languages, and why not.""" + + def __init__(self, supported, reason=""): + self.supported = supported + self.reason = reason + + def __bool__(self): + return self.supported + + +def _probe_call(call, label): + """Run one capability call, or return None if the cluster cannot answer.""" + try: + return call() + except opensearch_exceptions.OpenSearchException as e: + logger.warning("Capability probe (%s) failed: %s", label, e) + return None + + +def _version_supported(raw_version): + """Compare a reported cluster version against the minimum. + + An unreadable or unparsable version is treated as supported. Taking a + working feature away because a probe could not answer is worse than + letting the cluster reject the query itself. + """ + if not isinstance(raw_version, str) or not raw_version: + logger.warning("Could not read the OpenSearch version; assuming support") + return DirectQuerySupport(True) + + try: + too_old = version.parse(raw_version) < version.parse(MINIMUM_OPENSEARCH_VERSION) + except version.InvalidVersion: + logger.warning( + "Unparsable OpenSearch version %s; assuming support", raw_version + ) + return DirectQuerySupport(True) + + if too_old: + return DirectQuerySupport( + False, + f"PPL and SQL require OpenSearch {MINIMUM_OPENSEARCH_VERSION} or " + f"later; this cluster reports {raw_version}.", + ) + return DirectQuerySupport(True) + + +def _plugin_supported(plugins): + """Decide support from a _cat/plugins document. + + A document that could not be read leaves support unchanged, for the same + reason an unreadable version does. + """ + if plugins is None: + return DirectQuerySupport(True) + + try: + present = any( + "sql" in (entry.get("component") or "").lower() for entry in plugins + ) + except AttributeError: + logger.warning("Unexpected shape from the OpenSearch plugin list") + return DirectQuerySupport(True) + + if not present: + return DirectQuerySupport( + False, + "The OpenSearch SQL plugin, which serves the PPL and SQL endpoints, " + "is not installed on this cluster.", + ) + return DirectQuerySupport(True) + + +def _probe_cluster(): + """Ask the cluster for its version and plugin list.""" + client = get_client() + + root = _probe_call( + lambda: client.info(request_timeout=MAPPING_TIMEOUT_SECONDS), "version" + ) + raw_version = None + if isinstance(root, dict): + raw_version = (root.get("version") or {}).get("number") + + supported = _version_supported(raw_version) + if not supported: + return supported + + plugins = _probe_call( + lambda: client.cat.plugins( + format="json", request_timeout=MAPPING_TIMEOUT_SECONDS + ), + "plugin list", + ) + return _plugin_supported(plugins) + + +def reset_cache(): + """Forget the cached probe. Used by tests and after a config change.""" + with _probe_lock: + _probe["checked_at"] = 0.0 + _probe["result"] = None + + +def direct_query_support(): + """Report whether this cluster can serve PPL and SQL. + + Returns: + A DirectQuerySupport carrying a reason when unsupported. Truthy when + the languages are available. + """ + # Mirrors the datastore's own version gate: the test suite has no cluster + # to ask, and a probe per sketch load would only buy a connection refusal. + if current_app.config.get("TESTING"): + return DirectQuerySupport(True) + + now = time.time() + with _probe_lock: + fresh = _probe["result"] is not None and now - _probe["checked_at"] < ( + PROBE_TTL_SECONDS + ) + if fresh: + return _probe["result"] + + # Probing outside the lock keeps a slow cluster from blocking every other + # request; the worst case is two workers probing at once, which is + # harmless. + result = _probe_cluster() + + with _probe_lock: + _probe["checked_at"] = now + _probe["result"] = result + return result diff --git a/timesketch/api/v1/resources/direct_query/dialect.py b/timesketch/api/v1/resources/direct_query/dialect.py new file mode 100644 index 0000000000..db1645c4db --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/dialect.py @@ -0,0 +1,85 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Dialect interface for direct PPL / SQL queries. + +A dialect owns everything that differs between the OpenSearch query languages: +read-only validation, sketch scoping, request payload shape, and export +pagination. The resource shell in `endpoints.py` owns everything they share. + +Dialect instances are created once at import time and reused for every +request, so dispatch costs a single dict lookup. +""" + + +class DirectQueryDialect: + """Base class describing one OpenSearch query language.""" + + # Language identifier used on the wire (request body and response). + name = "" + + def api(self, client): + """Return the plugin client namespace that serves this dialect. + + Both languages are served by the OpenSearch SQL plugin, which the + client exposes as two namespaces carrying the same ``query`` and + ``explain`` calls. Selecting one here is what binds a dialect to its + endpoints. + """ + raise NotImplementedError + + def validate(self, query): + """Return an error message if the query is not read-only, else None. + + The shared shell has already rejected empty queries. + """ + raise NotImplementedError + + def scope(self, query, index_pattern, timeline_ids, time_range=None): + """Restrict a query to the sketch's indices, timelines and time range. + + ``time_range`` is a ``(start_micros, end_micros)`` pair, either side of + which may be None for an open end. It is applied here rather than left + to the caller's query text so the predicate is built the one way that + is reliable across both plugins. + + Returns: + Tuple of (scoped_query, error_message). Exactly one is set. + """ + raise NotImplementedError + + def execute_payload( + self, scoped_query, req_json + ): # pylint: disable=unused-argument + """Build the request body for executing a query. + + Dialects that honour request options such as ``fetch_size`` read them + from ``req_json``. + + Raises: + ValueError: if a request option is unusable. The resource shell + turns this into a 400. + """ + return {"query": scoped_query} + + def explain_payload(self, scoped_query): + """Build the request body for explaining a query.""" + return {"query": scoped_query} + + def stream(self, client, scoped_query): + """Yield NDJSON lines for the full result set. + + Implementations must stay generators so results are streamed to the + client rather than buffered in memory. + """ + raise NotImplementedError diff --git a/timesketch/api/v1/resources/direct_query/endpoints.py b/timesketch/api/v1/resources/direct_query/endpoints.py new file mode 100644 index 0000000000..194c7948bc --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/endpoints.py @@ -0,0 +1,373 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""REST resources for direct PPL / SQL queries. + +Each dialect gets its own endpoints (``/explore/ppl/``, ``/explore/sql/``), +mirroring how wildcard search is separated from query-string search +(`/explore/` and `/explore_wildcard/`). A resource pins its dialect as a class +attribute, so the language is fixed by the route and can never be steered by +the request body. + +All six resources share one implementation, so ACL enforcement, query scoping, +and response shaping cannot drift between the dialects. +""" + +import collections +import logging + +from flask import Response +from flask import abort +from flask import request +from flask import stream_with_context +from flask_login import current_user +from flask_login import login_required +from flask_restful import Resource +from opensearchpy import exceptions as opensearch_exceptions + +from timesketch.api.v1 import resources +from timesketch.api.v1.resources.direct_query.base import EXECUTE_TIMEOUT_SECONDS +from timesketch.api.v1.resources.direct_query.base import columns_from_schema +from timesketch.api.v1.resources.direct_query.base import empty_result +from timesketch.api.v1.resources.direct_query.base import error_message +from timesketch.api.v1.resources.direct_query.base import get_client +from timesketch.api.v1.resources.direct_query.base import get_sketch_scope +from timesketch.api.v1.resources.direct_query.base import parse_time_range +from timesketch.api.v1.resources.direct_query.base import ( + index_pattern_has_timeline_field, +) +from timesketch.api.v1.resources.direct_query.base import validate_query +from timesketch.api.v1.resources.direct_query.base import verify_scope_with_explain +from timesketch.api.v1.resources.direct_query.capability import direct_query_support +from timesketch.api.v1.resources.direct_query.registry import PPL_DIALECT +from timesketch.api.v1.resources.direct_query.registry import SQL_DIALECT +from timesketch.lib.definitions import HTTP_STATUS_CODE_BAD_REQUEST +from timesketch.lib.definitions import HTTP_STATUS_CODE_FORBIDDEN +from timesketch.lib.definitions import HTTP_STATUS_CODE_NOT_FOUND +from timesketch.models.sketch import Sketch + +logger = logging.getLogger("timesketch.direct_query_api") + +HTTP_STATUS_CODE_BAD_GATEWAY = 502 + +PreparedQuery = collections.namedtuple( + "PreparedQuery", ["sketch", "dialect", "scoped_query", "index_pattern", "req_json"] +) + + +def _parse_timeline_ids(raw): + """Normalise the request's ``timeline_ids`` to a list of ints, or None. + + These only ever narrow the sketch's own timelines, so a bad value is a + client error. Aborts with a 400 rather than letting the value fail deeper + in scoping. + """ + if raw is None: + return None + if not isinstance(raw, list): + abort(HTTP_STATUS_CODE_BAD_REQUEST, "timeline_ids must be a list of integers.") + try: + return [int(timeline_id) for timeline_id in raw] + except (TypeError, ValueError): + abort(HTTP_STATUS_CODE_BAD_REQUEST, "timeline_ids must be a list of integers.") + return None + + +def _prepare_query(sketch_id, dialect): + """Run ACL checks and scope the request's query to the sketch. + + Args: + sketch_id (int): primary key for a sketch database model. + dialect (DirectQueryDialect): dialect pinned by the resource handling + the route. + + Returns: + A PreparedQuery. Aborts with an HTTP error on failure. + """ + sketch = Sketch.get_with_acl(sketch_id) + if not sketch: + abort(HTTP_STATUS_CODE_NOT_FOUND, "No sketch found with this ID.") + + if not sketch.has_permission(current_user, "read"): + abort( + HTTP_STATUS_CODE_FORBIDDEN, + "User does not have read access controls on sketch.", + ) + + if sketch.get_status.status == "archived": + abort( + HTTP_STATUS_CODE_BAD_REQUEST, + "Unable to query on an archived sketch.", + ) + + # The UI hides these languages on a cluster that cannot serve them, but an + # API client reaches this route directly, and an explicit refusal reads + # better than whatever the cluster would return. + support = direct_query_support() + if not support: + abort(HTTP_STATUS_CODE_BAD_REQUEST, support.reason) + + req_json = request.get_json(silent=True) + if not req_json: + abort(HTTP_STATUS_CODE_BAD_REQUEST, "Request body must be JSON.") + + query = req_json.get("query", "") + if not isinstance(query, str): + abort(HTTP_STATUS_CODE_BAD_REQUEST, "Query must be a string.") + query = query.strip() + + validation_error = validate_query(query, dialect) + if validation_error: + abort(HTTP_STATUS_CODE_BAD_REQUEST, validation_error) + + requested_timeline_ids = _parse_timeline_ids(req_json.get("timeline_ids")) + + try: + time_range = parse_time_range(req_json) + except ValueError as e: + abort(HTTP_STATUS_CODE_BAD_REQUEST, str(e)) + + # Scope to the active timelines (not just their indices) so shared/orphaned + # rows from other or deleted timelines are excluded. + index_pattern, filter_timeline_ids = get_sketch_scope( + sketch, requested_timeline_ids + ) + if not index_pattern: + abort( + HTTP_STATUS_CODE_BAD_REQUEST, + "No valid indices found for this sketch. " + "Make sure at least one timeline is selected.", + ) + + if filter_timeline_ids and not index_pattern_has_timeline_field(index_pattern): + # No index here carries __ts_timeline_id, so the predicate cannot be + # used: the Calcite engine treats an unmapped field as an error, not as + # null. Dropping it loses nothing, because every row in such an index + # predates the field and would match the predicate's isnull() branch. + filter_timeline_ids = [] + + scoped_query, scope_error = dialect.scope( + query, index_pattern, filter_timeline_ids, time_range + ) + if scope_error: + abort(HTTP_STATUS_CODE_FORBIDDEN, scope_error) + + # Second opinion from the engine's own planner, which sees index references + # the dialect's regexes may not know to look for. + plan_error = verify_scope_with_explain( + dialect, + scoped_query, + [name.strip() for name in index_pattern.split(",") if name.strip()], + ) + if plan_error: + abort(HTTP_STATUS_CODE_FORBIDDEN, plan_error) + + return PreparedQuery(sketch, dialect, scoped_query, index_pattern, req_json) + + +def _call_opensearch(call, dialect, sketch_id, action, result_type): + """Run one plugin call and return ``(data, error_response)``. + + Exactly one of the tuple members is set. ``error_response`` is a ready to + return ``(body, status)`` pair. + """ + try: + return call(), None + except ( + opensearch_exceptions.ConnectionError, + opensearch_exceptions.SerializationError, + ) as e: + logger.error( + "OpenSearch %s %s failed for sketch %s: %s", + dialect.name.upper(), + action, + sketch_id, + e, + exc_info=True, + ) + return None, ( + empty_result(dialect, f"Failed to connect to OpenSearch: {e}", result_type), + HTTP_STATUS_CODE_BAD_GATEWAY, + ) + except opensearch_exceptions.TransportError as e: + # A query the plugin rejects is a user error, not a gateway failure, so + # it is reported in the envelope with a 200. + return None, ( + empty_result(dialect, error_message(e), result_type), + 200, + ) + + +class _BaseDirectQueryResource(resources.ResourceMixin, Resource): + """Shared ACL and scoping behaviour for the direct query endpoints. + + Subclasses pin ``dialect`` to the language their route serves. + """ + + dialect = None + + def prepare(self, sketch_id): + return _prepare_query(sketch_id, self.dialect) + + +class _BaseExecuteResource(_BaseDirectQueryResource): + """Execute a query and return a tabular result set.""" + + @login_required + def post(self, sketch_id): + """Handles POST request to execute a PPL or SQL query. + + Args: + sketch_id (int): primary key for a sketch database model + + Returns: + JSON with query results in a tabular format + """ + prepared = self.prepare(sketch_id) + dialect = prepared.dialect + + try: + payload = dialect.execute_payload(prepared.scoped_query, prepared.req_json) + except ValueError as e: + # A dialect rejects unusable request options (e.g. fetch_size) this + # way, which is a client error rather than a server fault. + abort(HTTP_STATUS_CODE_BAD_REQUEST, str(e)) + + data, error_response = _call_opensearch( + lambda: dialect.api(get_client()).query( + body=payload, request_timeout=EXECUTE_TIMEOUT_SECONDS + ), + dialect, + sketch_id, + "query", + "direct", + ) + if error_response: + return error_response + + columns = columns_from_schema(data) + datarows = data.get("datarows", []) + return { + "result_type": "direct", + "language": dialect.name, + "columns": columns, + "datarows": datarows, + "total": data.get("total", len(datarows)), + "size": data.get("size", len(datarows)), + "error": None, + } + + +class _BaseExplainResource(_BaseDirectQueryResource): + """Return the OpenSearch execution plan without running the query.""" + + @login_required + def post(self, sketch_id): + """Handles POST request to explain a PPL or SQL query. + + Applies the same ACL checks and query scoping as the execute endpoint. + + Args: + sketch_id (int): primary key for a sketch database model + + Returns: + JSON with the query execution plan + """ + prepared = self.prepare(sketch_id) + dialect = prepared.dialect + + data, error_response = _call_opensearch( + lambda: dialect.api(get_client()).explain( + body=dialect.explain_payload(prepared.scoped_query), + request_timeout=EXECUTE_TIMEOUT_SECONDS, + ), + dialect, + sketch_id, + "explain", + "direct_explain", + ) + if error_response: + return error_response + + return { + "result_type": "direct_explain", + "language": dialect.name, + "plan": data, + "error": None, + } + + +class _BaseExportResource(_BaseDirectQueryResource): + """Stream a full result set as NDJSON.""" + + @login_required + def post(self, sketch_id): + """Handles POST request to stream PPL/SQL results. + + Uses SQL cursor-based pagination or PPL size/from pagination to stream + all results as NDJSON. + + Args: + sketch_id: Integer primary key for a sketch database model + + Returns: + Streaming NDJSON response + """ + prepared = self.prepare(sketch_id) + generator = prepared.dialect.stream(get_client(), prepared.scoped_query) + return Response( + stream_with_context(generator), + mimetype="application/x-ndjson", + ) + + +# --------------------------------------------------------------------------- +# PPL resources (/explore/ppl/) +# --------------------------------------------------------------------------- +class PplQueryResource(_BaseExecuteResource): + """Handler for /api/v1/sketches/:sketch_id/explore/ppl/""" + + dialect = PPL_DIALECT + + +class PplQueryExplainResource(_BaseExplainResource): + """Handler for /api/v1/sketches/:sketch_id/explore/ppl/explain/""" + + dialect = PPL_DIALECT + + +class PplQueryExportResource(_BaseExportResource): + """Handler for /api/v1/sketches/:sketch_id/explore/ppl/export/""" + + dialect = PPL_DIALECT + + +# --------------------------------------------------------------------------- +# SQL resources (/explore/sql/) +# --------------------------------------------------------------------------- +class SqlQueryResource(_BaseExecuteResource): + """Handler for /api/v1/sketches/:sketch_id/explore/sql/""" + + dialect = SQL_DIALECT + + +class SqlQueryExplainResource(_BaseExplainResource): + """Handler for /api/v1/sketches/:sketch_id/explore/sql/explain/""" + + dialect = SQL_DIALECT + + +class SqlQueryExportResource(_BaseExportResource): + """Handler for /api/v1/sketches/:sketch_id/explore/sql/export/""" + + dialect = SQL_DIALECT diff --git a/timesketch/api/v1/resources/direct_query/ppl.py b/timesketch/api/v1/resources/direct_query/ppl.py new file mode 100644 index 0000000000..10b2af103d --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/ppl.py @@ -0,0 +1,371 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""PPL dialect for direct queries.""" + +import json +import logging +import re + +from opensearchpy import exceptions as opensearch_exceptions + +from timesketch.api.v1.resources.direct_query.base import DIRECT_QUERY_EXPORT_PAGE_SIZE +from timesketch.api.v1.resources.direct_query.base import columns_from_schema +from timesketch.api.v1.resources.direct_query.base import error_message +from timesketch.api.v1.resources.direct_query.base import time_range_predicate +from timesketch.api.v1.resources.direct_query.dialect import DirectQueryDialect + +logger = logging.getLogger("timesketch.direct_query_api") + +# Each export page re-runs the pipeline with a new offset, so a page can take +# considerably longer than a single execute. +EXPORT_TIMEOUT_SECONDS = 120 + +# Leading "search source=" clause of an already-scoped query. +_PPL_SOURCE_HEAD = re.compile( + r"^(search\s+source\s*=\s*(?:`[^`]+`|\S+))(.*)$", re.IGNORECASE | re.DOTALL +) + +# Same clause, but only the index identifier is captured. +_PPL_SEARCH_SOURCE = re.compile(r"^search\s+source\s*=\s*(`[^`]+`|\S+)", re.IGNORECASE) +_PPL_BARE_SOURCE = re.compile(r"^source\s*=\s*(`[^`]+`|\S+)", re.IGNORECASE) + +# A user-supplied "| head" stage, which already bounds the result set. +_PPL_HEAD_STAGE = re.compile(r"\|\s*head\b", re.IGNORECASE) + +# Commands that can name an index. Each is anchored to a command position -- +# the start of the query, just after a pipe, or just inside a subsearch -- so a +# field that happens to be called `source` or `lookup` is not mistaken for one. +_PPL_COMMAND_HEAD = r"(?:^|\||\[)\s*" +_PPL_IDENTIFIER = r"(`[^`]+`|[^\s|\]]+)" + +_PPL_SOURCE_REF = re.compile( + rf"{_PPL_COMMAND_HEAD}(?:search\s+)?source\s*=\s*{_PPL_IDENTIFIER}", re.IGNORECASE +) +_PPL_LOOKUP_REF = re.compile( + rf"{_PPL_COMMAND_HEAD}lookup\s+{_PPL_IDENTIFIER}", re.IGNORECASE +) +_PPL_DESCRIBE_REF = re.compile( + rf"{_PPL_COMMAND_HEAD}describe\s+{_PPL_IDENTIFIER}", re.IGNORECASE +) + +# A join, with any combination of the type keywords that may precede it. +_PPL_JOIN_HEAD = re.compile( + rf"{_PPL_COMMAND_HEAD}" + r"(?:(?:inner|left|right|full|cross|semi|anti|outer)\s+)*join\b", + re.IGNORECASE, +) + + +def _mask_ppl(query): + """Blank out quoted literals with spaces, preserving offsets. + + A value like ``'x | lookup other y'`` must not read as pipeline structure. + OpenSearch rejects a quoted index name outright (only bare and backticked + identifiers are accepted), so nothing an index reference needs is lost by + masking, and backticks are left intact so names stay readable. + """ + out = [] + quote = None + for char in query: + if quote is not None: + out.append(" ") + if char == quote: + quote = None + continue + if char in ("'", '"'): + quote = char + out.append(" ") + continue + out.append(char) + return "".join(out) + + +def _ppl_stage_end(segment): + """Return the offset of the pipe ending this stage, ignoring subsearches. + + A pipe inside ``[ ... ]`` belongs to the subsearch, not to the stage that + contains it. + """ + depth = 0 + for offset, char in enumerate(segment): + if char == "[": + depth += 1 + elif char == "]": + depth = max(depth - 1, 0) + elif char == "|" and depth == 0: + return offset + return len(segment) + + +def _strip_subsearches(segment): + """Blank out ``[ ... ]`` spans, leaving the enclosing stage's own tokens. + + A join's ON criteria may itself contain a subsearch, which would otherwise + supply the trailing token where the right-hand dataset is expected. + """ + out = [] + depth = 0 + for char in segment: + if char == "[": + depth += 1 + out.append(" ") + continue + if char == "]": + depth = max(depth - 1, 0) + out.append(" ") + continue + out.append(" " if depth else char) + return "".join(out) + + +def _ppl_join_targets(masked): + """Return ``(indices, unresolved)`` for the query's join commands. + + A join names its right-hand dataset at the end of the clause, after the ON + criteria, so the trailing token of the stage is the index. When the stage + ends in ``]`` that dataset is a subsearch instead, and the index sits + inside it where the ``source=`` scan picks it up. + + The ON criteria may also hold a subsearch of its own, with the dataset + still trailing it (``join ... on l.a in [ ... ] idx``). Those spans are + blanked before the trailing token is read, so the index is still found; + the subsearch's own ``source=`` is validated separately. + """ + indices = set() + unresolved = False + + for match in _PPL_JOIN_HEAD.finditer(masked): + segment = masked[match.end() :] + segment = segment[: _ppl_stage_end(segment)].rstrip() + if segment.endswith("]"): + continue + tokens = _strip_subsearches(segment).split() + if not tokens: + unresolved = True + continue + indices.add(tokens[-1].strip("`")) + + return indices, unresolved + + +def _ppl_index_references(query): + """Return ``(indices, unresolved)`` for every index the query names. + + From OpenSearch 3.0 a pipeline can reach a second index through ``lookup``, + ``join`` or a subsearch, none of which go through the leading ``source=``. + Validating only the first reference would let any of those read an index + outside the sketch, so every reference is collected here. + + ``unresolved`` is True when a reference could not be reduced to an + identifier. Callers must treat that as a scoping failure, the same way the + SQL dialect does: a reference the allowlist never saw would otherwise pass + straight through. + """ + masked = _mask_ppl(query) + + indices = set() + for pattern in (_PPL_SOURCE_REF, _PPL_LOOKUP_REF, _PPL_DESCRIBE_REF): + for match in pattern.finditer(masked): + indices.add(match.group(1).strip("`")) + + join_indices, unresolved = _ppl_join_targets(masked) + return indices | join_indices, unresolved + + +def _backtick_quote(index_name): + """Backtick-quote an index name if it needs quoting for PPL.""" + if index_name.startswith("`") and index_name.endswith("`"): + return index_name + return f"`{index_name}`" + + +def _ppl_timeline_predicate(timeline_ids): + """PPL predicate scoping rows to the given timelines. + + Keeps rows whose ``__ts_timeline_id`` is in the set, plus legacy rows that + predate the field, mirroring the explore datastore's behaviour. The caller + only injects this when some index in the pattern maps the field; where none + does, the predicate is both an error under Calcite and a no-op. + """ + ids = ", ".join(str(int(t)) for t in timeline_ids) + return f"__ts_timeline_id in ({ids}) or isnull(__ts_timeline_id)" + + +def _ppl_scope_predicate(timeline_ids, time_range): + """Combine the timeline and time-range predicates into one expression. + + Each part is parenthesised before being ANDed: the timeline predicate holds + an ``or``, which would otherwise bind more loosely than the ``and`` joining + it to the time bounds and widen the result set. + """ + parts = [] + if timeline_ids: + parts.append(_ppl_timeline_predicate(timeline_ids)) + range_predicate = time_range_predicate(time_range) + if range_predicate: + parts.append(range_predicate) + + if not parts: + return "" + if len(parts) == 1: + return parts[0] + return " and ".join(f"({part})" for part in parts) + + +def _inject_ppl_filter(query, timeline_ids, time_range): + """Insert the scoping filter as the first PPL pipe stage. + + The scoped query always starts with ``search source=<...>``; the new + ``| where`` runs before any user stage, so it filters orphaned/co-located + rows and out-of-range rows regardless of the rest of the pipeline. + """ + predicate = _ppl_scope_predicate(timeline_ids, time_range) + if not predicate: + return query + match = _PPL_SOURCE_HEAD.match(query) + if not match: + return query + head, rest = match.group(1), match.group(2).lstrip() + if rest.startswith("|"): + rest = rest[1:].lstrip() + injected = f"{head} | where {predicate}" + if rest: + injected += f" | {rest}" + return injected + + +def _scope_ppl_query(query, index_pattern, timeline_ids=None, time_range=None): + """Ensure PPL query targets the sketch indices (and timelines). + + Every index the pipeline names is validated, including those reached by + ``lookup``, ``join`` or a subsearch. If the query doesn't start with + 'search source=', prepend it. Index names are backtick-quoted for PPL + compatibility (UUID-style names starting with digits are not valid bare + identifiers). When timeline_ids is given, a ``__ts_timeline_id`` filter is + injected so results are scoped to those timelines, not the whole (possibly + shared) index. A time_range adds a numeric bound on ``timestamp`` in the + same stage. + """ + stripped = query.strip() + allowed = set(index_pattern.split(",")) + + # Validate every index the pipeline names, not just the leading source, so + # a lookup/join/subsearch cannot reach outside the sketch. + referenced, unresolved = _ppl_index_references(stripped) + if unresolved: + return None, ( + "Unable to determine which indices this PPL query targets. " + "Name the sketch index explicitly." + ) + for index_name in referenced: + if index_name not in allowed and index_name != index_pattern: + return None, "PPL query targets indices outside this sketch." + + if _PPL_SEARCH_SOURCE.match(stripped): + return _inject_ppl_filter(stripped, timeline_ids, time_range), None + + if stripped.lower().startswith("source") and _PPL_BARE_SOURCE.match(stripped): + return ( + _inject_ppl_filter(f"search {stripped}", timeline_ids, time_range), + None, + ) + + quoted = _backtick_quote(index_pattern) + scoped = f"search source={quoted} | {stripped}" + return _inject_ppl_filter(scoped, timeline_ids, time_range), None + + +def _truncated_export(error, emitted, offset): + """Build the trailing NDJSON line for an export that stopped early. + + Every page re-runs the pipeline and throws away ``offset`` rows, so a deep + export gets slower page by page until it exceeds the timeout. That failure + would otherwise look like a completed download, so the line says plainly + that the file is short, how many rows it holds, and what to do instead. + """ + logger.warning( + "PPL export stopped at offset %s after %s rows: %s", offset, emitted, error + ) + return ( + json.dumps( + { + "error": str(error), + "incomplete": True, + "rows_returned": emitted, + "failed_at_offset": offset, + "detail": ( + "This export is incomplete. PPL has no cursor, so each page " + "re-runs the whole query and skips the rows before it, which " + "gets slower the deeper it goes. Narrow the query, or use the " + "SQL export, which pages through a cursor instead." + ), + } + ) + + "\n" + ) + + +class PplDialect(DirectQueryDialect): + """OpenSearch Piped Processing Language.""" + + name = "ppl" + + def api(self, client): + return client.plugins.ppl + + def validate(self, query): + """PPL has no write commands, so any non-empty query is read-only.""" + return None + + def scope(self, query, index_pattern, timeline_ids, time_range=None): + return _scope_ppl_query(query, index_pattern, timeline_ids, time_range) + + def stream(self, client, scoped_query): + """Stream results using ``head N from M`` pagination.""" + api = self.api(client) + page_size = DIRECT_QUERY_EXPORT_PAGE_SIZE + offset = 0 + emitted = 0 + columns = None + # A user-supplied head already bounds the result set, so paginating on + # top of it would silently re-run the same rows. + has_user_head = bool(_PPL_HEAD_STAGE.search(scoped_query)) + + try: + while True: + if has_user_head: + paginated = scoped_query + else: + paginated = f"{scoped_query} | head {page_size} from {offset}" + + data = api.query( + body={"query": paginated}, + request_timeout=EXPORT_TIMEOUT_SECONDS, + ) + + if columns is None: + columns = columns_from_schema(data) + yield json.dumps({"columns": columns}) + "\n" + + rows = data.get("datarows", []) + for row in rows: + yield json.dumps(dict(zip(columns, row))) + "\n" + emitted += len(rows) + + if has_user_head or len(rows) < page_size: + break + offset += page_size + + except opensearch_exceptions.OpenSearchException as e: + yield _truncated_export(error_message(e), emitted, offset) diff --git a/timesketch/api/v1/resources/direct_query/registry.py b/timesketch/api/v1/resources/direct_query/registry.py new file mode 100644 index 0000000000..d570440313 --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/registry.py @@ -0,0 +1,28 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Dialect singletons for direct queries. + +Dialects are stateless, so each is instantiated once here and shared across +requests. Resources bind one of these to a class attribute, which is what pins +a language to its route. + +They live in their own module rather than in `endpoints.py` so the dialect +modules and the resource shell can both import them without a cycle. +""" + +from timesketch.api.v1.resources.direct_query.ppl import PplDialect +from timesketch.api.v1.resources.direct_query.sql import SqlDialect + +PPL_DIALECT = PplDialect() +SQL_DIALECT = SqlDialect() diff --git a/timesketch/api/v1/resources/direct_query/sql.py b/timesketch/api/v1/resources/direct_query/sql.py new file mode 100644 index 0000000000..33677895e5 --- /dev/null +++ b/timesketch/api/v1/resources/direct_query/sql.py @@ -0,0 +1,523 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""SQL dialect for direct queries.""" + +import json +import logging +import re + +from opensearchpy import exceptions as opensearch_exceptions + +from timesketch.api.v1.resources.direct_query.base import DIRECT_QUERY_EXPORT_PAGE_SIZE +from timesketch.api.v1.resources.direct_query.base import columns_from_schema +from timesketch.api.v1.resources.direct_query.base import error_message +from timesketch.api.v1.resources.direct_query.base import time_range_predicate +from timesketch.api.v1.resources.direct_query.dialect import DirectQueryDialect + +logger = logging.getLogger("timesketch.direct_query_api") + +# `fetch_size` is a SQL plugin concept; PPL has no equivalent. +DIRECT_QUERY_MAX_ROWS = 10000 +DIRECT_QUERY_DEFAULT_FETCH_SIZE = 1000 + +# Cursor pages are served from a held context, so they return faster than a +# PPL export page, which re-runs the pipeline each time. +EXPORT_TIMEOUT_SECONDS = 60 + +# Read-only SQL allowlist: the _sql plugin only accepts SELECT/SHOW/DESCRIBE. +# Allowlisting the leading keyword avoids false-positiving on words like DELETE +# that appear as string literals (e.g. `WHERE message LIKE '%delete%'`). +_SQL_READ_ONLY_LEADING = re.compile( + r"^\s*\(*\s*(SELECT|SHOW|DESCRIBE)\b", re.IGNORECASE +) + +# Matches a FROM or JOIN keyword (the start of a table reference / list). +_SQL_FROM_JOIN = re.compile(r"\b(?:FROM|JOIN)\b", re.IGNORECASE) + +# Keywords that terminate a FROM/JOIN table list. +_SQL_TABLE_LIST_STOP = re.compile( + r"\b(WHERE|GROUP|ORDER|HAVING|LIMIT|ON|UNION)\b", re.IGNORECASE +) + +# Leading identifier (backtick-quoted or bare) of a single table reference. +_SQL_TABLE_IDENT = re.compile(r"`([^`]+)`|([A-Za-z0-9_.*+-]+)") + +# Opens a sub-query rather than naming a table, so the reference to validate is +# the one in its own nested FROM. +_SQL_SUBQUERY_HEAD = re.compile(r"(SELECT|VALUES|WITH)\b", re.IGNORECASE) + +# Top-level SQL clause keywords used to position an injected FROM clause. +_SQL_CLAUSE_KEYWORD = re.compile(r"\b(WHERE|GROUP|ORDER|HAVING|LIMIT)\b", re.IGNORECASE) + +_SQL_FROM = re.compile(r"\bFROM\b", re.IGNORECASE) +_SQL_WHERE = re.compile(r"\bWHERE\b", re.IGNORECASE) +_SQL_UNION_OR_JOIN = re.compile(r"\b(UNION|JOIN)\b", re.IGNORECASE) +_SQL_UNION = re.compile(r"\bUNION\b", re.IGNORECASE) +_SQL_FROM_STOP = re.compile(r"\b(WHERE|GROUP|HAVING|ORDER|LIMIT|ON)\b", re.IGNORECASE) +_SQL_WHERE_BOUNDARY = re.compile(r"\b(GROUP|HAVING|ORDER|LIMIT)\b", re.IGNORECASE) +_SQL_BACKTICKED = re.compile(r"`[^`]*`") + +# A GROUP BY makes the response an aggregation, which the plugin will only +# return in full when fetch_size is 0 (cursor pagination does not apply). +_SQL_GROUP_BY = re.compile(r"\bGROUP\s+BY\b", re.IGNORECASE) + + +def _parse_fetch_size(raw): + """Clamp a client-supplied ``fetch_size`` to a usable page size. + + Raises: + ValueError: if the value is not a positive integer, so the resource can + answer with a 400 rather than failing inside the comparison. + """ + if raw is None: + return DIRECT_QUERY_DEFAULT_FETCH_SIZE + if isinstance(raw, bool) or not isinstance(raw, (int, str)): + raise ValueError("fetch_size must be a positive integer.") + try: + fetch_size = int(raw) + except ValueError as exc: + raise ValueError("fetch_size must be a positive integer.") from exc + if fetch_size < 1: + raise ValueError("fetch_size must be a positive integer.") + return min(fetch_size, DIRECT_QUERY_MAX_ROWS) + + +def _sql_timeline_predicate(timeline_ids): + """SQL predicate scoping rows to the given timelines. + + Keeps rows whose ``__ts_timeline_id`` is in the set, plus legacy rows that + predate the field, mirroring the explore datastore's behaviour. The caller + only injects this when some index in the pattern maps the field; where none + does, the predicate is both an error under Calcite and a no-op. + """ + ids = ", ".join(str(int(t)) for t in timeline_ids) + return f"(__ts_timeline_id IN ({ids}) OR __ts_timeline_id IS NULL)" + + +def _backtick_quote_sql(index_pattern): + """Quote indices as one SQL multi-index pattern: ``FROM `idx1,idx2```. + + A single backtick-quoted comma list is a union scan in OpenSearch SQL. + Separate-quoted identifiers (```idx1`, `idx2```) are treated as a JOIN, + which makes GROUP BY/ORDER BY silently return zero rows on multi-index + sketches while COUNT(*) still works -- a confusing partial failure. + """ + indices = [idx.strip() for idx in index_pattern.split(",") if idx.strip()] + return "`" + ",".join(indices) + "`" + + +def _mask_sql(query, mask_parens): + """Blank out literals, comments (and parens if ``mask_parens``) with spaces. + + Offsets are preserved so regex matches map back onto ``query``. Masking + strings stops a value like ``'%from table%'`` looking like SQL syntax, and + masking comments stops ``FROM /*x*/ idx`` hiding a table reference from + :func:`_referenced_indices`. Masking parens leaves only top-level keywords. + + Backtick-quoted identifiers are copied verbatim so index names stay + readable, and so a name containing ``--`` cannot open a comment. + """ + out = [] + depth = 0 + quote = None + comment = None + in_backtick = False + i = 0 + end = len(query) + + while i < end: + char = query[i] + pair = query[i : i + 2] + masked_here = mask_parens and depth > 0 + + if comment == "line": + # A newline ends the comment and is kept so line offsets survive. + out.append(char if char == "\n" else " ") + if char == "\n": + comment = None + i += 1 + continue + + if comment == "block": + if pair == "*/": + comment = None + out.append(" ") + i += 2 + continue + out.append(char if char == "\n" else " ") + i += 1 + continue + + if quote is not None: + out.append(" ") + if char == quote: + quote = None + i += 1 + continue + + if in_backtick: + out.append(" " if masked_here else char) + if char == "`": + in_backtick = False + i += 1 + continue + + if char == "`": + in_backtick = True + out.append(" " if masked_here else char) + i += 1 + continue + + if char in ("'", '"'): + quote = char + out.append(" ") + i += 1 + continue + + if pair == "--": + comment = "line" + out.append(" ") + i += 2 + continue + + if pair == "/*": + comment = "block" + out.append(" ") + i += 2 + continue + + if char == "(": + depth += 1 + out.append(" " if mask_parens else char) + i += 1 + continue + + if char == ")": + if depth > 0: + depth -= 1 + out.append(" " if mask_parens else char) + i += 1 + continue + + out.append(" " if masked_here else char) + i += 1 + + return "".join(out) + + +def _split_table_list(segment): + """Split a FROM/JOIN table list on commas outside backticks. + + ``FROM `a,b`` is one multi-index pattern rather than two tables, so its + internal commas must not split the list. + """ + parts = [] + current = [] + in_backtick = False + for char in segment: + if char == "`": + in_backtick = not in_backtick + elif char == "," and not in_backtick: + parts.append("".join(current)) + current = [] + continue + current.append(char) + parts.append("".join(current)) + return parts + + +def _referenced_indices(query): + """Return ``(indices, unresolved)`` for the query's FROM/JOIN clauses. + + Takes the leading identifier of each comma segment so every index in a list + (``FROM a, b``) or aliased list (``FROM a x, b y``) is validated, not just + the first. Strings and comments are masked, so neither a value nor a + comment can pose as (or hide) a table reference. + + ``unresolved`` is True when a table reference could not be reduced to an + identifier. Callers must treat that as a scoping failure: an empty index + set means "nothing to check" only when nothing was unresolved, otherwise a + reference the allowlist never saw would pass straight through. + + Parenthesised items are unwrapped rather than skipped. A nested SELECT is + ignored here because its own FROM is matched separately by this same loop, + but a parenthesised plain table (``FROM(idx)``) still has to be checked. + """ + masked = _mask_sql(query, mask_parens=False) + indices = set() + unresolved = False + + for keyword in _SQL_FROM_JOIN.finditer(masked): + rest = masked[keyword.end() :] + # Cut the table list at the next clause keyword or nested FROM/JOIN. + stop = _SQL_TABLE_LIST_STOP.search(rest) + segment = rest[: stop.start()] if stop else rest + nxt = _SQL_FROM_JOIN.search(segment) + if nxt: + segment = segment[: nxt.start()] + + if not segment.strip(): + unresolved = True + continue + + for part in _split_table_list(segment): + part = part.strip() + if not part: + continue + while part.startswith("("): + part = part[1:].lstrip() + if not part: + unresolved = True + continue + if _SQL_SUBQUERY_HEAD.match(part): + continue + match = _SQL_TABLE_IDENT.match(part) + if match: + indices.add(match.group(1) or match.group(2)) + else: + unresolved = True + + return indices, unresolved + + +def _sql_single_from_target(query): + """Return the single top-level FROM table, or None if not scope-able. + + A ``__ts_timeline_id`` filter can only be injected safely when there is one + plain index table at top level. JOINs (ambiguous column), UNIONs (multiple + SELECTs), comma table lists, and sub-query FROMs are left untouched. Commas + inside backticks (the ``\\`a,b\\``` multi-index pattern) are one table. + """ + masked = _mask_sql(query, mask_parens=True) + if _SQL_UNION_OR_JOIN.search(masked): + return None + from_match = _SQL_FROM.search(masked) + if not from_match: + return None + stop = _SQL_FROM_STOP.search(masked[from_match.end() :]) + end = from_match.end() + stop.start() if stop else len(query) + segment = query[from_match.end() : end].strip() + if not segment or segment.startswith("("): + return None + if "," in _SQL_BACKTICKED.sub("", segment): + return None + return segment + + +def _sql_scope_predicate(timeline_ids, time_range): + """Combine the timeline and time-range predicates into one expression. + + ``_sql_timeline_predicate`` already parenthesises itself, and the range is + a conjunction of comparisons, so ANDing the two needs no further grouping. + """ + parts = [] + if timeline_ids: + parts.append(_sql_timeline_predicate(timeline_ids)) + range_predicate = time_range_predicate(time_range, conjunction="AND") + if range_predicate: + parts.append(range_predicate) + return " AND ".join(parts) + + +def _inject_sql_filter(query, timeline_ids, time_range): + """AND the scoping filter into a single-FROM SQL query. + + Merges into an existing top-level WHERE (wrapping the original predicate to + preserve OR precedence) or inserts a new WHERE before GROUP/HAVING/ORDER/ + LIMIT. Queries that aren't a single plain-index SELECT are returned + unchanged (see :func:`_sql_single_from_target`). + """ + predicate = _sql_scope_predicate(timeline_ids, time_range) + if not predicate or _sql_single_from_target(query) is None: + return query + masked = _mask_sql(query, mask_parens=True) + + where_match = _SQL_WHERE.search(masked) + if where_match: + stop = _SQL_WHERE_BOUNDARY.search(masked, where_match.end()) + end = stop.start() if stop else len(query) + where_expr = query[where_match.end() : end].strip() + rest = query[end:].strip() + scoped = f"{query[: where_match.end()]} {predicate} AND ({where_expr})" + return f"{scoped} {rest}" if rest else scoped + + from_match = _SQL_FROM.search(masked) + stop = _SQL_WHERE_BOUNDARY.search(masked, from_match.end()) + insert_at = stop.start() if stop else len(query) + head = query[:insert_at].rstrip() + tail = query[insert_at:].strip() + scoped = f"{head} WHERE {predicate}" + return f"{scoped} {tail}" if tail else scoped + + +def _scope_sql_query(query, index_pattern, timeline_ids=None, time_range=None): + """Ensure an SQL query only targets the sketch's own indices (and timelines). + + Supports sub-queries, JOIN, UNION and window functions (a naive first-FROM + match would mis-scope these). Validates every referenced index, then either + passes through a self-scoped query or injects a FROM clause. When + timeline_ids is given, a ``__ts_timeline_id`` filter is added so a shared + index returns only the requested timelines' rows. A time_range adds a + numeric bound on ``timestamp`` to the same WHERE clause. + """ + allowed = set(idx.strip() for idx in index_pattern.split(",") if idx.strip()) + + # Validate every referenced index (a one-identifier pattern matches + # index_pattern directly). A reference we could not parse is rejected + # rather than ignored, so an unfamiliar FROM shape cannot slip past the + # allowlist unchecked. + referenced, unresolved = _referenced_indices(query) + if unresolved: + return None, ( + "Unable to determine which indices this SQL query targets. " + "Name the sketch index explicitly in the FROM clause." + ) + for table in referenced: + if table not in allowed and table != index_pattern: + return None, "SQL query targets indices outside this sketch." + + top_level = _mask_sql(query, mask_parens=True) + if _SQL_FROM.search(top_level): + # A top-level FROM means the query already scopes itself. + scoped = query + elif _SQL_UNION.search(top_level): + # A FROM-less UNION has multiple SELECTs that cannot be auto-scoped. + return None, ( + "UNION queries must include an explicit FROM `` clause for " + "each SELECT." + ) + else: + # Single SELECT without a FROM: inject one before the first clause + # keyword (or at the end) to keep clause ordering valid. + clause_match = _SQL_CLAUSE_KEYWORD.search(top_level) + insert_at = clause_match.start() if clause_match else len(query) + quoted = _backtick_quote_sql(index_pattern) + head = query[:insert_at].rstrip() + tail = query[insert_at:].strip() + scoped = f"{head} FROM {quoted}" + if tail: + scoped += f" {tail}" + + return _inject_sql_filter(scoped, timeline_ids, time_range), None + + +def _truncated_export(error, emitted): + """Build the trailing NDJSON line for an export that stopped early. + + A half-finished download otherwise looks like a finished one, so the line + states outright that the file is short and how many rows it holds. + """ + logger.warning("SQL export stopped after %s rows: %s", emitted, error) + return ( + json.dumps( + { + "error": str(error), + "incomplete": True, + "rows_returned": emitted, + "detail": ( + "This export is incomplete. The cursor stopped returning " + "pages before the result set was exhausted." + ), + } + ) + + "\n" + ) + + +def _close_cursor(api, cursor): + """Release a cursor's search context on the cluster. + + An export that is cancelled or fails part way leaves its cursor holding a + context until the cluster's keep-alive expires it. Closing is best effort: + every row it was going to deliver has already been sent or lost, so a + failure here is nothing the caller can act on. + """ + if not cursor: + return + try: + api.close(body={"cursor": cursor}) + except opensearch_exceptions.OpenSearchException as e: + logger.warning("Could not close the SQL export cursor: %s", e) + + +class SqlDialect(DirectQueryDialect): + """OpenSearch SQL.""" + + name = "sql" + + def api(self, client): + return client.plugins.sql + + def validate(self, query): + if not _SQL_READ_ONLY_LEADING.match(query): + return ( + "SQL queries must begin with SELECT, SHOW, or DESCRIBE. " + "Only read operations are allowed." + ) + return None + + def scope(self, query, index_pattern, timeline_ids, time_range=None): + return _scope_sql_query(query, index_pattern, timeline_ids, time_range) + + def execute_payload(self, scoped_query, req_json): + if _SQL_GROUP_BY.search(scoped_query): + return {"query": scoped_query, "fetch_size": 0} + return { + "query": scoped_query, + "fetch_size": _parse_fetch_size(req_json.get("fetch_size")), + } + + def stream(self, client, scoped_query): + """Stream results using cursor-based pagination.""" + api = self.api(client) + emitted = 0 + cursor = None + try: + data = api.query( + body={ + "query": scoped_query, + "fetch_size": DIRECT_QUERY_EXPORT_PAGE_SIZE, + }, + request_timeout=EXPORT_TIMEOUT_SECONDS, + ) + # Taken from every page before its rows are handed out, so that a + # download abandoned mid-page leaves the live cursor here rather + # than the spent one that fetched the rows being yielded. + cursor = data.get("cursor") + + columns = columns_from_schema(data) + yield json.dumps({"columns": columns}) + "\n" + + rows = data.get("datarows", []) + for row in rows: + yield json.dumps(dict(zip(columns, row))) + "\n" + emitted += len(rows) + + while cursor: + data = api.query( + body={"cursor": cursor}, + request_timeout=EXPORT_TIMEOUT_SECONDS, + ) + cursor = data.get("cursor") + rows = data.get("datarows", []) + for row in rows: + yield json.dumps(dict(zip(columns, row))) + "\n" + emitted += len(rows) + + except opensearch_exceptions.OpenSearchException as e: + yield _truncated_export(error_message(e), emitted) + finally: + # Reached on a cancelled download too, which is the case that would + # otherwise hold a context open for nothing. + _close_cursor(api, cursor) diff --git a/timesketch/api/v1/resources/direct_query_scoping_test.py b/timesketch/api/v1/resources/direct_query_scoping_test.py new file mode 100644 index 0000000000..eabc96042f --- /dev/null +++ b/timesketch/api/v1/resources/direct_query_scoping_test.py @@ -0,0 +1,974 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Tests for the helpers that scope a direct query to a sketch. + +Read-only validation, index and timeline scoping, time range handling and the +cluster capability probe. The dialects, the request shell and the export paths +are covered by `direct_query_test.py`. +""" + +import pytest +from unittest import mock + +from flask import Flask + +from timesketch.api.v1.resources.direct_query import base as base_module +from timesketch.api.v1.resources.direct_query.base import ( + configure_client, + parse_time_range, + time_range_predicate, + validate_query, +) +from timesketch.api.v1.resources.direct_query import capability +from timesketch.api.v1.resources.direct_query.capability import ( + MINIMUM_OPENSEARCH_VERSION, + _plugin_supported, + _version_supported, + direct_query_support, +) +from timesketch.api.v1.resources.direct_query.ppl import ( + _backtick_quote, + _inject_ppl_filter, + _ppl_timeline_predicate, + _scope_ppl_query, +) +from timesketch.api.v1.resources.direct_query.registry import PPL_DIALECT +from timesketch.api.v1.resources.direct_query.registry import SQL_DIALECT +from timesketch.api.v1.resources.direct_query.sql import ( + _inject_sql_filter, + _scope_sql_query, + _sql_single_from_target, + _sql_timeline_predicate, +) + + +# -------------------------------------------------------------------------- +# _backtick_quote +# -------------------------------------------------------------------------- +class TestBacktickQuote: + def test_bare_name(self): + assert _backtick_quote("abc123") == "`abc123`" + + def test_digit_starting_name(self): + assert _backtick_quote("21819512ab7649eb") == "`21819512ab7649eb`" + + def test_already_quoted(self): + assert _backtick_quote("`already_quoted`") == "`already_quoted`" + + def test_comma_separated(self): + result = _backtick_quote("idx1,idx2") + assert result == "`idx1,idx2`" + + +# -------------------------------------------------------------------------- +# validate_query +# -------------------------------------------------------------------------- +class TestValidateQuery: + def test_empty_query(self): + assert validate_query("", PPL_DIALECT) is not None + assert validate_query(" ", "sql") is not None + + def test_valid_ppl(self): + assert validate_query("stats count() by data_type", PPL_DIALECT) is None + + def test_valid_sql(self): + assert validate_query("SELECT * LIMIT 10", SQL_DIALECT) is None + + def test_valid_sql_show(self): + assert validate_query("SHOW TABLES LIKE foo", SQL_DIALECT) is None + + def test_valid_sql_describe(self): + assert validate_query("DESCRIBE TABLES LIKE foo", SQL_DIALECT) is None + + def test_valid_sql_leading_whitespace_and_paren(self): + assert validate_query(" (SELECT 1)", SQL_DIALECT) is None + + # Write/DDL statements are rejected because they do not begin with an + # allowlisted read-only keyword (SELECT/SHOW/DESCRIBE). + def test_forbidden_delete(self): + result = validate_query("DELETE FROM foo", SQL_DIALECT) + assert result is not None + assert "read operations" in result.lower() + + def test_forbidden_drop(self): + result = validate_query("DROP INDEX myindex", SQL_DIALECT) + assert result is not None + + def test_forbidden_insert(self): + result = validate_query("INSERT INTO foo VALUES (1)", SQL_DIALECT) + assert result is not None + + def test_forbidden_update(self): + result = validate_query("UPDATE foo SET bar=1", SQL_DIALECT) + assert result is not None + + def test_forbidden_create_index(self): + result = validate_query("CREATE INDEX idx ON foo", SQL_DIALECT) + assert result is not None + + def test_forbidden_alter(self): + result = validate_query("ALTER INDEX foo", SQL_DIALECT) + assert result is not None + + def test_keyword_substring_allowed(self): + """'updated' inside a field name should not trigger the filter.""" + assert validate_query("SELECT last_updated LIMIT 10", SQL_DIALECT) is None + + # A write verb is only a write verb in the leading keyword. Appearing in a + # string literal or a value makes it data, and the query stays readable. + def test_sql_literal_delete_allowed(self): + assert ( + validate_query("SELECT message WHERE message LIKE '%delete%'", SQL_DIALECT) + is None + ) + + def test_sql_literal_drop_in_value_allowed(self): + assert ( + validate_query("SELECT message WHERE action = 'DROP'", SQL_DIALECT) is None + ) + + def test_ppl_literal_drop_allowed(self): + """PPL is query-only; literals like 'drop' must not be rejected.""" + assert validate_query("where like(message, '%drop%')", PPL_DIALECT) is None + + def test_ppl_not_keyword_restricted(self): + """PPL has no leading-keyword allowlist (many valid first commands).""" + assert validate_query("stats count() by action", PPL_DIALECT) is None + + +# -------------------------------------------------------------------------- +# _scope_ppl_query +# -------------------------------------------------------------------------- +class TestScopePplQuery: + INDEX = "1111aaaa2222bbbb3333cccc4444dddd" + + def test_auto_prepend_source(self): + query, err = _scope_ppl_query("stats count()", self.INDEX) + assert err is None + assert query.startswith("search source=`") + assert self.INDEX in query + assert "| stats count()" in query + + def test_auto_prepend_preserves_pipe(self): + query, err = _scope_ppl_query( + "where message LIKE '%error%' | head 10", self.INDEX + ) + assert err is None + assert "| where message" in query + + def test_existing_search_source_valid(self): + raw = f"search source=`{self.INDEX}` | stats count()" + query, err = _scope_ppl_query(raw, self.INDEX) + assert err is None + assert query == raw + + def test_existing_search_source_invalid(self): + query, err = _scope_ppl_query( + "search source=`otherindex` | stats count()", self.INDEX + ) + assert query is None + assert "outside this sketch" in err + + def test_source_shorthand(self): + raw = f"source={self.INDEX} | head 100" + query, err = _scope_ppl_query(raw, self.INDEX) + assert err is None + assert query.startswith("search source=") + + def test_source_shorthand_invalid(self): + query, err = _scope_ppl_query("source=badindex | head 10", self.INDEX) + assert query is None + assert "outside this sketch" in err + + def test_multi_index_quoting(self): + multi = "idx1,idx2,idx3" + query, err = _scope_ppl_query("stats count()", multi) + assert err is None + assert "`idx1,idx2,idx3`" in query + + def test_existing_source_bare_unquoted(self): + raw = f"search source={self.INDEX} | head 10" + query, err = _scope_ppl_query(raw, self.INDEX) + assert err is None + assert query == raw + + +# -------------------------------------------------------------------------- +# _scope_ppl_query: indices reached without the leading source= +# +# SECURITY: from OpenSearch 3.0 a pipeline can read a second index through +# lookup, join or a subsearch. Validating only the leading source= let those +# cross sketch boundaries, so every reference has to be checked. +# -------------------------------------------------------------------------- +class TestScopePplCrossIndex: + INDEX = "1111aaaa2222bbbb3333cccc4444dddd" + OTHER = "aaaa1111bbbb2222cccc3333dddd4444" + + @pytest.mark.parametrize( + "raw", + [ + "search source=`{idx}` | lookup `{other}` message", + "search source=`{idx}` | lookup {other} message", + "search source=`{idx}` | join left=l right=r on l.a = r.a `{other}`", + "search source=`{idx}` | left join on l.a = r.a {other}", + "search source=`{idx}` | join left=l right=r on l.a = r.a " + "[ source=`{other}` ]", + "search source=`{idx}` | where a in [ source=`{other}` | fields a ]", + "search source=`{idx}` | where a in [ search source={other} ]", + ], + ) + def test_foreign_index_rejected(self, raw): + query, err = _scope_ppl_query( + raw.format(idx=self.INDEX, other=self.OTHER), self.INDEX + ) + assert query is None + assert "outside this sketch" in err + + @pytest.mark.parametrize( + "raw", + [ + "search source=`{idx}` | lookup `{idx}` message", + "search source=`{idx}` | join left=l right=r on l.a = r.a `{idx}`", + "search source=`{idx}` | where a in [ source=`{idx}` | fields a ]", + ], + ) + def test_same_index_allowed(self, raw): + query, err = _scope_ppl_query(raw.format(idx=self.INDEX), self.INDEX) + assert err is None + assert query is not None + + def test_join_without_a_dataset_is_rejected(self): + """An unparseable reference fails closed, as on the SQL side.""" + query, err = _scope_ppl_query( + f"search source=`{self.INDEX}` | join ", self.INDEX + ) + assert query is None + assert "Unable to determine which indices" in err + + # A join's ON criteria can hold its own subsearch, with the right-hand + # dataset still trailing it. Both positions have to be checked. + def test_join_with_subsearch_in_criteria_checks_trailing_index(self): + query, err = _scope_ppl_query( + f"search source=`{self.INDEX}` | join left=l right=r on l.a in " + f"[ source=`{self.INDEX}` | fields a ] `{self.OTHER}`", + self.INDEX, + ) + assert query is None + assert "outside this sketch" in err + + def test_join_with_subsearch_in_criteria_checks_the_subsearch(self): + query, err = _scope_ppl_query( + f"search source=`{self.INDEX}` | join left=l right=r on l.a in " + f"[ source=`{self.OTHER}` | fields a ] `{self.INDEX}`", + self.INDEX, + ) + assert query is None + assert "outside this sketch" in err + + def test_join_with_subsearch_in_criteria_allowed_when_both_in_sketch(self): + query, err = _scope_ppl_query( + f"search source=`{self.INDEX}` | join left=l right=r on l.a in " + f"[ source=`{self.INDEX}` | fields a ] `{self.INDEX}`", + self.INDEX, + ) + assert err is None + assert query is not None + + def test_describe_of_foreign_index_rejected(self): + query, err = _scope_ppl_query(f"describe {self.OTHER}", self.INDEX) + assert query is None + assert "outside this sketch" in err + + # A command name inside a literal is data, not pipeline structure. + @pytest.mark.parametrize( + "raw", + [ + "search source=`{idx}` | where message = 'x | lookup evil y'", + 'search source=`{idx}` | where message = "a | join on b evil"', + "search source=`{idx}` | where message = 'source=evil'", + ], + ) + def test_literal_mentioning_a_command_is_not_a_reference(self, raw): + query, err = _scope_ppl_query(raw.format(idx=self.INDEX), self.INDEX) + assert err is None + assert query is not None + + def test_field_named_source_is_not_a_reference(self): + query, err = _scope_ppl_query( + f"search source=`{self.INDEX}` | where source = 5", self.INDEX + ) + assert err is None + assert query is not None + + def test_multi_index_pattern_allowed_in_lookup(self): + multi = f"{self.INDEX},{self.OTHER}" + query, err = _scope_ppl_query( + f"search source=`{multi}` | lookup `{self.OTHER}` message", multi + ) + assert err is None + assert query is not None + + +# -------------------------------------------------------------------------- +# _scope_sql_query +# -------------------------------------------------------------------------- +class TestScopeSqlQuery: + INDEX = "1111aaaa2222bbbb3333cccc4444dddd" + + def test_auto_inject_from(self): + query, err = _scope_sql_query( + "SELECT data_type, COUNT(*) GROUP BY data_type", self.INDEX + ) + assert err is None + assert f"FROM `{self.INDEX}`" in query + + def test_existing_from_valid(self): + raw = f"SELECT * FROM {self.INDEX} LIMIT 10" + query, err = _scope_sql_query(raw, self.INDEX) + assert err is None + assert query == raw + + def test_existing_from_invalid(self): + query, err = _scope_sql_query("SELECT * FROM otherindex LIMIT 10", self.INDEX) + assert query is None + assert "outside this sketch" in err + + def test_select_with_where(self): + query, err = _scope_sql_query( + "SELECT message WHERE message LIKE '%error%' LIMIT 10", self.INDEX + ) + assert err is None + assert f"FROM `{self.INDEX}`" in query + assert "WHERE" in query + + def test_select_with_order(self): + query, err = _scope_sql_query( + "SELECT datetime ORDER BY datetime DESC LIMIT 10", self.INDEX + ) + assert err is None + assert f"FROM `{self.INDEX}`" in query + + def test_select_with_group(self): + query, err = _scope_sql_query( + "SELECT data_type, COUNT(*) as cnt GROUP BY data_type", self.INDEX + ) + assert err is None + assert f"FROM `{self.INDEX}`" in query + assert "GROUP BY" in query + + # --- clause ordering: an injected FROM has to precede HAVING --- + def test_having_without_group_injects_from_before_having(self): + query, err = _scope_sql_query( + "SELECT COUNT(*) c HAVING COUNT(*) > 1", self.INDEX + ) + assert err is None + assert query == f"SELECT COUNT(*) c FROM `{self.INDEX}` HAVING COUNT(*) > 1" + + def test_no_clause_appends_from_at_end(self): + query, err = _scope_sql_query("SELECT COUNT(*)", self.INDEX) + assert err is None + assert query == f"SELECT COUNT(*) FROM `{self.INDEX}`" + + # --- FROM-sub-query: the index sits in the inner FROM, not the outer --- + def test_from_subquery_inner_index_valid(self): + raw = ( + f"SELECT t.dt FROM (SELECT data_type AS dt FROM `{self.INDEX}` " + "LIMIT 5) t LIMIT 2" + ) + query, err = _scope_sql_query(raw, self.INDEX) + assert err is None + assert query == raw + + def test_from_subquery_inner_index_outside_sketch(self): + raw = "SELECT t.dt FROM (SELECT data_type AS dt FROM `evilindex` LIMIT 5) t" + query, err = _scope_sql_query(raw, self.INDEX) + assert query is None + assert "outside this sketch" in err + + # --- IN-sub-query without an outer FROM: outer SELECT must get a FROM --- + def test_in_subquery_without_outer_from(self): + raw = ( + "SELECT data_type WHERE data_type IN " + f"(SELECT data_type FROM `{self.INDEX}` LIMIT 1) LIMIT 2" + ) + query, err = _scope_sql_query(raw, self.INDEX) + assert err is None + assert query == ( + f"SELECT data_type FROM `{self.INDEX}` WHERE data_type IN " + f"(SELECT data_type FROM `{self.INDEX}` LIMIT 1) LIMIT 2" + ) + + # --- UNION / JOIN with explicit FROMs pass through unchanged --- + def test_union_with_explicit_from_passthrough(self): + raw = ( + f"SELECT data_type FROM `{self.INDEX}` LIMIT 1 " + f"UNION SELECT data_type FROM `{self.INDEX}` LIMIT 1" + ) + query, err = _scope_sql_query(raw, self.INDEX) + assert err is None + assert query == raw + + def test_self_join_with_explicit_from_passthrough(self): + raw = ( + f"SELECT a.data_type FROM `{self.INDEX}` a " + f"JOIN `{self.INDEX}` b ON a.data_type=b.data_type LIMIT 1" + ) + query, err = _scope_sql_query(raw, self.INDEX) + assert err is None + assert query == raw + + def test_join_outside_sketch_rejected(self): + raw = ( + f"SELECT a.data_type FROM `{self.INDEX}` a " + "JOIN `otherindex` b ON a.data_type=b.data_type" + ) + query, err = _scope_sql_query(raw, self.INDEX) + assert query is None + assert "outside this sketch" in err + + # --- FROM-less UNION cannot be auto-scoped: clear error, no broken SQL --- + def test_union_without_from_rejected(self): + query, err = _scope_sql_query( + "SELECT data_type LIMIT 1 UNION SELECT data_type LIMIT 1", self.INDEX + ) + assert query is None + assert "UNION" in err + + # --- string literals must never be treated as table refs / keywords --- + def test_literal_from_in_value_not_a_table(self): + query, err = _scope_sql_query( + "SELECT message WHERE message LIKE '%from table%'", self.INDEX + ) + assert err is None + assert query == ( + f"SELECT message FROM `{self.INDEX}` WHERE message LIKE '%from table%'" + ) + + # --- multi-index: comma FROM list is a multi-index scan in OpenSearch --- + INDEX_B = "9f1c2d3e4a5b6c7d8e9f0a1b2c3d4e5f" + + def test_multi_index_comma_list_all_validated(self): + multi = f"{self.INDEX},{self.INDEX_B}" + raw = f"SELECT count(*) FROM `{self.INDEX}`, `{self.INDEX_B}`" + query, err = _scope_sql_query(raw, multi) + assert err is None + assert query == raw + + def test_multi_index_comma_list_with_aliases(self): + multi = f"{self.INDEX},{self.INDEX_B}" + raw = f"SELECT * FROM `{self.INDEX}` x, `{self.INDEX_B}` y LIMIT 1" + query, err = _scope_sql_query(raw, multi) + assert err is None + assert query == raw + + def test_multi_index_injection(self): + # Injected multi-index FROM must be a single backtick-quoted pattern + # (`a,b`), a union scan. Separate-quoted identifiers (`a`, `b`) are a + # JOIN in OpenSearch SQL and make GROUP BY silently return no rows. + multi = f"{self.INDEX},{self.INDEX_B}" + query, err = _scope_sql_query("SELECT count(*) c", multi) + assert err is None + assert query == f"SELECT count(*) c FROM `{self.INDEX},{self.INDEX_B}`" + + def test_single_pattern_identifier_passthrough(self): + multi = f"{self.INDEX},{self.INDEX_B}" + raw = f"SELECT count(*) FROM `{self.INDEX},{self.INDEX_B}`" + query, err = _scope_sql_query(raw, multi) + assert err is None + assert query == raw + + # SECURITY: every index in a comma list is scanned, so checking only the + # first would let the second read outside the sketch. + def test_comma_list_second_index_outside_sketch_rejected(self): + raw = f"SELECT count(*) FROM `{self.INDEX}`, `{self.INDEX_B}`" + query, err = _scope_sql_query(raw, self.INDEX) # sketch only has INDEX + assert query is None + assert "outside this sketch" in err + + # SECURITY: comments must not hide a table reference from the allowlist. + @pytest.mark.parametrize( + "raw", + [ + "SELECT * FROM /*x*/ {other}", + "SELECT * FROM {other} /*x*/", + "SELECT * FROM\n-- pick one\n{other}", + "SELECT * FROM /* multi\nline */ {other}", + "SELECT * FROM `{allowed}`, /*x*/ {other}", + ], + ) + def test_commented_index_outside_sketch_rejected(self, raw): + query, err = _scope_sql_query( + raw.format(other=self.INDEX_B, allowed=self.INDEX), self.INDEX + ) + assert query is None + assert "outside this sketch" in err + + # SECURITY: a parenthesised plain table is a table, not a sub-query. + @pytest.mark.parametrize( + "raw", ["SELECT * FROM({other})", "SELECT * FROM ( {other} )"] + ) + def test_parenthesised_index_outside_sketch_rejected(self, raw): + query, err = _scope_sql_query(raw.format(other=self.INDEX_B), self.INDEX) + assert query is None + assert "outside this sketch" in err + + def test_parenthesised_index_inside_sketch_allowed(self): + query, err = _scope_sql_query(f"SELECT * FROM ({self.INDEX})", self.INDEX) + assert err is None + assert query is not None + + # A comment is still allowed as long as the real table is in the sketch. + def test_comment_with_allowed_index_passes(self): + query, err = _scope_sql_query( + f"SELECT * /* pick */ FROM `{self.INDEX}`", self.INDEX + ) + assert err is None + assert query is not None + + # SECURITY: an unparseable table reference must fail closed rather than + # being read as "no indices referenced". + def test_unresolvable_from_target_rejected(self): + query, err = _scope_sql_query("SELECT * FROM ", self.INDEX) + assert query is None + assert "Unable to determine which indices" in err + + # A backticked name containing -- must not open a comment. + def test_double_dash_inside_backticks_is_not_a_comment(self): + odd = "idx--name" + query, err = _scope_sql_query(f"SELECT * FROM `{odd}` LIMIT 1", odd) + assert err is None + assert query is not None + + +# -------------------------------------------------------------------------- +# __ts_timeline_id filter injection (orphaned-record guard) +# -------------------------------------------------------------------------- +class TestTimelinePredicates: + def test_sql_predicate(self): + assert _sql_timeline_predicate([3, 7]) == ( + "(__ts_timeline_id IN (3, 7) OR __ts_timeline_id IS NULL)" + ) + + def test_ppl_predicate(self): + assert _ppl_timeline_predicate([3, 7]) == ( + "__ts_timeline_id in (3, 7) or isnull(__ts_timeline_id)" + ) + + def test_sql_predicate_coerces_int(self): + # Guard against injection via non-numeric timeline IDs. + assert _sql_timeline_predicate([3]) == ( + "(__ts_timeline_id IN (3) OR __ts_timeline_id IS NULL)" + ) + + +class TestSqlSingleFromTarget: + IDX = "abc123" + IDX_B = "def456" + + def test_single_index(self): + assert _sql_single_from_target(f"SELECT * FROM `{self.IDX}` LIMIT 1") + + def test_multi_index_pattern_is_single(self): + # `a,b` (comma inside one backtick) is one union-scan target. + assert _sql_single_from_target(f"SELECT * FROM `{self.IDX},{self.IDX_B}`") + + def test_join_returns_none(self): + raw = f"SELECT * FROM `{self.IDX}` x JOIN `{self.IDX_B}` y ON x.a=y.a" + assert _sql_single_from_target(raw) is None + + def test_comma_table_list_returns_none(self): + raw = f"SELECT * FROM `{self.IDX}`, `{self.IDX_B}`" + assert _sql_single_from_target(raw) is None + + def test_subquery_from_returns_none(self): + raw = f"SELECT t.c FROM (SELECT a c FROM `{self.IDX}`) t" + assert _sql_single_from_target(raw) is None + + def test_no_from_returns_none(self): + assert _sql_single_from_target("SELECT 1") is None + + +class TestInjectSqlTimelineFilter: + IDX = "abc123" + + def test_no_timeline_ids_passthrough(self): + raw = f"SELECT * FROM `{self.IDX}` LIMIT 1" + assert _inject_sql_filter(raw, None, None) == raw + + def test_inject_without_where(self): + out = _inject_sql_filter(f"SELECT COUNT(*) c FROM `{self.IDX}`", [5], None) + assert out == ( + f"SELECT COUNT(*) c FROM `{self.IDX}` WHERE " + "(__ts_timeline_id IN (5) OR __ts_timeline_id IS NULL)" + ) + + def test_inject_before_group_by(self): + raw = f"SELECT data_type, COUNT(*) c FROM `{self.IDX}` GROUP BY data_type" + out = _inject_sql_filter(raw, [5], None) + assert out == ( + f"SELECT data_type, COUNT(*) c FROM `{self.IDX}` WHERE " + "(__ts_timeline_id IN (5) OR __ts_timeline_id IS NULL) " + "GROUP BY data_type" + ) + + def test_merge_existing_where_preserves_or_precedence(self): + raw = f"SELECT * FROM `{self.IDX}` WHERE a=1 OR b=2 LIMIT 10" + out = _inject_sql_filter(raw, [5], None) + assert out == ( + f"SELECT * FROM `{self.IDX}` WHERE " + "(__ts_timeline_id IN (5) OR __ts_timeline_id IS NULL) " + "AND (a=1 OR b=2) LIMIT 10" + ) + + def test_join_left_unchanged(self): + raw = f"SELECT * FROM `{self.IDX}` x JOIN `def456` y ON x.a=y.a LIMIT 1" + assert _inject_sql_filter(raw, [5], None) == raw + + def test_subquery_left_unchanged(self): + raw = f"SELECT t.c FROM (SELECT a c FROM `{self.IDX}`) t LIMIT 2" + assert _inject_sql_filter(raw, [5], None) == raw + + def test_string_literal_not_mistaken_for_clause(self): + raw = f"SELECT msg FROM `{self.IDX}` WHERE msg LIKE '%group by%' LIMIT 5" + out = _inject_sql_filter(raw, [5], None) + assert out == ( + f"SELECT msg FROM `{self.IDX}` WHERE " + "(__ts_timeline_id IN (5) OR __ts_timeline_id IS NULL) " + "AND (msg LIKE '%group by%') LIMIT 5" + ) + + +class TestInjectPplTimelineFilter: + IDX = "abc123" + + def test_no_timeline_ids_passthrough(self): + raw = f"search source=`{self.IDX}` | stats count()" + assert _inject_ppl_filter(raw, None, None) == raw + + def test_inject_first_stage(self): + raw = f"search source=`{self.IDX}` | stats count() as c" + out = _inject_ppl_filter(raw, [5], None) + assert out == ( + f"search source=`{self.IDX}` | where " + "__ts_timeline_id in (5) or isnull(__ts_timeline_id) | stats count() as c" + ) + + def test_inject_bare_source(self): + out = _inject_ppl_filter(f"search source=`{self.IDX}`", [5], None) + assert out == ( + f"search source=`{self.IDX}` | where " + "__ts_timeline_id in (5) or isnull(__ts_timeline_id)" + ) + + +class TestScopeWithTimelineFilter: + IDX = "abc123" + + def test_sql_scope_injects_filter(self): + query, err = _scope_sql_query( + "SELECT data_type, COUNT(*) c GROUP BY data_type", self.IDX, [5, 6] + ) + assert err is None + assert "__ts_timeline_id IN (5, 6)" in query + assert query.index("WHERE") < query.index("GROUP BY") + + def test_sql_scope_no_filter_when_ids_none(self): + query, err = _scope_sql_query("SELECT COUNT(*) c", self.IDX) + assert err is None + assert "__ts_timeline_id" not in query + + def test_ppl_scope_injects_filter(self): + query, err = _scope_ppl_query("stats count() as c", self.IDX, [5, 6]) + assert err is None + assert "where __ts_timeline_id in (5, 6)" in query + + def test_ppl_scope_no_filter_when_ids_none(self): + query, err = _scope_ppl_query("stats count() as c", self.IDX) + assert err is None + assert "__ts_timeline_id" not in query + + +# -------------------------------------------------------------------------- +# Time range parsing +# -------------------------------------------------------------------------- + +# 2026-04-07T00:00:00Z and the last microsecond of 2026-04-07, in the +# microsecond epoch Timesketch writes to the `timestamp` field. +APR_7_START = 1775520000000000 +APR_7_END = 1775606399999999 + + +class TestParseTimeRange: + def test_absent_returns_none(self): + assert parse_time_range({}) is None + + def test_date_only_start_snaps_to_midnight(self): + assert parse_time_range({"start_time": "2026-04-07"}) == (APR_7_START, None) + + def test_date_only_end_covers_the_whole_day(self): + # An end of "2026-04-07" means through the end of the 7th, not the + # instant it began, otherwise a single-day range matches nothing. + assert parse_time_range({"end_time": "2026-04-07"}) == (None, APR_7_END) + + def test_datetime_with_zulu_suffix(self): + parsed = parse_time_range({"start_time": "2026-04-07T00:00:00Z"}) + assert parsed == (APR_7_START, None) + + def test_datetime_with_offset(self): + parsed = parse_time_range({"start_time": "2026-04-07T01:00:00+01:00"}) + assert parsed == (APR_7_START, None) + + def test_naive_datetime_is_read_as_utc(self): + parsed = parse_time_range({"start_time": "2026-04-07T00:00:00"}) + assert parsed == (APR_7_START, None) + + def test_both_boundaries(self): + parsed = parse_time_range( + {"start_time": "2026-04-07", "end_time": "2026-04-07"} + ) + assert parsed == (APR_7_START, APR_7_END) + + def test_inverted_range_rejected(self): + with pytest.raises(ValueError, match="not be later than"): + parse_time_range({"start_time": "2026-04-08", "end_time": "2026-04-07"}) + + def test_unparseable_rejected(self): + with pytest.raises(ValueError, match="not a valid ISO 8601"): + parse_time_range({"start_time": "last tuesday"}) + + def test_non_string_rejected(self): + with pytest.raises(ValueError, match="must be an ISO 8601"): + parse_time_range({"start_time": 1775520000}) + + +class TestTimeRangePredicate: + def test_none_is_empty(self): + assert time_range_predicate(None) == "" + + def test_open_ended_start(self): + assert time_range_predicate((APR_7_START, None)) == ( + f"timestamp >= {APR_7_START}" + ) + + def test_open_ended_end(self): + assert time_range_predicate((None, APR_7_END)) == f"timestamp <= {APR_7_END}" + + def test_both_bounds(self): + assert time_range_predicate((APR_7_START, APR_7_END)) == ( + f"timestamp >= {APR_7_START} and timestamp <= {APR_7_END}" + ) + + def test_conjunction_is_configurable(self): + assert time_range_predicate((APR_7_START, APR_7_END), conjunction="AND") == ( + f"timestamp >= {APR_7_START} AND timestamp <= {APR_7_END}" + ) + + +class TestScopeWithTimeRange: + IDX = "abc123" + RANGE = (APR_7_START, APR_7_END) + + def test_ppl_range_only(self): + query, err = _scope_ppl_query("stats count() as c", self.IDX, None, self.RANGE) + assert err is None + assert query == ( + f"search source=`{self.IDX}` | where timestamp >= {APR_7_START} " + f"and timestamp <= {APR_7_END} | stats count() as c" + ) + + def test_ppl_range_and_timelines_are_parenthesised(self): + # The timeline predicate contains an `or`; without the parentheses it + # would bind more loosely than the `and` and widen the result set. + query, err = _scope_ppl_query("stats count() as c", self.IDX, [5], self.RANGE) + assert err is None + assert query == ( + f"search source=`{self.IDX}` | where " + "(__ts_timeline_id in (5) or isnull(__ts_timeline_id)) " + f"and (timestamp >= {APR_7_START} and timestamp <= {APR_7_END}) " + "| stats count() as c" + ) + + def test_sql_range_only(self): + query, err = _scope_sql_query("SELECT COUNT(*) c", self.IDX, None, self.RANGE) + assert err is None + assert query == ( + f"SELECT COUNT(*) c FROM `{self.IDX}` WHERE " + f"timestamp >= {APR_7_START} AND timestamp <= {APR_7_END}" + ) + + def test_sql_range_and_timelines(self): + query, err = _scope_sql_query("SELECT COUNT(*) c", self.IDX, [5], self.RANGE) + assert err is None + assert query == ( + f"SELECT COUNT(*) c FROM `{self.IDX}` WHERE " + "(__ts_timeline_id IN (5) OR __ts_timeline_id IS NULL) " + f"AND timestamp >= {APR_7_START} AND timestamp <= {APR_7_END}" + ) + + def test_sql_range_merges_with_user_where(self): + query, err = _scope_sql_query( + "SELECT * WHERE a=1 OR b=2 LIMIT 10", self.IDX, None, self.RANGE + ) + assert err is None + assert query == ( + f"SELECT * FROM `{self.IDX}` WHERE " + f"timestamp >= {APR_7_START} AND timestamp <= {APR_7_END} " + "AND (a=1 OR b=2) LIMIT 10" + ) + + def test_no_range_leaves_query_untouched(self): + query, err = _scope_ppl_query("stats count() as c", self.IDX, None, None) + assert err is None + assert "timestamp" not in query + + +# -------------------------------------------------------------------------- +# Cluster capability +# -------------------------------------------------------------------------- +class TestVersionSupported: + def test_meets_minimum(self): + assert _version_supported(MINIMUM_OPENSEARCH_VERSION) + + def test_newer_is_supported(self): + assert _version_supported("3.9.1") + + def test_older_is_not(self): + support = _version_supported("2.16.0") + assert not support + assert "3.7.0" in support.reason + assert "2.16.0" in support.reason + + def test_prerelease_below_minimum_is_not(self): + assert not _version_supported("3.6.0-rc1") + + # An unreadable version must not take a working feature away. + @pytest.mark.parametrize("raw", [None, "", "not-a-version", 3.7]) + def test_unreadable_version_assumes_support(self, raw): + assert _version_supported(raw) + + +class TestPluginSupported: + def test_sql_plugin_present(self): + assert _plugin_supported([{"component": "opensearch-sql"}]) + + def test_sql_plugin_absent(self): + support = _plugin_supported([{"component": "opensearch-alerting"}]) + assert not support + assert "SQL plugin" in support.reason + + def test_unreadable_list_assumes_support(self): + assert _plugin_supported(None) + + def test_null_component_is_skipped(self): + assert not _plugin_supported([{"component": None}]) + + +class TestDirectQuerySupport: + def setup_method(self): + capability.reset_cache() + + def teardown_method(self): + capability.reset_cache() + + @staticmethod + def _app(): + app = Flask(__name__) + app.config["OPENSEARCH_HOST"] = "localhost" + app.config["OPENSEARCH_PORT"] = 9200 + return app + + def test_supported_cluster(self): + responses = [ + {"version": {"number": "3.7.0"}}, + [{"component": "opensearch-sql"}], + ] + with self._app().app_context(): + with mock.patch.object(capability, "_probe_call", side_effect=responses): + assert direct_query_support() + + def test_old_cluster_is_refused_without_probing_plugins(self): + with self._app().app_context(): + with mock.patch.object( + capability, + "_probe_call", + side_effect=[{"version": {"number": "2.16.0"}}], + ) as probe: + support = direct_query_support() + assert not support + assert "2.16.0" in support.reason + # The plugin list is irrelevant once the version has ruled the cluster out. + assert probe.call_count == 1 + + def test_result_is_cached(self): + responses = [ + {"version": {"number": "3.7.0"}}, + [{"component": "opensearch-sql"}], + ] + with self._app().app_context(): + with mock.patch.object( + capability, "_probe_call", side_effect=responses + ) as probe: + direct_query_support() + direct_query_support() + assert probe.call_count == 2 + + def test_unreachable_cluster_assumes_support(self): + with self._app().app_context(): + with mock.patch.object(capability, "_probe_call", return_value=None): + assert direct_query_support() + + +# -------------------------------------------------------------------------- +# The shared OpenSearch client +# +# Credentials, TLS and the node list are the datastore's to derive. What +# matters here is that the direct-query path uses the client it hands back, +# and holds a single one rather than building a fresh pool per request. +# -------------------------------------------------------------------------- +class TestSharedClient: + def setup_method(self): + base_module.reset_client() + + teardown_method = setup_method + + @staticmethod + def _patch_builder(): + """Stand in for the datastore's client builder.""" + return mock.patch.object( + base_module, "build_opensearch_client", return_value=mock.MagicMock() + ) + + def test_configure_client_uses_the_datastore_builder(self): + with self._patch_builder() as builder: + configure_client(Flask(__name__)) + assert base_module.get_client() is builder.return_value + + def test_the_client_is_built_once_and_reused(self): + with self._patch_builder() as builder: + configure_client(Flask(__name__)) + for _ in range(3): + base_module.get_client() + assert builder.call_count == 1 + + def test_first_use_builds_a_client_when_startup_did_not(self): + with self._patch_builder() as builder: + with Flask(__name__).app_context(): + assert base_module.get_client() is builder.return_value + + def test_reset_forces_a_rebuild(self): + with self._patch_builder() as builder: + configure_client(Flask(__name__)) + base_module.reset_client() + configure_client(Flask(__name__)) + assert builder.call_count == 2 diff --git a/timesketch/api/v1/resources/direct_query_test.py b/timesketch/api/v1/resources/direct_query_test.py new file mode 100644 index 0000000000..bb0165810c --- /dev/null +++ b/timesketch/api/v1/resources/direct_query_test.py @@ -0,0 +1,968 @@ +# Copyright 2026 Google Inc. All rights reserved. +# +# Licensed 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. +"""Tests for the direct query dialects and resource shell. + +Query scoping helpers are covered by `direct_query_scoping_test.py`; this module +covers the dialect objects, the shared request shell, and the export streaming +paths. +""" + +import json +from unittest import mock + +import pytest + +from flask import Flask +from opensearchpy import exceptions as opensearch_exceptions +from werkzeug.exceptions import HTTPException + +from timesketch.api.v1.resources.direct_query import ( + PplQueryExplainResource, + PplQueryExportResource, + PplQueryResource, + SqlQueryExplainResource, + SqlQueryExportResource, + SqlQueryResource, +) +from timesketch.api.v1.resources.direct_query.base import ( + columns_from_schema, + empty_result, + format_opensearch_error, + get_sketch_scope, +) +from timesketch.api.v1.resources.direct_query.registry import ( + PPL_DIALECT, + SQL_DIALECT, +) +from timesketch.api.v1.resources.direct_query.sql import DIRECT_QUERY_MAX_ROWS +from timesketch.api.v1.resources.direct_query import base as base_module +from timesketch.api.v1.resources.direct_query import endpoints +from timesketch.api.v1.resources.direct_query import ppl as ppl_module + + +@pytest.fixture(autouse=True) +def _stub_opensearch_preflight(): + """Stub the two pre-flight calls _prepare_query makes to OpenSearch. + + Defaults are "field is mapped" and "the plan raises no objection", which is + the ordinary case. Tests covering either path re-patch locally. + """ + with mock.patch.object( + endpoints, "index_pattern_has_timeline_field", return_value=True + ), mock.patch.object(endpoints, "verify_scope_with_explain", return_value=None): + yield + + +@pytest.fixture(autouse=True) +def _reset_shared_client(): + """Keep a client built by one test from leaking into the next.""" + base_module.reset_client() + yield + base_module.reset_client() + + +class _CountingList(list): + """List that records how many times it has been iterated.""" + + def __init__(self, items): + super().__init__(items) + self.iter_count = 0 + + def __iter__(self): + self.iter_count += 1 + return super().__iter__() + + +def _make_timeline(timeline_id, index_name): + timeline = mock.MagicMock() + timeline.id = timeline_id + timeline.searchindex.index_name = index_name + return timeline + + +def _make_sketch(timelines, archived=False): + sketch = mock.MagicMock() + sketch.timelines = timelines + sketch.get_status.status = "archived" if archived else "ready" + return sketch + + +def _make_client(): + """An OpenSearch client whose plugin namespaces the test drives.""" + return mock.MagicMock() + + +def _rejected(reason="bad query", status=400): + """The exception the client raises when the plugin refuses a query.""" + return opensearch_exceptions.TransportError( + status, "SyntaxError", {"error": {"type": "SyntaxError", "reason": reason}} + ) + + +def _unreachable(message="down"): + """The exception the client raises when it cannot reach a node. + + Built the way the connection layer builds it, so that the status code and + the underlying cause sit where the handling code looks for them. + """ + return opensearch_exceptions.ConnectionError("N/A", message, Exception(message)) + + +def _timed_out(message="too deep"): + """The exception the client raises when a request exceeds its timeout.""" + return opensearch_exceptions.ConnectionTimeout( + "TIMEOUT", message, Exception(message) + ) + + +# -------------------------------------------------------------------------- +# Dialect registry +# -------------------------------------------------------------------------- +class TestRegistry: + def test_ppl_singleton_identity(self): + assert PPL_DIALECT.name == "ppl" + + def test_sql_singleton_identity(self): + assert SQL_DIALECT.name == "sql" + + def test_dialects_are_distinct(self): + assert PPL_DIALECT is not SQL_DIALECT + + def test_reimport_returns_same_instances(self): + """Resources bind these at import; a per-request instance would leak.""" + # pylint: disable=import-outside-toplevel,reimported + from timesketch.api.v1.resources.direct_query.registry import ( + PPL_DIALECT as ppl_again, + ) + from timesketch.api.v1.resources.direct_query.registry import ( + SQL_DIALECT as sql_again, + ) + + assert ppl_again is PPL_DIALECT + assert sql_again is SQL_DIALECT + + +# -------------------------------------------------------------------------- +# Dialect / client binding +# -------------------------------------------------------------------------- +class TestDialectApi: + """A dialect is bound to its language by the namespace it selects.""" + + def test_ppl_uses_the_ppl_namespace(self): + client = _make_client() + assert PPL_DIALECT.api(client) is client.plugins.ppl + + def test_sql_uses_the_sql_namespace(self): + client = _make_client() + assert SQL_DIALECT.api(client) is client.plugins.sql + + def test_dialects_do_not_share_a_namespace(self): + client = _make_client() + assert PPL_DIALECT.api(client) is not SQL_DIALECT.api(client) + + +# -------------------------------------------------------------------------- +# Payload construction +# -------------------------------------------------------------------------- +class TestPayloads: + def test_ppl_execute_payload(self): + assert PPL_DIALECT.execute_payload("search source=`a`", {}) == { + "query": "search source=`a`" + } + + def test_ppl_explain_payload(self): + assert PPL_DIALECT.explain_payload("search source=`a`") == { + "query": "search source=`a`" + } + + def test_sql_group_by_uses_zero_fetch_size(self): + """Aggregations are only returned in full when fetch_size is 0.""" + payload = SQL_DIALECT.execute_payload( + "SELECT a, COUNT(*) FROM `i` GROUP BY a", {} + ) + assert payload["fetch_size"] == 0 + + def test_sql_group_by_case_and_spacing(self): + payload = SQL_DIALECT.execute_payload("SELECT a FROM `i` group by a", {}) + assert payload["fetch_size"] == 0 + + def test_sql_default_fetch_size(self): + payload = SQL_DIALECT.execute_payload("SELECT * FROM `i`", {}) + assert payload["fetch_size"] == 1000 + + def test_sql_respects_requested_fetch_size(self): + payload = SQL_DIALECT.execute_payload("SELECT * FROM `i`", {"fetch_size": 25}) + assert payload["fetch_size"] == 25 + + def test_sql_clamps_fetch_size_to_max(self): + payload = SQL_DIALECT.execute_payload( + "SELECT * FROM `i`", {"fetch_size": 10_000_000} + ) + assert payload["fetch_size"] == DIRECT_QUERY_MAX_ROWS + + def test_sql_null_fetch_size_falls_back_to_default(self): + payload = SQL_DIALECT.execute_payload("SELECT * FROM `i`", {"fetch_size": None}) + assert payload["fetch_size"] == 1000 + + def test_sql_numeric_string_fetch_size(self): + payload = SQL_DIALECT.execute_payload("SELECT * FROM `i`", {"fetch_size": "25"}) + assert payload["fetch_size"] == 25 + + @pytest.mark.parametrize("value", ["abc", "", 0, -5, 3.9, [10], {"n": 1}, True]) + def test_sql_unusable_fetch_size_raises_value_error(self, value): + """A bad option must surface as a 400, not a comparison TypeError.""" + with pytest.raises(ValueError): + SQL_DIALECT.execute_payload("SELECT * FROM `i`", {"fetch_size": value}) + + +# -------------------------------------------------------------------------- +# get_sketch_scope +# -------------------------------------------------------------------------- +class TestGetSketchScope: + def test_returns_pattern_and_timeline_ids(self): + sketch = _make_sketch([_make_timeline(1, "aaa"), _make_timeline(2, "bbb")]) + pattern, timeline_ids = get_sketch_scope(sketch) + assert pattern == "aaa,bbb" + assert timeline_ids == [1, 2] + + def test_filters_by_timeline_ids(self): + sketch = _make_sketch( + [ + _make_timeline(1, "aaa"), + _make_timeline(2, "bbb"), + _make_timeline(3, "ccc"), + ] + ) + pattern, timeline_ids = get_sketch_scope(sketch, timeline_ids=[1, 3]) + assert pattern == "aaa,ccc" + assert timeline_ids == [1, 3] + + def test_deduplicates_shared_index(self): + """Two timelines can share one index; the pattern must list it once.""" + sketch = _make_sketch( + [_make_timeline(1, "shared"), _make_timeline(2, "shared")] + ) + pattern, timeline_ids = get_sketch_scope(sketch) + assert pattern == "shared" + assert timeline_ids == [1, 2] + + def test_skips_timeline_without_searchindex(self): + orphan = mock.MagicMock() + orphan.id = 9 + orphan.searchindex = None + sketch = _make_sketch([orphan, _make_timeline(1, "aaa")]) + pattern, timeline_ids = get_sketch_scope(sketch) + assert pattern == "aaa" + assert timeline_ids == [1] + + def test_empty_sketch(self): + assert get_sketch_scope(_make_sketch([])) == ("", []) + + def test_no_timeline_matches_the_filter(self): + sketch = _make_sketch([_make_timeline(1, "aaa")]) + assert get_sketch_scope(sketch, timeline_ids=[99]) == ("", []) + + def test_walks_timelines_once(self): + """The relationship is lazy-loaded, so it must not be iterated twice.""" + tracker = _CountingList([_make_timeline(1, "aaa")]) + sketch = mock.MagicMock() + sketch.timelines = tracker + get_sketch_scope(sketch) + assert tracker.iter_count == 1 + + +# -------------------------------------------------------------------------- +# Response helpers +# -------------------------------------------------------------------------- +class TestResponseHelpers: + def test_columns_from_schema(self): + data = {"schema": [{"name": "a"}, {"name": "b"}]} + assert columns_from_schema(data) == ["a", "b"] + + def test_columns_fallback_for_unnamed(self): + data = {"schema": [{}, {"name": "b"}]} + assert columns_from_schema(data) == ["col_0", "b"] + + def test_columns_missing_schema(self): + assert columns_from_schema({}) == [] + + def test_format_error_with_details(self): + data = { + "error": { + "type": "SyntaxError", + "reason": "bad token", + "details": "line 1", + } + } + assert format_opensearch_error(data) == "SyntaxError: bad token\nline 1" + + def test_format_error_without_details(self): + data = {"error": {"type": "SyntaxError", "reason": "bad token"}} + assert format_opensearch_error(data) == "SyntaxError: bad token" + + def test_format_error_non_dict(self): + assert format_opensearch_error({"error": "boom"}) == "boom" + + def test_empty_result_direct_has_row_fields(self): + envelope = empty_result(PPL_DIALECT, "boom", "direct") + assert envelope["language"] == "ppl" + assert envelope["error"] == "boom" + assert envelope["columns"] == [] + assert envelope["datarows"] == [] + assert envelope["total"] == 0 + assert envelope["size"] == 0 + + def test_empty_result_explain_omits_row_fields(self): + envelope = empty_result(SQL_DIALECT, "boom", "direct_explain") + assert envelope["result_type"] == "direct_explain" + assert envelope["language"] == "sql" + assert "datarows" not in envelope + + +# -------------------------------------------------------------------------- +# Resource wiring +# -------------------------------------------------------------------------- +class TestResourceWiring: + @pytest.mark.parametrize( + "resource", + [PplQueryResource, PplQueryExplainResource, PplQueryExportResource], + ) + def test_ppl_resources_pin_ppl(self, resource): + assert resource.dialect is PPL_DIALECT + + @pytest.mark.parametrize( + "resource", + [SqlQueryResource, SqlQueryExplainResource, SqlQueryExportResource], + ) + def test_sql_resources_pin_sql(self, resource): + assert resource.dialect is SQL_DIALECT + + +class TestExecuteResourceOptions: + """The execute resource turns unusable request options into a 400.""" + + def setup_method(self): + self.app = Flask(__name__) + # Short-circuits @login_required without wiring a LoginManager. + self.app.config["LOGIN_DISABLED"] = True + # Short-circuits the cluster capability probe: there is no cluster + # here for it to ask. + self.app.config["TESTING"] = True + self.sketch = _make_sketch([_make_timeline(1, "idx")]) + + def _post(self, body): + with self.app.test_request_context(json=body): + with mock.patch.object( + endpoints.Sketch, "get_with_acl", return_value=self.sketch + ): + return SqlQueryResource().post(1) + + def test_bad_fetch_size_is_bad_request(self): + with pytest.raises(HTTPException) as excinfo: + self._post({"query": "SELECT a", "fetch_size": "abc"}) + assert excinfo.value.code == 400 + + def test_negative_fetch_size_is_bad_request(self): + with pytest.raises(HTTPException) as excinfo: + self._post({"query": "SELECT a", "fetch_size": -1}) + assert excinfo.value.code == 400 + + +# -------------------------------------------------------------------------- +# _prepare_query +# -------------------------------------------------------------------------- +class TestPrepareQuery: + def setup_method(self): + self.app = Flask(__name__) + # Short-circuits the cluster capability probe: there is no cluster + # here for it to ask. + self.app.config["TESTING"] = True + self.sketch = _make_sketch([_make_timeline(1, "idx")]) + + def _prepare(self, body, dialect=PPL_DIALECT, sketch=None): + with self.app.test_request_context(json=body): + with mock.patch.object( + endpoints.Sketch, "get_with_acl", return_value=sketch or self.sketch + ): + return endpoints._prepare_query(1, dialect) + + def test_scopes_ppl_query(self): + prepared = self._prepare({"query": "stats count()"}) + assert prepared.dialect is PPL_DIALECT + assert prepared.scoped_query.startswith("search source=`idx`") + assert "__ts_timeline_id in (1)" in prepared.scoped_query + + def test_scopes_sql_query(self): + prepared = self._prepare({"query": "SELECT a"}, dialect=SQL_DIALECT) + assert prepared.dialect is SQL_DIALECT + assert "FROM `idx`" in prepared.scoped_query + assert "__ts_timeline_id IN (1)" in prepared.scoped_query + + def test_body_language_is_ignored(self): + """The route pins the dialect; a `language` field must not re-point it.""" + prepared = self._prepare( + {"query": "stats count()", "language": "sql"}, dialect=PPL_DIALECT + ) + assert prepared.dialect is PPL_DIALECT + + def test_empty_query_is_bad_request(self): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": " "}) + assert excinfo.value.code == 400 + + def test_sql_write_is_bad_request(self): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "DELETE FROM idx"}, dialect=SQL_DIALECT) + assert excinfo.value.code == 400 + + def test_missing_sketch_is_not_found(self): + with self.app.test_request_context(json={"query": "x"}): + with mock.patch.object(endpoints.Sketch, "get_with_acl", return_value=None): + with pytest.raises(HTTPException) as excinfo: + endpoints._prepare_query(1, PPL_DIALECT) + assert excinfo.value.code == 404 + + def test_archived_sketch_is_bad_request(self): + archived = _make_sketch([_make_timeline(1, "idx")], archived=True) + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "x"}, sketch=archived) + assert excinfo.value.code == 400 + + def test_sketch_without_indices_is_bad_request(self): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "x"}, sketch=_make_sketch([])) + assert excinfo.value.code == 400 + + def test_out_of_sketch_index_is_forbidden(self): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "source=other | head 1"}) + assert excinfo.value.code == 403 + + @pytest.mark.parametrize("query", [5, ["stats count()"], {"q": 1}, True]) + def test_non_string_query_is_bad_request(self, query): + """A wrong-typed query must 400, not fail inside .strip().""" + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": query}) + assert excinfo.value.code == 400 + + @pytest.mark.parametrize("timeline_ids", [1, "1", {"a": 1}, ["abc"], [None]]) + def test_bad_timeline_ids_is_bad_request(self, timeline_ids): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "stats count()", "timeline_ids": timeline_ids}) + assert excinfo.value.code == 400 + + def test_numeric_string_timeline_ids_are_accepted(self): + prepared = self._prepare({"query": "stats count()", "timeline_ids": ["1"]}) + assert "__ts_timeline_id in (1)" in prepared.scoped_query + + def test_timeline_ids_may_be_omitted(self): + prepared = self._prepare({"query": "stats count()"}) + assert "__ts_timeline_id in (1)" in prepared.scoped_query + + def _prepare_without_field(self, body, dialect=PPL_DIALECT): + with mock.patch.object( + endpoints, "index_pattern_has_timeline_field", return_value=False + ): + return self._prepare(body, dialect=dialect) + + def test_ppl_predicate_dropped_when_index_lacks_the_field(self): + """Naming an unmapped field is a hard error under Calcite.""" + prepared = self._prepare_without_field({"query": "stats count()"}) + assert "__ts_timeline_id" not in prepared.scoped_query + assert prepared.scoped_query.startswith("search source=`idx`") + + def test_sql_predicate_dropped_when_index_lacks_the_field(self): + prepared = self._prepare_without_field( + {"query": "SELECT a"}, dialect=SQL_DIALECT + ) + assert "__ts_timeline_id" not in prepared.scoped_query + assert "FROM `idx`" in prepared.scoped_query + + def test_plan_objection_is_forbidden(self): + """The plan check can veto even when the dialect was satisfied.""" + with mock.patch.object( + endpoints, + "verify_scope_with_explain", + return_value="PPL query targets indices outside this sketch.", + ): + with pytest.raises(HTTPException) as excinfo: + self._prepare({"query": "stats count()"}) + assert excinfo.value.code == 403 + + def test_plan_check_sees_the_scoped_query_and_allowlist(self): + with mock.patch.object( + endpoints, "verify_scope_with_explain", return_value=None + ) as verify: + self._prepare({"query": "stats count()"}) + scoped_query, allowed = verify.call_args[0][1:] + assert scoped_query.startswith("search source=`idx`") + assert allowed == ["idx"] + + def test_probe_is_asked_about_the_sketch_indices(self): + with mock.patch.object( + endpoints, "index_pattern_has_timeline_field", return_value=True + ) as probe: + self._prepare({"query": "stats count()"}) + assert probe.call_args[0][0] == "idx" + + def test_non_json_body_is_bad_request(self): + with self.app.test_request_context(data="not json"): + with mock.patch.object( + endpoints.Sketch, "get_with_acl", return_value=self.sketch + ): + with pytest.raises(HTTPException) as excinfo: + endpoints._prepare_query(1, PPL_DIALECT) + assert excinfo.value.code == 400 + + +# -------------------------------------------------------------------------- +# Timeline field mapping probe +# -------------------------------------------------------------------------- +class TestTimelineFieldProbe: + def setup_method(self): + base_module._timeline_field_cache.clear() + + teardown_method = setup_method + + @staticmethod + def _mapping(*present): + return { + name: {"mappings": {"__ts_timeline_id": {}} if is_present else {}} + for name, is_present in present + } + + @staticmethod + def _client(mapping=None, error=None): + client = _make_client() + if error is not None: + client.indices.get_field_mapping.side_effect = error + else: + client.indices.get_field_mapping.return_value = mapping + return client + + def _probe(self, index_pattern="idx", **client_kwargs): + client = self._client(**client_kwargs) + with mock.patch.object(base_module, "get_client", return_value=client): + result = base_module.index_pattern_has_timeline_field(index_pattern) + return result, client.indices.get_field_mapping + + def test_true_when_the_index_maps_the_field(self): + result, _ = self._probe(mapping=self._mapping(("idx", True))) + assert result is True + + def test_false_when_no_index_maps_the_field(self): + result, _ = self._probe(mapping=self._mapping(("idx", False))) + assert result is False + + def test_false_when_the_response_is_empty(self): + result, _ = self._probe(mapping={}) + assert result is False + + def test_true_when_any_index_in_the_pattern_maps_it(self): + """A multi-index query resolves the field from the union of mappings.""" + result, _ = self._probe( + index_pattern="a,b", mapping=self._mapping(("a", False), ("b", True)) + ) + assert result is True + + # Failing open keeps the predicate on: a loud query error beats silently + # widening a query's scope because a mapping call did not come back. + def test_true_when_opensearch_errors(self): + result, _ = self._probe(error=_rejected(status=500)) + assert result is True + + def test_true_when_the_response_cannot_be_deserialised(self): + result, _ = self._probe( + error=opensearch_exceptions.SerializationError("not json") + ) + assert result is True + + def test_true_when_the_request_raises(self): + result, _ = self._probe(error=_unreachable("no")) + assert result is True + + def _cached_probe(self, *patterns): + """Run the probe over several patterns against one client.""" + client = self._client(mapping=self._mapping(("idx", True))) + with mock.patch.object(base_module, "get_client", return_value=client): + for pattern in patterns: + base_module.index_pattern_has_timeline_field(pattern) + return client.indices.get_field_mapping + + def test_result_is_cached(self): + assert self._cached_probe("idx", "idx", "idx").call_count == 1 + + def test_distinct_patterns_are_cached_separately(self): + assert self._cached_probe("a", "b").call_count == 2 + + def test_expired_entry_is_refetched(self): + client = self._client(mapping=self._mapping(("idx", True))) + with mock.patch.object(base_module, "get_client", return_value=client): + base_module.index_pattern_has_timeline_field("idx") + expires_at, value = base_module._timeline_field_cache["idx"] + base_module._timeline_field_cache["idx"] = (expires_at - 10_000, value) + base_module.index_pattern_has_timeline_field("idx") + assert client.indices.get_field_mapping.call_count == 2 + + def test_cache_growth_is_bounded(self): + limit = base_module.TIMELINE_FIELD_CACHE_MAX_ENTRIES + self._cached_probe(*[f"idx{i}" for i in range(limit + 5)]) + assert len(base_module._timeline_field_cache) <= limit + + def test_probe_asks_only_for_the_timeline_field(self): + _, get_field_mapping = self._probe(mapping=self._mapping(("idx", True))) + assert get_field_mapping.call_args.kwargs == { + "fields": "__ts_timeline_id", + "index": "idx", + "ignore_unavailable": True, + "request_timeout": base_module.MAPPING_TIMEOUT_SECONDS, + } + + +# -------------------------------------------------------------------------- +# Execution-plan index extraction +# +# The plan bodies below are trimmed copies of real 3.7.0 responses, so the +# extractor is tested against shapes the cluster actually returns. +# -------------------------------------------------------------------------- +def _calcite_plan(*indices): + scans = "\n".join( + f" CalciteLogicalIndexScan(table=[[OpenSearch, {name}]])" for name in indices + ) + return {"calcite": {"logical": f"LogicalAggregate(group=[{{}}])\n{scans}\n"}} + + +class TestPlanIndices: + def test_calcite_single_index(self): + assert base_module.plan_indices(_calcite_plan("idx")) == {"idx"} + + def test_calcite_multi_index_pattern_is_split(self): + """One scan over `a,b` is two indices, both of which need checking.""" + assert base_module.plan_indices(_calcite_plan("a,b")) == {"a", "b"} + + def test_calcite_join_reports_both_scans(self): + assert base_module.plan_indices(_calcite_plan("a", "b")) == {"a", "b"} + + def test_v2_request_string(self): + plan = { + "root": { + "name": "ProjectOperator", + "children": [ + { + "name": "OpenSearchIndexScan", + "description": { + "request": "OpenSearchQueryRequest(indexName=idx, " + 'sourceBuilder={"from":0})' + }, + } + ], + } + } + assert base_module.plan_indices(plan) == {"idx"} + + def test_v2_request_string_multi_index(self): + plan = { + "root": { + "description": { + "request": "OpenSearchQueryRequest(indexName=a,b, " + 'sourceBuilder={"from":0})' + } + } + } + assert base_module.plan_indices(plan) == {"a", "b"} + + def test_v2_join_table_names(self): + """A SQL join names its tables in tableName, not a request string.""" + plan = { + "Logical Plan": { + "Join [ conditions=( a.m = b.m ) ]": { + "Group": [ + {"TableScan": {"tableAlias": "a", "tableName": "idx_a"}}, + {"TableScan": {"tableAlias": "b", "tableName": "idx_b"}}, + ] + } + } + } + assert base_module.plan_indices(plan) == {"idx_a", "idx_b"} + + def test_unrecognised_plan_yields_nothing(self): + assert base_module.plan_indices({"something": "else"}) == set() + + +class TestVerifyScopeWithExplain: + @staticmethod + def _verify(allowed, plan=None, error=None): + client = _make_client() + if error is not None: + client.plugins.ppl.explain.side_effect = error + else: + client.plugins.ppl.explain.return_value = plan + with mock.patch.object(base_module, "get_client", return_value=client): + result = base_module.verify_scope_with_explain( + PPL_DIALECT, "search source=`idx`", allowed + ) + return result, client + + def test_no_objection_when_plan_stays_in_the_sketch(self): + result, _ = self._verify(["idx"], plan=_calcite_plan("idx")) + assert result is None + + # SECURITY: this is the case the dialect regexes can miss. + def test_rejects_an_index_the_dialect_did_not_catch(self): + error, _ = self._verify(["idx"], plan=_calcite_plan("idx", "other")) + assert error is not None + assert "outside this sketch" in error + + def test_multi_index_pattern_all_in_sketch(self): + result, _ = self._verify(["a", "b"], plan=_calcite_plan("a,b")) + assert result is None + + def test_multi_index_pattern_partly_outside(self): + result, _ = self._verify(["a"], plan=_calcite_plan("a,b")) + assert result is not None + + # Falling back to the dialect's own (fail-closed) result beats taking the + # feature down when a plan cannot be read. + def test_unreadable_plan_raises_no_objection(self): + result, _ = self._verify(["idx"], plan={"unknown": "shape"}) + assert result is None + + def test_explain_error_raises_no_objection(self): + result, _ = self._verify(["idx"], error=_rejected()) + assert result is None + + def test_undeserialisable_plan_raises_no_objection(self): + result, _ = self._verify( + ["idx"], error=opensearch_exceptions.SerializationError("not json") + ) + assert result is None + + def test_connection_failure_raises_no_objection(self): + result, _ = self._verify(["idx"], error=_unreachable("no")) + assert result is None + + def test_it_explains_rather_than_executing(self): + _, client = self._verify(["idx"], plan=_calcite_plan("idx")) + client.plugins.ppl.explain.assert_called_once() + client.plugins.ppl.query.assert_not_called() + + +# -------------------------------------------------------------------------- +# PPL export streaming +# -------------------------------------------------------------------------- +class TestPplStream: + @staticmethod + def _client(pages=None, error=None): + client = _make_client() + if error is not None: + client.plugins.ppl.query.side_effect = error + elif isinstance(pages, list): + client.plugins.ppl.query.side_effect = pages + else: + client.plugins.ppl.query.return_value = pages + return client + + def test_single_page(self): + client = self._client({"schema": [{"name": "a"}], "datarows": [["x"], ["y"]]}) + lines = list(PPL_DIALECT.stream(client, "search source=`i`")) + assert json.loads(lines[0]) == {"columns": ["a"]} + assert json.loads(lines[1]) == {"a": "x"} + assert json.loads(lines[2]) == {"a": "y"} + + def test_paginates_until_short_page(self): + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"], ["2"]]}, + {"schema": [{"name": "a"}], "datarows": [["3"]]}, + ] + ) + with mock.patch.object(ppl_module, "DIRECT_QUERY_EXPORT_PAGE_SIZE", 2): + lines = list(PPL_DIALECT.stream(client, "search source=`i`")) + + assert [json.loads(x) for x in lines[1:]] == [ + {"a": "1"}, + {"a": "2"}, + {"a": "3"}, + ] + first, second = client.plugins.ppl.query.call_args_list + assert "head 2 from 0" in first.kwargs["body"]["query"] + assert "head 2 from 2" in second.kwargs["body"]["query"] + + def test_user_head_is_not_paginated(self): + """A user-supplied head already bounds the result set.""" + client = self._client({"schema": [{"name": "a"}], "datarows": [["1"], ["2"]]}) + with mock.patch.object(ppl_module, "DIRECT_QUERY_EXPORT_PAGE_SIZE", 2): + list(PPL_DIALECT.stream(client, "search source=`i` | head 2")) + query = client.plugins.ppl.query + assert query.call_count == 1 + assert "head 2 from" not in query.call_args.kwargs["body"]["query"] + + def test_rejected_query_yields_error_line(self): + client = self._client(error=_rejected("bad query")) + lines = list(PPL_DIALECT.stream(client, "search source=`i`")) + assert "bad query" in json.loads(lines[0])["error"] + + def test_failure_mid_export_is_marked_incomplete(self): + """A short download must not be mistakable for a finished one.""" + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [[i] for i in range(2)]}, + _timed_out(), + ] + ) + with mock.patch.object(ppl_module, "DIRECT_QUERY_EXPORT_PAGE_SIZE", 2): + lines = list(PPL_DIALECT.stream(client, "search source=`i`")) + + trailer = json.loads(lines[-1]) + assert trailer["incomplete"] is True + assert trailer["rows_returned"] == 2 + assert trailer["failed_at_offset"] == 2 + assert "SQL export" in trailer["detail"] + + def test_connection_error_yields_error_line(self): + client = self._client(error=_unreachable()) + lines = list(PPL_DIALECT.stream(client, "search source=`i`")) + assert "down" in json.loads(lines[0])["error"] + + def test_stream_is_lazy(self): + """Building the generator must not issue a request.""" + client = self._client({}) + PPL_DIALECT.stream(client, "search source=`i`") + client.plugins.ppl.query.assert_not_called() + + +# -------------------------------------------------------------------------- +# SQL export streaming +# -------------------------------------------------------------------------- +class TestSqlStream: + @staticmethod + def _client(pages=None, error=None): + client = _make_client() + if error is not None: + client.plugins.sql.query.side_effect = error + elif isinstance(pages, list): + client.plugins.sql.query.side_effect = pages + else: + client.plugins.sql.query.return_value = pages + return client + + def test_single_page(self): + client = self._client({"schema": [{"name": "a"}], "datarows": [["x"]]}) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + assert json.loads(lines[0]) == {"columns": ["a"]} + assert json.loads(lines[1]) == {"a": "x"} + + def test_follows_cursor(self): + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"]], "cursor": "c1"}, + {"datarows": [["2"]], "cursor": "c2"}, + {"datarows": [["3"]]}, + ] + ) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + + assert [json.loads(x) for x in lines[1:]] == [ + {"a": "1"}, + {"a": "2"}, + {"a": "3"}, + ] + calls = client.plugins.sql.query.call_args_list + assert calls[1].kwargs["body"] == {"cursor": "c1"} + assert calls[2].kwargs["body"] == {"cursor": "c2"} + + def test_cursor_reuses_first_page_columns(self): + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [], "cursor": "c1"}, + {"datarows": [["2"]]}, + ] + ) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + assert json.loads(lines[-1]) == {"a": "2"} + + def test_rejected_query_yields_error_line(self): + client = self._client(error=_rejected("bad sql")) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + assert "bad sql" in json.loads(lines[0])["error"] + + def test_cursor_failure_is_marked_incomplete(self): + """Rows already streamed are counted, so a short file is obvious.""" + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"]], "cursor": "c1"}, + _rejected("cursor is gone", status=500), + ] + ) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + + trailer = json.loads(lines[-1]) + assert "cursor is gone" in trailer["error"] + assert trailer["incomplete"] is True + assert trailer["rows_returned"] == 1 + + def test_connection_error_yields_error_line(self): + client = self._client(error=_unreachable()) + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + assert "down" in json.loads(lines[0])["error"] + + def test_stream_is_lazy(self): + client = self._client({}) + SQL_DIALECT.stream(client, "SELECT a FROM `i`") + client.plugins.sql.query.assert_not_called() + + def test_exhausted_cursor_needs_no_closing(self): + """The last page returns no cursor, so there is no context left.""" + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"]], "cursor": "c1"}, + {"datarows": [["2"]]}, + ] + ) + list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + client.plugins.sql.close.assert_not_called() + + @pytest.mark.parametrize( + "consumed,expected_cursor", + # Abandoning during the first page must close it too, not only a page + # reached through the cursor loop. + [(1, "c1"), (2, "c1"), (3, "c2")], + ) + def test_abandoned_download_closes_the_live_cursor(self, consumed, expected_cursor): + """A cancelled export must not leave a context held on the cluster.""" + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"]], "cursor": "c1"}, + {"datarows": [["2"]], "cursor": "c2"}, + ] + ) + generator = SQL_DIALECT.stream(client, "SELECT a FROM `i`") + for _ in range(consumed): + next(generator) + generator.close() + + client.plugins.sql.close.assert_called_once_with( + body={"cursor": expected_cursor} + ) + + def test_a_failed_close_does_not_break_the_export(self): + client = self._client( + [ + {"schema": [{"name": "a"}], "datarows": [["1"]], "cursor": "c1"}, + _unreachable(), + ] + ) + client.plugins.sql.close.side_effect = _unreachable("also down") + lines = list(SQL_DIALECT.stream(client, "SELECT a FROM `i`")) + assert json.loads(lines[-1])["incomplete"] is True diff --git a/timesketch/api/v1/resources/sketch.py b/timesketch/api/v1/resources/sketch.py index 388d99d01a..108a8d0684 100644 --- a/timesketch/api/v1/resources/sketch.py +++ b/timesketch/api/v1/resources/sketch.py @@ -34,6 +34,7 @@ from timesketch.api.v1 import resources from timesketch.api.v1 import utils +from timesketch.api.v1.resources.direct_query.capability import direct_query_support from timesketch.lib import forms from timesketch.lib.definitions import HTTP_STATUS_CODE_OK from timesketch.lib.definitions import HTTP_STATUS_CODE_CREATED @@ -500,6 +501,12 @@ def get(self, sketch_id): except ValueError: pass + # PPL and SQL come from the OpenSearch SQL plugin rather than the search + # API, so their availability is a cluster property. The frontend drops + # them from the search-mode menu when this is false, the same way it + # drops wildcard on a sketch without wildcard mappings. + supports_direct_query = bool(direct_query_support()) + views = [] for view in sketch.get_named_views: if not view.user: @@ -565,6 +572,7 @@ def get(self, sketch_id): else [] ), "supports_wildcard": supports_wildcard, + "supports_direct_query": supports_direct_query, } return self.to_json(sketch, meta=meta) diff --git a/timesketch/api/v1/routes.py b/timesketch/api/v1/routes.py index b518f01af4..45989a233e 100644 --- a/timesketch/api/v1/routes.py +++ b/timesketch/api/v1/routes.py @@ -25,6 +25,12 @@ from .resources.analysis import AnalyzerSessionResource from .resources.attribute import AttributeResource from .resources.explore import ExploreResource +from .resources.direct_query import PplQueryResource +from .resources.direct_query import PplQueryExplainResource +from .resources.direct_query import PplQueryExportResource +from .resources.direct_query import SqlQueryResource +from .resources.direct_query import SqlQueryExplainResource +from .resources.direct_query import SqlQueryExportResource from .resources.explore import SearchHistoryResource from .resources.explore import SearchHistoryTreeResource from .resources.explore import ExploreWildcardResource @@ -132,6 +138,14 @@ "/sketches//aggregation//", ), (ExploreResource, "/sketches//explore/"), + # Dialect-specific endpoints, mirroring how explore_wildcard is separated + # from explore. The language is fixed by the route, never by the body. + (PplQueryResource, "/sketches//explore/ppl/"), + (PplQueryExplainResource, "/sketches//explore/ppl/explain/"), + (PplQueryExportResource, "/sketches//explore/ppl/export/"), + (SqlQueryResource, "/sketches//explore/sql/"), + (SqlQueryExplainResource, "/sketches//explore/sql/explain/"), + (SqlQueryExportResource, "/sketches//explore/sql/export/"), ( ExploreWildcardResource, "/sketches//explore_wildcard/", diff --git a/timesketch/app.py b/timesketch/app.py index f2d641a3cf..7faf50f18a 100644 --- a/timesketch/app.py +++ b/timesketch/app.py @@ -219,6 +219,16 @@ def load_user(user_id): # Setup CSRF protection for the whole application CSRFProtect(app) + # The direct-query package holds its own long-lived OpenSearch client so + # that a paged export does not rebuild the connection pool on every page. + # Built here rather than on first use so that a malformed cluster + # configuration is reported at startup. Imported inside the factory to + # keep the resource package out of the module import graph. + # pylint: disable=import-outside-toplevel + from timesketch.api.v1.resources.direct_query.base import configure_client + + configure_client(app) + if app.config.get("ENABLE_PROFILING", False) and not app.config.get("TESTING"): # pylint: disable=import-outside-toplevel from werkzeug.middleware.profiler import ProfilerMiddleware diff --git a/timesketch/frontend-ng/setup-tests.js b/timesketch/frontend-ng/setup-tests.js index 790b8c5b6a..75d301b237 100644 --- a/timesketch/frontend-ng/setup-tests.js +++ b/timesketch/frontend-ng/setup-tests.js @@ -1,3 +1,14 @@ +// Components read the bare localStorage global from data(), which happy-dom +// does not define. Without this the component throws before the test body runs. +if (typeof globalThis.localStorage === 'undefined') { + const store = new Map() + globalThis.localStorage = { + getItem: (key) => (store.has(key) ? store.get(key) : null), + setItem: (key, value) => store.set(key, String(value)), + removeItem: (key) => store.delete(key), + clear: () => store.clear(), + } +} const node = document.createElement("meta"); node.textContent = 'test'; diff --git a/timesketch/frontend-ng/src/components/Explore/DirectQueryEditor.vue b/timesketch/frontend-ng/src/components/Explore/DirectQueryEditor.vue new file mode 100644 index 0000000000..8b91c3b2ca --- /dev/null +++ b/timesketch/frontend-ng/src/components/Explore/DirectQueryEditor.vue @@ -0,0 +1,250 @@ + + + + + + diff --git a/timesketch/frontend-ng/src/components/Explore/DirectQueryError.vue b/timesketch/frontend-ng/src/components/Explore/DirectQueryError.vue new file mode 100644 index 0000000000..588002e8b3 --- /dev/null +++ b/timesketch/frontend-ng/src/components/Explore/DirectQueryError.vue @@ -0,0 +1,59 @@ + + + + diff --git a/timesketch/frontend-ng/src/components/Explore/DirectQueryPanel.vue b/timesketch/frontend-ng/src/components/Explore/DirectQueryPanel.vue new file mode 100644 index 0000000000..791cdb66ad --- /dev/null +++ b/timesketch/frontend-ng/src/components/Explore/DirectQueryPanel.vue @@ -0,0 +1,338 @@ + + + + + + diff --git a/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.test.js b/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.test.js new file mode 100644 index 0000000000..bcf37efdff --- /dev/null +++ b/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.test.js @@ -0,0 +1,72 @@ +import { shallowMount, createLocalVue } from '@vue/test-utils' +import Vuetify from 'vuetify' +import Vue from 'vue' +import { expect, it, describe, beforeEach } from 'vitest' +import DirectQueryTable from './DirectQueryTable.vue' +import EventBus from '../../event-bus.js' + +const localVue = createLocalVue() +Vue.use(Vuetify) + +const mountTable = (columns, datarows) => + shallowMount(DirectQueryTable, { + localVue, + vuetify: new Vuetify(), + propsData: { columns, datarows, total: datarows.length, language: 'ppl' }, + }) + +// The component keys rows by column position, so a fixture row is addressed the +// same way the template addresses it. +const rowItem = (wrapper, rowIndex) => wrapper.vm.tableRows[rowIndex] + +describe('DirectQueryTable.vue pivot', () => { + let wrapper + + beforeEach(() => { + wrapper = mountTable(['username', 'cnt'], [['analyst@example.com', 42], ['', 7]]) + }) + + it('offers a pivot on a field column', () => { + expect(wrapper.vm.canPivot(rowItem(wrapper, 0), 0)).toBe(true) + }) + + it('does not offer a pivot on a numeric aggregate column', () => { + expect(wrapper.vm.canPivot(rowItem(wrapper, 0), 1)).toBe(false) + }) + + it('does not offer a pivot on an empty value', () => { + expect(wrapper.vm.canPivot(rowItem(wrapper, 1), 0)).toBe(false) + }) + + it('does not offer a pivot on a value past the keyword ignore_above limit', () => { + const w = mountTable(['path'], [['a'.repeat(256)], ['b'.repeat(257)]]) + expect(w.vm.canPivot(rowItem(w, 0), 0)).toBe(true) + expect(w.vm.canPivot(rowItem(w, 1), 0)).toBe(false) + }) + + it('does not offer a pivot on an aggregate expression column', () => { + const w = mountTable(['count()', 'span(datetime,1h)'], [['x', 'y']]) + expect(w.vm.canPivot(rowItem(w, 0), 0)).toBe(false) + expect(w.vm.canPivot(rowItem(w, 0), 1)).toBe(false) + }) + + it('emits a term chip that the event list understands', () => { + let received = null + EventBus.$once('setQueryAndFilter', (event) => { + received = event + }) + + wrapper.vm.pivot(rowItem(wrapper, 0), 0) + + expect(received).toEqual({ + doSearch: true, + chip: { + field: 'username', + value: 'analyst@example.com', + type: 'term', + operator: 'must', + active: true, + }, + }) + }) +}) diff --git a/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.vue b/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.vue new file mode 100644 index 0000000000..589e906677 --- /dev/null +++ b/timesketch/frontend-ng/src/components/Explore/DirectQueryTable.vue @@ -0,0 +1,416 @@ + + + + + + diff --git a/timesketch/frontend-ng/src/components/Explore/SearchGuideCard.vue b/timesketch/frontend-ng/src/components/Explore/SearchGuideCard.vue index c264a7e7b1..ff9c667457 100644 --- a/timesketch/frontend-ng/src/components/Explore/SearchGuideCard.vue +++ b/timesketch/frontend-ng/src/components/Explore/SearchGuideCard.vue @@ -23,6 +23,87 @@ limitations under the License.
+
+

PPL (Piped Processing Language) Examples

+

The sketch's index is added automatically — just type your pipe commands.

+ + + +
+ +
+

SQL Examples

+

The FROM clause is added automatically — just type your SELECT query.

+ + + +
+ + +
@@ -309,6 +391,10 @@ export default { type: Boolean, default: false, }, + queryLanguage: { + type: String, + default: 'lucene', + }, searchMode: { type: String, default: 'query_string', diff --git a/timesketch/frontend-ng/src/components/Explore/SearchHelpCard.vue b/timesketch/frontend-ng/src/components/Explore/SearchHelpCard.vue index 8baa27051e..bdff0a76aa 100644 --- a/timesketch/frontend-ng/src/components/Explore/SearchHelpCard.vue +++ b/timesketch/frontend-ng/src/components/Explore/SearchHelpCard.vue @@ -19,6 +19,7 @@ limitations under the License. :show-tags="false" :show-data-types="false" :show-saved-searches="false" + :query-language="queryLanguage" @search-triggered="$emit('close-dialog')" >