From 0bf1f842743f54dfcbc1b74affaa96d43449880f Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Mon, 12 Oct 2026 21:05:53 +0800 Subject: [PATCH] Keep secrets out of run records at cancels, cuts and flushes Several writes put a team secret, or a fragment of one, where every member of the team (Viewers included) can read it, or into storage that redaction is supposed to keep clean. A cancelled external call wrote its raw target to external_call.failed (the records endpoint, the SSE tail, the timeline). Every other branch already wrote the redacted target. Both cancellation branches of TraceExternalCallAsync and TraceExternalStreamAsync now do the same. agent.run_command cut its output (the inline cap, or the runner's capture budget) before the engine's exact-match redaction. A cut inside a secret left a fragment that no exact match recognises, and the artifact stored the raw output with no encrypted sidecar. The engine now hands each node its scope redactor on NodeRunContext. The node redacts first and cuts second, and masks a trailing fragment of a budget-cut capture that could start a secret. An uncut capture keeps its runtime value: the engine still masks it in node.completed, and downstream nodes, resumes and reruns read the real value from the sidecar. PersistArtifactAsync masks text and JSON for the same reason. Streamed interaction.delta rows were redacted one at a time, so a secret split across a 32 KiB or one-second flush was stored in two halves and the live view joined them back together. A flush now releases only text that no secret can still run across, and carries the rest. When the stream ends, the held text is released, so the rows still join up to the masked output. When a cancel or a provider fault cuts the stream, the held text is masked instead: it is exactly the possible start of a secret whose rest never arrived. The call's presence delta reports that its capture was masked. The agent plane had the same gap at its reader's 8 MiB pass. A stdout or stderr line longer than one pass arrives as a cut frame plus its continuation, and each was redacted on its own. A key that the cut landed in was stored as a head in one native record (and transcript line) and a tail in the next. The executor now carries the end of a cut frame that could start a secret into the frame that continues it, on the live tail, the re-attached tail and the diagnostics drain. Both redactors now take the settle rule from one shared SecretBoundary helper. http.request logged URL userinfo before any redaction ran, and the startup placeholder guard printed the whole value of a partly rendered setting to stderr. Both now name only what is needed: the scheme, host and port; the key and its placeholders. --- .../Services/Agents/AgentRunExecutor.cs | 9 +- .../Services/Agents/SecretRedactor.cs | 36 ++ .../Services/Redaction/SecretBoundary.cs | 51 +++ .../Workflows/Engine/WorkflowEngine.cs | 4 +- .../Llm/RecordingLLMClientDecorator.cs | 2 +- ...ngStreamingStructuredLLMClientDecorator.cs | 92 ++++-- .../Nodes/Builtin/AgentRunCommandNode.cs | 31 +- .../Nodes/Builtin/HttpRequestNode.cs | 3 +- .../Workflows/Nodes/NodeObservability.cs | 42 ++- .../Workflows/Nodes/NodeRunContext.cs | 9 + .../Runtime/PersistenceSecretRedactor.cs | 31 ++ .../Settings/ConfigurationPlaceholderGuard.cs | 10 +- .../NodeRecordRedactionEndpointE2ETests.cs | 309 ++++++++++++++++++ .../Infrastructure/PostgresFixture.cs | 8 + .../Workflows/AgentRunExecutorTests.cs | 64 ++++ .../Workflows/AgentRunReattachFlowTests.cs | 56 +++- .../SecretEchoStreamingLlmClient.cs | 63 ++++ .../Workflows/NodeRecordRedactionFlowTests.cs | 283 ++++++++++++++++ .../Agents/SecretRedactorTests.cs | 48 +++ .../ConfigurationPlaceholderGuardTests.cs | 35 +- .../Workflows/AgentRunCommandNodeTests.cs | 70 ++++ .../HttpRequestNodeLogTargetTests.cs | 64 ++++ .../Llm/RecordingLLMClientDecoratorTests.cs | 205 +++++++++++- .../Workflows/NodeObservabilityTests.cs | 76 +++++ .../Workflows/SecretBoundaryRedactionTests.cs | 89 +++++ 25 files changed, 1641 insertions(+), 49 deletions(-) create mode 100644 backend/src/CodeSpace.Core/Services/Redaction/SecretBoundary.cs create mode 100644 backend/tests/CodeSpace.E2ETests/Workflows/NodeRecordRedactionEndpointE2ETests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Workflows/Infrastructure/SecretEchoStreamingLlmClient.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Workflows/NodeRecordRedactionFlowTests.cs create mode 100644 backend/tests/CodeSpace.UnitTests/Workflows/HttpRequestNodeLogTargetTests.cs create mode 100644 backend/tests/CodeSpace.UnitTests/Workflows/SecretBoundaryRedactionTests.cs diff --git a/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs b/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs index aa8cef97d..cda2402d9 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs @@ -1075,11 +1075,12 @@ public async Task ReattachAsync(AgentRunReattachReservation reservation, Cancell var writer = NewEventWriter(context.Owner, context.TeamId); // same batched-append + flush-at-checkpoint path as the live tail var native = await OpenResumedCaptureAsync(context, redactor, cancellationToken).ConfigureAwait(false); // G1: the RESUMED frame stream of the same process, continuing its source cursor and the execution's reduction var applicationSourceHead = context.Handle.StdoutOffset; + var frames = redactor.CreateFrameStream(); // a line the reader cut is redacted with its continuation, as on the live tail async Task PersistFrameAsync(SandboxOutputFrame output) { var line = output.Text; - var redactedLine = redactor.Redact(line); + var redactedLine = frames.Redact(line, output.IsComplete); // A best-effort native flush may have failed after the normalized events were durable and before this // source head was checkpointed. Rewind the reader to the plane head to recover those native frames, but do @@ -4549,10 +4550,11 @@ private async Task RunHarnessAsync(HarnessRunContext context, Ca var facts = AgentRunFacts.For(context.Harness); // the three facts a forced terminal reports without folding, read with THIS harness's declared spellings (or the fallback union when it declares none) var writer = NewEventWriter(context.Owner, context.TeamId); // batches the DB inserts; flushed at each spool checkpoint + once at the end var native = await OpenNativeCaptureAsync(context, cancellationToken).ConfigureAwait(false); // G1: the lossless frame plane, dual-written beside the log; a plane that won't open leaves this path unchanged + var frames = context.Redactor.CreateFrameStream(); // a line longer than one read pass arrives cut; its two frames must not each keep half of a key async Task PersistAsync(string line, SandboxOutputFrame? output) { - var redactedLine = context.Redactor.Redact(line); + var redactedLine = frames.Redact(line, output?.IsComplete ?? true); // Capture the faithful transcript FIRST — redact the raw line, then keep it whether or not ParseEvents // surfaces any event. ParseEvents drops blank/unrecognized lines; the transcript keeps them so a replay @@ -5731,11 +5733,12 @@ private async Task RecordDiagnosticsAsync(DiagnosticCapture capture, Cancellatio if (!pump.IsCapturing) return; var delivered = 0; + var frames = capture.Redactor.CreateFrameStream(); // a diagnostic the reader cut must not keep the start of a key its continuation finishes async Task RecordAsync(SandboxDiagnosticLine line, CancellationToken token) { delivered++; - await pump.CaptureDiagnosticAsync(line.Text, capture.Redactor.Redact(line.Text), line.IsComplete, token).ConfigureAwait(false); + await pump.CaptureDiagnosticAsync(line.Text, frames.Redact(line.Text, line.IsComplete), line.IsComplete, token).ConfigureAwait(false); } var budget = new SandboxDiagnosticBudget { MaxLines = MaxDiagnosticFrames, MaxBytes = MaxDiagnosticBytes }; diff --git a/backend/src/CodeSpace.Core/Services/Agents/SecretRedactor.cs b/backend/src/CodeSpace.Core/Services/Agents/SecretRedactor.cs index a4cb5dfc2..6a06c425a 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/SecretRedactor.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/SecretRedactor.cs @@ -1,6 +1,7 @@ using System.Buffers; using System.Security.Cryptography; using System.Text; +using CodeSpace.Core.Services.Redaction; namespace CodeSpace.Core.Services.Agents; @@ -80,6 +81,41 @@ public string Redact(string text) /// source bytes with each transformed result. /// public SecretUtf8RedactionStream CreateUtf8Stream() => new(_secrets); + + /// + /// Creates a stateful redactor for decoded output frames, each persisted on its own (a native record, a transcript + /// line). A line longer than one read pass arrives as a frame the reader CUT — marked incomplete — and its + /// continuation; exact matching per frame cannot see a secret the cut split, so the end of a cut frame that a secret + /// could still run across is held back and redacted together with the frame that continues it. + /// + public SecretFrameRedactionStream CreateFrameStream() => new(this); + + /// How much of , a text that may still continue, can be redacted on its own (). + internal int SettledLength(string text) => SecretBoundary.SettledLength(text, _secrets); +} + +public sealed class SecretFrameRedactionStream +{ + private readonly SecretRedactor _redactor; + private string _carry = ""; + + internal SecretFrameRedactionStream(SecretRedactor redactor) => _redactor = redactor; + + /// + /// The redacted text to persist for one frame: the end held back from the cut frame before it, then this frame — + /// all of it when the frame ends its line, and when it was cut only what no secret can still run across, the rest + /// held for the frame that continues it. A stream that stops on a cut frame never persists what it held, so the + /// frames persisted always concatenate to a prefix of the line, redacted. + /// + public string Redact(string frame, bool isComplete) + { + var text = _carry + frame; + var settled = isComplete ? text.Length : _redactor.SettledLength(text); + + _carry = text[settled..]; + + return _redactor.Redact(text[..settled]); + } } public sealed class SecretUtf8RedactionStream diff --git a/backend/src/CodeSpace.Core/Services/Redaction/SecretBoundary.cs b/backend/src/CodeSpace.Core/Services/Redaction/SecretBoundary.cs new file mode 100644 index 000000000..ace5d7ff7 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Redaction/SecretBoundary.cs @@ -0,0 +1,51 @@ +namespace CodeSpace.Core.Services.Redaction; + +/// +/// Where a text that may still continue can be cut for exact-match redaction. A writer that persists a growing text in +/// pieces — a stream flushed by size or by time, a reader's bounded pass over one long line — redacts each piece on its +/// own, and exact matching cannot recognise a secret split between two of them. It releases the settled prefix and +/// carries the rest into the next piece. Shared by every redactor that writes in pieces, whatever its marker. +/// +internal static class SecretBoundary +{ + /// The longest prefix of that no secret runs across and after which nothing could yet become one. + public static int SettledLength(string text, IReadOnlyList secrets) + { + var settled = text.Length - LongestOpenSecretStart(text, secrets); + + while (StartOfSecretAcross(text, settled, secrets) is { } start) settled = start; + + return settled; + } + + /// The length of the longest end of that is the start, but not the whole, of a secret. + private static int LongestOpenSecretStart(string text, IReadOnlyList secrets) + { + var longest = 0; + + foreach (var secret in secrets) + { + for (var length = Math.Min(secret.Length - 1, text.Length); length > longest; length--) + { + if (!text.AsSpan(text.Length - length).SequenceEqual(secret.AsSpan(0, length))) continue; + + longest = length; + break; + } + } + + return longest; + } + + /// Where a whole secret that runs across starts, or null when none does. + private static int? StartOfSecretAcross(string text, int cut, IReadOnlyList secrets) + { + foreach (var secret in secrets) + { + for (var start = Math.Max(0, cut - secret.Length + 1); start < cut; start++) + if (text.AsSpan(start).StartsWith(secret, StringComparison.Ordinal)) return start; + } + + return null; + } +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs b/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs index cd08334be..c83976bea 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs @@ -3664,7 +3664,8 @@ private sealed record NodeExecution(WorkflowRun Run, NodeDefinition Node, INodeR private NodeRunContext BuildNodeRunContext(NodeExecution exec) { var nodeLogger = _loggerFactory.CreateLogger($"Workflow.{exec.Runtime.TypeKey}.{exec.Node.Id}"); - var observability = new NodeObservability(new NodeObservationBinding(_recordLogger, _artifactStore, exec.Run.Id, exec.Node.Id, exec.Run.TeamId, exec.ParentRecordId, PersistenceSecretRedactor.FromScope(exec.Scope), _completenessWriter)); + var redactor = PersistenceSecretRedactor.FromScope(exec.Scope); + var observability = new NodeObservability(new NodeObservationBinding(_recordLogger, _artifactStore, exec.Run.Id, exec.Node.Id, exec.Run.TeamId, exec.ParentRecordId, redactor, _completenessWriter)); return new NodeRunContext { @@ -3676,6 +3677,7 @@ private NodeRunContext BuildNodeRunContext(NodeExecution exec) Scope = exec.Scope, Logger = nodeLogger, Observability = observability, + PersistenceRedactor = redactor, ResumePayload = exec.ResumePayload, PriorAttemptPayload = exec.PriorAttemptPayload, NodeId = exec.Node.Id, diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingLLMClientDecorator.cs b/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingLLMClientDecorator.cs index 8e8c66122..ecefe17e2 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingLLMClientDecorator.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingLLMClientDecorator.cs @@ -227,7 +227,7 @@ protected static JsonElement FailedPayload(LlmCallScope scope, string provider, } /// Fold one redaction into THIS call's masking observation and hand its value back, so the flag the presence delta carries is a fact about the bytes that reached storage rather than about the configuration. A scope minted outside observes nothing and reads back verbatim, which is the conservative answer. - private static T Observe(LlmCallScope scope, PersistenceRedaction redaction) => scope.Masking is { } masking ? masking.Observe(redaction) : redaction.Value; + protected static T Observe(LlmCallScope scope, PersistenceRedaction redaction) => scope.Masking is { } masking ? masking.Observe(redaction) : redaction.Value; /// A plain-text field (a prompt / a text completion): the inline string when small, else a content-addressed $artifact_id ref. Null/empty rides as-is. protected static async Task OffloadTextAsync(LlmCallScope scope, string? text, CancellationToken cancellationToken) diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingStreamingStructuredLLMClientDecorator.cs b/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingStreamingStructuredLLMClientDecorator.cs index 0dcb7b733..4d7708fff 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingStreamingStructuredLLMClientDecorator.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingStreamingStructuredLLMClientDecorator.cs @@ -1,6 +1,7 @@ using System.Runtime.CompilerServices; using System.Text; using System.Text.Json; +using CodeSpace.Core.Services.Workflows.Runtime; using CodeSpace.Messages.Constants; namespace CodeSpace.Core.Services.Workflows.Llm; @@ -57,9 +58,7 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat () => StartedPayloadAsync(scope, Provider, request.Model, request.SystemPrompt, request.UserPrompt, request.Temperature, request.MaxOutputTokens, cancellationToken), cancellationToken).ConfigureAwait(false); var accumulator = new LlmCompletionAccumulator(); - var pending = new StringBuilder(); - var pendingBytes = 0; - var deltaOrdinal = 0; + var pending = new PendingDelta(PersistenceSecretRedactor.Union(scope.NativeCredentialRedactor, scope.CaptureRedactor)); await using var enumerator = _streamingInner.StreamAsync(request, cancellationToken).GetAsyncEnumerator(cancellationToken); using var timer = new PeriodicTimer(DeltaFlushInterval, _timeProvider); @@ -71,7 +70,7 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat bool moved; try { - if (pending.Length > 0) + if (!pending.IsEmpty) { var completed = await Task.WhenAny(moveNextTask, tickTask).ConfigureAwait(false); if (completed == tickTask) @@ -79,8 +78,7 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat cancellationToken.ThrowIfCancellationRequested(); if (!await tickTask.ConfigureAwait(false)) break; - await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, cancellationToken).ConfigureAwait(false); - pendingBytes = 0; + await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Settled, cancellationToken).ConfigureAwait(false); tickTask = timer.WaitForNextTickAsync(cancellationToken).AsTask(); continue; } @@ -92,8 +90,8 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat { // The caller's token is commonly already cancelled here. Capture uses its own non-cancelled attempt so // cancellation itself remains a typed terminal fact; SafeRecord still swallows any ledger failure. - if (pending.Length > 0) - await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, CancellationToken.None).ConfigureAwait(false); + if (!pending.IsEmpty) + await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Cut, CancellationToken.None).ConfigureAwait(false); var failureRecorded = await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionFailed, correlationId, () => Task.FromResult(FailedPayload(scope, Provider, ex)), CancellationToken.None).ConfigureAwait(false); if (failureRecorded && declared) await MarkCapturePresentAsync(scope).ConfigureAwait(false); @@ -109,12 +107,8 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat if (current is LlmStreamEvent.TextDelta delta) { pending.Append(delta.Text); - pendingBytes += Encoding.UTF8.GetByteCount(delta.Text); - if (pendingBytes >= DeltaFlushBytes) - { - await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, cancellationToken).ConfigureAwait(false); - pendingBytes = 0; - } + if (pending.Bytes >= DeltaFlushBytes) + await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Settled, cancellationToken).ConfigureAwait(false); } yield return current; @@ -123,21 +117,79 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat // Keep the final tail in the ordered live projection. This is at most one extra row per streamed call, preserves // exact concatenation, and does not change authority: completed below remains the canonical whole output. - if (pending.Length > 0) - await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal, cancellationToken).ConfigureAwait(false); + if (!pending.IsEmpty) + await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.End, cancellationToken).ConfigureAwait(false); var recorded = await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionCompleted, correlationId, async () => CompletedPayload(scope, Provider, accumulator.ResolveModel(request.Model), accumulator.Usage, await OffloadTextAsync(scope, accumulator.Text, CancellationToken.None).ConfigureAwait(false)), CancellationToken.None).ConfigureAwait(false); if (recorded && declared) await MarkCapturePresentAsync(scope).ConfigureAwait(false); } - private async Task RecordPendingDeltaAsync(LlmCallScope scope, Guid correlationId, StringBuilder pending, int ordinal, CancellationToken cancellationToken) + private async Task RecordPendingDeltaAsync(LlmCallScope scope, Guid correlationId, PendingDelta pending, DeltaFlush flush, CancellationToken cancellationToken) { - var fragment = pending.ToString(); - pending.Clear(); + if (pending.Take(flush) is not { } fragment) return; + + var text = Observe(scope, fragment.Text); await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionDelta, correlationId, - async () => DeltaPayload(scope, Provider, ordinal, await OffloadTextAsync(scope, fragment, cancellationToken).ConfigureAwait(false)), cancellationToken).ConfigureAwait(false); + async () => DeltaPayload(scope, Provider, fragment.Ordinal, await OffloadTextAsync(scope, text, cancellationToken).ConfigureAwait(false)), cancellationToken).ConfigureAwait(false); + } + + /// What a flush may release of the held text. + private enum DeltaFlush + { + /// Mid-stream: only the settled text; the rest is carried into the next flush. + Settled, + + /// The stream ended: everything, verbatim — a held tail nothing followed never became a secret. + End, + + /// + /// The stream was cut — a cancel or a fault: everything, with a held tail that could start a secret masked. The + /// carry holds exactly that text, the rest of it was never streamed, and exact matching never sees it whole. + /// + Cut, + } + + /// + /// The text streamed since the last delta row, and the ordinal the next row takes. Each row is redacted on its own, so + /// a flush mid-stream releases only the SETTLED text — none that a secret could still run across — and carries the + /// rest into the next flush; otherwise a secret the model prints across a flush (by size or by the one-second + /// cadence) is stored as two halves no exact match recognises, and the live view concatenates them back. The end + /// of the stream releases everything, so the rows still concatenate to exactly the streamed text, masked; a cut + /// releases everything too, but masks the end it was holding back. + /// + private sealed class PendingDelta(PersistenceSecretRedactor boundary) + { + private readonly StringBuilder _text = new(); + private int _ordinal; + + /// UTF-8 size of the held text — the size bound is in bytes, not UTF-16 chars. + public int Bytes { get; private set; } + + public bool IsEmpty => _text.Length == 0; + + public void Append(string text) + { + _text.Append(text); + Bytes += Encoding.UTF8.GetByteCount(text); + } + + /// The text one row may carry and that row's ordinal, or null when nothing is settled yet. + public (PersistenceRedaction Text, int Ordinal)? Take(DeltaFlush flush) + { + var text = _text.ToString(); + var length = flush == DeltaFlush.Settled ? boundary.SettledLength(text) : text.Length; + + if (length == 0) return null; + + _text.Remove(0, length); + Bytes = Encoding.UTF8.GetByteCount(text.AsSpan(length)); + + var released = flush == DeltaFlush.Cut ? boundary.RedactCut(text) : new PersistenceRedaction(text[..length], false); + + return (released, _ordinal++); + } } /// The interaction.delta payload: the ordinal (monotonic within the call) + the coalesced text fragment (inline-or-$artifact_id). diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/AgentRunCommandNode.cs b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/AgentRunCommandNode.cs index f201cf674..39938aeaa 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/AgentRunCommandNode.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/AgentRunCommandNode.cs @@ -160,8 +160,8 @@ public async Task RunAsync(NodeRunContext context, CancellationToken var maxOutputChars = TryReadPositiveInt(context, "maxOutputChars", out var cap) ? cap : 0; var captures = new[] { - new CommandOutputCapture("stdout", result.Stdout, OutputCap.Apply(result.Stdout, maxOutputChars), result.Observation?.Stdout), - new CommandOutputCapture("stderr", result.Stderr, OutputCap.Apply(result.Stderr, maxOutputChars), result.Observation?.Stderr), + CommandOutputCapture.Of("stdout", result.Stdout, result.Observation?.Stdout, maxOutputChars, context.PersistenceRedactor), + CommandOutputCapture.Of("stderr", result.Stderr, result.Observation?.Stderr, maxOutputChars, context.PersistenceRedactor), }; var outputs = new Dictionary { @@ -192,7 +192,7 @@ public async Task RunAsync(NodeRunContext context, CancellationToken if ((!capture.Inline.Truncated && capture.Complete) || teamId is not { } tid || string.IsNullOrEmpty(capture.Text)) return null; try { - return await _artifacts.PutAsync(tid, Encoding.UTF8.GetBytes(capture.Text), "text/plain", cancellationToken).ConfigureAwait(false); + return await _artifacts.PutAsync(tid, Encoding.UTF8.GetBytes(capture.Redacted), "text/plain", cancellationToken).ConfigureAwait(false); } catch (Exception ex) { @@ -210,10 +210,31 @@ private static async Task NoticeOutputLossAsync(NodeRunContext context, string d catch (Exception ex) { context.Logger.LogWarning(ex, "Command output capture loss could not be recorded; the command outcome is unchanged"); } } - private sealed record CommandOutputCapture(string Name, string Text, OutputCap.Result Inline, SandboxStreamObservation? Observation) + /// One captured stream: the runner's text, that text redacted, and the inline value the node returns. + private sealed record CommandOutputCapture(string Name, string Text, string Redacted, OutputCap.Result Inline, SandboxStreamObservation? Observation) { - public bool Complete => Observation is null or { ReachedEndOfStream: true, CaptureComplete: true }; + public bool Complete => IsComplete(Observation); public long ObservedBytes => Observation?.ObservedBytes ?? Encoding.UTF8.GetByteCount(Text); + + /// + /// Redact first, then cut. A capture nothing cut keeps its runtime value: the engine masks it whole in node.completed + /// and keeps the real one in the encrypted sidecar for downstream nodes, resumes and reruns. Any cut — the inline cap, + /// or the runner's budget ending the capture mid-stream — is taken from the redacted text instead, because a cut + /// inside a secret leaves a fragment exact matching never recognises again. The artifact always holds the redacted + /// text: it has no sidecar, and every member downloads exactly what is stored. + /// + public static CommandOutputCapture Of(string name, string text, SandboxStreamObservation? observation, int maxOutputChars, PersistenceSecretRedactor redactor) + { + var complete = IsComplete(observation); + var redacted = (complete ? redactor.Redact(text) : redactor.RedactCut(text)).Value ?? ""; + var inline = OutputCap.Apply(text, maxOutputChars); + + if (inline.Truncated || !complete) inline = OutputCap.Apply(redacted, maxOutputChars); + + return new CommandOutputCapture(name, text, redacted, inline, observation); + } + + private static bool IsComplete(SandboxStreamObservation? observation) => observation is null or { ReachedEndOfStream: true, CaptureComplete: true }; } private static bool TryReadNonEmpty(NodeRunContext context, string key, out string text) diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/HttpRequestNode.cs b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/HttpRequestNode.cs index e0ae4285f..8892b2985 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/HttpRequestNode.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/Builtin/HttpRequestNode.cs @@ -192,8 +192,9 @@ public async Task RunAsync(NodeRunContext context, CancellationToken }); } + /// Where the call goes — scheme, host and port — for the operator log. Never the userinfo (user:password@), path or query: this line reaches the shared log sink before any redaction runs. private static string SafeLogTarget(string url) => Uri.TryCreate(url, UriKind.Absolute, out var parsed) - ? parsed.GetLeftPart(UriPartial.Authority) + ? parsed.GetComponents(UriComponents.SchemeAndServer, UriFormat.UriEscaped) : "[invalid-or-relative-url]"; private static HttpRequestMessage BuildRequest(HttpMethod method, string url, IReadOnlyDictionary inputs) diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeObservability.cs b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeObservability.cs index 82701cafa..359c98a31 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeObservability.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeObservability.cs @@ -1,3 +1,4 @@ +using System.Text; using System.Text.Json; using CodeSpace.Core.Services.Workflows.Artifacts; using CodeSpace.Core.Services.Workflows.Lifecycle; @@ -101,7 +102,7 @@ public async Task TraceExternalCallAsync(string target, string // We still want to leave a trace so the timeline shows "this call was in flight when // we cancelled" — emit failed with the cancellation message, then re-throw. var duration = DateTimeOffset.UtcNow - startedAt; - await _recordLogger.ExternalCallFailedAsync(_runId, _nodeId, correlationId, target, "Operation cancelled.", duration, category: null, CancellationToken.None).ConfigureAwait(false); + await _recordLogger.ExternalCallFailedAsync(_runId, _nodeId, correlationId, persistedTarget, "Operation cancelled.", duration, category: null, CancellationToken.None).ConfigureAwait(false); throw; } catch (Exception ex) @@ -133,7 +134,7 @@ public async IAsyncEnumerable TraceExternalStreamAsync(string ta } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { - await _recordLogger.ExternalCallFailedAsync(_runId, _nodeId, correlationId, target, "Operation cancelled.", DateTimeOffset.UtcNow - startedAt, category: null, CancellationToken.None).ConfigureAwait(false); + await _recordLogger.ExternalCallFailedAsync(_runId, _nodeId, correlationId, persistedTarget, "Operation cancelled.", DateTimeOffset.UtcNow - startedAt, category: null, CancellationToken.None).ConfigureAwait(false); throw; } catch (Exception ex) @@ -150,6 +151,7 @@ public async IAsyncEnumerable TraceExternalStreamAsync(string ta public async Task PersistArtifactAsync(string contentType, ReadOnlyMemory bytes, CancellationToken cancellationToken = default) { + bytes = RedactStoredContent(contentType, bytes); var artifactId = await _artifactStore.PutAsync(_teamId, bytes, contentType, cancellationToken).ConfigureAwait(false); // Build the canonical artifact-ref shape. Operators reading the ledger see: @@ -166,6 +168,42 @@ public async Task PersistArtifactAsync(string contentType, ReadOnly return refJson; } + /// + /// An artifact has no encrypted sidecar: what is stored is what every member downloads. So text and JSON are masked + /// the way the ledger is, JSON through its parsed values so an escaped spelling of a secret is caught as well. Opaque + /// bytes cannot be read for a secret and are stored as given; unchanged content keeps its exact bytes. + /// + private ReadOnlyMemory RedactStoredContent(string contentType, ReadOnlyMemory bytes) + { + if (IsJson(contentType) && TryParseJson(bytes) is { } json) + { + var redactedJson = _redactor.Redact(json); + return redactedJson.Changed ? Encoding.UTF8.GetBytes(redactedJson.Value.GetRawText()) : bytes; + } + + if (!IsText(contentType)) return bytes; + + var redactedText = _redactor.Redact(Encoding.UTF8.GetString(bytes.Span)); + return redactedText.Changed ? Encoding.UTF8.GetBytes(redactedText.Value!) : bytes; + } + + private static bool IsJson(string contentType) => contentType.Split(';')[0].Trim().EndsWith("json", StringComparison.OrdinalIgnoreCase); + + private static bool IsText(string contentType) => IsJson(contentType) || contentType.StartsWith("text/", StringComparison.OrdinalIgnoreCase); + + private static JsonElement? TryParseJson(ReadOnlyMemory bytes) + { + try + { + using var document = JsonDocument.Parse(bytes); + return document.RootElement.Clone(); + } + catch (JsonException) + { + return null; + } + } + /// /// Test-only no-op. Calls the action straight through, no records, no artifacts. Never /// used in production — engine always supplies a real . diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeRunContext.cs b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeRunContext.cs index 4392f262b..ae8f541c1 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeRunContext.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Nodes/NodeRunContext.cs @@ -95,6 +95,15 @@ public sealed record NodeRunContext /// public required INodeObservability Observability { get; init; } + /// + /// Masks this node's in-scope secret values — the same exact-match redactor the engine applies to node.completed and + /// applies to its records. The engine's mask covers only the outputs it persists; text a + /// node stores itself (an artifact, which has no encrypted sidecar) or cuts down (an excerpt, whose cut can land + /// inside a secret and leave a fragment no exact match recognises) must be redacted by the node FIRST, then cut. + /// off the engine path, where no secrets are in scope. + /// + public PersistenceSecretRedactor PersistenceRedactor { get; init; } = PersistenceSecretRedactor.None; + /// /// Set ONLY when the node is being re-run after a suspension was resolved — carries the /// resume signal's payload (timer wake marker, the approver's decision, the callback body). diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/PersistenceSecretRedactor.cs b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/PersistenceSecretRedactor.cs index c4c395052..3cb36609d 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/PersistenceSecretRedactor.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/PersistenceSecretRedactor.cs @@ -1,4 +1,5 @@ using System.Text.Json; +using CodeSpace.Core.Services.Redaction; namespace CodeSpace.Core.Services.Workflows.Runtime; @@ -18,6 +19,12 @@ public PersistenceSecretRedactor(IEnumerable secrets) => _secrets = secr .OrderByDescending(value => value.Length) .ToArray(); + /// A redactor with no secrets: every value passes through unchanged. + public static PersistenceSecretRedactor None { get; } = new([]); + + /// One redactor masking every secret of each given one; an absent redactor contributes nothing. + public static PersistenceSecretRedactor Union(params PersistenceSecretRedactor?[] redactors) => new(redactors.OfType().SelectMany(redactor => redactor._secrets)); + public static PersistenceSecretRedactor FromScope(NodeRunScope scope) { var secrets = new List(scope.InheritedSecretValues); @@ -51,6 +58,30 @@ public IReadOnlyList FoundIn(JsonElement value) return new PersistenceRedaction(redacted, !string.Equals(value, redacted, StringComparison.Ordinal)); } + /// + /// How much of , a text that may still continue, can be redacted on its own: the longest prefix + /// that no secret runs across and after which nothing could yet become one. A writer persisting a growing text in + /// pieces releases this much and carries the rest into the next piece, because exact matching cannot recognise a + /// secret split between two of them. + /// + public int SettledLength(string text) => SecretBoundary.SettledLength(text, _secrets); + + /// + /// Redacts a text whose END was cut: a capture that stopped at its budget while the source went on. Besides every + /// whole secret, a trailing fragment that could be the start of one is masked, because the rest of it was cut away + /// and exact matching will never see it whole. + /// + public PersistenceRedaction RedactCut(string? value) + { + if (string.IsNullOrEmpty(value)) return new PersistenceRedaction(value, false); + + var settled = SettledLength(value); + + if (settled == value.Length) return Redact(value); + + return new PersistenceRedaction(Redact(value[..settled]).Value + Marker, true); + } + public PersistenceRedaction Redact(JsonElement value) { var changed = false; diff --git a/backend/src/CodeSpace.Core/Settings/ConfigurationPlaceholderGuard.cs b/backend/src/CodeSpace.Core/Settings/ConfigurationPlaceholderGuard.cs index 5dc3c23b6..c4a7dbc25 100644 --- a/backend/src/CodeSpace.Core/Settings/ConfigurationPlaceholderGuard.cs +++ b/backend/src/CodeSpace.Core/Settings/ConfigurationPlaceholderGuard.cs @@ -20,8 +20,10 @@ namespace CodeSpace.Core.Settings; /// /// So an unrendered placeholder is refused everywhere, not only in Production: it is never a /// value anyone intended, and the environment it appears in does not change that. The failure names -/// the key and shows the value, which is safe — a template is not a secret, and it is precisely the -/// text the operator needs to find in their release definition. +/// the key and the placeholders, the text the operator searches their release definition for, and +/// never the value: a partly rendered value (a connection string whose password was substituted and +/// whose database name was not) carries a real credential beside the template, and this message is +/// written to stderr before logging exists. /// public static class ConfigurationPlaceholderGuard { @@ -50,9 +52,9 @@ public static IReadOnlyList Violations(IConfiguration configuration) { if (string.IsNullOrEmpty(value)) continue; - var match = Unrendered.Match(value); + var placeholders = Unrendered.Matches(value).Select(match => match.Value).Distinct(StringComparer.Ordinal).ToList(); - if (match.Success) violations.Add($"{key} = {value} (unrendered {Describe(match.Value)})"); + if (placeholders.Count > 0) violations.Add($"{key} contains {string.Join(", ", placeholders)} (unrendered {Describe(placeholders[0])})"); } return violations; diff --git a/backend/tests/CodeSpace.E2ETests/Workflows/NodeRecordRedactionEndpointE2ETests.cs b/backend/tests/CodeSpace.E2ETests/Workflows/NodeRecordRedactionEndpointE2ETests.cs new file mode 100644 index 000000000..3da2711f3 --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Workflows/NodeRecordRedactionEndpointE2ETests.cs @@ -0,0 +1,309 @@ +using System.Net; +using System.Net.Http.Headers; +using System.Net.Http.Json; +using System.Net.Sockets; +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Workflows.Llm; +using CodeSpace.Core.Settings.OutboundHttp; +using CodeSpace.E2ETests.Infrastructure; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Enums; +using Microsoft.AspNetCore.Hosting; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Shouldly; + +namespace CodeSpace.E2ETests.Workflows; + +/// +/// What a Viewer of the team reads over HTTP after a run had a team secret in scope: the raw records, the live SSE tail, +/// and the artifacts the run stored. Three node-owned writes used to put the secret there: the failed record of a call +/// cancelled in flight, the excerpt and artifact of a command output cut inside the secret, and the streamed +/// interaction.delta rows of a model that printed the secret across a flush. +/// +/// Tier: High fidelity. The real app host (JWT, team scope, controllers, the records reader and SSE streamer, the +/// artifact download), the real engine drained behind POST /api/workflows/{id}/run, a real loopback socket held +/// open for the http.request, a real /bin/sh for the command. The one stand-in is the model: a streaming client +/// under its own provider tag, wrapped by the production recording decorators. +/// +[Trait("Category", "E2E")] +[Trait("Surface", "Http")] +public sealed class NodeRecordRedactionEndpointE2ETests : IClassFixture +{ + private readonly LoopbackModelApiFactory _factory; + + public NodeRecordRedactionEndpointE2ETests(LoopbackModelApiFactory factory) { _factory = factory; } + + [Fact] + public async Task A_viewer_reading_a_run_cancelled_mid_http_request_finds_no_secret_in_its_records_or_its_sse_tail() + { + var world = await SeedWorldAsync(); + + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + Guid runId; + try + { + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + var accepted = listener.AcceptTcpClientAsync(); + runId = await LaunchAsync(world, new { id = "call", typeKey = "http.request", config = new { timeoutSeconds = 60 }, inputs = new { url = $"http://127.0.0.1:{port}/hold?api_key={{{{team.API_KEY}}}}", method = "GET" } }); + + var engine = Task.Run(() => _factory.JobClient.DrainAsync()); + using var held = await accepted.WaitAsync(TimeSpan.FromSeconds(30)); + + (await SendAsync(HttpMethod.Post, $"/api/workflows/runs/{runId}/cancel", world.OwnerId, world.TeamId)).StatusCode.ShouldBe(HttpStatusCode.OK, "the owner cancels the run while its call is held open"); + + try { await engine.WaitAsync(TimeSpan.FromSeconds(60)); } + catch (OperationCanceledException) { /* the engine surfaces the operator's cancel */ } + } + finally { listener.Stop(); } + + var records = await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records"); + records.ShouldContain(WorkflowRunRecordTypes.ExternalCallFailed, customMessage: "precondition: the cancelled call left its failed record"); + ShouldHoldNoFragment(records, world.Canary, "GET /records"); + ShouldHoldNoFragment(await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records/stream?after=0"), world.Canary, "the SSE tail"); + } + + [Fact] + public async Task A_viewer_downloading_a_command_output_cut_inside_the_secret_finds_no_fragment_of_it() + { + var world = await SeedWorldAsync(); + + var script = "printf 'deploying with key=%s\\n' \"$1\"; head -c 4000 /dev/zero | tr '\\000' x"; + var runId = await LaunchAsync(world, new { id = "command", typeKey = "agent.run_command", config = new { }, inputs = new { command = "/bin/sh", args = new[] { "-c", script, "sh", "{{team.API_KEY}}" }, maxOutputChars = 40 } }); + await _factory.JobClient.DrainAsync(); + + var detail = await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}"); + var artifactId = await NodeOutputGuidAsync(runId, "command", "stdoutArtifactId"); + var artifact = await ReadAsViewerAsync(world, $"/api/artifacts/{artifactId}"); + + artifact.ShouldStartWith("deploying with key=[REDACTED]", customMessage: "the downloaded artifact is the whole output, redacted"); + ShouldHoldNoFragment(detail, world.Canary, "GET /runs/{id}"); + ShouldHoldNoFragment(artifact, world.Canary, "GET /api/artifacts/{id}"); + ShouldHoldNoFragment(await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records"), world.Canary, "GET /records"); + } + + [Fact] + public async Task A_viewer_tailing_model_deltas_streamed_across_a_flush_finds_no_fragment_of_the_secret() + { + var world = await SeedWorldAsync(); + await SeedEchoModelAsync(world.TeamId); + + var runId = await LaunchAsync(world, new { id = "llm", typeKey = "llm.complete", config = new { provider = SecretEchoStreamingLlmClient.ProviderTag, maxTokens = 60000 }, inputs = new { userPrompt = SecretEchoStreamingLlmClient.EchoMarker + "{{team.API_KEY}}" } }); + await _factory.JobClient.DrainAsync(); + + var tail = await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records/stream?after=0"); + var deltas = DeltaTexts(tail); + deltas.Count.ShouldBeGreaterThanOrEqualTo(2, "precondition: the 32-KiB bound flushed mid-stream, so the echo was split across rows"); + + var resolved = new List(); + foreach (var delta in deltas) resolved.Add(delta.ValueKind == JsonValueKind.String ? delta.GetString()! : await ReadAsViewerAsync(world, $"/api/artifacts/{delta.GetProperty("$artifact_id").GetString()}")); + + string.Concat(resolved).ShouldBe(SecretEchoStreamingLlmClient.Filler + "[REDACTED] done", "the live view concatenates the rows into the masked output"); + foreach (var text in resolved) ShouldHoldNoFragment(text, world.Canary, "an interaction.delta row as the Viewer's live view folds it"); + ShouldHoldNoFragment(tail, world.Canary, "the SSE tail"); + } + + [Fact] + public async Task A_viewer_tailing_a_run_cancelled_while_its_model_streams_the_start_of_the_secret_finds_no_fragment_of_it() + { + var world = await SeedWorldAsync(); + await SeedEchoModelAsync(world.TeamId); + + var runId = await LaunchAsync(world, new { id = "llm", typeKey = "llm.complete", config = new { provider = SecretEchoStreamingLlmClient.ProviderTag, maxTokens = 60000 }, inputs = new { userPrompt = SecretEchoStreamingLlmClient.HoldMarker + "{{team.API_KEY}}" } }); + var engine = Task.Run(() => _factory.JobClient.DrainAsync()); + await WaitForRecordAsync(runId, WorkflowRunRecordTypes.InteractionDelta, "the one-second flush writes the settled text while the stream is held"); + + (await SendAsync(HttpMethod.Post, $"/api/workflows/runs/{runId}/cancel", world.OwnerId, world.TeamId)).StatusCode.ShouldBe(HttpStatusCode.OK, "the owner cancels the run while its model is part-way into the secret"); + + try { await engine.WaitAsync(TimeSpan.FromSeconds(60)); } + catch (OperationCanceledException) { /* the engine surfaces the operator's cancel */ } + + // The cut's own row lands after run.cancelled, where the SSE tail of a terminal run stops; GET /records has it. + var records = await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records"); + RecordDeltaTexts(records).ShouldBe(new[] { SecretEchoStreamingLlmClient.HeldLead, "[REDACTED]" }, "the settled text, then the held start of the secret masked by the cut"); + ShouldHoldNoFragment(records, world.Canary, "GET /records"); + ShouldHoldNoFragment(await ReadAsViewerAsync(world, $"/api/workflows/runs/{runId}/records/stream?after=0"), world.Canary, "the SSE tail"); + } + + // ─── World ────────────────────────────────────────────────────────────────── + + private sealed record World(Guid OwnerId, Guid ViewerId, Guid TeamId, string Canary); + + /// An Owner who authors and runs, a Viewer who only reads, and the team secret the run resolves — set through the real variables endpoint. + private async Task SeedWorldAsync() + { + var teamId = Guid.NewGuid(); + var ownerId = await SeedUserAsync(teamId, TeamRole.Owner, createTeam: true); + var viewerId = await SeedUserAsync(teamId, TeamRole.Viewer, createTeam: false); + var canary = "CANARY-" + string.Concat(Enumerable.Range(0, 24).Select(_ => "BCDFGHJKLMNPQRSTVWZ"[Random.Shared.Next(19)])); + + var set = await SendAsync(HttpMethod.Put, "/api/team-variables/API_KEY", ownerId, teamId, new { name = "API_KEY", valueType = nameof(VariableValueType.Secret), value = canary }); + set.StatusCode.ShouldBe(HttpStatusCode.NoContent, await set.Content.ReadAsStringAsync()); + + return new World(ownerId, viewerId, teamId, canary); + } + + private async Task SeedUserAsync(Guid teamId, TeamRole role, bool createTeam) + { + using var scope = _factory.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var userId = Guid.NewGuid(); + + if (createTeam) db.Team.Add(new Team { Id = teamId, Slug = $"f2-{teamId:N}"[..20], Name = "F2", Kind = TeamKind.Workspace, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + db.User.Add(new User { SecurityStamp = TestToken.SeedStamp, Id = userId, Email = $"f2-{userId:N}@test.local", Name = role.ToString(), CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + db.TeamMembership.Add(new TeamMembership { Id = Guid.NewGuid(), TeamId = teamId, UserId = userId, Role = role, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + + await db.SaveChangesAsync(); + return userId; + } + + /// The team's pool row for the echo model: keyless, because the stand-in never authenticates. + private async Task SeedEchoModelAsync(Guid teamId) + { + using var scope = _factory.Services.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var credentialId = Guid.NewGuid(); + + db.ModelCredential.Add(new ModelCredential { Id = credentialId, TeamId = teamId, Provider = SecretEchoStreamingLlmClient.ProviderTag, DisplayName = "echo", Status = CredentialStatus.Active, CreatedBy = SystemUsers.SeederId, LastModifiedBy = SystemUsers.SeederId }); + db.ModelCredentialModel.Add(new ModelCredentialModel { Id = Guid.NewGuid(), ModelCredentialId = credentialId, ModelId = "secret-echo-model", Source = ModelSource.Manual, Enabled = true }); + + await db.SaveChangesAsync(); + } + + /// The Owner authors trigger → node → terminal over HTTP and launches it; the run executes when the test drains the job client. + private async Task LaunchAsync(World world, object node) + { + var nodeId = JsonSerializer.SerializeToElement(node).GetProperty("id").GetString()!; + var workflow = new + { + name = $"f2-{Guid.NewGuid():N}"[..20], + definition = new + { + schemaVersion = 1, + nodes = new[] { new { id = "start", typeKey = "trigger.manual", config = new { }, inputs = new { } }, node, new { id = "end", typeKey = "builtin.terminal", config = new { }, inputs = new { } } }, + edges = new object[] { new { from = "start", to = nodeId }, new { from = nodeId, to = "end" } }, + }, + activations = Array.Empty(), + enabled = true, + }; + + var created = await SendAsync(HttpMethod.Post, "/api/workflows", world.OwnerId, world.TeamId, workflow); + created.IsSuccessStatusCode.ShouldBeTrue(await created.Content.ReadAsStringAsync()); + var workflowId = JsonDocument.Parse(await created.Content.ReadAsStringAsync()).RootElement.GetProperty("id").GetGuid(); + + var launched = await SendAsync(HttpMethod.Post, $"/api/workflows/{workflowId}/run", world.OwnerId, world.TeamId, new { }); + launched.IsSuccessStatusCode.ShouldBeTrue(await launched.Content.ReadAsStringAsync()); + + return JsonDocument.Parse(await launched.Content.ReadAsStringAsync()).RootElement.GetProperty("runId").GetGuid(); + } + + private async Task WaitForRecordAsync(Guid runId, string recordType, string because) + { + var deadline = DateTime.UtcNow.AddSeconds(30); + + while (true) + { + using (var scope = _factory.Services.CreateScope()) + if (await scope.ServiceProvider.GetRequiredService().WorkflowRunRecord.AsNoTracking().AnyAsync(r => r.RunId == runId && r.RecordType == recordType)) return; + + if (DateTime.UtcNow > deadline) throw new ShouldAssertException($"{because}: no {recordType} record for run {runId} within 30s. Check the run's workflow_run_record rows and its node's status for why the stream never flushed."); + await Task.Delay(50); + } + } + + private async Task NodeOutputGuidAsync(Guid runId, string nodeId, string output) + { + using var scope = _factory.Services.CreateScope(); + var node = await scope.ServiceProvider.GetRequiredService().WorkflowRunNode.AsNoTracking().SingleAsync(n => n.RunId == runId && n.NodeId == nodeId); + + node.Status.ShouldBe(NodeStatus.Success, node.Error); + return JsonDocument.Parse(node.OutputsJson).RootElement.GetProperty(output).GetGuid(); + } + + /// The text of every interaction.delta frame on the SSE tail: inline, or the $artifact_id ref the Room fetches. + private static List DeltaTexts(string tail) + { + var texts = new List(); + var lines = tail.Split('\n'); + + for (var i = 0; i + 1 < lines.Length; i++) + { + if (lines[i].Trim() != "event: " + WorkflowRunRecordTypes.InteractionDelta) continue; + + var data = lines.Skip(i + 1).First(line => line.StartsWith("data: ", StringComparison.Ordinal))["data: ".Length..]; + texts.Add(DeltaPayload(JsonDocument.Parse(data).RootElement).GetProperty("text").Clone()); + } + + return texts; + } + + /// The inline text of every interaction.delta in a GET /records page, in sequence order. + private static string[] RecordDeltaTexts(string page) => JsonDocument.Parse(page).RootElement.GetProperty("records").EnumerateArray() + .Where(record => record.GetProperty("recordType").GetString() == WorkflowRunRecordTypes.InteractionDelta) + .Select(record => DeltaPayload(record).GetProperty("text").GetString()!) + .ToArray(); + + /// An SSE frame's data is the record; its payload rides as a JSON string or an object depending on the projection. + private static JsonElement DeltaPayload(JsonElement frame) + { + var payload = frame.TryGetProperty("payloadJson", out var json) ? json : frame.GetProperty("payload"); + + return payload.ValueKind == JsonValueKind.String ? JsonDocument.Parse(payload.GetString()!).RootElement : payload; + } + + private async Task ReadAsViewerAsync(World world, string url) + { + using var response = await SendAsync(HttpMethod.Get, url, world.ViewerId, world.TeamId); + var body = await response.Content.ReadAsStringAsync(); + + response.StatusCode.ShouldBe(HttpStatusCode.OK, $"the Viewer reads {url}: {body}"); + return body; + } + + private async Task SendAsync(HttpMethod method, string url, Guid userId, Guid teamId, object? body = null) + { + var client = _factory.CreateClient(); + client.Timeout = TimeSpan.FromSeconds(60); // a terminal run's SSE tail closes itself; this only bounds a broken one + + var request = new HttpRequestMessage(method, url); + request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", TestToken.Mint(userId, TestToken.SeedStamp)); + request.Headers.Add("X-Team-Id", teamId.ToString()); + if (body is not null) request.Content = JsonContent.Create(body); + + return await client.SendAsync(request); + } + + /// Any six consecutive characters of the secret: the whole value, and the head a cut would leave behind. + private static void ShouldHoldNoFragment(string text, string canary, string where) + { + for (var start = 0; start + 6 <= canary.Length; start++) + { + var fragment = canary.Substring(start, 6); + text.Contains(fragment, StringComparison.Ordinal).ShouldBeFalse($"{where} gives a Viewer '{fragment}', a fragment of the team secret"); + } + } + + /// + /// The app with two seams: a streaming stand-in model under its own provider tag (wrapped by the production recording + /// decorators like any client), and loopback admitted for http.request so a test can hold a call open on a local socket. + /// + public sealed class LoopbackModelApiFactory : TaskLaunchApiFactory + { + protected override void ConfigureWebHost(IWebHostBuilder builder) + { + base.ConfigureWebHost(builder); + builder.ConfigureAppConfiguration((_, configuration) => configuration.AddInMemoryCollection(new Dictionary { [$"{OutboundHttpAllowlistSetting.ConfigurationKey}:0"] = "127.0.0.1" })); + } + + protected override void ConfigureContractTestServices(ContainerBuilder builder) + { + builder.RegisterType().As().As().As().SingleInstance(); + } + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Infrastructure/PostgresFixture.cs b/backend/tests/CodeSpace.IntegrationTests/Infrastructure/PostgresFixture.cs index c470b79d5..3c2d6bbb8 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Infrastructure/PostgresFixture.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Infrastructure/PostgresFixture.cs @@ -368,6 +368,14 @@ private static void RegisterTestAssemblyTypes(ContainerBuilder builder) .As() .SingleInstance(); + // A streaming model that echoes a prompted value across the recorder's 32-KiB delta flush, for the + // interaction.delta redaction flows. Under its own tag; inert for every test that does not pin it. + builder.RegisterType() + .As() + .As() + .As() + .SingleInstance(); + // A2: the model-plane OUTAGE fake, under its own tag ("TestFaulting"). Root-registered rather than // scope-overridden because the node holds ILLMClientRegistry from the root container — a scope override // would never reach it. Inert until a test dials its script, so sibling flows are unaffected. diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunExecutorTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunExecutorTests.cs index c7848bcee..8b4a7ae5f 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunExecutorTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunExecutorTests.cs @@ -739,6 +739,52 @@ public async Task Redacts_an_echoed_model_key_from_the_append_only_log_and_resul (run.Error ?? "").ShouldNotContain(key); } + /// + /// A line longer than one 8 MiB read pass is delivered as a frame the reader CUT and its continuation, and each is + /// persisted on its own: a native record, a transcript line. A key the cut lands inside used to be stored as a head + /// in one record and a tail in the next — exact matching saw neither. The real executor, the real durable runner and + /// the production native-record plane; the harness drops non-JSON lines as the production CLI parsers do. + /// + [Fact] + public async Task A_model_key_split_by_the_read_pass_cut_of_a_long_line_leaves_no_fragment_in_native_records_or_the_transcript() + { + if (OperatingSystem.IsWindows()) return; + + var key = "sk-split-" + string.Concat(Enumerable.Range(0, 32).Select(_ => "BCDFGHJKLMNPQRSTVWZ23456789"[Random.Shared.Next(27)])); + var teamId = await SeedTeamAsync(); + var credId = await SeedModelCredentialAsync(teamId, "scripted-provider", key); + var runId = await CreateRunWithCredentialAsync(teamId, credId); + + // The filler stops 12 bytes short of the 8 MiB pass, so the reader cuts the line 12 characters into the key — + // on stdout and, written again, on stderr. A short control line carries the key whole. + var script = $"line() {{ head -c {8 * 1024 * 1024 - 12} /dev/zero | tr '\\000' x; printf '%s\\n' \"$SCRIPTED_MODEL_KEY\"; }}; line; line >&2; printf 'short line %s\\n' \"$SCRIPTED_MODEL_KEY\""; + await ExecuteAsync(runId, new LongLineKeyEchoingHarness("scripted-provider", "SCRIPTED_MODEL_KEY", script), productionCapturePlanes: true); + + using var scope = _fixture.BeginScope(); + var run = await scope.Resolve().GetAsync(runId, CancellationToken.None); + run.Status.ShouldBe(AgentRunStatus.Succeeded, run.Error); + + var records = await scope.Resolve().WorkflowRunNativeRecord.AsNoTracking().Where(r => r.AgentRunId == runId).OrderBy(r => r.Channel).ThenBy(r => r.Ordinal).ToListAsync(); + var cut = records.Single(r => r.Channel == NativeRecordChannel.Stdout && !r.IsFinal); + records.ShouldContain(r => r.Channel == NativeRecordChannel.Stderr && !r.IsFinal, "precondition: the diagnostics reader cut the long stderr line too"); + records.ShouldAllBe(r => r.PayloadRefJson == null, "precondition: every payload is inline, where this test reads it"); + + records.Single(r => r.Channel == NativeRecordChannel.Stdout && r.Ordinal == cut.Ordinal + 1).InlinePayload.ShouldBe(SecretRedactor.Placeholder, "the key is masked whole in the record that continues the cut, not dropped"); + foreach (var record in records) ShouldHoldNoFragmentOf(key, record.InlinePayload ?? "", $"native record {record.Channel} #{record.Ordinal}"); + + var result = JsonSerializer.Deserialize(run.ResultJson!, AgentJson.Options)!; + var transcript = result.TranscriptArtifactId is { } transcriptId ? System.Text.Encoding.UTF8.GetString((await scope.Resolve().GetBytesAsync(teamId, transcriptId, CancellationToken.None))!.Bytes) : result.Transcript; + transcript.ShouldContain("\n" + SecretRedactor.Placeholder + "\nshort line " + SecretRedactor.Placeholder, Case.Sensitive); + ShouldHoldNoFragmentOf(key, transcript, "the transcript"); + } + + /// Any six consecutive characters of the key: the whole value, and the head or tail a cut would leave behind. + private static void ShouldHoldNoFragmentOf(string key, string text, string where) + { + for (var start = 0; start + 6 <= key.Length; start++) + text.Contains(key.Substring(start, 6), StringComparison.Ordinal).ShouldBeFalse($"{where} holds '{key.Substring(start, 6)}', a fragment of the model key"); + } + [Fact] public async Task Redacts_an_echoed_author_supplied_env_token_from_the_error_and_the_append_only_log() { @@ -2273,6 +2319,24 @@ public IReadOnlyDictionary ProjectToEnv(ResolvedModelCredential new Dictionary { [_envVar] = credential.ApiKey ?? "" }; } + /// A projecting harness running a given script whose parser, like the production CLI parsers on a non-JSON line, makes an event of the short control line only. + private sealed class LongLineKeyEchoingHarness(string provider, string envVar, string script) : IAgentHarness, IModelCredentialProjector + { + public string Kind => "scripted-projector"; + public string Version => "test"; + public IReadOnlyList Models { get; } = new[] { "test-model" }; + + public SandboxSpec BuildInvocation(AgentTask task) => new() { Command = "/bin/sh", Args = new[] { "-c", script }, WorkingDirectory = task.WorkspaceDirectory, Environment = task.Environment, TimeoutSeconds = task.TimeoutSeconds }; + + public IReadOnlyList ParseEvents(string rawLine) => rawLine.StartsWith("short line", StringComparison.Ordinal) ? new[] { new AgentEvent { Kind = AgentEventKind.AssistantMessage, Text = rawLine } } : Array.Empty(); + + public IAgentEventFolder CreateFolder() => new TestEventFolder((fold, exitCode) => new() { Status = AgentRunStatus.Succeeded, ExitReason = "completed", Summary = fold.LastText }); + + public IReadOnlyList SupportedProviders => new[] { provider }; + + public IReadOnlyDictionary ProjectToEnv(ResolvedModelCredential credential) => new Dictionary { [envVar] = credential.ApiKey ?? "" }; + } + /// A projecting harness whose script ECHOES its injected key to stdout, and whose events carry that line in BOTH Text and a structured Data payload — the leak the redactor must mask before the append-only log persists it. private sealed class KeyEchoingHarness : IAgentHarness, IModelCredentialProjector { diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs index 23650f706..0755354c6 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs @@ -516,6 +516,55 @@ public async Task Re_attach_redacts_an_echoed_model_key_in_the_resumed_tail() } } + /// + /// The resumed tail reads the spool in 8 MiB passes like the live one, so a line longer than a pass arrives as a frame + /// the reader cut and its continuation. A key the cut lands inside must be masked whole in the transcript, not kept + /// as a head on one line and a tail on the next. + /// + [Fact] + public async Task Re_attach_masks_a_model_key_the_read_pass_cut_splits_in_the_resumed_tail() + { + if (OperatingSystem.IsWindows()) return; + + var key = "sk-split-" + string.Concat(Enumerable.Range(0, 32).Select(_ => "BCDFGHJKLMNPQRSTVWZ23456789"[Random.Shared.Next(27)])); + var teamId = await SeedTeamAsync(); + var credId = await SeedModelCredentialAsync(teamId, "scripted-provider", key); + var runId = await CreateRunWithCredentialAsync(teamId, credId); + using (var scope = _fixture.BeginScope()) + await scope.Resolve().MarkRunningAsync(runId, CancellationToken.None); + + // The dead observer checkpointed after the first line. The resumed pass then cuts the long line 12 characters into the key. + var spoolDir = NewSpoolDir(); + const string preCrash = "pre-crash-line\n"; + await File.WriteAllTextAsync(Path.Combine(spoolDir, "out.log"), preCrash + new string('x', 8 * 1024 * 1024 - 12) + key + "\npost-crash-line\n"); + await File.WriteAllTextAsync(Path.Combine(spoolDir, "exit"), "0"); + + var handle = new SandboxHandle { Kind = "local", ProcessId = 2147480000, SpoolDirectory = spoolDir, Deadline = DateTimeOffset.UtcNow.AddMinutes(10), StdoutOffset = preCrash.Length, InjectedKeyFingerprint = new SecretRedactor(new[] { key }).Fingerprint }; + using (var scope = _fixture.BeginScope()) + await scope.Resolve().SetRunnerHandleAsync(runId, JsonSerializer.Serialize(handle, AgentJson.Options), CancellationToken.None); + + using (var scope = _fixture.BeginScope()) + { + await scope.Resolve().Database.ExecuteSqlInterpolatedAsync($"UPDATE agent_run SET lease_expires_at = clock_timestamp() - interval '1 hour' WHERE id = {runId}"); + _reservations[runId] = (await scope.Resolve().ReserveReattachAsync(runId, CancellationToken.None))!; + } + + await ReattachAsync(runId, new ProjectingHarness("scripted-provider", "SCRIPTED_MODEL_KEY", maxLineChars: 1024)); + + using (var scope = _fixture.BeginScope()) + { + var run = await scope.Resolve().GetAsync(runId, CancellationToken.None); + run.Status.ShouldBe(AgentRunStatus.Succeeded, run.Error); + + var result = JsonSerializer.Deserialize(run.ResultJson!, AgentJson.Options)!; + var transcript = result.TranscriptArtifactId is { } transcriptId ? System.Text.Encoding.UTF8.GetString((await scope.Resolve().GetBytesAsync(teamId, transcriptId, CancellationToken.None))!.Bytes) : result.Transcript; + + transcript.ShouldEndWith("x\n" + SecretRedactor.Placeholder + "\npost-crash-line\n", Case.Sensitive, "the key is masked whole on the line that continues the cut, not dropped"); + for (var start = 0; start + 6 <= key.Length; start++) + transcript.Contains(key.Substring(start, 6), StringComparison.Ordinal).ShouldBeFalse($"the resumed transcript holds '{key.Substring(start, 6)}', a fragment of the model key"); + } + } + [Fact] public async Task A_re_attached_brokered_run_records_that_its_model_access_died_with_the_minting_worker() { @@ -1192,11 +1241,14 @@ private sealed class ProjectingHarness : IAgentHarness, IModelCredentialProjecto { private readonly string _provider; private readonly string _envVar; + private readonly int _maxLineChars; - public ProjectingHarness(string provider, string envVar) + /// A longer line becomes no event, as a production CLI parser drops a line it cannot parse. + public ProjectingHarness(string provider, string envVar, int maxLineChars = int.MaxValue) { _provider = provider; _envVar = envVar; + _maxLineChars = maxLineChars; } public string Kind => "scripted-projector"; @@ -1208,7 +1260,7 @@ public ProjectingHarness(string provider, string envVar) public IReadOnlyList ParseEvents(string rawLine) { var line = rawLine.Trim(); - return line.Length == 0 ? Array.Empty() : new[] { new AgentEvent { Kind = AgentEventKind.AssistantMessage, Text = line, Data = JsonSerializer.SerializeToElement(new { line }) } }; + return line.Length == 0 || line.Length > _maxLineChars ? Array.Empty() : new[] { new AgentEvent { Kind = AgentEventKind.AssistantMessage, Text = line, Data = JsonSerializer.SerializeToElement(new { line }) } }; } // Via ScriptedFolders, as above. Its exit reason on a non-zero exit becomes "non-zero-exit" rather than diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/Infrastructure/SecretEchoStreamingLlmClient.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/Infrastructure/SecretEchoStreamingLlmClient.cs new file mode 100644 index 000000000..9a4fabf64 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/Infrastructure/SecretEchoStreamingLlmClient.cs @@ -0,0 +1,63 @@ +using System.Runtime.CompilerServices; +using System.Text.Json; +using CodeSpace.Core.Services.Workflows.Llm; + +namespace CodeSpace.IntegrationTests.Workflows.Infrastructure; + +/// +/// A streaming model that prints back what it was told to, paced so the echo straddles the recorder's 32-KiB delta +/// flush: everything after in the user prompt is streamed with only its first ten characters +/// before the size bound and the rest after it. This is the prompt-injected model that splits a secret across two +/// interaction.delta rows on purpose. After instead, it streams only the start of the +/// value and holds the stream open until the call is cancelled: the model an operator cancels mid-secret. Registered +/// under its own provider tag, so it sits beside every other fake without a collision, and through the production +/// recording decorators like any real client. +/// +public sealed class SecretEchoStreamingLlmClient : ILLMClient, IStructuredLLMClient, IStreamingLLMClient +{ + public const string ProviderTag = "TestSecretEchoStream"; + + public const string EchoMarker = "ECHO:"; + + public const string HoldMarker = "HOLD:"; + + /// What a held stream sends before the start of the value: text no secret can begin with, so a flush can settle it. + public const string HeldLead = "progress "; + + /// How much of the value a held stream sends before it holds: the start of it, never the whole. + public const int HeldValueChars = 14; + + /// The filler streamed before the echo, sized so the first ten echoed characters complete the flush bound. + public static readonly string Filler = new('x', RecordingStreamingStructuredLLMClientDecorator.DeltaFlushBytes - 10); + + public string Provider => ProviderTag; + + public Task CompleteAsync(LLMCompletionRequest request, CancellationToken cancellationToken) => + Task.FromResult(new LLMCompletion { Text = Filler + EchoOf(request.UserPrompt), Model = request.Model, Usage = new() { InputTokens = 1, OutputTokens = 1, FinishReason = "stop" } }); + + public Task CompleteStructuredAsync(StructuredLLMCompletionRequest request, CancellationToken cancellationToken) => + Task.FromResult(new StructuredLLMCompletion { Json = JsonSerializer.SerializeToElement(new { }), Model = request.Model }); + + public async IAsyncEnumerable StreamAsync(LLMCompletionRequest request, [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (request.UserPrompt.Contains(HoldMarker, StringComparison.Ordinal)) + { + yield return new LlmStreamEvent.Meta(Model: request.Model, InputTokens: 7); + yield return new LlmStreamEvent.TextDelta(HeldLead + After(request.UserPrompt, HoldMarker)[..HeldValueChars]); + await Task.Delay(Timeout.Infinite, cancellationToken); + yield break; + } + + var echo = EchoOf(request.UserPrompt); + + yield return new LlmStreamEvent.Meta(Model: request.Model, InputTokens: 7); + yield return new LlmStreamEvent.TextDelta(Filler + echo[..10]); + await Task.Yield(); + yield return new LlmStreamEvent.TextDelta(echo[10..] + " done"); + yield return new LlmStreamEvent.Meta(OutputTokens: 4, FinishReason: "end_turn"); + } + + private static string EchoOf(string userPrompt) => After(userPrompt, EchoMarker); + + private static string After(string userPrompt, string marker) => userPrompt[(userPrompt.IndexOf(marker, StringComparison.Ordinal) + marker.Length)..]; +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/NodeRecordRedactionFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/NodeRecordRedactionFlowTests.cs new file mode 100644 index 000000000..9784d26d7 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/NodeRecordRedactionFlowTests.cs @@ -0,0 +1,283 @@ +using System.Net; +using System.Net.Sockets; +using System.Text; +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Variables; +using CodeSpace.Core.Services.Workflows.Artifacts; +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.Core.Services.Workflows.Runtime; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Commands.Workflows; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Workflows; + +/// +/// The records a node writes for itself, through the real engine against real Postgres, with a team secret in scope: +/// the failed record of a call cancelled in flight, the inline excerpt and the artifact of a command whose output was +/// cut inside the secret, and the streamed interaction.delta rows of a model that printed the secret across a +/// flush. Every one is read by any member of the team (the records endpoint, the SSE tail, the artifact download), so +/// none may hold the secret or any fragment of it. +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public sealed class NodeRecordRedactionFlowTests(PostgresFixture fixture) +{ + /// A shutdown or an operator cancel while the call is in flight writes the failed record from the cancellation branch. + [Fact] + public async Task Cancelling_a_run_while_its_http_request_is_in_flight_keeps_the_secret_out_of_every_record() + { + var canary = Canary(); + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(fixture); + await SetTeamSecretAsync(teamId, userId, canary); + + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + Guid runId; + try + { + var accepted = listener.AcceptTcpClientAsync(); + runId = await SeedRunAsync(teamId, userId, Definition(new NodeDefinition + { + Id = "call", TypeKey = "http.request", Config = WorkflowsTestSeed.Json("""{ "timeoutSeconds": 60 }"""), + Inputs = WorkflowsTestSeed.Json($$$"""{ "url": "http://127.0.0.1:{{{port}}}/hold?api_key={{team.API_KEY}}", "method": "GET" }"""), + })); + + var engine = Task.Run(() => RunEngineAsync(runId)); + using var held = await accepted.WaitAsync(TimeSpan.FromSeconds(30)); + + using (var operatorScope = fixture.BeginScopeAs(userId, teamId, Roles.Admin)) + (await operatorScope.Resolve().Send(new CancelRunCommand { RunId = runId })).ShouldNotBeNull(); + + try { await engine.WaitAsync(TimeSpan.FromSeconds(60)); } + catch (OperationCanceledException) { /* the engine surfaces the operator's cancel */ } + } + finally { listener.Stop(); } + + var records = await RecordsAsync(runId); + var failed = records.Single(r => r.RecordType == WorkflowRunRecordTypes.ExternalCallFailed); + JsonDocument.Parse(failed.PayloadJson).RootElement.GetProperty("target").GetString() + .ShouldBe($"http://127.0.0.1:{port}/hold?api_key={PersistenceSecretRedactor.Marker}", customMessage: "the call in flight at the cancel is named exactly as external_call.started named it"); + ShouldHoldNoFragment(records, canary); + } + + [Fact] + public async Task A_command_whose_inline_cap_lands_inside_the_secret_keeps_no_fragment_inline_or_in_its_artifact() + { + var canary = Canary(); + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(fixture); + await SetTeamSecretAsync(teamId, userId, canary); + + // "deploying with key=" is 19 characters; the 40-character cap keeps a 26-character head, 7 into the secret. + var runId = await SeedRunAsync(teamId, userId, Definition(Command("printf 'deploying with key=%s\\n' \"$1\"; head -c 4000 /dev/zero | tr '\\000' x", maxOutputChars: 40))); + await RunEngineAsync(runId); + + var outputs = await NodeOutputsAsync(runId); + outputs.GetProperty("stdout").GetString()!.ShouldStartWith("deploying with key="); + var artifact = await ArtifactTextAsync(teamId, outputs.GetProperty("stdoutArtifactId").GetGuid()); + artifact.ShouldStartWith("deploying with key=" + PersistenceSecretRedactor.Marker + "\n", customMessage: "the artifact is the whole output, redacted"); + + ShouldHoldNoFragment(outputs.GetRawText(), canary, "node outputs"); + ShouldHoldNoFragment(artifact, canary, "the stdout artifact"); + ShouldHoldNoFragment(await RecordsAsync(runId), canary); + } + + [Fact] + public async Task A_command_whose_capture_budget_ends_inside_the_secret_keeps_no_fragment_of_it() + { + var canary = Canary(); + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(fixture); + await SetTeamSecretAsync(teamId, userId, canary); + + // The runner keeps a 1 MiB prefix. Ten bytes short of it the secret begins, so the capture ends ten characters in. + var script = $"head -c {SandboxCaptureBudget.DefaultBytes - 10} /dev/zero | tr '\\000' x; printf '%s' \"$1\"; head -c 100 /dev/zero | tr '\\000' y"; + var runId = await SeedRunAsync(teamId, userId, Definition(Command(script, maxOutputChars: 64))); + await RunEngineAsync(runId); + + var outputs = await NodeOutputsAsync(runId); + outputs.GetProperty("stdoutCaptureComplete").GetBoolean().ShouldBeFalse("precondition: the runner's budget cut the capture"); + outputs.GetProperty("stdout").GetString()!.ShouldEndWith(PersistenceSecretRedactor.Marker, customMessage: "the excerpt's tail is the masked start of the secret"); + var excerpt = await ArtifactTextAsync(teamId, outputs.GetProperty("stdoutCapturedArtifactId").GetGuid()); + excerpt.ShouldEndWith("x" + PersistenceSecretRedactor.Marker); + + ShouldHoldNoFragment(outputs.GetRawText(), canary, "node outputs"); + ShouldHoldNoFragment(excerpt, canary, "the captured-excerpt artifact"); + } + + /// + /// The engine's own path: llm.complete above the streaming threshold, the production recording decorators, the node + /// scope's capture redactor built from the run's secret paths, and the real ledger the SSE tail reads. + /// + [Fact] + public async Task A_model_streaming_the_secret_across_a_delta_flush_leaves_no_fragment_in_any_delta_row() + { + var canary = Canary(); + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(fixture); + await WorkflowsTestSeed.SeedCredentialedModelAsync(fixture, teamId, "secret-echo-model", SecretEchoStreamingLlmClient.ProviderTag); + await SetTeamSecretAsync(teamId, userId, canary); + + var runId = await SeedRunAsync(teamId, userId, Definition(new NodeDefinition + { + Id = "llm", TypeKey = "llm.complete", + Config = JsonSerializer.SerializeToElement(new { provider = SecretEchoStreamingLlmClient.ProviderTag, maxTokens = 60000 }), + Inputs = JsonSerializer.SerializeToElement(new { userPrompt = SecretEchoStreamingLlmClient.EchoMarker + "{{team.API_KEY}}" }), + })); + await RunEngineAsync(runId); + + var records = await RecordsAsync(runId); + var deltas = new List(); + foreach (var row in records.Where(r => r.RecordType == WorkflowRunRecordTypes.InteractionDelta && r.NodeId == "llm")) + deltas.Add(await CapturedTextAsync(teamId, JsonDocument.Parse(row.PayloadJson).RootElement.GetProperty("text"))); + deltas.Count.ShouldBeGreaterThanOrEqualTo(2, "precondition: the 32-KiB bound flushed mid-stream, so the echo was split across rows"); + string.Concat(deltas).ShouldBe(SecretEchoStreamingLlmClient.Filler + PersistenceSecretRedactor.Marker + " done", "the rows concatenate to the masked output: nothing lost, nothing duplicated"); + + foreach (var delta in deltas) ShouldHoldNoFragment(delta, canary, "an interaction.delta row"); + ShouldHoldNoFragment(records, canary); + } + + /// + /// An operator's cancel while the model is part-way into the secret: the one-second flush has written the settled + /// text and is holding back the start of the secret, whose rest never arrives. The cancel's own flush must not + /// release that start verbatim. + /// + [Fact] + public async Task Cancelling_a_run_while_its_model_streams_the_start_of_the_secret_leaves_no_fragment_in_any_delta_row() + { + var canary = Canary(); + var (teamId, userId) = await WorkflowsTestSeed.SeedTeamAsync(fixture); + await WorkflowsTestSeed.SeedCredentialedModelAsync(fixture, teamId, "secret-echo-model", SecretEchoStreamingLlmClient.ProviderTag); + await SetTeamSecretAsync(teamId, userId, canary); + + var runId = await SeedRunAsync(teamId, userId, Definition(new NodeDefinition + { + Id = "llm", TypeKey = "llm.complete", + Config = JsonSerializer.SerializeToElement(new { provider = SecretEchoStreamingLlmClient.ProviderTag, maxTokens = 60000 }), + Inputs = JsonSerializer.SerializeToElement(new { userPrompt = SecretEchoStreamingLlmClient.HoldMarker + "{{team.API_KEY}}" }), + })); + + var engine = Task.Run(() => RunEngineAsync(runId)); + await WaitForRecordAsync(runId, WorkflowRunRecordTypes.InteractionDelta, "the one-second flush writes the settled text while the stream is held"); + + using (var operatorScope = fixture.BeginScopeAs(userId, teamId, Roles.Admin)) + (await operatorScope.Resolve().Send(new CancelRunCommand { RunId = runId })).ShouldNotBeNull(); + + try { await engine.WaitAsync(TimeSpan.FromSeconds(60)); } + catch (OperationCanceledException) { /* the engine surfaces the operator's cancel */ } + + var records = await RecordsAsync(runId); + records.ShouldContain(r => r.RecordType == WorkflowRunRecordTypes.InteractionFailed, "precondition: the cancel cut the stream"); + var deltas = new List(); + foreach (var row in records.Where(r => r.RecordType == WorkflowRunRecordTypes.InteractionDelta && r.NodeId == "llm")) + deltas.Add(await CapturedTextAsync(teamId, JsonDocument.Parse(row.PayloadJson).RootElement.GetProperty("text"))); + + deltas.ToArray().ShouldBe(new[] { SecretEchoStreamingLlmClient.HeldLead, PersistenceSecretRedactor.Marker }, "the settled text, then the held start of the secret masked by the cut"); + ShouldHoldNoFragment(records, canary); + } + + // ─── Helpers ──────────────────────────────────────────────────────────────── + + private async Task WaitForRecordAsync(Guid runId, string recordType, string because) + { + var deadline = DateTime.UtcNow.AddSeconds(30); + while (!(await RecordsAsync(runId)).Any(r => r.RecordType == recordType)) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException($"{because}: no {recordType} record for run {runId} within 30s. Check the run's workflow_run_record rows and its node's status for why the stream never flushed."); + await Task.Delay(50); + } + } + + /// A unique secret whose random part is upper-case consonants, so no six of its characters can turn up by chance in a GUID, a hex digest or prose elsewhere in a record. + private static string Canary() => "CANARY-" + string.Concat(Enumerable.Range(0, 24).Select(_ => "BCDFGHJKLMNPQRSTVWZ"[Random.Shared.Next(19)])); + + private static NodeDefinition Command(string script, int maxOutputChars) => new() + { + Id = "command", TypeKey = "agent.run_command", Config = WorkflowsTestSeed.EmptyJson(), + Inputs = JsonSerializer.SerializeToElement(new { command = "/bin/sh", args = new[] { "-c", script, "sh", "{{team.API_KEY}}" }, maxOutputChars }), + }; + + private static WorkflowDefinition Definition(NodeDefinition node) => new() + { + SchemaVersion = 1, + Nodes = new List + { + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + node, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + }, + Edges = new List { new() { From = "start", To = node.Id }, new() { From = node.Id, To = "end" } }, + }; + + private async Task SetTeamSecretAsync(Guid teamId, Guid userId, string value) + { + using var scope = fixture.BeginScope(); + await scope.Resolve().SetAsync(VariableScope.Team, teamId, teamId, "API_KEY", VariableValueType.Secret, JsonSerializer.SerializeToElement(value), null, userId, CancellationToken.None); + } + + private async Task SeedRunAsync(Guid teamId, Guid userId, WorkflowDefinition definition) + { + Guid workflowId; + using (var scope = fixture.BeginScopeAs(userId, teamId, Roles.Admin)) + { + workflowId = await scope.Resolve().Send(new CreateWorkflowCommand { Name = "f2-redact-" + Guid.NewGuid().ToString("N")[..8], Definition = definition, Activations = new List(), Enabled = true }); + } + + return await WorkflowsTestSeed.SeedManualRunAsync(fixture, workflowId, teamId); + } + + private async Task RunEngineAsync(Guid runId) + { + using var scope = fixture.BeginScope(); + await scope.Resolve().ExecuteRunAsync(runId, CancellationToken.None); + } + + private async Task> RecordsAsync(Guid runId) + { + using var scope = fixture.BeginScope(); + return await scope.Resolve().WorkflowRunRecord.AsNoTracking().Where(r => r.RunId == runId).OrderBy(r => r.Sequence).ToListAsync(); + } + + private async Task NodeOutputsAsync(Guid runId) + { + using var scope = fixture.BeginScope(); + var node = await scope.Resolve().WorkflowRunNode.AsNoTracking().SingleAsync(n => n.RunId == runId && n.NodeId == "command"); + node.Status.ShouldBe(NodeStatus.Success, node.Error); + return JsonDocument.Parse(node.OutputsJson).RootElement.Clone(); + } + + private async Task ArtifactTextAsync(Guid teamId, Guid artifactId) + { + using var scope = fixture.BeginScope(); + var artifact = (await scope.Resolve().GetBytesAsync(teamId, artifactId, CancellationToken.None)).ShouldNotBeNull($"artifact {artifactId} was stored for the team"); + return Encoding.UTF8.GetString(artifact.Bytes); + } + + /// A captured text field: inline, or offloaded to the team's artifact store behind a $artifact_id ref. + private async Task CapturedTextAsync(Guid teamId, JsonElement field) => + field.ValueKind == JsonValueKind.String ? field.GetString()! : await ArtifactTextAsync(teamId, field.GetProperty("$artifact_id").GetGuid()); + + private static void ShouldHoldNoFragment(IEnumerable records, string canary) + { + foreach (var record in records) ShouldHoldNoFragment(record.PayloadJson, canary, $"workflow_run_record {record.RecordType} (node {record.NodeId})"); + } + + /// Any six consecutive characters of the secret: a whole value AND the head a cut would leave behind. + private static void ShouldHoldNoFragment(string text, string canary, string where) + { + for (var start = 0; start + 6 <= canary.Length; start++) + { + var fragment = canary.Substring(start, 6); + text.Contains(fragment, StringComparison.Ordinal).ShouldBeFalse($"{where} holds '{fragment}', a fragment of the team secret"); + } + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Agents/SecretRedactorTests.cs b/backend/tests/CodeSpace.UnitTests/Agents/SecretRedactorTests.cs index 58a892b24..2578384da 100644 --- a/backend/tests/CodeSpace.UnitTests/Agents/SecretRedactorTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Agents/SecretRedactorTests.cs @@ -141,4 +141,52 @@ public void A_chunk_with_no_patterns_is_returned_without_copying() MemoryMarshal.TryGetArray(transformed.Bytes, out var segment).ShouldBeTrue(); ReferenceEquals(segment.Array, source).ShouldBeTrue("a zero-pattern redactor must hand back the caller's own buffer, never a copy of it"); } + + private const string FrameKey = "sk-frame-BR6FZM29FGMQJMBZ262GMMFL6KSJH3ZJ"; + + /// + /// A line longer than one read pass arrives as a frame the reader cut (incomplete) and its continuation, and each + /// is persisted on its own — a native record, a transcript line. Exact matching per frame cannot see a key the cut + /// split, so the end of a cut frame that could start one is held back and redacted with its continuation. Every + /// split point of the line is tried, including several cuts in a row. + /// + [Fact] + public void Frame_stream_masks_a_key_split_across_cut_frames_wherever_the_cuts_land() + { + var redactor = new SecretRedactor(new[] { FrameKey }); + var line = "xxxx " + FrameKey + " yyyy"; + + for (var first = 0; first <= line.Length; first++) + { + for (var second = first; second <= line.Length; second += 7) + { + var stream = redactor.CreateFrameStream(); + var persisted = new[] { stream.Redact(line[..first], isComplete: false), stream.Redact(line[first..second], isComplete: false), stream.Redact(line[second..], isComplete: true) }; + + string.Concat(persisted).ShouldBe("xxxx *** yyyy", $"the frames concatenate to the masked line (cuts at {first} and {second})"); + foreach (var frame in persisted) + for (var start = 0; start + 6 <= FrameKey.Length; start++) + frame.ShouldNotContain(FrameKey.Substring(start, 6), Case.Sensitive, $"a frame keeps '{FrameKey.Substring(start, 6)}' of the key (cuts at {first} and {second})"); + } + } + } + + [Fact] + public void Frame_stream_holds_back_only_what_a_cut_frame_could_still_turn_into_a_key() + { + var stream = new SecretRedactor(new[] { FrameKey }).CreateFrameStream(); + + stream.Redact("progress sk-fr", isComplete: false).ShouldBe("progress ", "the start of the key is held for the continuation"); + stream.Redact("ame-nope", isComplete: false).ShouldBe("sk-frame-nope", "a continuation that does not finish the key releases the held text verbatim"); + stream.Redact("done sk-", isComplete: true).ShouldBe("done sk-", "a complete frame ends its line: nothing can continue it, so nothing is held"); + } + + [Fact] + public void Frame_stream_with_no_secrets_passes_every_frame_through() + { + var stream = SecretRedactor.None.CreateFrameStream(); + const string frame = "an incomplete frame"; + + ReferenceEquals(stream.Redact(frame, isComplete: false), frame).ShouldBeTrue("a run with nothing to mask must not copy its frames"); + } } diff --git a/backend/tests/CodeSpace.UnitTests/Settings/ConfigurationPlaceholderGuardTests.cs b/backend/tests/CodeSpace.UnitTests/Settings/ConfigurationPlaceholderGuardTests.cs index a22a8b81d..e07af1ebf 100644 --- a/backend/tests/CodeSpace.UnitTests/Settings/ConfigurationPlaceholderGuardTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Settings/ConfigurationPlaceholderGuardTests.cs @@ -18,20 +18,39 @@ namespace CodeSpace.UnitTests.Settings; public class ConfigurationPlaceholderGuardTests { [Theory] - [InlineData("#{Agents:RunSpoolDirectory}")] // the one that shipped - [InlineData("/app/#{Agents:RunSpoolDirectory}")] // embedded in an otherwise real path - [InlineData("#{Db:Password}")] - [InlineData("#{DB_PASSWORD}")] // SHOUTING_CASE, no colon - [InlineData("__CONNECTION_STRING__")] // Azure DevOps token replacement - [InlineData("https://#{Api:Host}/api/webhooks")] - public void An_unrendered_placeholder_refuses_the_start(string value) + [InlineData("#{Agents:RunSpoolDirectory}", "#{Agents:RunSpoolDirectory}")] // the one that shipped + [InlineData("/app/#{Agents:RunSpoolDirectory}", "#{Agents:RunSpoolDirectory}")] // embedded in an otherwise real path + [InlineData("#{Db:Password}", "#{Db:Password}")] + [InlineData("#{DB_PASSWORD}", "#{DB_PASSWORD}")] // SHOUTING_CASE, no colon + [InlineData("__CONNECTION_STRING__", "__CONNECTION_STRING__")] // Azure DevOps token replacement + [InlineData("https://#{Api:Host}/api/webhooks", "#{Api:Host}")] + public void An_unrendered_placeholder_refuses_the_start(string value, string placeholder) { var act = () => ConfigurationPlaceholderGuard.ThrowIfUnrendered(ConfigurationOf(("Agents:RunSpoolDirectory", value))); var thrown = act.ShouldThrow(); thrown.Message.ShouldContain("Agents:RunSpoolDirectory", customMessage: "the operator has to be told which key to go and define"); - thrown.Message.ShouldContain(value, customMessage: "and the literal text to search their release definition for"); + thrown.Message.ShouldContain(placeholder, customMessage: "and the literal text to search their release definition for"); + } + + /// + /// A partly rendered value. The release step substituted the password and missed the database name, so the value + /// now holds a real credential beside the template. This message goes to stderr before logging exists, so it names + /// the key and the placeholder and never the value. + /// + [Fact] + public void A_partly_rendered_value_is_reported_by_key_and_placeholder_and_never_by_its_value() + { + var configuration = ConfigurationOf(("CodeSpaceStore:ConnectionString", "Host=pg.internal;Username=codespace;Password=CANARY-placeholder-7e21;Database=#{CodeSpace:DbName};Search Path=__DB_SCHEMA__")); + + var thrown = Should.Throw(() => ConfigurationPlaceholderGuard.ThrowIfUnrendered(configuration)); + + thrown.Message.ShouldContain("CodeSpaceStore:ConnectionString"); + thrown.Message.ShouldContain("#{CodeSpace:DbName}"); + thrown.Message.ShouldContain("__DB_SCHEMA__", customMessage: "every placeholder in the value is named, so one redeploy fixes them all"); + thrown.Message.ShouldNotContain("CANARY-placeholder-7e21", Case.Sensitive, "the rendered part of the value may be a credential"); + thrown.Message.ShouldNotContain("pg.internal"); } /// diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/AgentRunCommandNodeTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/AgentRunCommandNodeTests.cs index 5cc6dedb9..4b42a4254 100644 --- a/backend/tests/CodeSpace.UnitTests/Workflows/AgentRunCommandNodeTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Workflows/AgentRunCommandNodeTests.cs @@ -404,6 +404,76 @@ public async Task Capture_loss_is_reported_without_overwriting_the_completed_com result.Outputs.ContainsKey("stdoutCapturedArtifactId").ShouldBe(failure != "artifact"); } + private const string Canary = "CANARY-run-command-5f0c2d7e9a1b"; + + /// + /// The cap used to cut the RAW capture and leave the engine to mask the excerpt: a head that ended inside a secret + /// kept its first characters, which exact matching never recognises again. The excerpt and the artifact (which has + /// no encrypted sidecar, so every member downloads exactly what is stored) are now cut from the redacted text. + /// + [Fact] + public async Task A_cap_that_lands_inside_a_secret_keeps_no_fragment_of_it_inline_or_in_the_artifact() + { + var team = Guid.NewGuid(); + var stdout = "deploying with key=" + Canary + "\n" + new string('x', 4000); + var stub = new StubRunCommandService { Result = new() { Status = SandboxStatus.Success, ExitCode = 0, Stdout = stdout, Stderr = "" } }; + var artifacts = new FakeArtifactStore(); + var context = ContextWithSys(new() { ["command"] = JsonSerializer.SerializeToElement("sh"), ["maxOutputChars"] = JsonSerializer.SerializeToElement(40) }, new() { [SystemScopeKeys.TeamId] = JsonSerializer.SerializeToElement(team) }) with { PersistenceRedactor = new PersistenceSecretRedactor([Canary]) }; + + var result = await new AgentRunCommandNode(stub, artifacts).RunAsync(context, CancellationToken.None); + + var inline = result.Outputs["stdout"].GetString()!; + inline.ShouldStartWith("deploying with key=", customMessage: "the 26-character head ends 7 characters into where the secret was"); + ShouldHoldNoFragmentOfTheCanary(inline, "the inline excerpt"); + var artifact = System.Text.Encoding.UTF8.GetString(artifacts.Puts.ShouldHaveSingleItem().Bytes); + artifact.ShouldBe("deploying with key=" + PersistenceSecretRedactor.Marker + "\n" + new string('x', 4000), "the artifact is the whole capture, redacted"); + result.Outputs.ContainsKey("stdoutArtifactId").ShouldBeTrue(); + } + + [Fact] + public async Task A_runner_capture_that_stopped_inside_a_secret_keeps_no_fragment_of_it() + { + var team = Guid.NewGuid(); + var captured = new string('x', 100) + " key=" + Canary[..12]; // the runner's budget ended the capture 12 characters into the secret + var observed = new SandboxStreamObservation { ObservedBytes = 5000, ReachedEndOfStream = true, CaptureComplete = false }; + var stub = new StubRunCommandService { Result = new() { Status = SandboxStatus.Success, ExitCode = 0, Stdout = captured, Stderr = "", Observation = new SandboxObservation { Stdout = observed } } }; + var artifacts = new FakeArtifactStore(); + var context = ContextWithSys(new() { ["command"] = JsonSerializer.SerializeToElement("sh") }, new() { [SystemScopeKeys.TeamId] = JsonSerializer.SerializeToElement(team) }) with { PersistenceRedactor = new PersistenceSecretRedactor([Canary]) }; + + var result = await new AgentRunCommandNode(stub, artifacts).RunAsync(context, CancellationToken.None); + + var inline = result.Outputs["stdout"].GetString()!; + inline.ShouldBe(new string('x', 100) + " key=" + PersistenceSecretRedactor.Marker); + ShouldHoldNoFragmentOfTheCanary(System.Text.Encoding.UTF8.GetString(artifacts.Puts.ShouldHaveSingleItem().Bytes), "the captured-excerpt artifact"); + result.Outputs.ContainsKey("stdoutCapturedArtifactId").ShouldBeTrue(); + result.Outputs["stdoutCapturedBytes"].GetInt64().ShouldBe(captured.Length, "the size reported is the runner's capture, not the masked copy"); + } + + /// + /// Nothing was cut, so the engine's exact match masks the whole value in node.completed and keeps the real one in the + /// encrypted sidecar. The node must not mask it first: a downstream node, a resume and a rerun read the runtime value. + /// + [Fact] + public async Task An_uncut_capture_keeps_its_runtime_value_for_the_engine_to_mask_at_persistence() + { + var stdout = "token=" + Canary; + var stub = new StubRunCommandService { Result = new() { Status = SandboxStatus.Success, ExitCode = 0, Stdout = stdout, Stderr = "" } }; + var artifacts = new FakeArtifactStore(); + var context = ContextWithSys(new() { ["command"] = JsonSerializer.SerializeToElement("sh") }, new() { [SystemScopeKeys.TeamId] = JsonSerializer.SerializeToElement(Guid.NewGuid()) }) with { PersistenceRedactor = new PersistenceSecretRedactor([Canary]) }; + + var result = await new AgentRunCommandNode(stub, artifacts).RunAsync(context, CancellationToken.None); + + result.Outputs["stdout"].GetString().ShouldBe(stdout); + artifacts.Puts.ShouldBeEmpty(); + } + + private static void ShouldHoldNoFragmentOfTheCanary(string text, string where) + { + const int window = 6; + for (var start = 0; start + window <= Canary.Length; start++) + text.ShouldNotContain(Canary.Substring(start, window), Case.Sensitive, $"{where} keeps '{Canary.Substring(start, window)}', a fragment of the secret: {text[..Math.Min(text.Length, 200)]}"); + } + private sealed class LossObserver : INodeObservability, INodeLossReporting { public List Notices { get; } = []; diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/HttpRequestNodeLogTargetTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/HttpRequestNodeLogTargetTests.cs new file mode 100644 index 000000000..af0edec1a --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Workflows/HttpRequestNodeLogTargetTests.cs @@ -0,0 +1,64 @@ +using System.Net; +using System.Text.Json; +using CodeSpace.Core.Services.Workflows.Nodes; +using CodeSpace.Core.Services.Workflows.Nodes.Builtin; +using CodeSpace.Core.Services.Workflows.Runtime; +using CodeSpace.Messages.Enums; +using Microsoft.Extensions.Logging; +using Shouldly; + +namespace CodeSpace.UnitTests.Workflows; + +/// +/// The operator log line http.request writes before the call. It names where the call goes and nothing more: a +/// URL's userinfo (https://user:{{team.PASSWORD}}@host) is a credential, and this line reaches the shared log +/// sink at Information, before the ledger's redaction or the outbound guard ever run. +/// +[Trait("Category", "Unit")] +public sealed class HttpRequestNodeLogTargetTests +{ + [Theory] + [InlineData("https://svc-user:CANARY-userinfo-61a0@api.example.test/v1/deploy?token=CANARY-query-61a0", "https://api.example.test")] + [InlineData("http://svc-user:CANARY-userinfo-61a0@api.example.test:8443/v1", "http://api.example.test:8443")] + public async Task The_log_target_is_scheme_host_and_port_only(string url, string expected) + { + var logger = new CapturingLogger(); + + var result = await new HttpRequestNode(new StubFactory()).RunAsync(Context(url, logger), CancellationToken.None); + + result.Status.ShouldBe(NodeStatus.Success, result.Error); + var line = logger.Lines.First(entry => entry.StartsWith("HTTP GET ", StringComparison.Ordinal)); + line.ShouldBe($"HTTP GET {expected} (timeout=30s)"); + string.Join('\n', logger.Lines).ShouldNotContain("CANARY", Case.Sensitive, "no part of a credential reaches the operator log"); + } + + private static NodeRunContext Context(string url, ILogger logger) => new() + { + Inputs = new Dictionary { ["url"] = JsonSerializer.SerializeToElement(url), ["method"] = JsonSerializer.SerializeToElement("GET") }, + Config = new Dictionary(), + RawInputs = JsonDocument.Parse("{}").RootElement, + RawConfig = JsonDocument.Parse("{}").RootElement, + Scope = new NodeRunScope { Trigger = new Dictionary() }, + Logger = logger, + Observability = NodeObservability.NoOp, + }; + + private sealed class StubFactory : IHttpClientFactory + { + public HttpClient CreateClient(string name) => new(new OkHandler()); + } + + private sealed class OkHandler : HttpMessageHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) => + Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("{}") }); + } + + private sealed class CapturingLogger : ILogger + { + public List Lines { get; } = []; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) => Lines.Add(formatter(state, exception)); + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/Llm/RecordingLLMClientDecoratorTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/Llm/RecordingLLMClientDecoratorTests.cs index 4d80a2396..f5afd9713 100644 --- a/backend/tests/CodeSpace.UnitTests/Workflows/Llm/RecordingLLMClientDecoratorTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Workflows/Llm/RecordingLLMClientDecoratorTests.cs @@ -451,6 +451,183 @@ public async Task A_slow_sub_threshold_stream_flushes_its_live_projection_within } } + private const string StreamSecret = "CANARY-delta-9b2e4f70c13a"; + + /// + /// Each interaction.delta used to be redacted on its own, so a secret the model printed across a flush landed + /// as two halves no exact match recognised, and the Room concatenates the deltas back into the whole secret. A flush + /// now releases only text no secret can still run across, and carries the rest into the next one. + /// + [Fact] + public async Task A_secret_split_by_a_size_flush_is_masked_in_the_deltas() + { + var head = new string('x', RecordingStreamingStructuredLLMClientDecorator.DeltaFlushBytes - 10); + var inner = new SplitStreamingClient([head + StreamSecret[..10]], [StreamSecret[10..] + " tail"]); + inner.Release.TrySetResult(); + var logger = new CapturingLogger(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret]))) + await ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, new FakeTimeProvider()), CancellationToken.None); + + ShouldMaskTheSecretInEveryDelta(logger, head + PersistenceSecretRedactor.Marker + " tail"); + DeltaTexts(logger).Count.ShouldBe(2, "the size flush still wrote its row, holding back only the start of the secret"); + } + + [Fact] + public async Task A_secret_split_by_a_time_flush_is_masked_in_the_deltas() + { + var inner = new SplitStreamingClient(["Bearer " + StreamSecret[..10]], [StreamSecret[10..] + " tail"]); + var time = new FakeTimeProvider(); + var logger = new CapturingLogger(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret]))) + { + var consume = ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, time), CancellationToken.None); + await inner.WaitingAtPause.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + time.Advance(RecordingStreamingStructuredLLMClientDecorator.DeltaFlushInterval); + await WaitForDeltasAsync(logger, 1, "the one-second flush writes what is settled"); + DeltaTexts(logger).ToArray().ShouldBe(new[] { "Bearer " }, "the live view shows the settled text and holds back the possible start of the secret"); + + inner.Release.TrySetResult(); + await consume; + } + + ShouldMaskTheSecretInEveryDelta(logger, "Bearer " + PersistenceSecretRedactor.Marker + " tail"); + } + + [Fact] + public async Task Secrets_split_by_a_size_flush_and_then_a_time_flush_are_both_masked() + { + const string second = "CANARY-delta-second-41d8e5"; + var head = new string('x', RecordingStreamingStructuredLLMClientDecorator.DeltaFlushBytes - 5); + var inner = new SplitStreamingClient([head + StreamSecret[..5], StreamSecret[5..] + " and " + second[..7]], [second[7..] + " end"]); + var time = new FakeTimeProvider(); + var logger = new CapturingLogger(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret, second]))) + { + var consume = ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, time), CancellationToken.None); + await inner.WaitingAtPause.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await WaitForDeltasAsync(logger, 1, "the size flush has written"); + + time.Advance(RecordingStreamingStructuredLLMClientDecorator.DeltaFlushInterval); + await WaitForDeltasAsync(logger, 2, "the time flush writes the now-whole first secret"); + + inner.Release.TrySetResult(); + await consume; + } + + ShouldMaskTheSecretInEveryDelta(logger, head + PersistenceSecretRedactor.Marker + " and " + PersistenceSecretRedactor.Marker + " end", second); + } + + [Fact] + public async Task The_final_flush_releases_a_held_tail_that_never_became_a_secret() + { + // The stream ends on "C", which could have been the start of the secret. Nothing follows, so it is not one. + var inner = new SplitStreamingClient(["done. " + StreamSecret[..1]], []); + inner.Release.TrySetResult(); + var logger = new CapturingLogger(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret]))) + await ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, new FakeTimeProvider()), CancellationToken.None); + + string.Concat(DeltaTexts(logger)).ShouldBe("done. C", "the carry is released at the final flush; no streamed text is lost"); + logger.Calls[^1].Payload.GetProperty("output").GetString().ShouldBe("done. C"); + } + + /// + /// A cancel or a provider fault cuts the stream while the carry holds exactly the text that could be the start of a + /// secret, and the rest of it never arrives. Released verbatim, it is a fragment exact matching never recognises + /// again; the cut's flush masks it, and the call's presence delta says its capture was masked. + /// + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task A_stream_cut_while_the_carry_holds_the_start_of_a_secret_masks_that_start(bool cancelled) + { + var inner = new SplitStreamingClient(["progress " + StreamSecret[..14]], [StreamSecret[14..]]); + var time = new FakeTimeProvider(); + var logger = new CapturingLogger(); + var completeness = new CapturingCompletenessWriter(); + using var cancellation = new CancellationTokenSource(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret]), completeness)) + { + var consume = ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, time), cancellation.Token); + await inner.WaitingAtPause.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + time.Advance(RecordingStreamingStructuredLLMClientDecorator.DeltaFlushInterval); + await WaitForDeltasAsync(logger, 1, "the one-second flush writes the settled text and holds back the start of the secret"); + + if (cancelled) cancellation.Cancel(); + else inner.Release.TrySetException(new IOException("connection reset mid-stream")); + + await Should.ThrowAsync(() => consume); + } + + DeltaTexts(logger).ToArray().ShouldBe(new[] { "progress ", PersistenceSecretRedactor.Marker }, "the cut's flush masks the held start of the secret instead of releasing it"); + ShouldHoldNoFragmentOf(StreamSecret, DeltaTexts(logger)); + logger.Snapshot()[^1].RecordType.ShouldBe(WorkflowRunRecordTypes.InteractionFailed); + completeness.Advances.Single(advance => advance.Present > 0).Masked.ShouldBeTrue("the stored delta was masked, so the presence delta may not claim a verbatim capture"); + } + + [Fact] + public async Task A_stream_cut_after_text_that_cannot_start_a_secret_keeps_that_text_verbatim() + { + var inner = new SplitStreamingClient(["progress so far"], []); + var logger = new CapturingLogger(); + var completeness = new CapturingCompletenessWriter(); + + using (PushScope(logger, new PersistenceSecretRedactor([StreamSecret]), completeness)) + { + var consume = ConsumeAsync(new RecordingStreamingStructuredLLMClientDecorator(inner, new FakeTimeProvider()), CancellationToken.None); + await inner.WaitingAtPause.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + inner.Release.TrySetException(new IOException("connection reset mid-stream")); + await Should.ThrowAsync(() => consume); + } + + DeltaTexts(logger).ToArray().ShouldBe(new[] { "progress so far" }, "a cut masks only a held tail that could start a secret, never text that cannot"); + completeness.Advances.Single(advance => advance.Present > 0).Masked.ShouldBeFalse(); + } + + private static void ShouldHoldNoFragmentOf(string secret, IEnumerable texts) + { + foreach (var text in texts) + for (var start = 0; start + 6 <= secret.Length; start++) + text.ShouldNotContain(secret.Substring(start, 6), Case.Sensitive, $"no delta row may keep '{secret.Substring(start, 6)}', a fragment of a secret"); + } + + private static IDisposable PushScope(IRunRecordLogger logger, PersistenceSecretRedactor redactor, IRunDataCompletenessWriter? completeness = null) => + LlmCallContext.Push(new LlmCallScope(Run, Team, "llm", "", "llm.complete", logger, new NoopOffloader(), CaptureRedactor: redactor, Completeness: completeness).Unbudgeted("recording test — budget not under test")); + + private static List DeltaTexts(CapturingLogger logger) => + logger.Snapshot().Where(c => c.RecordType == WorkflowRunRecordTypes.InteractionDelta).Select(c => c.Payload.GetProperty("text").GetString()!).ToList(); + + private static void ShouldMaskTheSecretInEveryDelta(CapturingLogger logger, string expectedConcatenation, params string[] otherSecrets) + { + var deltas = logger.Snapshot().Where(c => c.RecordType == WorkflowRunRecordTypes.InteractionDelta).ToList(); + deltas.Select(d => d.Payload.GetProperty("ordinal").GetInt32()).ShouldBe(Enumerable.Range(0, deltas.Count), "ordinals stay monotonic from 0"); + + foreach (var secret in otherSecrets.Prepend(StreamSecret)) + for (var start = 0; start + 6 <= secret.Length; start++) + deltas.ShouldAllBe(d => !d.Payload.GetProperty("text").GetString()!.Contains(secret.Substring(start, 6)), $"no delta row may keep '{secret.Substring(start, 6)}', a fragment of a secret"); + + string.Concat(DeltaTexts(logger)).ShouldBe(expectedConcatenation, "concatenated, the deltas are the masked output: nothing lost, nothing duplicated"); + logger.Snapshot()[^1].Payload.GetProperty("output").GetString().ShouldBe(expectedConcatenation, "and they agree with the authoritative completed row"); + } + + private static async Task WaitForDeltasAsync(CapturingLogger logger, int count, string because) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (Volatile.Read(ref logger.DeltaCount) < count) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException($"{because}: expected {count} interaction.delta rows within 5s, saw {Volatile.Read(ref logger.DeltaCount)}. The streaming recorder's flush did not run — check RecordPendingDeltaAsync."); + await Task.Delay(10); + } + } + [Fact] public async Task A_stream_capture_failure_never_changes_or_faults_the_provider_stream() { @@ -746,6 +923,25 @@ public async IAsyncEnumerable StreamAsync(LLMCompletionRequest r } } + /// Streams beforePause, waits for , then streams afterPause — so a test can hold the stream open across a fake-clock flush. + private sealed class SplitStreamingClient(IReadOnlyList beforePause, IReadOnlyList afterPause) : ILLMClient, IStructuredLLMClient, IStreamingLLMClient + { + public TaskCompletionSource WaitingAtPause { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource Release { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public string Provider => "streaming"; + public Task CompleteAsync(LLMCompletionRequest request, CancellationToken ct) => Task.FromResult(new LLMCompletion { Text = "", Model = "m" }); + public Task CompleteStructuredAsync(StructuredLLMCompletionRequest request, CancellationToken ct) => Task.FromResult(new StructuredLLMCompletion { Json = JsonDocument.Parse("{}").RootElement, Model = "m" }); + + public async IAsyncEnumerable StreamAsync(LLMCompletionRequest request, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken ct) + { + foreach (var part in beforePause) yield return new LlmStreamEvent.TextDelta(part); + WaitingAtPause.TrySetResult(); + await Release.Task.WaitAsync(ct); + foreach (var part in afterPause) yield return new LlmStreamEvent.TextDelta(part); + yield return new LlmStreamEvent.Meta(Model: "m", OutputTokens: 1, FinishReason: "stop"); + } + } + private sealed class NoopOffloader : IArtifactOffloader { public Task OffloadIfLargeAsync(Guid teamId, string? text, string contentType, CancellationToken ct) => Task.FromResult(new OffloadedText(text ?? "", null)); @@ -764,17 +960,22 @@ private sealed record Captured(Guid RunId, string RecordType, string? NodeId, st private sealed class CapturingLogger : IRunRecordLogger { + private readonly object _gate = new(); public List Calls { get; } = new(); public TaskCompletionSource DeltaRecorded { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public int DeltaCount; public bool ThrowOnRecord { get; set; } public bool HonorCancellation { get; set; } + /// A copy taken under the lock, for a test that reads while the stream is still recording on another thread. + public List Snapshot() { lock (_gate) return Calls.ToList(); } + public Task RecordInteractionAsync(Guid runId, string recordType, string? nodeId, string iterationKey, Guid correlationId, Guid? parentRecordId, JsonElement payload, CancellationToken cancellationToken) { if (HonorCancellation) cancellationToken.ThrowIfCancellationRequested(); if (ThrowOnRecord) throw new InvalidOperationException("ledger down"); - Calls.Add(new Captured(runId, recordType, nodeId, iterationKey, correlationId, payload.Clone())); - if (recordType == WorkflowRunRecordTypes.InteractionDelta) DeltaRecorded.TrySetResult(); + lock (_gate) Calls.Add(new Captured(runId, recordType, nodeId, iterationKey, correlationId, payload.Clone())); + if (recordType == WorkflowRunRecordTypes.InteractionDelta) { Interlocked.Increment(ref DeltaCount); DeltaRecorded.TrySetResult(); } return Task.FromResult(Guid.NewGuid()); } diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/NodeObservabilityTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/NodeObservabilityTests.cs index cea39654e..510d9595a 100644 --- a/backend/tests/CodeSpace.UnitTests/Workflows/NodeObservabilityTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Workflows/NodeObservabilityTests.cs @@ -216,6 +216,82 @@ await Should.ThrowAsync(() => observability.TraceExte logger.FailedCalls.Single().Target.ShouldBe(PersistenceSecretRedactor.Marker); } + /// + /// A host shutdown or an operator cancel while the call is in flight. The cancellation branch used to persist the raw + /// target while every other branch persisted the redacted one, so a ?api_key={{team.SECRET}} reached the + /// failed record every member reads through the records endpoint, the SSE tail and the timeline. + /// + [Fact] + public async Task TraceExternalCallAsync_cancelled_mid_call_persists_the_redacted_target() + { + const string secret = "CANARY-cancel-call-7d1f0b"; + var logger = new RecordingLogger(); + var observability = new NodeObservability(new NodeObservationBinding(logger, new RecordingArtifactStore(), RunId, NodeId, TeamId, ParentRecordId, new PersistenceSecretRedactor([secret]))); + using var cancellation = new CancellationTokenSource(); + var inFlight = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var call = observability.TraceExternalCallAsync( + target: "https://api.example.test/hold?api_key=" + secret, + method: "GET", + requestPayload: null, + action: async ct => { inFlight.TrySetResult(); await Task.Delay(Timeout.Infinite, ct); return 0; }, + cancellationToken: cancellation.Token); + await inFlight.Task.WaitAsync(TimeSpan.FromSeconds(5)); + cancellation.Cancel(); + + await Should.ThrowAsync(() => call); + var failed = logger.FailedCalls.ShouldHaveSingleItem("the cancelled call still leaves its failed record"); + failed.Target.ShouldBe("https://api.example.test/hold?api_key=" + PersistenceSecretRedactor.Marker); + failed.Target.ShouldBe(logger.StartedCalls.Single().Target, "the failed record names the call exactly as the started record did"); + } + + [Fact] + public async Task TraceExternalStreamAsync_cancelled_mid_stream_persists_the_redacted_target() + { + const string secret = "CANARY-cancel-stream-3e9a42"; + var logger = new RecordingLogger(); + var observability = new NodeObservability(new NodeObservationBinding(logger, new RecordingArtifactStore(), RunId, NodeId, TeamId, ParentRecordId, new PersistenceSecretRedactor([secret]))); + using var cancellation = new CancellationTokenSource(); + + await using var events = observability.TraceExternalStreamAsync("https://api.example.test/stream?api_key=" + secret, "POST", null, HoldAfterFirstEvent, cancellation.Token).GetAsyncEnumerator(); + (await events.MoveNextAsync()).ShouldBeTrue(); + cancellation.Cancel(); + + await Should.ThrowAsync(async () => await events.MoveNextAsync()); + var failed = logger.FailedCalls.ShouldHaveSingleItem("the cancelled stream still leaves its failed record"); + failed.Target.ShouldNotContain(secret); + failed.Target.ShouldBe(logger.StartedCalls.Single().Target, "the failed record names the stream exactly as the started record did"); + } + + private static async IAsyncEnumerable HoldAfterFirstEvent([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) + { + yield return 1; + await Task.Delay(Timeout.Infinite, cancellationToken); + yield return 2; + } + + /// + /// An artifact has no encrypted sidecar: the bytes stored are the bytes every member downloads. So a node's text or + /// JSON is masked before it is stored, and opaque bytes, which exact matching cannot read, pass through untouched. + /// + [Theory] + [InlineData("text/plain", "token=CANARY-artifact-55c1 in a log", "token=[REDACTED] in a log")] + [InlineData("application/json", "{\"auth\":\"CANARY-artifact-55c1\",\"n\":1}", "{\"auth\":\"[REDACTED]\",\"n\":1}")] + [InlineData("application/json", "{\"auth\":\"CANARY\\u002Dartifact\\u002D55c1\"}", "{\"auth\":\"[REDACTED]\"}")] // an escaped spelling of the same value + [InlineData("application/octet-stream", "token=CANARY-artifact-55c1", "token=CANARY-artifact-55c1")] + [InlineData("text/plain", "nothing secret", "nothing secret")] + public async Task PersistArtifactAsync_masks_text_and_json_before_storing_and_keeps_opaque_bytes(string contentType, string content, string expected) + { + var store = new RecordingArtifactStore(); + var observability = new NodeObservability(new NodeObservationBinding(new RecordingLogger(), store, RunId, NodeId, TeamId, ParentRecordId, new PersistenceSecretRedactor(["CANARY-artifact-55c1"]))); + + var reference = await observability.PersistArtifactAsync(contentType, System.Text.Encoding.UTF8.GetBytes(content), CancellationToken.None); + + var stored = System.Text.Encoding.UTF8.GetString(store.PutCalls.Single().Bytes.Span); + stored.ShouldBe(expected); + reference.GetProperty("size_bytes").GetInt32().ShouldBe(System.Text.Encoding.UTF8.GetByteCount(stored), "the ref describes the bytes that were stored"); + } + [Fact] public async Task PersistArtifactAsync_delegates_to_store_with_bound_team_id() { diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/SecretBoundaryRedactionTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/SecretBoundaryRedactionTests.cs new file mode 100644 index 000000000..e2e1b1b40 --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Workflows/SecretBoundaryRedactionTests.cs @@ -0,0 +1,89 @@ +using CodeSpace.Core.Services.Workflows.Runtime; +using Shouldly; + +namespace CodeSpace.UnitTests.Workflows; + +/// +/// A secret that meets a cut. Exact-match redaction recognises only a whole value, so text persisted in pieces (a stream +/// flushed by size or by time, a capture the runner stopped at its budget, a capped excerpt) must never split a secret +/// between two pieces, and a piece whose end was cut away must not keep the head of one. +/// +[Trait("Category", "Unit")] +public sealed class SecretBoundaryRedactionTests +{ + private const string Secret = "CANARY-boundary-0123456789abcdef"; + + [Theory] + [InlineData("nothing to see here", 19)] // no secret can start anywhere in it + [InlineData("key=CANA", 4)] // the tail could still become the secret + [InlineData("key=" + Secret, 36)] // a whole secret is settled: exact matching sees it + [InlineData("key=" + Secret + " and C", 41)] // only the trailing "C" is held back + [InlineData("", 0)] + public void SettledLength_holds_back_only_what_could_still_become_a_secret(string text, int expected) + { + new PersistenceSecretRedactor([Secret]).SettledLength(text).ShouldBe(expected); + } + + [Fact] + public void SettledLength_never_cuts_through_a_secret_that_overlaps_itself() + { + // "abXab" ends with its own first two characters. The tail "ab" alone would put the cut at 4 and release + // "zabX" — three characters of the secret — so a whole occurrence across the cut moves it back to its start. + new PersistenceSecretRedactor(["abXab"]).SettledLength("zabXab").ShouldBe(1); + } + + [Fact] + public void SettledLength_with_no_secrets_settles_everything() + { + new PersistenceSecretRedactor([]).SettledLength("anything at all").ShouldBe(15); + } + + [Fact] + public void Releasing_settled_pieces_leaves_no_fragment_of_a_secret_wherever_the_text_was_split() + { + var redactor = new PersistenceSecretRedactor([Secret]); + const string text = "before " + Secret + " after"; + + for (var split = 0; split <= text.Length; split++) + { + var arrived = text[..split]; + var settled = redactor.SettledLength(arrived); + + var persisted = redactor.Redact(arrived[..settled]).Value + redactor.Redact(arrived[settled..] + text[split..]).Value; + + persisted.ShouldBe("before [REDACTED] after", $"the text split at {split}"); + } + } + + [Theory] + [InlineData("deploy key=CANARY-boundary-012", "deploy key=[REDACTED]", true)] // cut inside the secret + [InlineData("ok " + Secret + " tail", "ok [REDACTED] tail", true)] // whole secret, clean end + [InlineData("all clear", "all clear", false)] + public void RedactCut_masks_a_trailing_fragment_that_could_be_the_start_of_a_secret(string text, string expected, bool changed) + { + var redaction = new PersistenceSecretRedactor([Secret]).RedactCut(text); + + redaction.Value.ShouldBe(expected); + redaction.Changed.ShouldBe(changed); + } + + [Fact] + public void RedactCut_passes_null_through() + { + new PersistenceSecretRedactor([Secret]).RedactCut(null).Value.ShouldBeNull(); + } + + [Fact] + public void Union_masks_every_secret_of_every_redactor_and_skips_absent_ones() + { + var union = PersistenceSecretRedactor.Union(new PersistenceSecretRedactor(["first-secret"]), null, new PersistenceSecretRedactor(["second-secret"])); + + union.Redact("first-secret then second-secret").Value.ShouldBe("[REDACTED] then [REDACTED]"); + } + + [Fact] + public void None_masks_nothing() + { + PersistenceSecretRedactor.None.Redact("plain").Changed.ShouldBeFalse(); + } +}