[DarshitChanpura]

Description

Replaces ThreadContext.stashContext with the plugin subject assigned through IdentityAwarePlugin, so sql reaches the indices it owns as its plugin subject rather than from a stashed context that skips privilege evaluation.

All 12 production stash calls were in three classes, each wrapping a call on the node client. SQLPlugin now builds a PluginClient around that client, hands it the subject in assignSubject, and passes it to the three:

• OpenSearchDataSourceMetadataStorage, 5 sites, reads and writes .ql-datasources
• StateStore, 6 sites, reads and writes .query_execution_request_<datasource>
• IndexCleanup, 1 site, runs the delete-by-query behind FlintIndexRetention

PluginClient is a final FilterClient that runs each action inside PluginSubject.runAs:

• The listener is wrapped with ContextPreservingActionListener.wrapPreservingContext, so it runs in the caller's thread context, keeps the response headers the action added, and leaves the responding thread's own context in place afterwards.
• A synchronous failure is thrown to the caller, as the wrapped client would throw it. It is not also sent to the listener, so a listener that throws is not notified a second time.
• The subject can be assigned once. Assigning the same subject again is a no-op, a different one is rejected, and null is rejected. An action run before a subject is assigned throws, so a wiring mistake fails at once rather than running unauthenticated.
• getRemoteClusterClient is not supported, because a plugin subject has no meaning on another cluster. Nothing calls it on this client; RemoteMappingsReader uses the node client.

PluginClient lives in the opensearch module, which already owns the server facing client code and is already on the compile classpath of all three call sites and of the plugin entry point, so this adds no module dependency. It is in its own commit so that moving to a shared class in common-utils, when that publishes, is a delete plus an import change.

The three client fields are typed PluginClient rather than Client, so a plain node client cannot be wired in by mistake. Nothing else moves. sql queries user indices by design, and the paths that do that, including OpenSearchJobExecutionResponseReader reading a query's results and OpenSearchAsyncQueryScheduler writing a job document, keep the node client and keep running with the caller's privileges.

Ordering: assignSubject runs after createComponents, so the client is built first and the subject arrives second. Nothing calls through it in between. The only index work createComponents does is createDataSource on the default datasource, which returns before it reaches storage, and if that ever changed the client would throw rather than run without a subject.

Index coverage and the permissions file

.ql-datasources is registered through getSystemIndexDescriptors, and .query_execution_request_<datasource> is covered by the registered .query_execution_request* pattern. A plugin subject is allowed every index action on its own registered system indices, so the gets, searches, indexes, updates, deletes and index creations need nothing declared.

IndexCleanup is also driven against query_execution_result*, which is not registered and should not be: the Spark job writes that index and the user reads their own results back from it, so making it a system index would hide it from both. On that pattern the subject is evaluated like any other role, so plugin-additional-permissions.yml grants what the retention purge runs there:

• indices:data/write/delete/byquery, the request itself
• indices:data/read/search, the search that opens its scroll. Its shard level phases are not evaluated again.
• indices:data/write/bulk* and indices:data/write/delete, for the shard level bulk that carries the deletes. Security checks a shard bulk against the action of each item in it as well as against bulk[s].
• indices:admin/refresh*, for setRefresh(true). The refresh runs the shard level indices:admin/refresh[s] through the action filters, so the literal indices:admin/refresh alone would not cover it.

indices:data/read/scroll and indices:data/read/scroll/clear are cluster permissions in security's classification even though they read an index, so they are listed under cluster_permissions. The top level bulk is cluster classified too, but security grants it to every plugin already.

Every entry is a literal action or a wildcard. Action group names such as read are not used, because they match no action in this file.

Delete-by-query failures are now reported

Delete-by-query completes normally when searches or deletes inside it were rejected, and lists them in the response. IndexCleanup read only the deleted count, so a rejected purge looked like a purge with nothing to delete. Under a stashed context privileges were not evaluated and this could not happen; under the plugin subject it can. IndexCleanup now fails its listener when the response lists a search or delete failure, and FlintIndexRetention logs it as an error and continues with the next index, as it already does for other failures. A version conflict is not counted: it means the document was written after the search found it, so it is not stale.

