Skip to content

Record the creator on a workflow's sharing record #2300

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DarshitChanpura wants to merge 7 commits into
base: main
Choose a base branch
from
Open

Record the creator on a workflow's sharing record #2300

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5909377
Cover all_shared_principals and search for monitors and workflows
DarshitChanpura Oct 12, 2026
069a8a3
Record the creator on a workflow's sharing record
DarshitChanpura Oct 12, 2026
60b457e
Declare the type value of workflow documents
DarshitChanpura Oct 14, 2026
e13ec29
Match workflow documents by the value they store
DarshitChanpura Oct 21, 2026
d88ce61
Merge remote-tracking branch 'upstream/main' into fix/workflow-resour…
DarshitChanpura Oct 22, 2026
f3dcd02
Merge remote-tracking branch 'upstream/main' into fix/workflow-resour…
DarshitChanpura Oct 30, 2026
bb19a0f
Reset the calling thread before the workflow write suspends
DarshitChanpura Oct 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions alerting/src/main/kotlin/org/opensearch/alerting/AlertingResourceSharingExtension.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,17 +9,17 @@ import org.opensearch.commons.alerting.model.ScheduledJob.Companion.SCHEDULED_JO
import org.opensearch.security.spi.resources.ResourceProvider
import org.opensearch.security.spi.resources.ResourceSharingExtension
import org.opensearch.security.spi.resources.client.ResourceSharingClient
import java.util.function.Function

