From 37c02c74ff901dfa3479242e378ce6a414dc5c77 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 20 May 2026 23:29:35 +0200 Subject: [PATCH 1/2] [codex] Authorize workflow imports before persistence (#7510) * Authorize workflow imports before persistence * Address import authorization review feedback * Fix import authorization test fixture data --- .../WorkflowDefinitions/Import/Endpoint.cs | 13 +-- .../ImportAuthorizationExtensions.cs | 92 +++++++++++++++ .../ImportFiles/Endpoint.cs | 53 +++++---- .../Import/ImportAuthorizationTests.cs | 108 ++++++++++++++++++ 4 files changed, 237 insertions(+), 29 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportAuthorizationExtensions.cs create mode 100644 test/component/Elsa.Workflows.ComponentTests/Scenarios/RestApis/Endpoints/WorkflowDefinitions/Import/ImportAuthorizationTests.cs diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Import/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Import/Endpoint.cs index 2f12131f2..2d3eec31f 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Import/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/Import/Endpoint.cs @@ -1,6 +1,4 @@ using Elsa.Abstractions; -using Elsa.Workflows.Api.Constants; -using Elsa.Workflows.Api.Requirements; using Elsa.Workflows.Api.Security; using Elsa.Workflows.Management; using Elsa.Workflows.Management.Models; @@ -15,6 +13,7 @@ namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.Import; [PublicAPI] internal class Import : ElsaEndpoint { + private readonly IWorkflowDefinitionStore _workflowDefinitionStore; private readonly IWorkflowDefinitionImporter _workflowDefinitionImporter; private readonly IWorkflowDefinitionLinker _linker; private readonly IAuthorizationService _authorizationService; @@ -22,11 +21,13 @@ internal class Import : ElsaEndpoint /// public Import( + IWorkflowDefinitionStore workflowDefinitionStore, IWorkflowDefinitionImporter workflowDefinitionImporter, IWorkflowDefinitionLinker linker, IAuthorizationService authorizationService, PythonWorkflowDefinitionAuthorizationService pythonAuthorizationService) { + _workflowDefinitionStore = workflowDefinitionStore; _workflowDefinitionImporter = workflowDefinitionImporter; _linker = linker; _authorizationService = authorizationService; @@ -54,10 +55,7 @@ internal class Import : ElsaEndpoint return; } - var result = await ImportSingleWorkflowDefinitionAsync(model, cancellationToken); - var definition = result.WorkflowDefinition; - - var authorizationResult = await _authorizationService.AuthorizeAsync(User, new NotReadOnlyResource(definition), AuthorizationPolicies.NotReadOnlyPolicy); + var authorizationResult = await _authorizationService.AuthorizeWorkflowDefinitionImportAsync(User, _workflowDefinitionStore, model, cancellationToken); if (!authorizationResult.Succeeded) { @@ -65,6 +63,8 @@ internal class Import : ElsaEndpoint return; } + var result = await ImportSingleWorkflowDefinitionAsync(model, cancellationToken); + var definition = result.WorkflowDefinition; var updatedModel = await _linker.MapAsync(definition, cancellationToken); if (result.Succeeded) @@ -98,5 +98,4 @@ internal class Import : ElsaEndpoint return result; } - } diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportAuthorizationExtensions.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportAuthorizationExtensions.cs new file mode 100644 index 000000000..bbe3cb9dc --- /dev/null +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportAuthorizationExtensions.cs @@ -0,0 +1,92 @@ +using System.Security.Claims; +using Elsa.Common.Models; +using Elsa.Workflows.Api.Constants; +using Elsa.Workflows.Api.Requirements; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Management.Models; +using Microsoft.AspNetCore.Authorization; + +namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions; + +internal static class ImportAuthorizationExtensions +{ + public static async Task AuthorizeWorkflowDefinitionImportAsync( + this IAuthorizationService authorizationService, + ClaimsPrincipal user, + IWorkflowDefinitionStore workflowDefinitionStore, + WorkflowDefinitionModel model, + CancellationToken cancellationToken) + { + var definition = await FindExistingDefinitionAsync(workflowDefinitionStore, model.DefinitionId, cancellationToken); + return await authorizationService.AuthorizeAsync(user, new NotReadOnlyResource(definition), AuthorizationPolicies.NotReadOnlyPolicy); + } + + public static async Task AuthorizeWorkflowDefinitionImportsAsync( + this IAuthorizationService authorizationService, + ClaimsPrincipal user, + IWorkflowDefinitionStore workflowDefinitionStore, + IEnumerable models, + CancellationToken cancellationToken) + { + var modelList = models.ToList(); + + if (modelList.Count == 0) + return await authorizationService.AuthorizeAsync(user, new NotReadOnlyResource(), AuthorizationPolicies.NotReadOnlyPolicy); + + var definitions = await FindExistingDefinitionsAsync(workflowDefinitionStore, modelList, cancellationToken); + + foreach (var model in modelList) + { + definitions.TryGetValue(model.DefinitionId ?? string.Empty, out var definition); + var authorizationResult = await authorizationService.AuthorizeAsync(user, new NotReadOnlyResource(definition), AuthorizationPolicies.NotReadOnlyPolicy); + + if (!authorizationResult.Succeeded) + return authorizationResult; + } + + return AuthorizationResult.Success(); + } + + private static async Task FindExistingDefinitionAsync( + IWorkflowDefinitionStore workflowDefinitionStore, + string? definitionId, + CancellationToken cancellationToken) + { + if (string.IsNullOrWhiteSpace(definitionId)) + return null; + + return await workflowDefinitionStore.FindAsync(new WorkflowDefinitionFilter + { + DefinitionId = definitionId, + VersionOptions = VersionOptions.Latest + }, cancellationToken); + } + + private static async Task> FindExistingDefinitionsAsync( + IWorkflowDefinitionStore workflowDefinitionStore, + IEnumerable models, + CancellationToken cancellationToken) + { + var definitionIds = models + .Select(x => x.DefinitionId) + .Where(x => !string.IsNullOrWhiteSpace(x)) + .Select(x => x!) + .Distinct(StringComparer.Ordinal) + .ToList(); + + if (definitionIds.Count == 0) + return new Dictionary(); + + var definitions = await workflowDefinitionStore.FindManyAsync(new WorkflowDefinitionFilter + { + DefinitionIds = definitionIds, + VersionOptions = VersionOptions.Latest + }, cancellationToken); + + return definitions + .GroupBy(x => x.DefinitionId, StringComparer.Ordinal) + .ToDictionary(x => x.Key, x => x.First(), StringComparer.Ordinal); + } +} diff --git a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportFiles/Endpoint.cs b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportFiles/Endpoint.cs index 61334a825..99bce9e04 100644 --- a/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportFiles/Endpoint.cs +++ b/src/modules/Elsa.Workflows.Api/Endpoints/WorkflowDefinitions/ImportFiles/Endpoint.cs @@ -1,9 +1,6 @@ using Elsa.Abstractions; -using Elsa.Workflows.Api.Constants; -using Elsa.Workflows.Api.Requirements; using Elsa.Workflows.Api.Security; using Elsa.Workflows.Management; -using Elsa.Workflows.Management.Mappers; using Elsa.Workflows.Management.Models; using JetBrains.Annotations; using Microsoft.AspNetCore.Authorization; @@ -17,25 +14,22 @@ namespace Elsa.Workflows.Api.Endpoints.WorkflowDefinitions.ImportFiles; [PublicAPI] internal class ImportFiles : ElsaEndpoint { - private readonly IWorkflowDefinitionService _workflowDefinitionService; + private readonly IWorkflowDefinitionStore _workflowDefinitionStore; private readonly IWorkflowDefinitionImporter _workflowDefinitionImporter; - private readonly WorkflowDefinitionMapper _workflowDefinitionMapper; private readonly IApiSerializer _apiSerializer; private readonly IAuthorizationService _authorizationService; private readonly PythonWorkflowDefinitionAuthorizationService _pythonAuthorizationService; /// public ImportFiles( - IWorkflowDefinitionService workflowDefinitionService, + IWorkflowDefinitionStore workflowDefinitionStore, IWorkflowDefinitionImporter workflowDefinitionImporter, - WorkflowDefinitionMapper workflowDefinitionMapper, IApiSerializer apiSerializer, IAuthorizationService authorizationService, PythonWorkflowDefinitionAuthorizationService pythonAuthorizationService) { - _workflowDefinitionService = workflowDefinitionService; + _workflowDefinitionStore = workflowDefinitionStore; _workflowDefinitionImporter = workflowDefinitionImporter; - _workflowDefinitionMapper = workflowDefinitionMapper; _apiSerializer = apiSerializer; _authorizationService = authorizationService; _pythonAuthorizationService = pythonAuthorizationService; @@ -52,17 +46,21 @@ internal class ImportFiles : ElsaEndpoint /// public override async Task HandleAsync(WorkflowDefinitionModel model, CancellationToken cancellationToken) { - var authorizationResult = await _authorizationService.AuthorizeAsync(User, new NotReadOnlyResource(), AuthorizationPolicies.NotReadOnlyPolicy); - - if (!authorizationResult.Succeeded) - { - await Send.ForbiddenAsync(cancellationToken); - return; - } - if (Files.Any()) { - var count = await ImportFilesAsync(Files, cancellationToken); + var models = await ReadWorkflowDefinitionModelsAsync(Files, cancellationToken); + if (!await AuthorizePythonUsageAsync(models, cancellationToken)) + return; + + var authorizationResult = await _authorizationService.AuthorizeWorkflowDefinitionImportsAsync(User, _workflowDefinitionStore, models, cancellationToken); + + if (!authorizationResult.Succeeded) + { + await Send.ForbiddenAsync(cancellationToken); + return; + } + + var count = await ImportWorkflowDefinitionsAsync(models, cancellationToken); if (!ValidationFailed && !HttpContext.Response.HasStarted) await Send.OkAsync(new { Count = count }, cancellationToken); @@ -72,28 +70,39 @@ internal class ImportFiles : ElsaEndpoint await Send.ErrorsAsync(400, cancellationToken); } - private async Task ImportFilesAsync(IFormFileCollection files, CancellationToken cancellationToken) + private async Task> ReadWorkflowDefinitionModelsAsync(IFormFileCollection files, CancellationToken cancellationToken) { var models = await WorkflowDefinitionImportFileReader.ReadAsync(files, _apiSerializer, () => HttpContext.Response.HasStarted, cancellationToken); + return models.ToList(); + } + private async Task AuthorizePythonUsageAsync(IEnumerable models, CancellationToken cancellationToken) + { foreach (var model in models) { var pythonAuthorizationResult = await _pythonAuthorizationService.AuthorizeAsync(model, User, cancellationToken); if (pythonAuthorizationResult != PythonWorkflowDefinitionAuthorizationResult.Allowed) { await PythonWorkflowDefinitionAuthorizationFailure.SendAsync(pythonAuthorizationResult, Send.ForbiddenAsync, message => AddError(message), Send.ErrorsAsync, cancellationToken); - return 0; + return false; } } + return true; + } + + private async Task ImportWorkflowDefinitionsAsync(IEnumerable models, CancellationToken cancellationToken) + { var count = 0; foreach (var model in models) { if (HttpContext.Response.HasStarted) return count; - await ImportSingleWorkflowDefinitionAsync(model, cancellationToken); - count++; + var result = await ImportSingleWorkflowDefinitionAsync(model, cancellationToken); + + if (result.Succeeded) + count++; } return count; diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/RestApis/Endpoints/WorkflowDefinitions/Import/ImportAuthorizationTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/RestApis/Endpoints/WorkflowDefinitions/Import/ImportAuthorizationTests.cs new file mode 100644 index 000000000..611b389bd --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/RestApis/Endpoints/WorkflowDefinitions/Import/ImportAuthorizationTests.cs @@ -0,0 +1,108 @@ +using System.Net; +using System.Text; +using System.Text.Json; +using Elsa.Api.Client.Resources.WorkflowDefinitions.Contracts; +using Elsa.Api.Client.Resources.WorkflowDefinitions.Models; +using Elsa.Workflows; +using Elsa.Workflows.Activities; +using Elsa.Workflows.ComponentTests.Abstractions; +using Elsa.Workflows.ComponentTests.Fixtures; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Management.Materializers; +using Microsoft.Extensions.DependencyInjection; +using Refit; +using WorkflowDefinitionEntity = Elsa.Workflows.Management.Entities.WorkflowDefinition; + +namespace Elsa.Workflows.ComponentTests.Scenarios.RestApis.Endpoints.WorkflowDefinitions.Import; + +public class ImportAuthorizationTests : AppComponentTest +{ + private readonly IWorkflowDefinitionStore _store; + private readonly IActivitySerializer _activitySerializer; + private readonly IWorkflowDefinitionsApi _client; + + public ImportAuthorizationTests(App app) : base(app) + { + _store = Scope.ServiceProvider.GetRequiredService(); + _activitySerializer = Scope.ServiceProvider.GetRequiredService(); + _client = WorkflowServer.CreateApiClient(); + } + + [Fact] + public async Task ImportExistingReadOnlyDefinition_ShouldReturnForbiddenAndLeaveStorageUnchanged() + { + var definitionId = $"readonly-import-{Guid.NewGuid():N}"; + await SaveDefinitionAsync(definitionId, "Original", isReadonly: true); + + var exception = await Assert.ThrowsAsync(() => _client.ImportAsync(CreateImportModel(definitionId, "Updated"))); + + Assert.Equal(HttpStatusCode.Forbidden, exception.StatusCode); + await AssertDefinitionUnchangedAsync(definitionId, "Original", isReadonly: true); + } + + [Fact] + public async Task ImportFilesWithReadOnlyTarget_ShouldReturnForbiddenAndLeaveStorageUnchanged() + { + var writableDefinitionId = $"writable-import-files-{Guid.NewGuid():N}"; + var readOnlyDefinitionId = $"readonly-import-files-{Guid.NewGuid():N}"; + await SaveDefinitionAsync(writableDefinitionId, "Writable Original"); + await SaveDefinitionAsync(readOnlyDefinitionId, "ReadOnly Original", isReadonly: true); + + await using var writableStream = CreateImportStream(writableDefinitionId, "Writable Updated"); + await using var readOnlyStream = CreateImportStream(readOnlyDefinitionId, "ReadOnly Updated"); + var files = new List + { + new(writableStream, "writable.json", "application/json"), + new(readOnlyStream, "readonly.json", "application/json") + }; + + var exception = await Assert.ThrowsAsync(() => _client.ImportFilesAsync(files)); + + Assert.Equal(HttpStatusCode.Forbidden, exception.StatusCode); + await AssertDefinitionUnchangedAsync(writableDefinitionId, "Writable Original"); + await AssertDefinitionUnchangedAsync(readOnlyDefinitionId, "ReadOnly Original", isReadonly: true); + } + + private async Task SaveDefinitionAsync(string definitionId, string name, bool isReadonly = false) + { + await _store.SaveAsync(new WorkflowDefinitionEntity + { + Id = Guid.NewGuid().ToString("N"), + DefinitionId = definitionId, + Name = name, + CreatedAt = DateTimeOffset.UtcNow, + IsLatest = true, + IsReadonly = isReadonly, + MaterializerName = JsonWorkflowMaterializer.MaterializerName, + StringData = _activitySerializer.Serialize(new Sequence()) + }); + } + + private async Task AssertDefinitionUnchangedAsync(string definitionId, string expectedName, bool isReadonly = false) + { + var definitions = (await _store.FindManyAsync(new WorkflowDefinitionFilter + { + DefinitionId = definitionId + })).ToList(); + + var definition = Assert.Single(definitions); + Assert.Equal(expectedName, definition.Name); + Assert.Equal(isReadonly, definition.IsReadonly); + } + + private static WorkflowDefinitionModel CreateImportModel(string definitionId, string name) + { + return new() + { + DefinitionId = definitionId, + Name = name + }; + } + + private static MemoryStream CreateImportStream(string definitionId, string name) + { + var json = JsonSerializer.Serialize(CreateImportModel(definitionId, name), new JsonSerializerOptions(JsonSerializerDefaults.Web)); + return new(Encoding.UTF8.GetBytes(json)); + } +} From 3fc87945fd6a7fefc030c5e82373ac82744db625 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Thu, 21 May 2026 00:31:34 +0200 Subject: [PATCH 2/2] Stabilize bulk dispatch fire-and-forget component test (#7520) * Stabilize bulk dispatch fire-and-forget test * Improve bulk dispatch timeout diagnostics * Stabilize bulk dispatch polling test * Reduce bulk dispatch test polling pressure --- .../BulkDispatchWorkflowsTests.cs | 61 +++++++++++++++---- 1 file changed, 50 insertions(+), 11 deletions(-) diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs index 928bc317a..5f4f636b3 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/Activities/Composition/BulkDispatchWorkflows/BulkDispatchWorkflowsTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using Elsa.Common.Models; using Elsa.Testing.Shared; using Elsa.Testing.Shared.Models; @@ -7,6 +8,8 @@ using Elsa.Workflows.ComponentTests.Abstractions; using Elsa.Workflows.ComponentTests.Fixtures; using Elsa.Workflows.ComponentTests.Scenarios.Activities.Composition.BulkDispatchWorkflows.Workflows; using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Models; using Elsa.Workflows.State; using Microsoft.Extensions.DependencyInjection; @@ -16,11 +19,17 @@ namespace Elsa.Workflows.ComponentTests.Scenarios.Activities.Composition.BulkDis public class BulkDispatchWorkflowsTests : AppComponentTest { private const int ChildWorkflowTimeoutSeconds = 30; + private const int InitialChildWorkflowPollingIntervalMilliseconds = 250; + private const int MaxChildWorkflowPollingIntervalMilliseconds = 1000; + // The runtime persists this marker under the activity property name; resume handlers read the same key from WorkflowState.Properties. + private const string WaitForCompletionPropertyName = nameof(Elsa.Workflows.Runtime.Activities.BulkDispatchWorkflows.WaitForCompletion); private readonly AsyncWorkflowRunner _workflowRunner; + private readonly IWorkflowInstanceStore _workflowInstanceStore; public BulkDispatchWorkflowsTests(App app) : base(app) { _workflowRunner = Scope.ServiceProvider.GetRequiredService(); + _workflowInstanceStore = Scope.ServiceProvider.GetRequiredService(); } [Fact(DisplayName = "BulkDispatchWorkflows should wait for all child workflows to complete")] @@ -37,21 +46,19 @@ public class BulkDispatchWorkflowsTests : AppComponentTest { var expectedChildCount = 3; - // Run the main workflow and wait for child workflows to complete - var (result, completedChildWorkflows) = await RunWorkflowAndWaitForChildWorkflowsAsync( - BulkDispatchFireAndForgetWorkflow.DefinitionId, + var result = await RunWorkflowAsync(BulkDispatchFireAndForgetWorkflow.DefinitionId); + + AssertWorkflowFinished(result); + var childWorkflowInstances = await WaitForChildWorkflowInstancesAsync( + result.WorkflowExecutionContext.Id, SlowBulkChildWorkflow.DefinitionId, expectedChildCount); - AssertWorkflowFinished(result); - var mainWorkflowCompletedAt = result.WorkflowExecutionContext.UpdatedAt; - - // Assert that all child workflows completed after the main workflow - Assert.Equal(expectedChildCount, completedChildWorkflows.Count); - foreach (var childContext in completedChildWorkflows) + Assert.Equal(expectedChildCount, childWorkflowInstances.Count); + foreach (var childWorkflowInstance in childWorkflowInstances) { - Assert.True(childContext.UpdatedAt > mainWorkflowCompletedAt, - $"Child workflow should complete after main workflow. Main: {mainWorkflowCompletedAt}, Child: {childContext.UpdatedAt}"); + Assert.Equal(result.WorkflowExecutionContext.Id, childWorkflowInstance.ParentWorkflowInstanceId); + Assert.False(childWorkflowInstance.WorkflowState.Properties.ContainsKey(WaitForCompletionPropertyName)); } } @@ -176,4 +183,36 @@ public class BulkDispatchWorkflowsTests : AppComponentTest workflowEvents.WorkflowStateCommitted -= OnWorkflowStateCommitted; } } + + private async Task> WaitForChildWorkflowInstancesAsync( + string parentWorkflowInstanceId, + string childWorkflowDefinitionId, + int expectedChildCount, + CancellationToken cancellationToken = default) + { + var timeout = TimeSpan.FromSeconds(ChildWorkflowTimeoutSeconds); + var pollingInterval = TimeSpan.FromMilliseconds(InitialChildWorkflowPollingIntervalMilliseconds); + var maxPollingInterval = TimeSpan.FromMilliseconds(MaxChildWorkflowPollingIntervalMilliseconds); + var stopwatch = Stopwatch.StartNew(); + var filter = new WorkflowInstanceFilter + { + DefinitionId = childWorkflowDefinitionId, + ParentWorkflowInstanceIds = [parentWorkflowInstanceId] + }; + var actualChildCount = 0; + + while (stopwatch.Elapsed < timeout) + { + var instances = (await _workflowInstanceStore.FindManyAsync(filter, cancellationToken)).ToList(); + actualChildCount = instances.Count; + + if (instances.Count >= expectedChildCount) + return instances; + + await Task.Delay(pollingInterval, cancellationToken); + pollingInterval = TimeSpan.FromMilliseconds(Math.Min(pollingInterval.TotalMilliseconds * 2, maxPollingInterval.TotalMilliseconds)); + } + + throw new TimeoutException($"Expected {expectedChildCount} child workflow instances of definition '{childWorkflowDefinitionId}' for parent '{parentWorkflowInstanceId}', but found {actualChildCount}."); + } }