Other behaviour checked for this change

• Reads of a just written document. StateStore creates and updates session, statement, job metadata and Flint index state documents and later reads them back by id, for example OpenSearchSessionStorageService.getSession, OpenSearchStatementStorageService.getStatement, OpenSearchAsyncQueryJobMetadataStorageService.getJobMetadata and OpenSearchFlintIndexStateModelService.getFlintIndexStateModel, all through StateStore.get. Under the plugin subject a get on a system index is evaluated and made non realtime. StateStore.get already sets refresh(true), and a non realtime get with refresh(true) refreshes the shard before reading, so it still sees the write. Every StateStore write also uses WAIT_UNTIL. OpenSearchDataSourceMetadataStorage reads only through search, after writes that use IMMEDIATE. So no read path changes, and no refresh workaround is added.
• Resource sharing. sql registers no resource types, and none of .ql-datasources, .query_execution_request* or query_execution_result* is a resource index, so resource sharing filtering does not apply to any migrated search whether resource sharing is on or off. The counts behind the session and refresh job limits in DefaultLeaseManager and the active session and statement gauges stay cluster wide.
• Nested actions. No migrated path calls a sql transport action. Each one issues only index, get, update, delete, search, create index or delete-by-query on the plugin's own indices.
• Without the security plugin, core assigns a no-op subject that only stashes the context, so every migrated path behaves as it did before.

Testing

spotlessCheck, and build for opensearch, datasources and async-query (compile, unit tests, jacocoTestCoverageVerification), plus test for plugin, pass. Unit test counts from the JUnit XML, against a baseline on main:

module	this branch	main
opensearch	1834	1822
datasources	106	106
async-query	162	156
plugin	64 (7 skipped)	64 (7 skipped)

The delta is PluginClientTest (12 tests) and IndexCleanupTest (6 tests).

PluginClientTest covers running inside the subject, the index() and admin() paths the stores use, the caller's context in the listener on success and on failure, the responding thread's context afterwards, response headers, a synchronous failure thrown and not sent to the listener, a throwing listener notified once, use before a subject is assigned, set once, null, and the remote cluster client. Each test was confirmed to fail on its own when the client is changed to the variant it guards against: action outside runAs, raw listener, newStoredContext(false) with runBefore, newRestorableContext(false), a catch that sends failures to the listener, an unchecked setSubject, a forwarding getRemoteClusterClient, no null check, no unassigned check, and an index() that forwards straight to the wrapped client. IndexCleanupTest covers the request shape, the deleted count, a rejected delete, a rejected search, an ignored version conflict and a transport failure. Its failure tests fail when the failure lists are ignored, and its conflict test fails when conflicts are not excluded.

Not verified by execution: the permissions file. Unit tests do not go through the security plugin. This repository runs integTestWithSecurity on every pull request, against a cluster with the security plugin installed and system indices enabled, but every test in that task is under org.opensearch.sql.security and none of them touch .ql-datasources, the async query state index or the retention cron. So nothing in CI exercises a migrated call site under security, and the entries in the permissions file are reasoned from security's evaluator rather than observed passing.

The async-query tests that build a real StateStore (StatementTest, InteractiveSessionTest, OpenSearchAsyncQueryJobMetadataStorageServiceTest and the AsyncQueryExecutorServiceSpec subclasses) are JUnit 4. The test task includes only the jupiter engine, and the junit4 task resolves to NO-SOURCE, so the 15 JUnit 4 classes in async-query do not run in CI, on main either. Run locally with the task temporarily pointed at the test classes, and not committed: StatementTest, InteractiveSessionTest and OpenSearchAsyncQueryJobMetadataStorageServiceTest pass, 27 of 27, against an in-memory cluster with the plugin client and core's no-op subject. The Spec classes and the house keeper tests fail with the same 76 failures on this branch and on main, from a settings registration error in their setup. Wiring that task back up and fixing those classes is left for a separate change.

Related

Tracking issue: opensearch-project/opensearch-plugins#238
RFC: opensearch-project/security#4439

Check List

• New functionality includes testing.
• Commits are signed per the DCO using --signoff.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

[github-actions]

PR Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit dd2a5f4.