class AlertingResourceSharingExtension : ResourceSharingExtension {
/**
* Monitors and workflows share [SCHEDULED_JOBS_INDEX]. Each provider declares its own
* type-specific field paths so the security plugin can distinguish them without a
* top-level discriminator on the stored doc:
*
* - [ResourceProvider.typeField] — used by the postIndex hook to classify new writes.
* `monitor.type` is present on monitor docs and absent on workflow docs (and vice versa),
* so the framework iterates matching providers and picks the one whose typeField
* resolves non-null.
* - [ResourceProvider.typeField] and [ResourceProvider.matchesResource]: used to classify
* documents. Workflow documents hold `workflow` in `workflow.type`, while the type is
* registered as `alerting-workflow`, so the workflow provider matches on that value.
* - [ResourceProvider.ownerNamePath] / [ResourceProvider.ownerBackendRolesPath] — used by
* the security plugin's `POST /_plugins/_security/api/resources/migrate` endpoint when
* seeding sharing entries for legacy docs, so a single admin call can attribute owners
Expand All @@ -38,6 +38,8 @@ class AlertingResourceSharingExtension : ResourceSharingExtension {
override fun resourceType(): String = ResourceSharingUtils.WORKFLOW_RESOURCE_TYPE
override fun resourceIndexName(): String = SCHEDULED_JOBS_INDEX
override fun typeField(): String = "workflow.type"
override fun matchesResource(fieldValue: Function<String, String>): Boolean =

@riysaxen-amzn riysaxen-amzn Oct 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: this only compiles against a security SPI that has matchesResource (security#6627), and build.gradle doesn't change, so it relies on the current security snapshot. Since the description says the two must ship together, would a minimum security version check or a release-notes entry be worth adding so that dependency is enforced rather than only described?

@DarshitChanpura DarshitChanpura Oct 30, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Alerting compiles against the security SPI of its own OpenSearch version (opensearch-security-spi:${opensearch_build}), and the two ship in the same release. A build against an SPI without matchesResource does not compile, so that is the check. Release notes are generated from the PR metadata now, so the description carries the dependency.

fieldValue.apply(typeField()) == "workflow"
override fun ownerNamePath(): String = "/workflow/user/name"
override fun ownerBackendRolesPath(): String = "/workflow/user/backend_roles"
}
Expand Down
25 changes: 11 additions & 14 deletions alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ import org.opensearch.alerting.util.indexStashed
import org.opensearch.alerting.util.isADMonitor
import org.opensearch.alerting.util.isQueryLevelMonitor
import org.opensearch.alerting.util.use
import org.opensearch.alerting.util.withStoredContext
import org.opensearch.alerting.workflow.CompositeWorkflowRunner
import org.opensearch.cluster.service.ClusterService
import org.opensearch.common.inject.Inject
Expand Down Expand Up @@ -227,6 +228,8 @@ class TransportIndexWorkflowAction @Inject constructor(
}

val tenantId = client.threadPool().threadContext.getHeader(AlertingPlugin.TENANT_ID_HEADER)
// Capture before launch: Dispatchers.IO drops the thread context, and the workflow write needs the caller's auth
val storedContext = client.threadPool().threadContext.newStoredContext(false)
scope.launch(TenantContext(tenantId)) {
try {
val triggerCount = transformedRequest.workflow.triggers.size
Expand All @@ -246,10 +249,6 @@ class TransportIndexWorkflowAction @Inject constructor(
client,
object : ActionListener<AcknowledgedResponse> {
override fun onResponse(response: AcknowledgedResponse) {
// Capture pre-stash context so the write path can restore the auth-user
// transient — the security plugin's ResourceIndexListener needs it to
// record the workflow's resource-share entry with createdBy=<caller>.
val storedContext = client.threadPool().threadContext.newStoredContext(false)
client.threadPool().threadContext.stashContext().use {
IndexWorkflowHandler(
client, actionListener, transformedRequest, user, tenantId, storedContext
Expand Down Expand Up @@ -285,7 +284,7 @@ class TransportIndexWorkflowAction @Inject constructor(
private val request: IndexWorkflowRequest,
private val user: User?,
private val tenantId: String?,
private val storedThreadContext: org.opensearch.common.util.concurrent.ThreadContext.StoredContext? = null,
private val storedThreadContext: org.opensearch.common.util.concurrent.ThreadContext.StoredContext,
) {
fun resolveUserAndStart() {
scope.launch(TenantContext(tenantId)) {
Expand Down Expand Up @@ -432,12 +431,11 @@ class TransportIndexWorkflowAction @Inject constructor(
.setIfPrimaryTerm(request.primaryTerm)
.timeout(indexTimeout)

// Restore the caller's persistent auth header so the shard-level ResourceIndexListener sees
// the caller when recording the workflow's share entry. [Client.indexStashed] stashes
// per-call so the internal-index write isn't gated by the caller's privileges.
storedThreadContext?.restore()
try {
val indexResponse: IndexResponse = client.indexStashed(indexRequest)
// The caller's context reaches only this write, where the security plugin records the creator
val indexResponse: IndexResponse = client.threadPool().threadContext.withStoredContext(storedThreadContext) {
client.indexStashed(indexRequest)
}
val failureReasons = checkShardsFailure(indexResponse)
if (failureReasons != null) {
log.error("Failed to create workflow: $failureReasons")
Expand Down Expand Up @@ -600,11 +598,10 @@ class TransportIndexWorkflowAction @Inject constructor(
.setIfPrimaryTerm(request.primaryTerm)
.timeout(indexTimeout)

// Restore the caller's persistent auth header for the shard-level ResourceIndexListener;
// [Client.indexStashed] stashes per-call so the internal-index write isn't gated by the caller.
storedThreadContext?.restore()
try {
val indexResponse: IndexResponse = client.indexStashed(indexRequest)
val indexResponse: IndexResponse = client.threadPool().threadContext.withStoredContext(storedThreadContext) {

@riysaxen-amzn riysaxen-amzn Oct 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question: on update, this restores the context captured on the request thread, which is the updater's, not the creator's. That's fine if the security plugin only records createdBy on the first index of a doc. Could you confirm that an update by a user the workflow is shared with doesn't rewrite the owner? I couldn't tell from the new tests whether "update" there means updating the workflow doc or updating the sharing.

@DarshitChanpura DarshitChanpura Oct 30, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed. The security plugin writes the sharing record only when the document is created. On a later write to the resource, ResourceIndexListener.postIndex only restores all_shared_principals from the record and does not change created_by.

"Update" in the existing test meant an update of the workflow document by the owner. In bb19a0f I added test workflow update by a read-write shared user keeps the owner: bob, with read-write access to the workflow and its delegate monitor, renames the workflow, and the sharing record still names alice as the owner.

client.indexStashed(indexRequest)
}
val failureReasons = checkShardsFailure(indexResponse)
if (failureReasons != null) {
actionListener.onFailure(
Expand Down
14 changes: 14 additions & 0 deletions alerting/src/main/kotlin/org/opensearch/alerting/util/SdkClientExtensions.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,20 @@ suspend fun SdkClient.deleteDataObjectStashed(
.await()
}

/**
* Runs [block] under [stored], then restores the context the coroutine had, on whichever thread it resumes. This keeps
* the caller's context from reaching the calls that follow [block].
*/
suspend fun <T> ThreadContext.withStoredContext(stored: ThreadContext.StoredContext, block: suspend () -> T): T {

@riysaxen-amzn riysaxen-amzn Oct 30, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can leave the caller's identity on a worker thread

ThreadContext is per thread, and block suspends. Tracing the write path:

  1. On IO worker A, this saves own, then stored.restore() puts the caller's context on A.
  2. indexStashed calls stashContext() on A. A stash drops transient headers but keeps persistent ones, and the caller's auth is a persistent header (the comment removed in this PR said the write relies on that). It then suspends on client.index.
  3. On Dispatchers.IO the coroutine may resume on worker B, so both stored.close() in indexStashed and own.restore() here run on B.
  4. Nothing resets A. It keeps the stash state, which still carries the caller's persistent auth header.

The next coroutine that captures context on A can then pick up this caller, which is the "Alice's workflow recorded as Bob's" symptom from the description. The KDoc's promise ("keeps the caller's context from reaching the calls that follow") holds for the coroutine, but not for thread A.

One way to avoid it is to do the restore, stash and dispatch synchronously on A before suspending, and reset A right away:

suspendCoroutine { cont ->
    val own = ctx.newStoredContext(false)
    stored.restore()
    try {
        ctx.stashContext().use { client.index(request, listenerFor(cont)) }
    } finally {
        own.restore()
    }
}

The listener resumes the coroutine, and no thread keeps the caller's context after this returns. I haven't reproduced the leak; this is from the ThreadContext semantics, so please correct me if the security plugin clears the persistent header somewhere I missed.

@DarshitChanpura DarshitChanpura Oct 30, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right. stashContext() keeps persistent headers, so after the coroutine suspended, the thread that started the write kept the caller's persistent headers, and the thread it resumed on received the starting thread's context.

In bb19a0f I replaced withStoredContext and indexStashed with indexWithContext, which follows your shape: it sets the caller's context, stashes, dispatches the index request, and gives the calling thread its own context back before suspending. SdkClientExtensionsTests starts the write on one thread and answers it on another. It fails on the previous helper (the starting thread still held the caller's header) and passes now.

val own = newStoredContext(false)
stored.restore()
try {
return block()
} finally {
own.restore()
}
}

/**
* Executes [Client.index] with the caller's [ThreadContext] stashed just before dispatch; the stash
* is closed via the completion callback so it works across the async / coroutine-resume boundary.
Expand Down
Loading
Loading