Path	Line	Severity	Description
plugin/src/main/resources/plugin-additional-permissions.yml	30	medium	The index_permissions block grants indices:data/write/bulk*, indices:data/write/delete/byquery, and indices:admin/refresh to any index matching 'query_execution_result*'. This wildcard could cover unintended indices if a user or another component creates an index whose name begins with that prefix, allowing the plugin subject to write to or delete from it without a per-datasource check. The comment acknowledges the intent, but the blast radius of the wildcard is wider than a pinned index name.
opensearch/src/main/java/org/opensearch/sql/opensearch/client/PluginClient.java	35	low	The subject field is a mutable volatile set post-construction via setSubject(). Between plugin startup and the assignSubject() callback from the identity framework, any call to doExecute() throws IllegalStateException, which is caught but surfaces as a listener failure rather than a startup guard. A race during early startup (e.g., a cluster-manager event firing before assignSubject completes) could silently drop operations. This is an operational correctness concern rather than deliberate malice, but it is worth verifying the startup ordering guarantee.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 0 | Medium: 1 | Low: 1

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit ca64ad2)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible Issue The failureOf method filters out CONFLICT status from bulkFailures but does not check if the list is empty before accessing the first element. If all bulk failures are conflicts and there are no search failures, bulkFailures.get(0) will throw IndexOutOfBoundsException. This occurs when documents are written after the search finds them but all deletes conflict. private static Exception failureOf(String indexName, BulkByScrollResponse response) { // A conflict means the doc was written after the search found it, so it is not stale. List<BulkItemResponse.Failure> bulkFailures = response.getBulkFailures().stream() .filter(failure -> failure.getStatus() != RestStatus.CONFLICT) .toList(); List<ScrollableHitSource.SearchFailure> searchFailures = response.getSearchFailures(); if (bulkFailures.isEmpty() && searchFailures.isEmpty()) { return null; } Throwable cause = bulkFailures.isEmpty() ? searchFailures.get(0).getReason() : bulkFailures.get(0).getCause(); return new IllegalStateException( Possible Issue The subject field is an AtomicReference read by action threads and written once by assignSubject on a different thread. While AtomicReference ensures visibility of the reference itself, there is no guarantee that the PluginSubject object's internal state is visible to reading threads if it is not immutable or does not establish its own happens-before relationship. If PluginSubject implementations have mutable state initialized after construction, action threads may see stale or partially constructed state. private final AtomicReference<PluginSubject> subject = new AtomicReference<>();

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to ca64ad2
Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Ensure listener notified exactly once The listener contract is violated when both onResponse and onFailure could be called. If failureOf returns non-null after logging deleted docs, the listener receives both a response (implicitly through the log) and a failure. Ensure the listener is notified exactly once by checking for failures before logging success. async-query/src/main/java/org/opensearch/sql/spark/cluster/IndexCleanup.java [52-71] client.execute( DeleteByQueryAction.INSTANCE, deleteRequest, ActionListener.wrap( response -> { - long deleted = response.getDeleted(); - if (deleted > 0) { - // if 0 docs get deleted, it means our query cannot find any matching doc - // or the index does not exist at all - LOG.info("{} docs are deleted for index:{}", deleted, indexName); - } Exception failure = failureOf(indexName, response); if (failure != null) { listener.onFailure(failure); } else { + long deleted = response.getDeleted(); + if (deleted > 0) { + // if 0 docs get deleted, it means our query cannot find any matching doc + // or the index does not exist at all + LOG.info("{} docs are deleted for index:{}", deleted, indexName); + } listener.onResponse(deleted); } }, listener::onFailure)); Suggestion importance[1-10]: 7 __ Why: Valid improvement to check for failures before logging success. This ensures cleaner separation of success and failure paths, though the original code already ensures the listener is called exactly once (either onFailure or onResponse). The reordering improves code clarity.	Medium
General	Prevent race condition in subject assignment The compareAndExchange operation may fail to detect concurrent assignments if two threads call setSubject simultaneously with different subjects. Use compareAndSet in a loop or synchronize the method to ensure thread-safe assignment and prevent race conditions. opensearch/src/main/java/org/opensearch/sql/opensearch/client/PluginClient.java [54-60] -public void setSubject(PluginSubject subject) { +public synchronized void setSubject(PluginSubject subject) { Objects.requireNonNull(subject, "subject"); - PluginSubject existing = this.subject.compareAndExchange(null, subject); + PluginSubject existing = this.subject.get(); if (existing != null && existing != subject) { throw new IllegalStateException("PluginClient already has a subject assigned."); } + this.subject.set(subject); } Suggestion importance[1-10]: 3 __ Why: The suggestion addresses a theoretical race condition, but compareAndExchange is specifically designed to handle concurrent access atomically. While synchronization would work, it's unnecessary overhead. The existing implementation correctly handles the case where the same subject is assigned twice and rejects different subjects.	Low

---

Previous suggestions

Suggestions up to commit dd2a5f4

Category	Suggestion	Impact
Possible issue	Fix resource leak in context management The stored context is not closed when runAs succeeds, creating a potential resource leak. The context should be closed in a finally block or the restoration callback should handle closure to ensure cleanup happens in all code paths. opensearch/src/main/java/org/opensearch/sql/opensearch/client/PluginClient.java [50-65] ThreadContext.StoredContext storedContext = threadPool().getThreadContext().newStoredContext(false); try { currentSubject.runAs( () -> { LOG.debug( "Running transport action as subject: {}", currentSubject.getPrincipal().getName()); super.doExecute( - action, request, ActionListener.runBefore(listener, storedContext::restore)); + action, request, ActionListener.runBefore(listener, () -> { + try { + storedContext.restore(); + } finally { + storedContext.close(); + } + })); }); } catch (Exception e) { - // Reported through the listener rather than thrown, so an async caller is not left waiting. storedContext.close(); listener.onFailure(e); } Suggestion importance[1-10]: 8 __ Why: This identifies a genuine resource leak where storedContext is not closed in the success path. The StoredContext should be closed after use to prevent resource leaks. The suggested fix properly ensures cleanup in both success and failure paths by closing the context after restoration.	Medium
General	Add index existence check before deletion The deleteDocsByQuery method should handle the case where the index doesn't exist before attempting deletion. Without this check, the operation may fail unnecessarily when the index is missing, which is a valid state according to the comment. async-query/src/main/java/org/opensearch/sql/spark/cluster/IndexCleanup.java [45-58] +if (!clusterService.state().routingTable().hasIndex(indexName)) { + LOG.debug("Index {} does not exist, skipping deletion", indexName); + listener.onResponse(0L); + return; +} + client.execute( DeleteByQueryAction.INSTANCE, deleteRequest, ActionListener.wrap( response -> { long deleted = response.getDeleted(); if (deleted > 0) { - // if 0 docs get deleted, it means our query cannot find any matching doc - // or the index does not exist at all LOG.info("{} docs are deleted for index:{}", deleted, indexName); } listener.onResponse(response.getDeleted()); }, listener::onFailure)); Suggestion importance[1-10]: 5 __ Why: The suggestion adds a defensive check for index existence before deletion. While this could improve robustness, the IndicesOptions.LENIENT_EXPAND_OPEN option on line 42 already handles missing indices gracefully, making this check redundant. The suggestion is valid but offers marginal improvement.	Low

[codecov]

Codecov Report

❌ Patch coverage is 62.19512% with 31 lines in your changes missing coverage. Please review.
✅ Project coverage is 75.28%. Comparing base (9249643) to head (ca64ad2).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...rch/sql/spark/execution/statestore/StateStore.java	0.00%	31 Missing ⚠️

❌ Your project check has failed because the head coverage (75.28%) is below the target coverage (99.00%). You can increase the head coverage or adjust the target coverage.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #5838      +/-   ##
============================================
+ Coverage     75.19%   75.28%   +0.08%     
- Complexity    10362    10383      +21     
============================================
  Files           962      963       +1     
  Lines         41098    41107       +9     
  Branches       4683     4687       +4     
============================================
+ Hits          30905    30948      +43     
+ Misses         8896     8862      -34     
  Partials       1297     1297              

Flag	Coverage Δ
sql-engine	75.28% <62.19%> (+0.08%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Persistent review updated to latest commit ca64ad2